该 Demo 的“系统功能链路”为用户发消息“北京今天气温怎么样要穿什么”StateGraph 判断是否提到“城市 气温”若提到城市→ 调用“气温 API”我们用模拟函数代替**调用 RAG伪搜索该气温下穿什么”返回最终回复给用户Flask API 返回 JSON1.同步执行代码# # 1. 导入# fromflaskimportFlask,request,jsonifyfromlanggraph.graphimportStateGraphfromtypingimportTypedDict,Optional# # 2. 定义 State系统的“全局记忆结构”# classWeatherState(TypedDict):user_input:strcity:Optional[str]temperature:Optional[int]advice:Optional[str]# # 3. ----------- 业务逻辑函数可被 StateGraph 调用-----------# # ----- step1NLU 识别是否出现城市和“气温”相关词 -----defextract_city_and_weather(state:WeatherState):textstate[user_input]# 简化的城市词表真实项目可接入实体识别模型cities[北京,上海,广州,深圳,杭州]found_cityNoneforcincities:ifcintext:found_citycbreak# 判断用户是否提到天气/气温has_weather_intentany(kintextforkin[天气,气温,热,冷,穿什么])iffound_cityandhas_weather_intent:return{city:found_city}else:# 没城市或没天气意图 → 不进入后续步骤return{city:None}# ----- step2获取城市气温模拟 API -----deffetch_temperature(state:WeatherState):citystate[city]ifnotcity:return{temperature:None}# 假设模拟温度fake_weather_db{北京:5,上海:13,广州:21,深圳:22,杭州:10,}return{temperature:fake_weather_db.get(city,15)}# ----- step3RAG 查询穿衣建议我们用简单规则代替 -----defrag_fetch_clothing_advice(state:WeatherState):tempstate[temperature]iftempisNone:return{advice:未识别到城市或天气相关内容。}iftemp5:a天气很冷建议穿厚羽绒服、围巾、手套。eliftemp15:a有点冷建议穿风衣、薄棉服、长袖。eliftemp25:a天气适中可以穿短袖或薄外套。else:a天气较热可以穿短袖短裤。return{advice:a}# # 4. ----------- 构建 LangGraph StateGraph -------------------# graphStateGraph(WeatherState)graph.add_node(extract_city,extract_city_and_weather)graph.add_node(get_temp,fetch_temperature)graph.add_node(rag_advice,rag_fetch_clothing_advice)graph.set_entry_point(extract_city)# 条件分支# 若识别到 city则进入天气查询否则直接返回 advicedefhas_city(state:WeatherState):returngoifstate[city]elsestopgraph.add_conditional_edges(extract_city,has_city,{go:get_temp,stop:rag_advice})# 获取温度之后一定进入 RAGgraph.add_edge(get_temp,rag_advice)# 编译图app_graphgraph.compile()# # 5. -------------------- Flask API ---------------------------# appFlask(__name__)app.route(/ask,methods[POST])defask_weather():datarequest.json user_inputdata.get(text,)# 执行 LangGraphresultapp_graph.invoke({user_input:user_input,city:None,temperature:None,advice:None})returnjsonify({city:result[city],temperature:result[temperature],advice:result[advice]})# # 6. 启动 Flask# if__name____main__:app.run(host0.0.0.0,port5000)使用启动服务后发送curl-XPOST http://127.0.0.1:5000/ask\-HContent-Type: application/json\-d{text:北京今天冷吗穿什么}返回结果示例{city:北京,temperature:5,advice:天气很冷建议穿厚羽绒服、围巾、手套。}2.异步流执行代码# # 1. 导入# fromflaskimportFlask,request,jsonify,Responsefromlanggraph.graphimportStateGraphfromtypingimportTypedDict,Optionalimportasyncioimportjson# # 2. 定义 State# classWeatherState(TypedDict):user_input:strcity:Optional[str]temperature:Optional[int]advice:Optional[str]# # 3. 业务逻辑节点# defextract_city_and_weather(state:WeatherState):textstate[user_input]cities[北京,上海,广州,深圳,杭州]found_cityNoneforcincities:ifcintext:found_citycbreakhas_weather_intentany(kintextforkin[天气,气温,热,冷,穿什么])iffound_cityandhas_weather_intent:return{city:found_city}else:return{city:None}asyncdeffetch_temperature(state:WeatherState):awaitasyncio.sleep(0.2)# 模拟延迟citystate[city]ifnotcity:return{temperature:None}fake_weather_db{北京:5,上海:13,广州:21,深圳:22,杭州:10,}return{temperature:fake_weather_db.get(city,15)}asyncdefrag_fetch_clothing_advice(state:WeatherState):awaitasyncio.sleep(0.2)# 模拟 RAG 查询延迟tempstate[temperature]iftempisNone:return{advice:未识别到城市或天气相关内容。}iftemp5:a天气很冷建议穿厚羽绒服、围巾、手套。eliftemp15:a有点冷建议穿风衣、薄棉服、长袖。eliftemp25:a天气适中可以穿短袖或薄外套。else:a天气较热可以穿短袖短裤。return{advice:a}# # 4. 构建 StateGraph# graphStateGraph(WeatherState)graph.add_node(extract_city,extract_city_and_weather)graph.add_node(get_temp,fetch_temperature)graph.add_node(rag_advice,rag_fetch_clothing_advice)graph.set_entry_point(extract_city)# 条件分支defhas_city(state:WeatherState):returngoifstate[city]elsestopgraph.add_conditional_edges(extract_city,has_city,{go:get_temp,stop:rag_advice})graph.add_edge(get_temp,rag_advice)# 编译图app_graphgraph.compile()# # 5. Flask API 异步流# appFlask(__name__)app.route(/ask_stream,methods[POST])defask_stream():user_inputrequest.json.get(text,)asyncdefevent_stream():asyncforchunkinapp_graph.astream({user_input:user_input,city:None,temperature:None,advice:None}):# 每个 chunk 是当前节点返回的数据yieldfdata:{json.dumps(chunk)}\n\nawaitasyncio.sleep(0.05)# Flask 支持 SSE (Server-Sent Events)returnResponse(event_stream(),mimetypetext/event-stream)# # 6. 启动 Flask# if__name____main__:app.run(host0.0.0.0,port5000)使用启动 Flaskpython async_weather_demo.py测试 SSE 异步流curl-N-XPOST http://127.0.0.1:5000/ask_stream\-HContent-Type: application/json\-d{text:北京今天冷吗穿什么}输出示例逐步返回data: {city: 北京} data: {temperature: 5} data: {advice: 天气很冷建议穿厚羽绒服、围巾、手套。}前端可以边接收边显示用户体验更像聊天机器人实时响应。3.区别同步调用invoke()用法resultapp_graph.invoke(state)特点一步执行完成StateGraph 内所有节点按图的顺序执行完毕后才返回结果。返回的是最终状态State 字典或对象。阻塞调用Python 会等待整个图跑完才能继续下一行代码。如果中间有耗时操作比如 API 调用、数据库查询调用线程会被阻塞。结果一次性返回无法看到节点执行的中间状态。对于长流程或多步骤任务用户必须等待所有节点完成才能获取响应。示例resultapp_graph.invoke({user_input:北京今天冷吗穿什么,city:None,temperature:None,advice:None})print(result)输出示例{city:北京,temperature:5,advice:天气很冷建议穿厚羽绒服、围巾、手套。}优点简单直接适合流程短、响应快的场景。缺点无法逐步显示执行过程耗时长时用户体验差。异步流调用astream()用法asyncforchunkinapp_graph.astream(state):print(chunk)特点异步迭代每个节点执行完成后就会yield当前节点的输出。可以边执行边获取中间结果。非阻塞可以在执行过程中处理其他任务适合 async/await 场景。适合长流程、API 请求或 RAG 查询等耗时操作。流式返回中间状态每个节点的输出都可以立即返回给前端或日志。适合前端实时显示“进度”或“逐步结果”。示例asyncforchunkinapp_graph.astream({user_input:北京今天冷吗穿什么,city:None,temperature:None,advice:None}):print(chunk)输出示例逐步返回{city: 北京} {temperature: 5} {advice: 天气很冷建议穿厚羽绒服、围巾、手套。}优点用户体验好可实现流式响应、进度显示、分步处理。缺点写法稍复杂需要异步支持async/await前端也需要支持 SSE 或 WebSocket 才能实时显示。对比总结特性invoke()同步astream()异步流执行方式阻塞同步异步迭代逐步返回返回结果最终状态一次性返回节点输出逐步返回适用场景快速流程节点少无长耗时操作长流程、多节点、耗时操作API、RAG、LLM用户体验等待所有节点完成才有结果可实时看到中间执行状态前端显示一次性渲染可流式渲染SSE、WebSocket什么时候选哪个同步invoke()简单场景、节点少、无耗时操作例如本地计算、简单逻辑判断异步流astream()流程长、节点多、含网络请求、RAG 或 LLM 调用需要实时显示中间结果给用户例如聊天机器人、问答系统、流程审批、实时推荐