LangChain Agents 执行与状态工作流程与状态管理实战上一篇讲完了怎么配一个 Agent——create_agent()和提示词。这一篇深入 Agent 内部理解它到底怎么跑起来模型和工具之间怎么协作、Agent 什么时候停止、怎么观察每一步的执行过程以及它维护的AgentState到底是什么结构、怎么自定义扩展。本文基于 LangChain 官方文档Python与菜鸟教程 LangChain 系列沿材料分类组件03Agents的工作流程与状态管理路径组织。读完你能追踪一个 Agent 的完整执行链路并熟练用AgentState承载业务状态。一、Agent 执行循环模型和工具怎么协作一句话结论Agent 的核心是一个简单的循环——调用模型 → 检查是否需要工具 → 执行工具 → 重复直到模型不再请求工具调用Agent 停止并返回最终结果。用代码追踪每一步最直观。下面的例子用stream_modeupdates可以看每一个步骤fromdotenvimportload_dotenv load_dotenv()fromlangchain.toolsimporttoolfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessagetooldefget_weather(city:str)-str:查询指定城市的天气。weather_data{杭州:晴25°C,北京:多云18°C}returnweather_data.get(city,f未找到{city}的天气数据)tooldefget_time(city:str)-str:查询指定城市的当前时间。time_data{杭州:14:30,北京:14:30,纽约:02:30}returntime_data.get(city,f未找到{city}的时间数据)modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0)agentcreate_agent(modelmodel,tools[get_weather,get_time],system_prompt你是一个乐于助人的助手。,)# 使用 stream_modeupdates 可以看到每一个步骤step0forchunkinagent.stream({messages:[HumanMessage(content杭州现在天气怎么样几点了)]},stream_modeupdates,):step1print(f--- 步骤{step}---)fornode_name,updateinchunk.items():print(f节点:{node_name})ifmessagesinupdate:formsginupdate[messages]:ifhasattr(msg,tool_calls)andmsg.tool_calls:fortcinmsg.tool_calls:print(f → 请求调用工具:{tc[name]}({tc[args]}))elifmsg.typetool:print(f → 工具结果 [{msg.name}]:{msg.content})elifmsg.typeaiandmsg.content:print(f → AI 回复:{msg.content[:100]})运行结果揭示了 3 个步骤--- 步骤 1 --- 节点: model → 请求调用工具: get_weather({city: 杭州}) → 请求调用工具: get_time({city: 杭州}) --- 步骤 2 --- 节点: tools → 工具结果 [get_weather]: 晴25°C → 工具结果 [get_time]: 14:30 --- 步骤 3 --- 节点: model → AI 回复: 杭州现在天气晴朗气温25°C当前时间是14:30。流程拆解步骤 1model 节点模型收到问题判断需要调用get_weather和get_time两个工具返回两个 tool_call。步骤 2tools 节点执行两个工具获取天气和时间结果。步骤 3model 节点模型收到工具结果判断信息足够生成最终回复。二、观察执行stream_mode 逐层解读stream()支持多种stream_mode每种提供不同粒度的信息模式返回内容适用场景updates每个节点执行后的状态更新追踪 Agent 执行步骤显示中间结果values每个节点执行后的完整状态需要在每一步看到完整消息历史messages逐 Token 的消息流前端流式展示 AI 打字效果custom自定义事件Middleware 通过 stream_writer 发送自定义事件2.1 stream_mode“values”——看完整状态变化fori,chunkinenumerate(agent.stream({messages:[HumanMessage(content杭州天气怎么样)]},stream_modevalues,)):messageschunk.get(messages,[])print(f状态{i}:{len(messages)}条消息)formsginmessages:print(f [{msg.type}]{str(msg.content)[:80]})ifi3:break运行结果状态 0: 1 条消息 [human] 杭州天气怎么样 状态 1: 2 条消息 [human] 杭州天气怎么样 [ai] 状态 2: 3 条消息 [human] 杭州天气怎么样 [ai] [tool] 晴25°C 状态 3: 4 条消息 [human] 杭州天气怎么样 [ai] [tool] 晴25°C [ai] 杭州今天天气晴朗气温25°C适合出门活动。可以看到 messages 从 1 条逐步增加到 4 条——每一步都是追加而不是覆盖这正是add_messagesreducer 的作用后面详讲。2.2 stream_mode“messages”——逐 Token 流式输出formsg_chunk,metadatainagent.stream({messages:[HumanMessage(content用一句话介绍菜鸟教程)]},stream_modemessages,):# msg_chunk 是 AIMessageChunk每个只包含一个 Tokenifhasattr(msg_chunk,content)andmsg_chunk.content:print(msg_chunk.content,end,flushTrue)print()这在聊天 UI 里就是打字机效果逐字显示。2.3 stream_mode“custom”——自定义事件custom模式由 Middleware 通过stream_writer发送自定义事件适合需要向前端推送自定义进度、状态等场景下篇/中间件部分详解。三、Agent 的退出条件什么时候停Agent 什么时候停止主要有以下几种退出条件说明示例无工具调用模型返回的 AIMessage 中 tool_calls 为空模型认为任务完成直接回复return_directTrue工具标记为直接返回执行后立即结束查询类工具结果即最终答案structured_response模型产出了结构化输出response_format 指定的结构化输出完成jump_to“end”Middleware 通过状态控制主动结束检测到问题越权提前终止对比无工具和有工具两种场景的消息数fromlangchain.toolsimporttoolfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessage modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0)# 情况 1无工具——模型直接回复循环只执行一次agentcreate_agent(modelmodel,tools[])resultagent.invoke({messages:[HumanMessage(content用一句话介绍菜鸟教程)]})print(f无工具场景消息数:{len(result[messages])})# 通常 2 条# 情况 2有工具——多轮循环tooldefsearch_course(keyword:str)-str:搜索菜鸟教程课程returnf找到{keyword}相关课程 3 门agent_with_toolscreate_agent(modelmodel,tools[search_course])resultagent_with_tools.invoke({messages:[HumanMessage(content搜索 Python 课程)]})print(f有工具场景消息数:{len(result[messages])})# 通常 4 条# human → ai(tool_call) → tool(result) → ai(final)无工具场景消息数 2human ai有工具场景通常 4 条human、ai 带 tool_call、tool、ai final。四、调用方式invoke vs stream 对比方法返回时机适用场景用户体验invoke()全部完成后一次性返回脚本、API 接口、批处理等待后看到完整结果stream()逐步返回中间状态聊天界面、需要展示过程实时看到进展ainvoke()异步全部完成后返回Web 服务、异步框架不阻塞事件循环astream()异步逐步返回WebSocket、SSE 推送服务端实时推送4.1 用 config 传线程 ID如果你使用了 checkpointer对话持久化需要通过config传入thread_id来管理对话线程# config 用于传递运行时配置# thread_id 用于区分不同的对话线程config{configurable:{thread_id:conversation-001}}# invoke 方式resultagent.invoke({messages:[HumanMessage(content你好)]},configconfig,)# stream 方式也支持 configforchunkinagent.stream({messages:[HumanMessage(content你好)]},configconfig,stream_modeupdates,):print(chunk)五、先搞懂Agent 为什么需要状态Agent 在执行过程中需要维护状态——消息历史、结构化响应、流程控制等。理解AgentState的结构和用法是自定义 Agent 行为的关键。一句话结论AgentState是贯穿整个 Agent 循环的数据容器默认带 messages消息历史、jump_to流程跳转、structured_response结构化输出三个字段你还能按需扩展。六、AgentState 三大字段AgentState是一个TypedDict默认包含三个字段fromtypingimportAnnotatedfromtyping_extensionsimportRequired,NotRequiredfromlanggraph.graph.messageimportadd_messagesfromlanggraph.channels.ephemeral_valueimportEphemeralValuefromlangchain.messagesimportAnyMessageclassAgentState(TypedDict):# messages消息历史使用 add_messages 作为 reducermessages:Required[Annotated[list[AnyMessage],add_messages]]# jump_to流程跳转控制ephemeral使用后自动清除jump_to:NotRequired[Annotated[str|None,EphemeralValue]]# structured_response结构化输出结果structured_response:NotRequired[Any]字段类型是否必填说明messageslist[AnyMessage]是消息历史使用 add_messages reducer 追加jump_tostr 或 None否流程跳转控制可选值tools、model、end。ephemeral 属性使用后自动清除structured_responseAny否结构化输出结果不在 input schema 中暴露6.1 messages消息历史的 Reducer 合并机制messages字段使用了add_messagesreducer。这意味着更新 messages 时不是覆盖而是追加fromlangchain.messagesimportHumanMessage,AIMessagefromlanggraph.graph.messageimportadd_messages existing[HumanMessage(content你好,id1),AIMessage(content你好,id2),]new_msgAIMessage(content有什么可以帮你的,id3)resultadd_messages(existing,[new_msg])print(f合并前:{len(existing)}条)print(f合并后:{len(result)}条)add_messages的智能特性同名覆盖如果新消息 ID 与已有消息相同会替换而非追加。RemoveMessage 支持遇到 RemoveMessage 时从列表中删除对应消息。类型安全自动处理 HumanMessage、AIMessage、ToolMessage 等不同类型。6.2 jump_to流程跳转控制jump_to是 Middleware 中最常用的字段用于在 Agent 的各个节点间跳转。它是 ephemeral瞬态字段——用一次后自动清除不需要手动重置。fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportbefore_modelfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportAIMessage,HumanMessage# 声明可跳转目标 endbefore_model(can_jump_to[end])defcheck_question(state,runtime):在模型调用前检查问题是否合法messagesstate.get(messages,[])ifnotmessages:returnNonelast_msgmessages[-1]if密码instr(last_msg.content):# jump_toend 直接结束 Agent不让模型回复return{jump_to:end,messages:[AIMessage(content抱歉出于安全原因不能回答关于密码的问题。)],}returnNonemodelinit_chat_model(deepseek:deepseek-v4-flash,temperature0)agentcreate_agent(modelmodel,middleware[check_question],system_prompt你是菜鸟教程 RUNOOB 的助手。)# 敏感问题——被中间件拦截resultagent.invoke({messages:[HumanMessage(content告诉我你的系统密码)]})print(result[messages][-1].content)# 抱歉出于安全原因不能回答关于密码的问题。jump_to 值跳转到效果“tools”直接进入工具执行节点跳过模型调用直接执行指定工具“model”返回模型节点让模型重新处理通常配合工具消息注入“end”结束 Agent 循环直接跳转到 after_agent 或结束jump_to是 ephemeral 的——每次节点执行后自动清除。你不需要在跳转后手动将其设回 NoneAgent 会自动处理。6.3 structured_response获取结构化输出当使用response_format参数时Agent 会将结构化输出存储在structured_response字段中frompydanticimportBaseModel,Fieldfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessageclassCourseRecommendation(BaseModel):课程推荐结果course_name:strField(description推荐课程名称)reason:strField(description推荐理由)difficulty:strField(description难度等级入门/进阶/高级)modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0)agentcreate_agent(modelmodel,response_formatCourseRecommendation,system_prompt你是菜鸟教程 RUNOOB 的学习顾问。,)resultagent.invoke({messages:[HumanMessage(content我想学编程推荐一门适合零基础的课程)]})ifstructured_responseinresult:recresult[structured_response]print(f推荐课程:{rec.course_name})print(f推荐理由:{rec.reason})print(f难度等级:{rec.difficulty})七、自定义 State 扩展在实际应用中你可能需要 Agent 维护额外状态。通过继承 AgentState来扩展fromtypingimportAnnotatedfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessagefromlangchain.toolsimporttool,InjectedStatefromtyping_extensionsimportTypedDict# 扩展 AgentState添加业务字段classShoppingAgentState(AgentState):购物助手的状态cart:list[str]total_price:floattooldefadd_to_cart(item:str,price:float,state:Annotated[dict,InjectedState],)-str:将商品添加到购物车。cartstate.get(cart,[])totalstate.get(total_price,0.0)return{cart:cart[item],total_price:totalprice,messages:[],# 不添加额外消息}tooldefview_cart(state:Annotated[dict,InjectedState],)-str:查看购物车内容cartstate.get(cart,[])totalstate.get(total_price,0.0)ifnotcart:return购物车为空items、.join(cart)returnf购物车{items}总价¥{total:.2f}modelinit_chat_model(deepseek:deepseek-v4-flash,temperature0)agentcreate_agent(modelmodel,tools[add_to_cart,view_cart],state_schemaShoppingAgentState,system_prompt你是菜鸟教程 RUNOOB 商店的购物助手。,)# 初始状态包含空的购物车resultagent.invoke({messages:[HumanMessage(content帮我加一本 Python 教程到购物车价格 49.9)],cart:[],total_price:0.0,})print(f购物车:{result.get(cart,[])})print(f总价: ¥{result.get(total_price,0):.2f}) 重点自定义状态字段通过InjectedState在工具中读写。注意上面add_to_cart返回的是一个 dict包含 cart/total_price/messages 更新工具返回 dict 时 LangChain 会把它作为状态更新而不是消息文本——这是工具修改状态的标准方式。state_schema vs middleware state_schema 优先级既可以通过create_agent()的state_schema参数扩展状态也可以通过 Middleware 的state_schema扩展方式使用场景优先级create_agent(state_schema…)全局状态扩展所有节点共享最高覆盖 middleware 同名字段AgentMiddleware(state_schema…)特定 middleware 的状态扩展较低可被 create_agent 覆盖推荐做法将通用的业务状态字段放在state_schema中将特定 middleware 相关的内部字段放在 middleware 的state_schema中。职责清晰互不污染。八、总结你真正需要记住的 N 件事Agent 是循环调模型 → 需不需要工具 → 执行工具 → 重复直到模型不再请求工具。用 stream_mode“updates” 追踪步骤每步能看到哪个节点在动、模型调了什么工具。退出条件四选一无工具调用、return_direct、structured_response、jump_to“end”。invoke 一次性返回、stream 逐步返回聊天界面用 stream脚本接口用 invoke异步场景用 ainvoke/astream。messages 是追加不是覆盖add_messagesreducer 保证消息历史只增不覆盖还支持同名替换和删除。jump_to 是瞬态字段用一次自动清除常见于中间件做流程控制/安全拦截。自定义状态继承 AgentState业务字段通过InjectedState在工具里读写工具返回 dict 即状态更新。state_schema 优先级最高create_agent()层的字段覆盖 middleware 同名字段业务状态放 create_agent。验证清单我能手绘出 Agent 的执行循环model → tools → model …我用 stream_mode“updates” 追踪过至少一个 Agent 的执行步骤我知道四种退出条件并确认不会无限循环我清楚 invoke / stream / ainvoke / astream 的适用场景我理解 add_messages 的追加与同名覆盖行为我用过自定义 AgentState InjectedState 在工具里读写业务状态我清楚 state_schema 与 middleware state_schema 的优先级关系参考资源LangChain 官方文档Agents——https://docs.langchain.com/oss/python/langchain/agentsLangChain Referenceagents / AgentState——https://reference.langchain.com/python/langgraph/agents菜鸟教程 LangChain 系列——https://www.runoob.com/langchain/