LangChain 系列写了这么多篇前面几篇把模型调用、提示词、RAG、记忆、工具都串过了但我们始终绕开了一个最关键的问题这些组件到底是怎么被组织起来形成一个能自主完成任务的 Agent 的这次就把langchain-6-9-agent执行流程这个主题彻底拆开。不聊概念堆砌直接看 Agent 在执行时内部发生了什么一轮任务进来LLM 拿着工具列表怎么思考、怎么决策、怎么调用工具、怎么根据观测结果收敛到最终答案。同时把 LangGraph 与 LangChain 的关系、不同 Agent 范式的选择、多 Agent 主从模式也一并讲清楚。这篇文章会覆盖Agent 执行流程的源码级拆解、一个可运行的最简 ReAct Agent 示例、工具注册与参数解析、记忆在流程中的生效位置、Plan-and-Execute 任务规划实现、不同 Agent 范式的对比、多 Agent 协作的 Subagent 调用方式以及调试和排查方法。适合已经会写基础 LangChain 代码、但对 Agent 内部执行机制还不够清楚的开发者。1. 核心能力速览能力项说明核心主题LangChain Agent 的执行流程、内部循环与组件协作核心组件LLM、工具Tool、记忆Memory、规划器Planner、执行器Executor默认流程范式ReAct思考Thought→ 行动Action→ 观察Observation→ 最终答案Final Answer变体范式OpenAI Tools Agent、Plan-and-Execute、自定义 Agent 范式与 LangGraph 关系LangGraph 是 LangChain 官方推荐的底层编排框架可以自定义更细粒度的 Agent 循环典型应用场景工具调用、信息查询、代码生成、多步任务规划、多 Agent 协作调试手段verbose 模式、中间步骤输出、LangSmith 链路追踪关键依赖langchain、langchain-core、langchain-openai或其他模型接入包说明一点本文不涉及具体显存、显卡、GPU 参数因为 Agent 执行流程属于应用层代码逻辑与模型推理的后端资源消耗关系不大。如果你把它接到本地模型上显存开销取决于你用的推理服务而不是 LangChain 本身。2. Agent 执行流程的本质从一个普通 LLM 调用说起为了理解 Agent 的执行流程先看一个普通的 LLM 调用是什么样的。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) result llm.invoke(今天北京天气怎么样) print(result.content)这段代码的问题很明显gpt-4o-mini的训练数据有截止时间它不联网、不能查实时天气、不能执行任何外部操作。你问它天气它只能根据常识编一个回答。这不是模型笨而是模型本身只有生成文本的能力没有“获取新信息”的能力。Agent 要解决的就是这个问题让模型在回答之前能够主动调用工具、获取外部数据、观察结果再基于新的信息继续推理。整个过程是一个循环不是一次 LLM 调用就结束的。2.1 Agent 与普通 LLM 调用的区别对比项普通 LLM 调用Agent 执行输入用户 prompt用户 prompt 工具列表 历史记忆输出文本最终回答或中间步骤序列是否有外部动作无可以调用工具、查询 API、执行代码是否多轮推理否是可以多轮迭代直到收敛失败处理直接输出错误或编造可重试、可换工具、可主动停止把这两者放在一起看Agent 的执行流程本质上是在每一轮迭代中LLM 接收当前状态决定是调用工具还是输出最终答案。3. LangChain Agent 执行流程源码视角拆解LangChain 中 Agent 执行的核心组件有三个Agent负责决定下一步做什么也就是 LLM 与提示词的组合。ToolsAgent 可以调用的外部能力。AgentExecutor负责协调循环逻辑把 Agent 的决策落地。3.1 AgentExecutor 的循环逻辑从源码角度看AgentExecutor 的_call方法内部是一个 while 循环核心流程如下1. 接收用户输入。 2. 将输入、工具描述、中间步骤、记忆拼装成 prompt。 3. 交给 LLM得到一个 AgentAction 或 AgentFinish。 4. 如果是 AgentFinish返回最终输出。 5. 如果是 AgentAction执行对应工具得到 Observation。 6. 把 Action 和 Observation 追加到中间步骤。 7. 回到第 2 步继续循环。 8. 设置最大迭代次数防止死循环。用伪代码表示就是# LangChain AgentExecutor 的简化逻辑 def execute(self, inputs): intermediate_steps [] iterations 0 while iterations self.max_iterations: # 1. 让 Agent 决定下一步 output self.agent.plan(inputs, intermediate_steps) # 2. 如果决定结束返回最终答案 if isinstance(output, AgentFinish): return output.return_values # 3. 否则执行工具 observation self.tools[output.tool].run(output.tool_input) # 4. 记录中间步骤 intermediate_steps.append((output, observation)) iterations 1 return {output: 达到最大迭代次数停止执行}这段伪代码对应了实际AgentExecutor的骨架。理解了这个循环你就理解了 LangChain Agent 执行流程的核心。3.2 ReAct 范式中的每一步以 ReActReason Act范式为例Agent 在每一轮都会输出一个包含三个部分的文本Thought: 我需要知道北京的天气 Action: search Action Input: 北京今天天气LangChain 通过输出解析器OutputParser把这些文本解析成结构化的AgentAction然后执行工具。工具返回内容后下一轮 prompt 中会追加Observation: 北京今天晴气温 18-28 摄氏度LLM 看到 Observation 后如果认为信息足够了就输出Final Answer: 北京今天天气晴朗气温 18-28 摄氏度。从 Thought 到 Final Answer 的整个过程就是一次完整的 Agent 执行流程。4. 最简可运行的 Agent 执行流程示例为了把流程讲透我们写一个真实可运行的例子。这个例子不接外部 API用一个本地假工具代替真实工具方便观察整个循环。4.1 环境准备pip install langchain langchain-openai如果你用的是 OpenAI 兼容接口包括本地模型、中转服务也可以通过base_url来配置。这里以 OpenAI 为例import os os.environ[OPENAI_API_KEY] your-api-key4.2 定义工具from langchain.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气情况。输入参数为城市名称返回天气描述。 # 这里模拟工具返回真实项目中可以替换为天气 API 调用 return f{city} 今天晴气温 22 摄氏度东南风 2 级。注意工具的函数名、参数名、docstring 都非常重要。LLM 会通过这些信息来理解工具的功能和参数格式。工具描述写得越清晰Agent 的决策越准确。4.3 创建 ReAct Agentfrom langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) agent create_react_agent(llmllm, tools[get_weather], promptagent_prompt) agent_executor AgentExecutor( agentagent, tools[get_weather], verboseTrue, max_iterations5, handle_parsing_errorsTrue, )这里注意create_react_agent需要传入一个promptLangChain 在langchain.agents模块中提供了PREFIX、FORMAT_INSTRUCTIONS、SUFFIX等模板片段我们可以直接拼接from langchain_core.prompts import PromptTemplate from langchain.agents import AgentExecutor, create_react_agent from langchain.agents.format_scratchpad.log import format_log_to_str prompt PromptTemplate.from_template( You are a helpful assistant. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: {agent_scratchpad} )LangChain 的create_react_agent内部会处理tools和tool_names的注入所以上面模板中的{tools}、{tool_names}、{input}、{agent_scratchpad}都需要保留。4.4 运行 Agent 并观察执行流程result agent_executor.invoke({input: 北京今天天气怎么样}) print(result[output])如果你开启了verboseTrue控制台会输出完整的执行过程类似这样 Entering new AgentExecutor chain... Thought: 我需要查询北京的天气。 Action: get_weather Action Input: 北京 Observation: 北京 今天晴气温 22 摄氏度东南风 2 级。 Thought: 我已经获得了北京的天气信息现在可以回答用户的问题。 Final Answer: 北京今天天气晴朗气温 22 摄氏度东南风 2 级。 Finished chain.这段输出就是 Agent 执行流程的可视化结果。你可以清楚地看到Thought是模型的内部推理Action是工具选择Action Input是传给工具的参数Observation是工具的返回结果最后Final Answer是面向用户的最终输出。如果你把verboseFalse那么外部调用者只能看到最终的result[output]而中间过程会被隐藏。这就是 Agent 执行流程的封装性。5. Agent 的工具注册与参数解析在实际项目中一个 Agent 往往有多个工具。比如一个客服 Agent 可能有查订单工具、查物流工具、退款工具、查商品信息工具。LLM 需要从多个工具中选择一个合适的来调用。5.1 多工具注册from langchain.tools import tool tool def get_order_status(order_id: str) - str: 根据订单号查询订单状态。 return f订单 {order_id} 已发货预计 3 天后送达。 tool def get_product_info(product_name: str) - str: 根据商品名称查询商品详情和价格。 return f{product_name} 售价 299 元库存充足。 tools [get_order_status, get_product_info]5.2 工具选择和参数解析当模型决定调用工具时它不仅要选择工具名还要按工具的参数 schema 填参数。比如get_order_status需要order_id模型会从用户输入中提取订单号填入。如果工具需要多个参数模型会以 JSON 格式输出LangChain 内部通过tool.args_schema定义的 Pydantic 模型来校验参数。from pydantic import BaseModel, Field class OrderQueryInput(BaseModel): order_id: str Field(description订单号) user_id: str Field(description用户ID) tool(args_schemaOrderQueryInput) def get_order_detail(order_id: str, user_id: str) - str: 根据订单号和用户ID查询订单详情。 return f订单 {order_id} 属于用户 {user_id}商品总价 599 元。参数 schema 定义得越严格模型越不容易产生幻觉参数。这也是 Agent 执行流程中容易被忽略的细节工具描述和参数描述决定了 Agent 的行为上限。5.3 工具调用失败的兜底实际执行中模型可能输出一个不存在的工具名或者参数格式错误。AgentExecutor的handle_parsing_errors参数可以控制这种行为agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, max_iterations5, handle_parsing_errors你的格式有误请重新按格式输出。, )当模型输出解析失败时这个字符串会作为 Observation 返回给模型让模型有机会修正自己的输出。这是一个非常实用的容错机制。6. 记忆如何在 Agent 执行流程中生效Agent 执行流程中如果用户连续提问Agent 需要记住之前的对话内容否则上下文就断了。6.1 对话记忆的注入位置在 LangChain 的 Agent 执行流程中历史消息通过chat_history字段注入到 prompt 中。以create_react_agent为例我们需要自己管理记忆from langchain.memory import ConversationBufferMemory memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, )注意create_react_agent生成的 prompt 中默认没有chat_history字段。更常见的做法是用initialize_agent配合AgentExecutor或者从langchain.agents的create_tool_calling_agent配带chat_history的 prompt。简单起见我们可以手动构造记忆注入from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手。), MessagesPlaceholder(variable_namechat_history), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])然后在调用时主动传入历史result agent_executor.invoke({ input: 北京天气怎么样, chat_history: memory.chat_memory.messages, })6.2 记忆在流程中的生效时机当用户第二轮提问“那上海呢”时Agent 需要知道“那”指的是“天气”。如果把对话历史塞进 prompt模型就能理解。如果没有历史模型会把这个句子当成独立问题回答就会很奇怪。所以记忆在 Agent 执行流程中的位置是在每一轮循环开始时历史消息作为上下文的一部分注入 LLM 输入而不是在执行过程中动态追加。如果你使用的是initialize_agentLangChain 会通过memory参数自动把历史注入到chat_history字段。但使用更底层的create_react_agent时需要自己处理。这也是从 0.1 到 0.2 版本迁移时最容易踩的坑。7. 任务规划Plan-and-Execute 流程的实现ReAct 流程的特点是“边想边做”每一步都根据观测结果重新决策。但有些任务步骤明确比如“分析竞品 → 生成报告 → 发送邮件”如果每一步都让 LLM 临时决策既慢又不稳定。Plan-and-Execute 范式把执行流程拆成两个阶段规划阶段LLM 先生成一个完整的任务计划列表。执行阶段按计划逐项执行每项执行完可以更新计划。7.1 用 LangGraph 实现 Plan-and-ExecuteLangGraph 是 LangChain 官方的底层编排框架用它实现 Plan-and-Execute 更灵活。这里给一个简化示例from langgraph.graph import StateGraph, END from typing import TypedDict, List class AgentState(TypedDict): input: str plan: List[str] step_index: int output: str def planner(state: AgentState): # 这里调用 LLM 生成计划 plan [搜索最新行业报告, 整理关键数据, 生成总结] return {plan: plan, step_index: 0} def executor(state: AgentState): # 执行当前步骤 current_step state[plan][state[step_index]] # 调用工具…… return {step_index: state[step_index] 1} def should_continue(state: AgentState): if state[step_index] len(state[plan]): return END return executor graph StateGraph(AgentState) graph.add_node(planner, planner) graph.add_node(executor, executor) graph.add_edge(planner, executor) graph.add_conditional_edges(executor, should_continue) graph.set_entry_point(planner) app graph.compile()这种实现方式把“规划”和“执行”拆成两个独立的节点在复杂任务中的可控性比 ReAct 更高。如果你想深入了解 LangGraph 与 LangChain 的区别核心要点是LangChain 提供的是 Agent 的高层封装LangGraph 提供的是 Agent 流程的自定义实现能力。7.2 怎么选择ReAct 还是 Plan-and-Execute维度ReActPlan-and-Execute任务类型开放域、不确定性高步骤明确、多阶段灵活性高每步自适应中按计划执行速度慢每步都推理快规划一次容错性高观测后可改变方向低计划可能有偏差适用场景客服、问答、自主搜索数据处理流水线、代码生成、报表生成没有绝对的好坏只有适合场景的区别。实际项目中甚至可以把两者结合先用 Plan-and-Execute 生成计划再对单个步骤用 ReAct 精细执行。8. Agent 不同范式的对比Tools Agent 与自定义范式热搜词里有一个高频问题LangChain 中 Agent 可以用不同的 Agent 范式吗答案是肯定的。8.1 常见 Agent 范式范式代表性接口核心机制ReActcreate_react_agent文本格式输出 Thought/Action/ObservationOpenAI Toolscreate_tool_calling_agent模型原生支持 function calling结构化输出Structured Chatcreate_structured_chat_agent以 JSON 结构化输出工具调用Plan-and-Execute自定义或 LangGraph先规划再执行Custom继承BaseSingleActionAgent完全自定义决策逻辑8.2 OpenAI Tools Agent 示例OpenAI 的模型原生支持工具调用参数直接以 JSON 结构传给模型比 ReAct 的文本解析更稳定from langchain.agents import create_tool_calling_agent, AgentExecutor prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({input: 帮我查一下订单 12345 的状态})create_tool_calling_agent不需要手动维护 Action/Action Input 的文本格式模型会直接返回结构化的tool_calls。在模型支持 function calling 的前提下这种方式的稳定性明显优于 ReAct。8.3 什么时候需要自定义范式当你需要实现状态机、人工审核节点、条件分支、多 Agent 路由时直接写一个自定义 Agent 或直接使用 LangGraph 是更合理的选择。自定义 Agent 需要实现以下方法from langchain_core.agents import AgentAction, AgentFinish class MyAgent: def plan(self, inputs, intermediate_steps): # 根据中间步骤决定下一步 if self._is_done(intermediate_steps): return AgentFinish(return_values{output: 最终结果}) return AgentAction(toolmy_tool, tool_input参数, log日志)核心就是实现plan方法返回AgentAction或AgentFinish。9. 多 Agent 执行流程主从模式与 Subagent 调用热搜词中还有一个高频话题多 Agent 设计中的主从模式。简单说就是把 Subagent 当作一种特殊的 Tool 来调用。9.1 主从模式的核心思想主 Agent 负责接收用户请求、拆解任务、协调各子 Agent。子 Agent 负责具体子任务的执行。这种模式下子 Agent 的调用方式和普通工具完全一致。9.2 用 Tool 包装 Subagentfrom langchain.agents import create_react_agent, AgentExecutor from langchain.tools import tool # 子 Agent sub_agent create_react_agent(llmllm, tools[get_weather], promptprompt) sub_executor AgentExecutor(agentsub_agent, tools[get_weather]) tool def weather_sub_agent(city: str) - str: 调用天气子 Agent 查询指定城市的天气。 result sub_executor.invoke({input: f{city} 的天气怎么样}) return result[output] # 主 Agent 只看到 weather_sub_agent 这个工具 main_agent create_react_agent(llmllm, tools[weather_sub_agent], promptprompt) main_executor AgentExecutor(agentmain_agent, tools[weather_sub_agent], verboseTrue) result main_executor.invoke({input: 北京和上海的天气分别怎么样}) print(result[output])这个例子里主 Agent 不知道天气子 Agent 内部用了什么工具、什么提示词它只知道有一个weather_sub_agent可以调用。这种封装让多 Agent 系统的复杂度大大降低。你完全可以在一个 Agent 系统里挂很多子 Agent每个子 Agent 负责一个领域。9.3 多 Agent 协作的注意事项子 Agent 的返回结果要尽量结构化方便主 Agent 理解。每个子 Agent 要有独立的 prompt 和工具边界避免职责模糊。主 Agent 的迭代次数上限要比单 Agent 高因为每次子 Agent 调用都是一次完整的嵌套循环。子 Agent 的错误要能向上传递否则主 Agent 会误判子任务已经完成。10. Agent 执行流程的调试与常见问题排查Agent 执行流程的调试难点在于多轮循环中可能在某一步突然崩掉或者模型绕圈子始终不输出最终答案。下面给出一套排查思路。10.1 开启 verbose 模式agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, )verboseTrue会打印每一步的 Thought、Action、Observation这是最直接的排查手段。10.2 中间步骤的代码级观察如果你不想看控制台可以通过return_intermediate_steps拿到中间步骤result agent_executor.invoke( {input: 查一下北京天气}, return_intermediate_stepsTrue, ) for step in result[intermediate_steps]: action, observation step print(Action:, action.tool, action.tool_input) print(Observation:, observation)这样就能在自己的代码里检查每一步的输入输出。10.3 常见问题排查表问题现象可能原因排查方式解决方案Agent 一直重复调用同一个工具工具返回结果不够明确模型无法收敛观察重复的 Observation改进工具返回信息的完整性增加最大迭代次数限制Agent 输出不存在的工具名模型幻觉或工具描述不清晰查看 verbose 输出让工具 docstring 更明确更换更强的模型工具参数解析失败模型生成的参数不符合 schema检查错误日志简化参数提供更具体的 Field 描述达到 max_iterations 停止任务复杂度超过迭代上限查看最后几步观察结果加大迭代次数拆分子任务Agent 最终答案与工具结果不一致模型没有正确利用 Observation检查 Final Answer 前后的 Thought在 prompt 中强调必须基于 Observation 作答对话历史没有被 Agent 使用没有正确注入 chat_history检查输入字典使用带 MessagesPlaceholder 的 prompt模型 API 返回超时输入 prompt 太长或推理服务不稳定查看模型服务日志精简工具描述和中间步骤使用更快的模型10.4 避免死循环的兜底策略在生产环境中max_iterations必须设置。否则遇到一个复杂任务Agent 可能陷入无限循环浪费大量 token 和时间。建议的默认值是 5 到 10根据任务复杂度调整。如果业务允许还可以加一个超时控制import asyncio result asyncio.run( asyncio.wait_for( asyncio.to_thread(agent_executor.invoke, {input: 复杂任务}), timeout60, ) )10.5 LangSmith 链路追踪如果项目已经引入了 LangSmith直接在环境变量中配置export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour-langsmith-api-key export LANGCHAIN_PROJECTagent-debugLangSmith 会把每一轮 Agent 执行的完整链路、token 消耗、耗时记录在后台面板里适合在开发环境排查复杂问题。11. Agent 执行流程的最佳实践与工程化建议最后给几组工程化建议都是实际项目中比较通用的做法。11.1 工具设计决定了 Agent 的上限Agent 的执行流程再完善工具不好用也白搭。工具设计要遵循几个原则每个工具只做一件事职责单一。工具名要具备提示性比如calculate_order_total比do_thing好得多。参数名要清晰order_id好过id。给工具和参数加上详细描述描述会直接进入模型上下文。返回结果尽量用结构化的短文本避免返回一大段无关信息。11.2 成本与延迟控制多轮 Agent 执行会消耗大量 token尤其是带长工具描述和长历史的时候。控制成本的手段精简工具描述只保留必要信息。设置max_iterations防止无谓循环。优先使用支持 function calling 的模型比 ReAct 文本解析更省 token。在调用链路上做缓存相同问题直接命中缓存。11.3 安全与合规边界Agent 可以自主调用工具这意味着它可能调用了你不希望它调用的工具或者传入了不该传的参数。建议在生产环境做好这几件事工具层做权限控制不同角色能调用的工具集合不同。对工具入参做校验不能把用户输入直接拼进 SQL、命令或 API 请求。涉及用户隐私数据的工具必须在提示词中明确说明只允许在授权范围内调用。所有调用记录留痕方便事后审计。如果 Agent 能操作外部系统比如发送邮件、执行转账务必增加人工确认节点。11.4 先小规模验证再上生产第一次使用某套 Agent 流程时建议先在一个小数据量的测试环境里跑重点观察工具的调用准确率。参数的解析成功率。多轮执行的平均步数。最终答案与工具结果的一致性。基于这些指标判断是否需要增加调试、调整提示词或更换模型。Agent 执行流程不是一路“通”就完事的需要反复迭代的往往是提示词和工具设计而不是代码框架本身。12. 总结与下一步这次把 LangChain Agent 执行流程的核心链路梳理了一遍从 AgentExecutor 的循环机制到 ReAct 的 Thought/Action/Observation 流转再到 Plan-and-Execute 和多 Agent 主从模式最后是调试方法和工程化建议。如果你之前只会调用AgentExecutor不清楚中间发生了什么这篇文章可以直接帮你建立心智模型。下一步建议按这个顺序验证先跑通一个最简 ReAct Agent然后给它加记忆再换成create_tool_calling_agent对比效果最后尝试用 LangGraph 实现一个自定义执行流程。每一步跑通之后再往工程化方向补齐工具权限、日志、缓存和成本控制。Agent 执行流程本身不复杂复杂的永远是你对任务、工具、模型三者的边界管理。