在早期Agent实现中,开发者需手动维护消息上下文列表,通过正则匹配解析工具指令,并用循环控制“思考→行动→观察→应答”流程。这种模式存在三大瓶颈:
状态管理碎片化:对话历史、工具调用记录分散在多个变量中
流程控制复杂:多轮工具调用需嵌套循环,错误处理代码臃肿
扩展性差:新增工具需修改核心逻辑,难以支持人工干预等场景
class AgentState(TypedDict): messages: Annotated[list[AnyMessage], operator.add] # 消息自动累积 intermediate_steps: Annotated[list[tuple], custom_merge] # 自定义状态合并
通过Annotated元数据声明状态合并策略:
operator.add:列表自动拼接(默认)
自定义函数:实现消息更新替换等高级逻辑
graph.add_node("llm", self.call_openai) # LLM节点 graph.add_node("action", self.take_action) # 工具执行节点
每个节点需满足:
输入:AgentState对象
输出:更新后的AgentState子集
职责单一:如工具节点仅处理执行逻辑
def exists_action(state: AgentState): return len(state['messages'][-1].tool_calls) > 0 # 检测工具调用 graph.add_conditional_edges( "llm", exists_action, {True: "action", False: END} # 动态路由 )
国内环境推荐使用博查搜索替代Tavily:
class BoChaSearchResults(BaseTool): def _run(self, query: str): payload = {"query": query, "count": 4} response = requests.post(API_URL, json=payload) return format_results(response.json()["data"]["webPages"]["value"])
关键适配技巧:
结果解析对齐LangChain的ToolMessage格式
错误处理返回结构化提示词
# 保存状态 checkpoint = graph.get_state(message_id) # 故障恢复 graph.recover_state(checkpoint)
支持从任意节点继续执行,保障长任务可靠性
def human_approve(state): if state["risk_level"] > 0.8: return "human_review" # 转人工审核 return "auto_process"
最后总结建议:
通过LangGraph,开发者能以“绘制流程图”的直观方式构建企业级Agent系统,显著降低认知负荷。其模块化设计更符合工程实践需求,是通向复杂智能体应用的必经之路。如果本次分享对你有所帮助,记得告诉身边有需要的朋友,"我们正在经历的不仅是技术迭代,而是认知革命。当人类智慧与机器智能形成共生关系,文明的火种将在新的维度延续。"在这场波澜壮阔的文明跃迁中,主动拥抱AI时代,就是掌握打开新纪元之门的密钥,让每个人都能在智能化的星辰大海中,找到属于自己的航向。