《AI大模型应用开发实战从入门到精通共60篇》 36、Agent实战：用LangGraph构建可复用的工作流

36、Agent实战用LangGraph构建可复用的工作流昨天凌晨三点我盯着终端里那个诡异的死循环——Agent在调用天气API和日历API之间反复横跳每次返回的结果都正确但就是停不下来。日志里最后一条消息是“Agent决定再次查询天气”而此刻窗外正在下暴雨。这个bug让我意识到单纯依赖LLM的“自主决策”来编排工作流就像让一个醉汉开车——方向感时好时坏但永远不知道他什么时候会撞上护栏。LangGraph就是那个安全带。它不是要限制Agent的创造力而是给那些天马行空的决策画一条清晰的赛道。今天这篇笔记我会从那个让我失眠的bug讲起带你手撸一个可复用的工作流框架。从死循环到状态机先还原那个坑爹的场景。我用LangChain的AgentExecutor跑一个日程助手Agent需要先查天气再安排会议。代码大概长这样fromlangchain.agentsimportAgentExecutor,create_openai_functions_agentfromlangchain.toolsimporttooltooldefget_weather(city:str)-str:查询指定城市的天气# 这里踩过坑返回格式太随意导致Agent误解returnf{city}当前气温28°C湿度65%tooldefschedule_meeting(time:str,topic:str)-str:安排会议returnf已安排{topic}会议在{time}agentcreate_openai_functions_agent(llm,[get_weather,schedule_meeting])executorAgentExecutor(agentagent,tools[get_weather,schedule_meeting])用户输入“帮我安排明天下午3点的会议先看看天气怎么样”。Agent先调用了get_weather返回了正确结果然后它又调了一次get_weather接着又调了一次……直到max_iterations耗尽。问题出在哪Agent认为“看看天气”这个动作需要反复确认因为工具返回的字符串里没有明确的“任务完成”标记。LangGraph的解决思路很暴力——把Agent的思考过程拆成显式的节点每个节点只做一件事节点之间的跳转由你定义的逻辑控制而不是让LLM自己决定下一步该干嘛。核心概念节点、边、状态LangGraph的核心就三个东西状态State、节点Node、边Edge。别被名字唬住说白了就是状态一个字典存着所有节点共享的数据。比如当前对话历史、工具调用结果、用户意图等。节点一个函数输入是状态输出是更新后的状态。每个节点只负责一件事比如“调用天气API”或“生成最终回复”。边定义节点之间的跳转逻辑。可以是固定跳转A做完必须去B也可以是条件跳转根据状态里的某个字段决定去C还是D。这种设计天然避免了死循环——因为每个节点执行完后必须通过边明确告诉系统下一步去哪没有“再想想”这个选项。手撸一个可复用的工作流直接上代码我会把那个日程助手用LangGraph重写。先装依赖pipinstalllanggraph langchain-openai第一步定义状态状态就是一个TypedDict别搞复杂了。我习惯把“中间结果”和“最终输出”分开存fromtypingimportTypedDict,List,Optionalfromlanggraph.graphimportStateGraph,ENDclassAgentState(TypedDict):messages:List[dict]# 对话历史每个元素是{role: user/assistant, content: ...}weather_result:Optional[str]# 天气查询结果None表示还没查meeting_result:Optional[str]# 会议安排结果next_action:str# 下一个要执行的节点名由条件边决定这里有个小技巧next_action字段是给条件边用的相当于一个“路标”。别把逻辑写在节点内部那样又回到AgentExecutor的老路上了。第二步定义节点函数每个节点函数接收状态返回一个字典字典的key就是要更新的状态字段。注意节点函数不能直接修改传入的状态必须返回一个新字典。defcall_weather_tool(state:AgentState)-dict:节点1调用天气工具# 从对话历史中提取城市名这里简化处理user_msgstate[messages][-1][content]# 别这样写直接用LLM解析太慢。我通常用正则或简单字符串匹配city北京if北京inuser_msgelse上海# 实际项目用NER# 模拟调用天气APIweatherf{city}当前气温28°C湿度65%return{weather_result:weather,messages:state[messages][{role:assistant,content:f天气查询结果{weather}}],next_action:decide_next# 告诉条件边我干完了你来决定下一步}defcall_meeting_tool(state:AgentState)-dict:节点2调用会议工具# 从天气结果和用户消息中提取时间time15:00# 简化处理topic项目评审meetingf已安排{topic}会议在{time}return{meeting_result:meeting,messages:state[messages][{role:assistant,content:f会议安排结果{meeting}}],next_action:decide_next}defdecide_next_action(state:AgentState)-dict:节点3决策节点——根据当前状态决定下一步# 这里踩过坑不要用LLM做决策太慢且不可控。用规则引擎或简单条件判断ifstate[weather_result]isNone:return{next_action:call_weather}elifstate[meeting_result]isNone:return{next_action:call_meeting}else:return{next_action:generate_response}defgenerate_response(state:AgentState)-dict:节点4生成最终回复# 组装最终输出final_msgf天气{state[weather_result]}\n会议{state[meeting_result]}return{messages:state[messages][{role:assistant,content:final_msg}],next_action:END# 告诉图结束了}注意decide_next_action这个节点——它才是整个工作流的“大脑”。别让LLM干这个活LLM适合生成内容不适合做流程控制。用硬编码的条件判断既快又稳。第三步构建图这是LangGraph最爽的部分——像搭积木一样把节点和边连起来# 初始化图指定状态类型workflowStateGraph(AgentState)# 添加节点workflow.add_node(call_weather,call_weather_tool)workflow.add_node(call_meeting,call_meeting_tool)workflow.add_node(decide_next,decide_next_action)workflow.add_node(generate_response,generate_response)# 设置入口节点workflow.set_entry_point(decide_next)# 添加条件边从decide_next出发根据状态里的next_action跳转workflow.add_conditional_edges(decide_next,lambdastate:state[next_action],# 读取状态中的路标{call_weather:call_weather,call_meeting:call_meeting,generate_response:generate_response})# 添加固定边工具节点执行完后必须回到决策节点workflow.add_edge(call_weather,decide_next)workflow.add_edge(call_meeting,decide_next)workflow.add_edge(generate_response,END)# 编译图appworkflow.compile()这段代码的逻辑清晰得像电路图从决策节点开始根据状态跳转到对应工具节点工具节点执行完回到决策节点直到所有任务完成才生成最终回复。没有死循环的可能因为每个工具节点执行后都会回到决策节点而决策节点只有三个出口没有“再查一次天气”这个选项。第四步运行工作流# 初始状态initial_state{messages:[{role:user,content:帮我安排明天下午3点的会议先看看北京天气}],weather_result:None,meeting_result:None,next_action:decide_next}# 执行resultapp.invoke(initial_state)# 输出最终消息print(result[messages][-1][content])输出天气北京当前气温28°C湿度65% 会议已安排项目评审会议在15:00让工作流可复用上面的例子是硬编码的实际项目中我们需要一个通用的工作流模板。我的做法是抽象出一个ToolNode和一个DecisionNode让它们根据配置动态执行classToolNode:def__init__(self,tool_func,tool_name):self.tool_functool_func self.tool_nametool_namedef__call__(self,state:AgentState)-dict:# 通用工具调用逻辑resultself.tool_func(state)return{f{self.tool_name}_result:result,next_action:decide_next}classDecisionNode:def__init__(self,rules:dict):# rules: {condition_key: next_node_name, ...}self.rulesrulesdef__call__(self,state:AgentState)-dict:forcondition,next_nodeinself.rules.items():ifstate.get(condition)isNone:return{next_action:next_node}return{next_action:generate_response}这样你只需要定义工具函数和决策规则就能快速组装不同的工作流。比如换一个“先查汇率再计算价格”的工作流只需要改rules字典和工具函数图结构完全复用。踩坑记录状态字段命名规范所有节点返回的字段名必须一致否则条件边会找不到路标。我习惯用next_action作为统一的路标字段。节点函数的幂等性每个节点应该能安全地重复执行。比如天气节点如果状态里已经有weather_result应该直接返回而不重复调用API。我通常会在节点函数开头加一个检查ifstate.get(weather_result)isnotNone:return{next_action:decide_next}# 跳过调试技巧LangGraph自带一个get_graph()方法可以打印出图的拓扑结构。我经常在编译前调用print(app.get_graph())确认节点和边的连接是否正确。与LLM的协作虽然决策节点我用规则但工具调用参数提取还是可以用LLM。我的做法是在工具节点内部调用一个小模型比如GPT-3.5-turbo来解析用户意图而不是让LLM控制流程。个人经验别把LangGraph当成万能药。如果你的工作流只有两三个步骤用简单的if-else比图更直接。LangGraph的价值在于当你有5个以上节点、节点之间有复杂的条件跳转、需要多人协作维护时它的结构化优势才会显现。另外我强烈建议把决策逻辑和业务逻辑分离。决策节点只负责“下一步去哪”不负责“怎么做”。这样当你需要修改流程时只需要改决策规则不用动工具函数。反之亦然。最后那个让我失眠的死循环bug用LangGraph重写后再也没出现过。不是因为LangGraph有多智能而是因为它强制我把“Agent的思考过程”变成了“工程师的流程图”。有时候限制才是真正的自由。