为什么需要 LangGraph 判断节点常规 AI 对话流程多为固定顺序无法根据用户意图动态调整执行路径。在工具型 Agent 中系统必须先判断用户问题是否需要调用外部工具工具执行后如何衔接结果输出非工具类问题直接结束流程LangGraph 的判断节点本质是工作流的 “大脑”基于对话状态实时决策下一步走向让 Agent 具备类人的思考与分支能力同时保持流程清晰、可扩展、易维护。二、核心设计消息图与判断节点架构本次实践采用 LangGraph 的MessageGraph消息图架构状态以对话消息列表为载体天然适配多轮交互场景。整体流程分为三层节点决策判断节点接收用户输入通过大模型输出标准化指令完成意图识别工具执行节点根据判断结果调用天气查询工具获取真实数据结果生成节点格式化工具返回信息输出友好回答判断节点作为流程入口承担全流程的路由指挥工作是整个 Agent 的逻辑核心。三、判断节点的核心能力1. 意图识别与指令标准化判断节点通过强约束提示词引导大模型输出纯 JSON 格式指令明确两种动作需查询天气触发工具调用无需工具直接结束流程指令格式统一、无冗余内容为后续路由提供稳定输入。2. 运行时动态路由判断节点执行完成后条件路由函数读取指令中的动作字段实时决定流程走向动作匹配天气查询 → 跳转工具节点动作匹配结束指令 → 终止工作流无需硬编码分支逻辑流程可随业务灵活扩展。3. 状态共享与上下文传递判断节点基于全局消息状态做决策工具节点与结果节点可无缝读取上下文保证多步骤间数据一致避免信息断层。四、判断节点实现逻辑拆解1. 节点定义与状态处理判断节点以消息状态为输入调用大模型生成决策指令对输出内容做轻量化清洗保证指令纯净可解析为路由环节筑牢基础。2. 条件边与路由映射通过条件边绑定判断节点与路由函数建立 “指令动作→执行节点” 的映射关系。普通边固定工具执行与结果生成的顺序形成完整闭环判断节点 →条件→ 工具节点 → 结果节点 → 结束判断节点 →条件→ 结束3. 图编译与流程可视化完成节点与边的定义后编译生成可执行工作流并支持 ASCII 流程图输出直观呈现判断节点的分支逻辑便于调试与维护。五、场景验证判断节点的实际效果测试 1天气查询意图用户询问城市天气时判断节点识别工具需求路由至天气工具执行查询后生成格式化回答流程完整闭环。测试 2非工具类问题用户提出无关问题时判断节点直接输出结束指令工作流立即终止避免无效工具调用提升响应效率。两种场景均由判断节点统一调度流程切换流畅逻辑清晰可控。六、总结与扩展价值LangGraph 的判断节点本质是将复杂分支逻辑图式化以节点 条件边的方式替代繁琐的条件嵌套代码。在本次天气 Agent 中判断节点实现了意图精准识别与流程动态调度工具调用与直接回答的自由切换可视化工作流与低耦合架构设计该模式可直接迁移至知识库问答、函数调用、多步骤任务编排等场景是构建可靠、灵活 AI Agent 的核心方案。代码实现123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156importjsonfromlangchain_core.messagesimportHumanMessage, AIMessage,SystemMessagefromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportMessageGraph, ENDimportredefremove_think_tags(text):移除字符串中think/think标签及其内容Args:text: 包含think标签的字符串Returns:移除think标签及其内容后的字符串# 支持多行内容和标签可能有的属性patternrthink[^]*.*?/thinkresultre.sub(pattern, , text, flagsre.DOTALL)# 清理空白字符resultresult.strip()resultre.sub(r\n{3,},\n\n, result)# 限制连续换行returnresult# 1. LLM 配置 # 配置 Deepseek 密钥和模型参数DEEPSEEK_API_KEYxxxxxxxxxxxxxxx# 替换为实际的 API KeyllmChatOpenAI(api_keyDEEPSEEK_API_KEY,base_urlhttps://xxxxxxxxxxxx/v1,# Deepseek 的 API 基础地址modeldeepseek-v3:671b,# Deepseek 对话模型可选deepseek-chat-pro 等高级模型temperature0.7,# 温度参数0-1越低越稳定max_tokens1024# 最大生成 tokens)# 2. 工具节点模拟天气 APIdefweather_tool(city):weather_map{北京:晴25°C微风,上海:多云28°C东南风3级,广州:雷阵雨30°C南风4级,深圳:大雨29°C西南风5级}returnweather_map.get(city, f{city}暂无天气数据)# 3. 定义图节点 defchain1_node(state):决策节点严格约束 LLM 输出纯 JSONprint(chain1_node--state:str(state))# 核心修复强化 Prompt 约束 降低 temperaturesystem_prompt(你是一个天气助手。请严格遵守1. 只输出纯JSON格式为{action:check_weather,action_input:城市名}或{action:Final Answer,action_input:结束end}2. 禁止输出任何其他字符包括空格、换行、标记、解释文字3. 必须使用双引号4. 只输出JSON不要任何其他文字。)# 调用 LLMresponsellm.invoke([SystemMessage(contentsystem_prompt)]state)# 最小清理仅去除首尾空白应对模型可能的空格cleaned_contentresponse.content.strip()#去掉 think部分cleaned_contentremove_think_tags(cleaned_content)# print(chai1cleaned_content:cleaned_content)# 返回清理后的纯文本消息returnAIMessage(contentcleaned_content)deftool_node(state):工具节点执行天气查询print(tool_node--state:str(state))last_msgstate[-1].content# 直接解析假设 chain1_node 已确保纯 JSONaction_datajson.loads(last_msg)# 无 try-except依赖 Prompt 约束cityaction_data.get(action_input,北京)or北京resultweather_tool(city)returnHumanMessage(contentf城市:{city}|天气:{result})defchain2_node(state):结果生成节点print(chain2_node--state:str(state))# state[-1]最后一个元素tool_resultstate[-1].contentcitytool_result.split(|)[0].replace(城市:, )weathertool_result.split(|)[1].replace(天气:, )returnAIMessage(contentf️ {city}当前天气{weather})# 4. 构建带条件边的图 graphMessageGraph()#定义点graph.add_node(chain1, chain1_node)# 决策节点LLM判断要查天气/直接结束graph.add_node(tool, tool_node)# 工具节点调用天气查询函数graph.add_node(chain2, chain2_node)# 结果节点格式化输出天气信息#定义分支graph.set_entry_point(chain1)# 条件边路由函数直接解析无复杂提取defrouter(state):last_msgstate[-1].content# 直接解析 JSON依赖 chain1_node 的纯输出保证actionjson.loads(last_msg)[action]print(f\n 路由决策: action{action})returntoolifactioncheck_weatherelseENDgraph.add_conditional_edges(chain1,router,{tool:tool, END: END})graph.add_edge(tool,chain2)graph.add_edge(chain2, END)appgraph.compile()# 画流程图print(app.get_graph().draw_ascii())# 5. 执行测试 if__name____main__:print(*60)print(️ 天气查询 Agent 测试纯 JSON 输出保障)print(*60)# 测试用例1天气查询print(\n_TestCase 1: 天气查询 → 触发工具调用_)inputs[HumanMessage(content北京天气怎么样)]foreventinapp.stream(inputs):nodelist(event.keys())[0]msgevent[node].contentprint(f\n[→ {node}])print(f {msg})
