Chainlit + Google Maps 实战:左侧对话、右侧实时地图的交互式 AI 画布教程
简介
本文主要讲解如何使用chainlit实现一个如何使用元素边栏作为画布的功能,即左侧是对话框,右侧是画布,实时展示左侧对话框中AI回答时需要展示的画面。以地图应用为例,当在对话框中,提问,北京的位置,ai会调用地图工具获取北京的地理坐标,然后在右侧画布的地图上展示AI的位置。
教程
1. 创建一个map-canvas项目文件夹
图片中.md类型文件可以忽略!
2. 创建一个app.py 类
代码如下:
import jsonimport chainlit as clfrom anthropic import AsyncAnthropicSYSTEM = \"you are a helpful assistant.\"MODEL_NAME = \"claude-3-5-sonnet-latest\"c = AsyncAnthropic()@cl.step(type=\"tool\")async def move_map_to(latitude: float, longitude: float): await open_map() fn = cl.CopilotFunction( name=\"move-map\", args={\"latitude\": latitude, \"longitude\": longitude} ) await fn.acall() return \"Map moved!\"tools = [ { \"name\": \"move_map_to\", \"description\": \"Move the map to the given latitude and longitude.\", \"input_schema\": { \"type\": \"object\", \"properties\": { \"latitude\": { \"type\": \"string\", \"description\": \"The latitude of the location to move the map to\", }, \"longitude\": { \"type\": \"string\", \"description\": \"The longitude of the location to move the map to\", }, }, \"required\": [\"latitude\", \"longitude\"], }, }]TOOL_FUNCTIONS = { \"move_map_to\": move_map_to,}async def call_claude(chat_messages): msg = cl.Message(content=\"\", author=\"Claude\") async with c.messages.stream( max_tokens=1024, system=SYSTEM, messages=chat_messages, tools=tools, model=MODEL_NAME, ) as stream: async for text in stream.text_stream: await msg.stream_token(text) await msg.send() response = await stream.get_final_message() return responseasync def call_tool(tool_use): tool_name = tool_use.name tool_input = tool_use.input tool_function = TOOL_FUNCTIONS.get(tool_name) if tool_function: try: return await tool_function(**tool_input) except TypeError: return json.dumps({\"error\": f\"Invalid input for {tool_name}\"}) else: return json.dumps({\"error\": f\"Invalid tool: {tool_name}\"})async def open_map(): map_props = {\"latitude\": 37.7749, \"longitude\": -122.4194, \"zoom\": 12} custom_element = cl.CustomElement(name=\"Map\", props=map_props, display=\"inline\") await cl.ElementSidebar.set_title(\"canvas\") await cl.ElementSidebar.set_elements([custom_element], key=\"map-canvas\")@cl.action_callback(\"close_map\")async def on_test_action(): await cl.ElementSidebar.set_elements([])@cl.set_startersasync def set_starters(): return [ cl.Starter( label=\"Paris\", message=\"Show me Paris.\", ), cl.Starter( label=\"NYC\", message=\"Show me NYC.\", ), cl.Starter( label=\"Tokyo\", message=\"Show me Tokyo.\", ), ]@cl.on_chat_startasync def on_start(): cl.user_session.set(\"chat_messages\", []) await open_map()@cl.on_messageasync def on_message(msg: cl.Message): chat_messages = cl.user_session.get(\"chat_messages\") chat_messages.append({\"role\": \"user\", \"content\": msg.content}) response = await call_claude(chat_messages) while response.stop_reason == \"tool_use\": tool_use = next(block for block in response.content if block.type == \"tool_use\") tool_result = await call_tool(tool_use) messages = [ {\"role\": \"assistant\", \"content\": response.content}, { \"role\": \"user\", \"content\": [ { \"type\": \"tool_result\", \"tool_use_id\": tool_use.id, \"content\": str(tool_result), } ], }, ] chat_messages.extend(messages) response = await call_claude(chat_messages) final_response = next( (block.text for block in response.content if hasattr(block, \"text\")), None, ) chat_messages = cl.user_session.get(\"chat_messages\") chat_messages.append({\"role\": \"assistant\", \"content\": final_response})
代码解读
这段代码是一个使用 Python 编写的 Chainlit 应用,结合了 Anthropic 的异步 API(AsyncAnthropic
)来实现一个基于聊天的交互式地图应用。它允许用户通过自然语言指令让 AI 指导系统移动地图到指定位置。
下面是对整个代码的详细解读:
🧠 1. 导入模块
import jsonimport chainlit as clfrom anthropic import AsyncAnthropic
json
: 用于处理 JSON 数据。chainlit as cl
: Chainlit 是一个用于构建 LLM 应用的框架,提供 UI 和工具支持。AsyncAnthropic
: 异步调用 Anthropic 提供的 Claude 大模型。
🧩 2. 系统提示与模型设置
SYSTEM = \"you are a helpful assistant.\"MODEL_NAME = \"claude-3-5-sonnet-latest\"c = AsyncAnthropic()
SYSTEM
: 定义系统的角色描述。MODEL_NAME
: 使用的模型名称。c
: 初始化一个异步客户端实例,用于调用 Claude 模型。
🛠️ 3. 自定义工具函数:move_map_to
@cl.step(type=\"tool\")async def move_map_to(latitude: float, longitude: float): await open_map() fn = cl.CopilotFunction( name=\"move-map\", args={\"latitude\": latitude, \"longitude\": longitude} ) await fn.acall() return \"Map moved!\"
- 这是一个 Chainlit 工具函数,用于将地图移动到指定经纬度。
- 使用
@cl.step
装饰器标记为一个工具步骤,可以在 UI 中显示。 - 调用
open_map()
打开地图界面。 - 使用
CopilotFunction
发送命令给前端的地图组件。
🧰 4. 工具描述定义
tools = [ { \"name\": \"move_map_to\", \"description\": \"Move the map to the given latitude and longitude.\", \"input_schema\": { \"type\": \"object\", \"properties\": { \"latitude\": { \"type\": \"string\", \"description\": \"The latitude of the location to move the map to\", }, \"longitude\": { \"type\": \"string\", \"description\": \"The longitude of the location to move the map to\", }, }, \"required\": [\"latitude\", \"longitude\"], }, }]
- 定义工具接口,供 Claude 模型识别并调用。
- 输入参数是字符串类型,虽然实际传入的是浮点数,但这里可能为了兼容性或格式统一而设为字符串。
🔗 5. 工具映射表
TOOL_FUNCTIONS = { \"move_map_to\": move_map_to,}
- 将工具名映射到对应的 Python 函数,方便在后续调用中查找执行。
🤖 6. 调用 Claude 模型
async def call_claude(chat_messages): msg = cl.Message(content=\"\", author=\"Claude\") async with c.messages.stream( max_tokens=1024, system=SYSTEM, messages=chat_messages, tools=tools, model=MODEL_NAME, ) as stream: async for text in stream.text_stream: await msg.stream_token(text) await msg.send() response = await stream.get_final_message() return response
- 向 Claude 发起请求,并流式返回响应。
- 支持工具调用(
tools=tools
)。 - 使用
stream
接口实现实时输出效果。
⚙️ 7. 工具调用函数
async def call_tool(tool_use): tool_name = tool_use.name tool_input = tool_use.input tool_function = TOOL_FUNCTIONS.get(tool_name) if tool_function: try: return await tool_function(**tool_input) except TypeError: return json.dumps({\"error\": f\"Invalid input for {tool_name}\"}) else: return json.dumps({\"error\": f\"Invalid tool: {tool_name}\"})
- 根据模型返回的
tool_use
决定调用哪个工具。 - 如果工具不存在或输入不合法,则返回错误信息。
🗺️ 8. 地图打开函数
async def open_map(): map_props = {\"latitude\": 37.7749, \"longitude\": -122.4194, \"zoom\": 12} custom_element = cl.CustomElement(name=\"Map\", props=map_props, display=\"inline\") await cl.ElementSidebar.set_title(\"canvas\") await cl.ElementSidebar.set_elements([custom_element], key=\"map-canvas\")
- 默认打开旧金山的地图视图。
- 使用
CustomElement
加载自定义组件(地图)。 - 设置侧边栏标题和元素。
🚫 9. 关闭地图动作
@cl.action_callback(\"close_map\")async def on_test_action(): await cl.ElementSidebar.set_elements([])
- 注册一个动作回调,当用户点击关闭地图按钮时清空侧边栏内容。
💬 10. 预设启动语句
@cl.set_startersasync def set_starters(): return [ cl.Starter(label=\"Paris\", message=\"Show me Paris.\"), cl.Starter(label=\"NYC\", message=\"Show me NYC.\"), cl.Starter(label=\"Tokyo\", message=\"Show me Tokyo.\"), ]
- 在界面上显示三个预设按钮,点击后发送相应消息给模型。
🟢 11. 初始化逻辑
@cl.on_chat_startasync def on_start(): cl.user_session.set(\"chat_messages\", []) await open_map()
- 当会话开始时初始化对话历史和打开地图。
📨 12. 主消息处理逻辑
@cl.on_messageasync def on_message(msg: cl.Message): chat_messages = cl.user_session.get(\"chat_messages\") chat_messages.append({\"role\": \"user\", \"content\": msg.content}) response = await call_claude(chat_messages) while response.stop_reason == \"tool_use\": tool_use = next(block for block in response.content if block.type == \"tool_use\") tool_result = await call_tool(tool_use) messages = [ {\"role\": \"assistant\", \"content\": response.content}, { \"role\": \"user\", \"content\": [ { \"type\": \"tool_result\", \"tool_use_id\": tool_use.id, \"content\": str(tool_result), } ], }, ] chat_messages.extend(messages) response = await call_claude(chat_messages) final_response = next( (block.text for block in response.content if hasattr(block, \"text\")), None, ) chat_messages = cl.user_session.get(\"chat_messages\") chat_messages.append({\"role\": \"assistant\", \"content\": final_response})
流程说明:
- 接收用户消息 → 添加到
chat_messages
- 调用 Claude 获取响应
- 如果模型要求调用工具:
- 解析
tool_use
- 调用对应函数(如
move_map_to
) - 构造
tool_result
并再次调用 Claude 继续推理
- 解析
- 直到没有工具需要调用:
- 提取最终文本回复
- 显示在界面上
✅ 总结功能亮点:
stream
实现逐字输出CustomElement
集成地图组件🧪 示例交互流程:
- 用户点击 “Paris”
- 系统收到
\"Show me Paris.\"
- Claude 判断需要调用
move_map_to(48.8566, 2.3522)
- 执行工具函数 → 地图跳转到巴黎
- 回复:“Map moved to Paris!”
3. 新建一个requirements.txt文件
内容如下:
chainlit=>2.4.301
4. 自定义地图前端组件
在map-canvas项目文件夹,下新建一个public文件夹,public文件夹下新建一个elements文件夹,elements文件夹新建一个Map.jsx文件,Map.jsx代码如下:
import React, { useEffect, useRef } from \"react\";import { useRecoilValue } from \"recoil\";import { callFnState } from \"@chainlit/react-client\";import { Button } from \"@/components/ui/button\";import { ArrowLeft } from \"lucide-react\";export default function GoogleMap() { const mapRef = useRef(null); const mapInstanceRef = useRef(null); const callFn = useRecoilValue(callFnState); useEffect(() => { // Check if API is already loaded if (window.google && window.google.maps) { // Maps API already loaded, initialize map directly initializeMap(); return; } // Check if the script is already in the process of loading const existingScript = document.querySelector( `script[src^=\"https://maps.googleapis.com/maps/api/js\"]` ); if (existingScript) { // Script is loading but not ready, wait for it const originalCallback = window.initMap; window.initMap = () => { if (originalCallback) originalCallback(); initializeMap(); }; return; } // Load the script only if it\'s not already loaded or loading const script = document.createElement(\"script\"); script.src = `https://maps.googleapis.com/maps/api/js?&callback=initMap`; script.async = true; script.defer = true; // Define the callback function window.initMap = initializeMap; document.head.appendChild(script); // Clean up return () => { // Don\'t remove the script as other components might be using it // Just clean up our callback if (window.initMap === initializeMap) { window.initMap = null; } }; }, []); useEffect(() => { if (callFn?.name === \"move-map\") { const { latitude, longitude } = callFn.args; moveMapTo(latitude, longitude); callFn.callback(); } }, [callFn]); const initializeMap = () => { if (mapRef.current && !mapInstanceRef.current) { mapInstanceRef.current = new window.google.maps.Map(mapRef.current, { center: { lat: props.latitude, lng: props.longitude }, zoom: props.zoom, }); } }; const moveMapTo = (newLat, newLng, newZoom = 12) => { if (!mapInstanceRef.current) return false; // Create a new LatLng object const newPosition = new window.google.maps.LatLng(newLat, newLng); // Pan the map to the new position mapInstanceRef.current.panTo(newPosition); // Update zoom if provided if (newZoom !== null) { mapInstanceRef.current.setZoom(newZoom); } return true; }; return ( <div className=\"h-full w-full relative\"> <Button className=\"absolute z-10\" style={{top: \".5rem\", left: \".5rem\"}} onClick={() => callAction({ name: \"close_map\", payload: {} })} size=\"icon\" > <ArrowLeft /> </Button> <div ref={mapRef} className=\"h-full w-full\" /> </div> );}
代码解读
这段代码是一个使用 React 编写的地图组件 GoogleMap
,它实现了以下核心功能:
- 使用
Google Maps JavaScript API
加载并初始化地图。 - 响应来自
Chainlit
后端的工具调用(如移动地图到指定经纬度)。 - 提供关闭地图按钮。
下面是详细的逐段解读和说明:
🧩 1. 引入依赖
import React, { useEffect, useRef } from \"react\";import { useRecoilValue } from \"recoil\";import { callFnState } from \"@chainlit/react-client\";import { Button } from \"@/components/ui/button\";import { ArrowLeft } from \"lucide-react\";
React
, useEffect
, useRef
useRecoilValue
, callFnState
Button
ArrowLeft
📌 2. 组件声明与状态绑定
export default function GoogleMap() { const mapRef = useRef(null); // DOM 容器引用 const mapInstanceRef = useRef(null); // 地图实例引用 const callFn = useRecoilValue(callFnState); // 接收 Chainlit 的函数调用
mapRef
: 用于挂载 Google Map 的元素。mapInstanceRef
: 存储实际的google.maps.Map
实例。callFn
: Chainlit 提供的工具调用数据源。当后端通过CopilotFunction
发送命令时,这里会更新。
🚀 3. 初始化地图:加载 Google Maps API 并创建地图
useEffect(() => { if (window.google && window.google.maps) { initializeMap(); return; } const existingScript = document.querySelector( `script[src^=\"https://maps.googleapis.com/maps/api/js\"]` ); if (existingScript) { const originalCallback = window.initMap; window.initMap = () => { if (originalCallback) originalCallback(); initializeMap(); }; return; } const script = document.createElement(\"script\"); script.src = `https://maps.googleapis.com/maps/api/js?&callback=initMap`; script.async = true; script.defer = true; window.initMap = initializeMap; document.head.appendChild(script); return () => { if (window.initMap === initializeMap) { window.initMap = null; } };}, []);
✅ 功能详解:
- 首先检查
google.maps
是否已经加载:- 如果已加载,直接调用
initializeMap()
创建地图。
- 如果已加载,直接调用
- 如果脚本正在加载但未完成:
- 替换全局回调函数
initMap
,等待加载完成后初始化地图。
- 替换全局回调函数
- 如果尚未加载:
- 动态插入 Google Maps 脚本标签。
- 设置异步加载,并注册全局回调
initMap
。
⚠️ 注意事项:
- 这个逻辑避免了重复加载 Google Maps API。
- 回调函数必须是全局的(
window.initMap
),这是 Google Maps API 的限制。
🔧 4. 工具响应处理:监听 Chainlit 的地图移动请求
useEffect(() => { if (callFn?.name === \"move-map\") { const { latitude, longitude } = callFn.args; moveMapTo(latitude, longitude); callFn.callback(); // 通知后端已完成操作 }}, [callFn]);
- 当接收到名为
\"move-map\"
的工具调用时,提取经纬度参数并调用moveMapTo()
移动地图。 - 执行完后调用
callFn.callback()
,告诉后端任务完成。
🗺️ 5. 初始化地图函数
const initializeMap = () => { if (mapRef.current && !mapInstanceRef.current) { mapInstanceRef.current = new window.google.maps.Map(mapRef.current, { center: { lat: props.latitude, lng: props.longitude }, zoom: props.zoom, }); }};
⚠️ 注意问题:
- 这里使用了
props.latitude
、props.longitude
和props.zoom
,但在当前代码中没有看到props
的定义。 - 应该是从父组件传入或默认值设定,否则会导致错误。
✅ 建议修改为:
const initializeMap = () => { if (mapRef.current && !mapInstanceRef.current) { mapInstanceRef.current = new window.google.maps.Map(mapRef.current, { center: { lat: 37.7749, lng: -122.4194 }, // 默认旧金山 zoom: 12, }); }};
或者添加
props
支持:function GoogleMap({ latitude = 37.7749, longitude = -122.4194, zoom = 12 }) { ... }
🔄 6. 地图移动函数
const moveMapTo = (newLat, newLng, newZoom = 12) => { if (!mapInstanceRef.current) return false; const newPosition = new window.google.maps.LatLng(newLat, newLng); mapInstanceRef.current.panTo(newPosition); if (newZoom !== null) { mapInstanceRef.current.setZoom(newZoom); } return true;};
panTo
: 平滑地将地图中心移动到新位置。setZoom
: 可选地调整缩放级别。
🧱 7. 渲染部分:地图容器 + 关闭按钮
return (
<Button className=\"absolute z-10\" style={{ top: \".5rem\", left: \".5rem\" }} onClick={() => callAction({ name: \"close_map\", payload: {} })} size=\"icon\" >);mapRef
是地图渲染的目标容器。Button
提供一个关闭地图的按钮,触发callAction
调用后端动作close_map
。
⚠️ 注意:
callAction
函数没有在当前组件中定义或导入,应该从 Chainlit 导出,例如:
import { callAction } from \"@chainlit/react-client\";
✅ 总结功能流程图:
#mermaid-svg-Sh9HKFLD4cB41cz8 {font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .error-icon{fill:#552222;}#mermaid-svg-Sh9HKFLD4cB41cz8 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-Sh9HKFLD4cB41cz8 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .marker.cross{stroke:#333333;}#mermaid-svg-Sh9HKFLD4cB41cz8 svg{font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .label{font-family:\"trebuchet ms\",verdana,arial,sans-serif;color:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .cluster-label text{fill:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .cluster-label span{color:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .label text,#mermaid-svg-Sh9HKFLD4cB41cz8 span{fill:#333;color:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .node rect,#mermaid-svg-Sh9HKFLD4cB41cz8 .node circle,#mermaid-svg-Sh9HKFLD4cB41cz8 .node ellipse,#mermaid-svg-Sh9HKFLD4cB41cz8 .node polygon,#mermaid-svg-Sh9HKFLD4cB41cz8 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .node .label{text-align:center;}#mermaid-svg-Sh9HKFLD4cB41cz8 .node.clickable{cursor:pointer;}#mermaid-svg-Sh9HKFLD4cB41cz8 .arrowheadPath{fill:#333333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-Sh9HKFLD4cB41cz8 .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-Sh9HKFLD4cB41cz8 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-Sh9HKFLD4cB41cz8 .cluster text{fill:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 .cluster span{color:#333;}#mermaid-svg-Sh9HKFLD4cB41cz8 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-Sh9HKFLD4cB41cz8 :root{--mermaid-font-family:\"trebuchet ms\",verdana,arial,sans-serif;} 用户点击预设消息 Chainlit 调用 move-map 工具 React 组件接收到 callFn 地图平移至目标坐标
💡 示例使用方式(Chainlit 端)
假设你在 Chainlit 后端有如下工具函数:
@cl.step(type=\"tool\")async def move_map_to(latitude: float, longitude: float): fn = cl.CopilotFunction(name=\"move-map\", args={\"latitude\": latitude, \"longitude\": longitude}) await fn.acall() return \"Moved map to location\"
前端 React 组件会自动接收到这个调用,并执行
moveMapTo()
。
5.安装依赖
在项目根目录文件夹下,执行以下命令安装依赖。
pip install -r .\\requirements.txt
6.运行应用程序
要启动 Chainlit 应用程序,请打开终端并导航到包含的目录app.py。然后运行以下命令:
chainlit run app.py -w
-
-w, --watch
:当模块发生变化时重新加载应用。当指定此选项时,将启动文件监视程序,对文件的任何更改都将导致服务器重新加载应用程序,从而允许更快的迭代。 -
-h,--headless
:阻止应用在浏览器中打开。 -
-d,--debug
:设置日志级别为debug。默认日志级别为error。 -
-c,--ci
:以ci模式运行。 -
--no-cache
:禁用第三方缓存,如langchain。 -
--host
:指定运行服务器的不同主机。 -
--port
:指定运行服务器的另一个端口。 -
--root-path
:指定运行服务器的子路径。
相关文章
《Chainlit 自定义元素开发指南:使用 JSX 和受限导入实现交互式界面》