python-LangGraph框架（3-28-LangGraph 判断节点实战）

矜持的左手

159人浏览 · 2026-07-03 14:33:22

矜持的左手 · 2026-07-03 14:33:22 发布

一、为什么需要 LangGraph 判断节点

常规 AI 对话流程多为固定顺序，无法根据用户意图动态调整执行路径。在工具型 Agent 中，系统必须先判断：

用户问题是否需要调用外部工具
工具执行后如何衔接结果输出
非工具类问题直接结束流程

LangGraph 的判断节点本质是工作流的 “大脑”，基于对话状态实时决策下一步走向，让 Agent 具备类人的思考与分支能力，同时保持流程清晰、可扩展、易维护。

二、核心设计：消息图与判断节点架构

本次实践采用 LangGraph 的MessageGraph消息图架构，状态以对话消息列表为载体，天然适配多轮交互场景。整体流程分为三层节点：

决策判断节点：接收用户输入，通过大模型输出标准化指令，完成意图识别
工具执行节点：根据判断结果调用天气查询工具，获取真实数据
结果生成节点：格式化工具返回信息，输出友好回答

判断节点作为流程入口，承担全流程的路由指挥工作，是整个 Agent 的逻辑核心。

三、判断节点的核心能力

1. 意图识别与指令标准化

判断节点通过强约束提示词，引导大模型输出纯 JSON 格式指令，明确两种动作：

需查询天气：触发工具调用
无需工具：直接结束流程

指令格式统一、无冗余内容，为后续路由提供稳定输入。

2. 运行时动态路由

判断节点执行完成后，条件路由函数读取指令中的动作字段，实时决定流程走向：

动作匹配天气查询 → 跳转工具节点
动作匹配结束指令 → 终止工作流

无需硬编码分支逻辑，流程可随业务灵活扩展。

3. 状态共享与上下文传递

判断节点基于全局消息状态做决策，工具节点与结果节点可无缝读取上下文，保证多步骤间数据一致，避免信息断层。

四、判断节点实现逻辑拆解

1. 节点定义与状态处理

判断节点以消息状态为输入，调用大模型生成决策指令，对输出内容做轻量化清洗，保证指令纯净可解析，为路由环节筑牢基础。

2. 条件边与路由映射

通过条件边绑定判断节点与路由函数，建立 “指令动作→执行节点” 的映射关系。普通边固定工具执行与结果生成的顺序，形成完整闭环：判断节点 →（条件）→ 工具节点 → 结果节点 → 结束判断节点 →（条件）→ 结束

3. 图编译与流程可视化

完成节点与边的定义后，编译生成可执行工作流，并支持 ASCII 流程图输出，直观呈现判断节点的分支逻辑，便于调试与维护。

五、场景验证：判断节点的实际效果

测试 1：天气查询意图

用户询问城市天气时，判断节点识别工具需求，路由至天气工具，执行查询后生成格式化回答，流程完整闭环。

测试 2：非工具类问题

用户提出无关问题时，判断节点直接输出结束指令，工作流立即终止，避免无效工具调用，提升响应效率。

两种场景均由判断节点统一调度，流程切换流畅，逻辑清晰可控。

六、总结与扩展价值

LangGraph 的判断节点，本质是将复杂分支逻辑图式化，以节点 + 条件边的方式，替代繁琐的条件嵌套代码。在本次天气 Agent 中，判断节点实现了：

意图精准识别与流程动态调度
工具调用与直接回答的自由切换
可视化工作流与低耦合架构设计

该模式可直接迁移至知识库问答、函数调用、多步骤任务编排等场景，是构建可靠、灵活 AI Agent 的核心方案。

代码实现：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

import json

from langchain_core.messages import HumanMessage, AIMessage,SystemMessage

from langchain_openai import ChatOpenAI

from langgraph.graph import MessageGraph, END

import re

def remove_think_tags(text):

"""

移除字符串中<think></think>标签及其内容

Args:

text: 包含think标签的字符串

Returns:

移除think标签及其内容后的字符串

"""

# 支持多行内容和标签可能有的属性

pattern = r'<think[^>]*>.*?</think>'

result = re.sub(pattern, '', text, flags=re.DOTALL)

# 清理空白字符

result = result.strip()

result = re.sub(r'\n{3,}', '\n\n', result) # 限制连续换行

return result

# ==================== 1. LLM 配置 ====================

# 配置 Deepseek 密钥和模型参数

DEEPSEEK_API_KEY = "xxxxxxxxxxxxxxx" # 替换为实际的 API Key

llm = ChatOpenAI(

api_key=DEEPSEEK_API_KEY,

base_url="https://xxxxxxxxxxxx/v1", # Deepseek 的 API 基础地址

model="deepseek-v3:671b", # Deepseek 对话模型（可选：deepseek-chat-pro 等高级模型）

temperature=0.7, # 温度参数（0-1，越低越稳定）

max_tokens=1024 # 最大生成 tokens

)

# ==================== 2. 工具节点（模拟天气 API）====================

def weather_tool(city):

weather_map = {

"北京": "晴，25°C，微风",

"上海": "多云，28°C，东南风3级",

"广州": "雷阵雨，30°C，南风4级",

"深圳": "大雨，29°C，西南风5级"

}

return weather_map.get(city, f"{city}暂无天气数据")

# ==================== 3. 定义图节点 ====================

def chain1_node(state):

"""决策节点：严格约束 LLM 输出纯 JSON"""

print('chain1_node-->state:' + str(state))

# 🔑 核心修复：强化 Prompt 约束 + 降低 temperature

system_prompt = (

"你是一个天气助手。请严格遵守："

"1. 只输出纯JSON，格式为{'action':'check_weather','action_input':'城市名'}或{'action':'Final Answer','action_input':'结束end'}；"

"2. 禁止输出任何其他字符（包括空格、换行、```标记、解释文字）；"

"3. 必须使用双引号；"

"4. 只输出JSON，不要任何其他文字。"

)

# 调用 LLM

response = llm.invoke([SystemMessage(content=system_prompt)] + state)

# 🔑 最小清理：仅去除首尾空白（应对模型可能的空格）

cleaned_content = response.content.strip()

#去掉 think部分

cleaned_content = remove_think_tags(cleaned_content)

# print('chai1+cleaned_content:'+cleaned_content)

# 返回清理后的纯文本消息

return AIMessage(content=cleaned_content)

def tool_node(state):

"""工具节点：执行天气查询"""

print('tool_node-->state:' + str(state))

last_msg = state[-1].content

# 直接解析（假设 chain1_node 已确保纯 JSON）

action_data = json.loads(last_msg) # 无 try-except，依赖 Prompt 约束

city = action_data.get("action_input", "北京") or "北京"

result = weather_tool(city)

return HumanMessage(content=f"城市:{city}|天气:{result}")

def chain2_node(state):

"""结果生成节点"""

print('chain2_node-->state:' + str(state))

# state[-1]最后一个元素

tool_result = state[-1].content

city = tool_result.split("|")[0].replace("城市:", "")

weather = tool_result.split("|")[1].replace("天气:", "")

return AIMessage(content=f"🌤️ {city}当前天气：{weather}")

# ==================== 4. 构建带条件边的图 ====================

graph = MessageGraph()

#定义点

graph.add_node("chain1", chain1_node) # 决策节点：LLM判断要查天气/直接结束

graph.add_node("tool", tool_node) # 工具节点：调用天气查询函数

graph.add_node("chain2", chain2_node) # 结果节点：格式化输出天气信息

#定义分支

graph.set_entry_point("chain1")

# 条件边路由函数（直接解析，无复杂提取）

def router(state):

last_msg = state[-1].content

# 直接解析 JSON（依赖 chain1_node 的纯输出保证）

action = json.loads(last_msg)["action"]

print(f"\n🔍 路由决策: action='{action}'")

return "tool" if action == "check_weather" else END

graph.add_conditional_edges(

"chain1",

router,

{"tool": "tool", END: END}

)

graph.add_edge("tool", "chain2")

graph.add_edge("chain2", END)

app = graph.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="北京天气怎么样")]

for event in app.stream(inputs):

node = list(event.keys())[0]

msg = event[node].content

print(f"\n[→ {node}]")

print(f" 💬 {msg}")

# 测试用例2：非天气问题

print("\n" + "=" * 60)

print("_TestCase 2: 非天气问题 → 直接返回答案_")

inputs = [HumanMessage(content="今天星期几")]

for event in app.stream(inputs):

node = list(event.keys())[0]

msg = event[node].content

print(f"\n[→ {node}]")

print(f" 💬 {msg}")

结果输出：

+-----------+
| __start__ |
+-----------+
*
*
*
+--------+
| chain1 |
+--------+
. .
.. .
. ..
+------+ .
| tool | .
+------+ .
* .
* .
* .
+--------+ .
| chain2 | ..
+--------+ .
* .
** ..
* .
+---------+
| __end__ |
+---------+
============================================================
🌤️ 天气查询 Agent 测试（纯 JSON 输出保障）
============================================================

_TestCase 1: 天气查询 → 触发工具调用_
chain1_node-->state:[HumanMessage(content='北京天气怎么样', id='5cf8d50d-fa84-4ff1-9c2c-418f8d426377')]

龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

让Codex给AI Agent装“眼睛和耳朵“：视频解析技能开发与踩坑修复全流程

本文探讨了如何为AI Agent添加视频/音频解析能力，通过FFmpeg+read_image+Whisper三段式架构实现。作者使用Codex生成技能包时遇到四个典型问题：Windows路径分隔符导致目录错误（改用pathlib.Path修复）、中文编码乱码（显式指定UTF-8编码）、沙箱禁止import openai（改用Groq免费API）、帧分析假数据（移交read_image处理）。最终

龙虾开发者社区

[智能体-636]：AI重构生产价值：从人才红利到数字智能资产的时代更迭

AI改变了这有传统的模式，它把每个行业每个领域的内的“人”的成功的知识、经验、技能、最佳实践数字化、共享化、商品化成AI智能体，AI数字人，并可以进行大规模复制、销售、传播，从大模型，到智能体，到技能Skill，到工具Tools，硅基智能的边界逐步延展到物理世界，在各个领域进行物质资料的加工和生产，组织只需要购买这些智能体，就可以完成生产。在这样的系统中，AI是工具，AI是资产，人不再是人才，组织

龙虾开发者社区

Cursor 的 Rules、Skills、Agent 到底怎么选？聊聊我的判断

Cursor 2.0 把多代理能力放出来之后，我身边不少人开始纠结一个问题：Rules、Commands、Skills、Sub-Agents，这些都能指挥 AI 干活，到底什么时候用哪个？什么时候写几条规则就够了，什么时候真得拉一个独立的 Agent 出来？我自己用了一阵子，踩了些坑，把判断逻辑理清楚了，写下来给你参考。