番摊机器人 LangGraph判断节点实战：用条件路由打造智能天气查询Agent
在AI Agent开发中，动态决策与流程分支是核心能力。传统线性调用难以应对“是否需要调用工具”“是否结束对话”
等灵活场景，而LangGraph的判断节点与条件边机制，能以极简方式实现工作流的智能路由。本文结合天气查询场景，
带您从零掌握LangGraph判断节点的实战技巧。
一、核心概念解析
1.1 判断节点与条件边
LangGraph中的判断节点是实现分支逻辑的关键，它本身不执行具体业务操作，仅根据当前状态决定流程走向。条件边则
是连接判断节点与业务节点的“智能枢纽”，通过路由函数实现动态流转：
普通边：类似固定铁轨，流程必须按指定路径执行
条件边：类似交通枢纽，根据状态数据动态选择下一个节点
1.2 状态管理的重要性
状态（State）是LangGraph的核心数据容器，所有节点通过状态共享信息。在天气查询场景中，状态需要包含：
用户的查询问题
已获取的天气数据
对话历史记录
流程控制标记
二、实战项目：智能天气查询Agent
2.1 需求分析
我们将构建一个具备以下能力的天气查询Agent：
理解用户的自然语言查询（如“北京明天天气怎么样？”）
自动判断是否需要调用天气API获取数据
处理多轮对话（如追问“那后天呢？”）
优雅处理无法回答的问题
2.2 环境准备
首先完成基础环境搭建：
# 安装依赖
!pip install langgraph langchain-openai python-dotenv
# 导入核心库
import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
2.3 定义状态结构
使用TypedDict定义清晰的状态模式：
class WeatherState(TypedDict):
"""天气查询Agent的状态结构"""
user_query: str # 用户的查询问题
weather_data: dict # 存储获取的天气数据
conversation_history: list # 对话历史记录
need_tool_call: bool # 是否需要调用工具的标记
current_city: str # 当前查询的城市
current_date: str # 当前查询的日期
2.4 实现业务节点
创建处理具体业务逻辑的节点函数：
2.4.1 意图理解节点
def understand_intent(state: WeatherState) -> WeatherState:
"""理解用户查询意图，提取城市和日期信息"""
prompt = f"""
分析用户查询意图：{state['user_query']}
请提取：
1. 目标城市（如果没有明确提及，使用默认值"北京"）
2. 查询日期（如果没有明确提及，使用默认值"今天"）
3. 是否需要调用天气API（是/否）
格式要求：JSON格式输出
"""
response = llm.invoke(prompt)
result = eval(response.content)
return {
"current_city": result["城市"],
"current_date": result["日期"],
"need_tool_call": result["是否需要调用API"] == "是",
"conversation_history": state["conversation_history"] + [
{"role": "user", "content": state["user_query"]},
{"role": "assistant", "content": "已理解查询意图"}
]
}
2.4.2 天气查询工具节点
def query_weather_api(state: WeatherState) -> WeatherState:
"""模拟调用天气API获取数据"""
# 这里使用模拟数据，实际开发中替换为真实API调用
mock_weather_data = {
"北京": {
"今天": {"temperature": "25℃", "condition": "晴", "wind": "微风"},
"明天": {"temperature": "28℃", "condition": "多云", "wind": "南风3级"}
},
"上海": {
"今天": {"temperature": "27℃", "condition": "阴", "wind": "东风2级"},
"明天": {"temperature": "30℃", "condition": "小雨", "wind": "东南风4级"}
}
}
weather_data = mock_weather_data.get(
state["current_city"],
mock_weather_data["北京"]
).get(
state["current_date"],
mock_weather_data["北京"]["今天"]
)
return {
"weather_data": weather_data,
"conversation_history": state["conversation_history"] + [
{"role": "assistant", "content": f"已获取{state['current_city']}{state['current_date']}的天气数据"}
]
}
2.4.3 结果生成节点
def generate_response(state: WeatherState) -> WeatherState:
"""生成自然语言回复"""
weather = state["weather_data"]
response_text = f"""
{state['current_city']}{state['current_date']}的天气情况：
🌡️ 温度：{weather['temperature']}
☁️ 天气：{weather['condition']}
💨 风力：{weather['wind']}
"""
return {
"conversation_history": state["conversation_history"] + [
{"role": "assistant", "content": response_text}
]
}
2.5 实现判断节点
创建核心的路由判断函数：
def should_call_tool(state: WeatherState) -> Literal["query_weather_api", "generate_response"]:
"""判断是否需要调用天气API"""
return "query_weather_api" if state["need_tool_call"] else "generate_response"
2.6 构建与编译图
# 创建状态图
workflow = StateGraph(WeatherState)
# 添加节点
workflow.add_node("understand_intent", understand_intent)
workflow.add_node("query_weather_api", query_weather_api)
workflow.add_node("generate_response", generate_response)
# 添加边
workflow.add_edge(START, "understand_intent")
workflow.add_conditional_edges(
"understand_intent",
should_call_tool,
{
"query_weather_api": "query_weather_api",
"generate_response": "generate_response"
}
)
workflow.add_edge("query_weather_api", "generate_response")
workflow.add_edge("generate_response", END)
# 编译图
app = workflow.compile()
2.7 测试Agent
# 测试用例1：需要调用工具的查询
test_input1 = {
"user_query": "上海明天的天气怎么样？",
"weather_data": {},
"conversation_history": [],
"need_tool_call": False,
"current_city": "",
"current_date": ""
}
result1 = app.invoke(test_input1)
print("测试1回复：", result1["conversation_history"][-1]["content"])
# 测试用例2：不需要调用工具的查询
test_input2 = {
"user_query": "重复一下刚才的天气信息",
"weather_data": result1["weather_data"],
"conversation_history": result1["conversation_history"],
"need_tool_call": False,
"current_city": "上海",
"current_date": "明天"
}
result2 = app.invoke(test_input2)
print("测试2回复：", result2["conversation_history"][-1]["content"])
三、高级技巧与优化
3.1 多条件判断
可以扩展路由函数实现更复杂的分支逻辑：
def advanced_router(state: WeatherState) -> Literal["query_api", "use_cache", "ask_clarification"]:
"""高级路由函数"""
if state["user_query"] in ["退出", "结束"]:
return END
elif state.get("weather_data"):
return "use_cache"
elif not state.get("current_city"):
return "ask_clarification"
else:
return "query_api"
3.2 加入人工干预
利用LangGraph的人类在环（Human-in-the-Loop）能力：
def human_review(state: WeatherState) -> WeatherState:
"""人工审核节点"""
print("请审核天气数据：", state["weather_data"])
approval = input("是否批准？(y/n)：")
return {"human_approved": approval.lower() == "y"}
def check_human_approval(state: WeatherState) -> Literal["generate_response", "human_review"]:
"""判断是否需要人工审核"""
return "generate_response" if state.get("human_approved", True) else "human_review"
3.3 状态持久化
实现对话状态的持久化存储：
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用SQLite存储状态
memory = SqliteSaver.from_conn_string("weather_agent.db")
app = workflow.compile(checkpointer=memory)
# 带会话ID的调用
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(test_input1, config=config)
四、常见问题与解决方案
4.1 路由逻辑错误
问题：流程走向不符合预期 解决方案：
在路由函数中添加详细日志
使用LangSmith追踪状态变化
测试边界条件
4.2 状态更新冲突
问题：多个节点同时修改同一状态字段 解决方案：
明确节点职责，避免交叉修改
使用状态合并策略
考虑使用不可变状态模式
4.3 性能优化
问题：复杂图执行速度慢 解决方案：
异步节点提升并发能力
缓存频繁调用的结果
优化大模型调用次数
五、总结
通过本次实战，我们掌握了LangGraph判断节点与条件边的核心用法，构建了一个具备动态决策能力的智能天气查询Agent。LangGraph的图思维模式让复杂工作流的开发变得直观可控，特别适合构建需要多轮决策、工具调用和状态管理的AI Agent应用。
未来可以扩展的方向包括：
集成真实的天气API服务
添加多语言支持
实现更复杂的对话管理逻辑
部署为可对外提供服务的API