LangChain 设计与实现

第14章 Agent 架构与执行循环

作者杨艺韬 · 9,867 字 · 发布于 2026-04-05 · 更新于 2026-09-25

开篇引言

在前面的章节中，我们剖析了 LangChain 的 Runnable 协议、回调系统、工具抽象和链式组合等基础设施。这些构件本身已经足够强大，但它们只是被动地执行预定义的流水线。真正让大语言模型从"输出生成器"进化为"自主决策者"的，是 Agent 架构。

Agent 的核心思想出奇地简单：给模型一组工具，让它自行决定每一步应该调用哪个工具、传入什么参数，然后根据工具返回的结果继续推理，直到它认为已经得到了最终答案。这个"思考-行动-观察"的循环，是当今几乎所有 AI Agent 系统的基石。

LangChain 的 Agent 实现分为两层：langchain_core.agents 定义了数据模型（AgentAction、AgentFinish、AgentStep），而 langchain_classic.agents.agent 提供了执行引擎（BaseSingleActionAgent、BaseMultiActionAgent、AgentExecutor）。这种分层确保了核心数据结构的稳定性，同时允许执行逻辑不断演进。

本章将深入 Agent 的内部机制，从数据模型到执行循环，从停止条件到错误处理，逐层拆解 LangChain Agent 的完整实现。

本章要点

AgentAction、AgentFinish、AgentStep 三个核心数据模型的设计与序列化机制
BaseSingleActionAgent 与 BaseMultiActionAgent 的抽象设计差异
RunnableAgent 如何将 Runnable 协议桥接到 Agent 接口
AgentExecutor 执行循环的完整流程：_call -> _take_next_step -> _iter_next_step
max_iterations、max_execution_time、early_stopping_method 三重停止保护
handle_parsing_errors 的多态错误处理策略

14.1 Agent 数据模型：行动、结束与步骤

Agent 系统的基石是三个定义在 libs/core/langchain_core/agents.py 中的数据模型。它们共同构成了 Agent 与外界交互的语言。

14.1.1 AgentAction -- 行动请求

AgentAction 表示 Agent 决定执行的一次工具调用。它的结构精简而完备：

# langchain_core/agents.py

class AgentAction(Serializable):
    """表示 Agent 请求执行的一个动作"""

    tool: str
    """要执行的工具名称"""

    tool_input: str | dict
    """传给工具的输入参数"""

    log: str
    """额外的日志信息，记录 LLM 的完整预测"""

    type: Literal["AgentAction"] = "AgentAction"

三个字段各有深意。tool 和 tool_input 共同决定了要执行什么操作。log 则保留了 LLM 的原始输出，这在两个场景下至关重要：一是审计追踪，让开发者能够回溯 LLM 的决策过程；二是后续迭代，将 LLM 之前的思考过程传回给它，帮助它做出更好的下一步决策。

AgentAction 继承自 Serializable，通过 is_lc_serializable 和 get_lc_namespace 支持序列化：

@classmethod
def is_lc_serializable(cls) -> bool:
    return True

@classmethod
def get_lc_namespace(cls) -> list[str]:
    return ["langchain", "schema", "agent"]

值得注意的是 messages 属性，它将 AgentAction 转换为消息格式，桥接了 Agent 系统与消息系统：

@property
def messages(self) -> Sequence[BaseMessage]:
    return _convert_agent_action_to_messages(self)

14.1.2 AgentActionMessageLog -- 聊天模型的行动记录

当底层 LLM 是聊天模型时，它返回的是消息对象而非纯文本。AgentActionMessageLog 扩展了 AgentAction，增加了 message_log 字段来保存原始消息序列：

class AgentActionMessageLog(AgentAction):
    message_log: Sequence[BaseMessage]
    """保存原始的 LLM 消息预测"""

    type: Literal["AgentActionMessageLog"] = "AgentActionMessageLog"

这个子类是连接"文本世界"和"消息世界"的桥梁。在消息转换函数中，如果发现是 AgentActionMessageLog 类型，就直接返回原始消息日志，避免了信息丢失：

# libs/core/langchain_core/agents.py:191
def _convert_agent_action_to_messages(
    agent_action: AgentAction,
) -> Sequence[BaseMessage]:
    if isinstance(agent_action, AgentActionMessageLog):
        return agent_action.message_log
    return [AIMessage(content=agent_action.log)]

对应地，观察结果的转换也会区别对待两种类型：

# libs/core/langchain_core/agents.py:209
def _convert_agent_observation_to_messages(
    agent_action: AgentAction, observation: Any
) -> Sequence[BaseMessage]:
    if isinstance(agent_action, AgentActionMessageLog):
        return [_create_function_message(agent_action, observation)]
    # 回退到 HumanMessage；非字符串先试 json.dumps，不可序列化再退到 str()
    content = observation
    if not isinstance(observation, str):
        try:
            content = json.dumps(observation, ensure_ascii=False)
        except Exception:
            content = str(observation)
    return [HumanMessage(content=content)]

14.1.3 AgentFinish -- 终止信号

当 Agent 认为任务已经完成时，它返回一个 AgentFinish 对象：

class AgentFinish(Serializable):
    return_values: dict
    """返回值字典"""

    log: str
    """额外的日志信息"""

    type: Literal["AgentFinish"] = "AgentFinish"

return_values 是一个字典，通常包含一个 "output" 键，对应最终答案。使用字典而非简单字符串，是为了支持多返回值的场景。messages 属性则简单地返回一个 AIMessage：

@property
def messages(self) -> Sequence[BaseMessage]:
    return [AIMessage(content=self.log)]

14.1.4 AgentStep -- 执行结果

AgentStep 将行动与其观察结果打包在一起：

class AgentStep(Serializable):
    action: AgentAction
    """执行的动作"""

    observation: Any
    """动作的执行结果"""

    @property
    def messages(self) -> Sequence[BaseMessage]:
        return _convert_agent_observation_to_messages(self.action, self.observation)

classDiagram
    class Serializable {
        <<abstract>>
        +is_lc_serializable() bool
        +get_lc_namespace() list
        +to_json() dict
    }
    class AgentAction {
        +tool: str
        +tool_input: str | dict
        +log: str
        +type: "AgentAction"
        +messages: Sequence~BaseMessage~
    }
    class AgentActionMessageLog {
        +message_log: Sequence~BaseMessage~
        +type: "AgentActionMessageLog"
    }
    class AgentFinish {
        +return_values: dict
        +log: str
        +type: "AgentFinish"
        +messages: Sequence~BaseMessage~
    }
    class AgentStep {
        +action: AgentAction
        +observation: Any
        +messages: Sequence~BaseMessage~
    }
    Serializable <|-- AgentAction
    Serializable <|-- AgentFinish
    Serializable <|-- AgentStep
    AgentAction <|-- AgentActionMessageLog
    AgentStep --> AgentAction : 包含

这四个数据模型共同定义了 Agent 的"通信协议"。AgentAction 是 Agent 发出的"请求"，AgentStep 是执行后的"回执"，AgentFinish 是最终的"结案报告"。它们都继承自 Serializable，但只有 AgentAction（连带子类 AgentActionMessageLog）和 AgentFinish 覆写了 is_lc_serializable() 返回 True；AgentStep 沿用基类默认的 False，dumpd() 对它只产出 {'type': 'not_implemented', ...} 占位（实跑）。

从设计原则上看，这四个类体现了"关注点分离"的思想。行动的请求（AgentAction）与行动的结果（AgentStep 中的 observation）被刻意分开，而非用一个大而全的类来表示。这种分离使得系统的每个部分只关心自己需要的信息：Agent 的 plan 方法只需要返回 AgentAction 或 AgentFinish，不需要关心工具如何执行；AgentExecutor 只需要将 AgentAction 传给对应的工具，不需要关心 Agent 的推理过程。

值得注意的是 tool_input 字段使用了联合类型 str | dict。这是为了同时支持两种工具调用风格：旧式的纯文本输入（如 ReAct Agent 从文本中提取的 Action Input）和新式的结构化输入（如 Tool Calling Agent 从 tool_calls 中提取的参数字典）。这种类型灵活性确保了数据模型可以同时服务于多种 Agent 实现。

type 字段使用 Literal 类型标记，这是判别联合（Discriminated Union）的写法：数据里带着 type，Pydantic 之类的消费方就能据此分辨一条记录是哪个类，源码注释也只说它"用于序列化"。不过 LangChain 自己的 load() 并不看这个字段，它按序列化结果里的 id（命名空间 + 类名，映射表在 libs/core/langchain_core/load/mapping.py）决定实例化哪个类。

14.2 Agent 基类：单动作与多动作

LangChain 定义了两个 Agent 基类，分别对应两种执行模式。

14.2.1 BaseSingleActionAgent

BaseSingleActionAgent 是最基本的 Agent 抽象。它每次决策只返回一个动作：

# langchain_classic/agents/agent.py

class BaseSingleActionAgent(BaseModel):
    @property
    def return_values(self) -> list[str]:
        return ["output"]

    def get_allowed_tools(self) -> list[str] | None:
        return None

    @abstractmethod
    def plan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> AgentAction | AgentFinish:
        """给定输入和历史步骤，决定下一步做什么"""

    @abstractmethod
    async def aplan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> AgentAction | AgentFinish:
        """plan 的异步版本"""

    @property
    @abstractmethod
    def input_keys(self) -> list[str]:
        """返回输入键列表"""

plan 方法是 Agent 的核心接口。它接收两个关键输入：

intermediate_steps: 之前执行的所有步骤及其结果，以 list[tuple[AgentAction, str]] 的形式传入
**kwargs: 用户输入参数，如 input、chat_history 等

返回值是一个联合类型：要么是 AgentAction（继续执行），要么是 AgentFinish（结束执行）。

return_stopped_response 方法处理 Agent 被强制停止的情况：

# libs/langchain/langchain_classic/agents/agent.py:110（BaseSingleActionAgent）
# :280 的 BaseMultiActionAgent 版签名相同，只是 force 返回的文案是
# "Agent stopped due to max iterations."
def return_stopped_response(
    self,
    early_stopping_method: str,
    intermediate_steps: list[tuple[AgentAction, str]],
    **_: Any,
) -> AgentFinish:
    if early_stopping_method == "force":
        return AgentFinish(
            {"output": "Agent stopped due to iteration limit or time limit."},
            "",
        )
    raise ValueError(
        f"Got unsupported early_stopping_method `{early_stopping_method}`"
    )

14.2.2 BaseMultiActionAgent

BaseMultiActionAgent 允许一次返回多个动作，实现并行工具调用：

# libs/langchain/langchain_classic/agents/agent.py:221
class BaseMultiActionAgent(BaseModel):
    @abstractmethod
    def plan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> list[AgentAction] | AgentFinish:
        """返回一个或多个动作"""

    @abstractmethod
    async def aplan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> list[AgentAction] | AgentFinish:
        """异步返回一个或多个动作"""

注意返回类型的差异：list[AgentAction] | AgentFinish。当 LLM 一次性返回多个工具调用时（例如 OpenAI 的 parallel tool calling），Agent 可以将它们全部返回，交给 AgentExecutor 执行：同步路径逐个执行，异步路径才用 asyncio.gather 并发（见 §14.6）。

classDiagram
    class BaseSingleActionAgent {
        <<abstract>>
        +return_values: list~str~
        +plan(intermediate_steps, callbacks, **kwargs) AgentAction|AgentFinish
        +aplan(intermediate_steps, callbacks, **kwargs) AgentAction|AgentFinish
        +return_stopped_response(method, steps) AgentFinish
        +get_allowed_tools() list|None
        +input_keys: list~str~
    }
    class BaseMultiActionAgent {
        <<abstract>>
        +return_values: list~str~
        +plan(steps, cb, **kw) list~AgentAction~|AgentFinish
        +aplan(steps, cb, **kw) list~AgentAction~|AgentFinish
        +return_stopped_response(method, steps) AgentFinish
    }
    class RunnableAgent {
        +runnable: Runnable
        +stream_runnable: bool
        +plan() AgentAction|AgentFinish
    }
    class RunnableMultiActionAgent {
        +runnable: Runnable
        +stream_runnable: bool
        +plan() list~AgentAction~|AgentFinish
    }
    class AgentOutputParser {
        <<abstract>>
        +parse(text) AgentAction|AgentFinish
    }
    class MultiActionAgentOutputParser {
        <<abstract>>
        +parse(text) list~AgentAction~|AgentFinish
    }
    BaseSingleActionAgent <|-- RunnableAgent
    BaseMultiActionAgent <|-- RunnableMultiActionAgent
    BaseSingleActionAgent <|-- Agent
    BaseSingleActionAgent <|-- LLMSingleActionAgent

14.2.3 RunnableAgent -- 桥接 Runnable 协议

RunnableAgent 是连接新旧两套架构的关键适配器。它将一个 Runnable[dict, AgentAction | AgentFinish] 适配为 BaseSingleActionAgent 接口：

# libs/langchain/langchain_classic/agents/agent.py:389
class RunnableAgent(BaseSingleActionAgent):
    runnable: Runnable[dict, AgentAction | AgentFinish]
    input_keys_arg: list[str] = []
    return_keys_arg: list[str] = []
    stream_runnable: bool = True

    def plan(
        self,
        intermediate_steps: list[tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> AgentAction | AgentFinish:
        inputs = {**kwargs, "intermediate_steps": intermediate_steps}
        final_output: Any = None
        if self.stream_runnable:
            # 使用流式调用，使中间 token 可以通过回调系统传出
            for chunk in self.runnable.stream(
                inputs, config={"callbacks": callbacks}
            ):
                if final_output is None:
                    final_output = chunk
                else:
                    final_output += chunk
        else:
            final_output = self.runnable.invoke(
                inputs, config={"callbacks": callbacks}
            )
        return final_output

stream_runnable=True 这个默认值很有意义。即使最终需要完整输出才能解析出 AgentAction，使用流式调用也能让中间的 token 通过回调系统逐个传出，支持 stream_log 等高级功能。这是一个"看似多余实则精妙"的设计决策：plan 方法的返回值始终是完整的 AgentAction，但调用过程中的流式 token 已经通过回调链传播出去了。

RunnableMultiActionAgent 的结构与之完全相同，仅返回类型不同（list[AgentAction] | AgentFinish）。

14.3 AgentExecutor：执行引擎

AgentExecutor 是整个 Agent 系统的执行引擎。它继承自 Chain，将 Agent 的决策与工具的执行编织成完整的循环。

14.3.1 配置参数

# libs/langchain/langchain_classic/agents/agent.py:1012
class AgentExecutor(Chain):
    agent: BaseSingleActionAgent | BaseMultiActionAgent | Runnable
    """要运行的 Agent"""
    tools: Sequence[BaseTool]
    """Agent 可以调用的工具集"""
    return_intermediate_steps: bool = False
    """是否返回中间步骤"""
    max_iterations: int | None = 15
    """最大迭代次数，None 表示无限制"""
    max_execution_time: float | None = None
    """最大执行时间（秒）"""
    early_stopping_method: str = "force"
    """早停策略：'force' 或 'generate'"""
    handle_parsing_errors: bool | str | Callable = False
    """输出解析错误的处理方式"""
    trim_intermediate_steps: int | Callable = -1
    """中间步骤的裁剪策略，-1 表示不裁剪"""

这些参数构成了 AgentExecutor 的控制面板。max_iterations=15 是源码里写死的默认值（libs/langchain/langchain_classic/agents/agent.py:1023），它的作用是给"LLM 陷进无效循环"设一个硬上限——具体多少步够用完全取决于任务，源码注释只说了一句"设成 None 可能导致死循环"。max_execution_time 提供了时间维度的保护，特别适合对延迟敏感的在线服务场景。handle_parsing_errors 的多态类型设计提供了从简单到高级的多层次错误处理能力，布尔值适合快速开发，自定义函数适合精细化的错误恢复策略。trim_intermediate_steps 解决了长时间运行 Agent 的上下文膨胀问题：正整数提供简单的滑动窗口（只留最近 N 步），函数值提供完全定制的裁剪逻辑，而默认值 -1 的含义是不裁剪（见 §14.5.3 的 _prepare_intermediate_steps）。return_intermediate_steps 虽然默认关闭以减少输出体积，但在调试阶段几乎总是应该开启，因为 Agent 的中间步骤是理解其行为的最重要线索。

14.3.2 Runnable 自动转换

AgentExecutor 的一个精巧设计是，它接受原始的 Runnable 作为 agent 参数，通过 model_validator 自动将其包装为合适的 Agent 类型：

# libs/langchain/langchain_classic/agents/agent.py:1110
@model_validator(mode="before")
@classmethod
def validate_runnable_agent(cls, values: dict) -> Any:
    agent = values.get("agent")
    if agent and isinstance(agent, Runnable):
        try:
            output_type = agent.OutputType
        except TypeError:
            multi_action = False
        # （源码另有 except Exception 分支：记日志后同样按单动作处理，此处省略）
        else:
            multi_action = output_type == list[AgentAction] | AgentFinish

        stream_runnable = values.pop("stream_runnable", True)
        if multi_action:
            values["agent"] = RunnableMultiActionAgent(
                runnable=agent, stream_runnable=stream_runnable,
            )
        else:
            values["agent"] = RunnableAgent(
                runnable=agent, stream_runnable=stream_runnable,
            )
    return values

这段代码通过检查 Runnable 的 OutputType 来判断应该使用单动作还是多动作适配器。例如以 ToolsAgentOutputParser（一个 MultiActionAgentOutputParser）结尾的管道，OutputType 恰为 list[AgentAction] | AgentFinish，会被包成 RunnableMultiActionAgent；create_react_agent 产出的管道则落到 RunnableAgent（实跑）。这种自动检测的设计使得用户可以直接传入 LCEL 链作为 agent，无需手动包装。

另一个验证器确保工具集与 Agent 的允许列表一致：

# libs/langchain/langchain_classic/agents/agent.py:1084
@model_validator(mode="after")
def validate_tools(self) -> Self:
    agent = self.agent
    tools = self.tools
    allowed_tools = agent.get_allowed_tools()
    if allowed_tools is not None and set(allowed_tools) != {
        tool.name for tool in tools
    }:
        raise ValueError(
            f"Allowed tools ({allowed_tools}) different than "
            f"provided tools ({[tool.name for tool in tools]})"
        )
    return self

14.3.3 执行循环的核心：_call 方法

_call 方法是整个执行引擎的心脏：

# libs/langchain/langchain_classic/agents/agent.py:1570
def _call(
    self,
    inputs: dict[str, str],
    run_manager: CallbackManagerForChainRun | None = None,
) -> dict[str, Any]:
    # 1. 构建工具映射表
    name_to_tool_map = {tool.name: tool for tool in self.tools}
    color_mapping = get_color_mapping(
        [tool.name for tool in self.tools],
        excluded_colors=["green", "red"],
    )

    # 2. 初始化状态
    intermediate_steps: list[tuple[AgentAction, str]] = []
    iterations = 0
    time_elapsed = 0.0
    start_time = time.time()

    # 3. 主循环
    while self._should_continue(iterations, time_elapsed):
        next_step_output = self._take_next_step(
            name_to_tool_map, color_mapping, inputs,
            intermediate_steps, run_manager=run_manager,
        )

        # 3a. Agent 决定结束
        if isinstance(next_step_output, AgentFinish):
            return self._return(
                next_step_output, intermediate_steps, run_manager,
            )

        # 3b. 累积中间步骤
        intermediate_steps.extend(next_step_output)

        # 3c. 检查 return_direct 工具
        if len(next_step_output) == 1:
            tool_return = self._get_tool_return(next_step_output[0])
            if tool_return is not None:
                return self._return(
                    tool_return, intermediate_steps, run_manager,
                )

        iterations += 1
        time_elapsed = time.time() - start_time

    # 4. 超出限制，触发早停
    output = self._action_agent.return_stopped_response(
        self.early_stopping_method, intermediate_steps, **inputs,
    )
    return self._return(output, intermediate_steps, run_manager)

flowchart TD
    A[用户输入] --> B[构建工具映射表]
    B --> C["初始化状态<br/>iterations=0<br/>intermediate_steps=[ ]"]
    C --> D{_should_continue?<br/>检查迭代次数和时间}
    D -->|是| E["_take_next_step<br/>调用 agent.plan() + 执行工具"]
    E --> F{返回 AgentFinish?}
    F -->|是| G["_return: 返回最终结果"]
    F -->|否| H[添加到 intermediate_steps]
    H --> I{工具标记了<br/>return_direct?}
    I -->|是| G
    I -->|否| J["iterations += 1<br/>更新 time_elapsed"]
    J --> D
    D -->|否: 超出限制| L["return_stopped_response<br/>触发早停策略"]
    L --> G

14.3.4 单步执行：_iter_next_step

_iter_next_step 是循环中每一步的详细实现，它是一个生成器函数：

# libs/langchain/langchain_classic/agents/agent.py:1301
def _iter_next_step(
    self, name_to_tool_map, color_mapping, inputs,
    intermediate_steps, run_manager=None,
) -> Iterator[AgentFinish | AgentAction | AgentStep]:
    try:
        # 准备中间步骤（可能被裁剪）
        intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)
        # 调用 Agent 的 plan 方法获取决策
        output = self._action_agent.plan(
            intermediate_steps,
            callbacks=run_manager.get_child() if run_manager else None,
            **inputs,
        )
    except OutputParserException as e:
        # 处理解析错误（详见 14.5 节）
        ...
        yield AgentStep(action=output, observation=observation)
        return

    # 如果是 AgentFinish，直接返回
    if isinstance(output, AgentFinish):
        yield output
        return

    # 统一为列表形式
    actions = [output] if isinstance(output, AgentAction) else output

    # 第一轮：先把所有动作 yield 出去（供 stream 即时产出 actions 片段）
    for agent_action in actions:
        yield agent_action

    # 第二轮：逐个执行工具（on_agent_action 回调在这里触发）
    for agent_action in actions:
        yield self._perform_agent_action(
            name_to_tool_map, color_mapping, agent_action, run_manager,
        )

注意 actions 被遍历了两次：第一次只 yield 所有 AgentAction，第二次才真正执行工具。先 yield 的动作供 stream() 背后的 AgentExecutorIterator 在工具开跑前就产出 actions 片段（§14.8）；_take_next_step 则把整个生成器 list() 收齐后交给 _consume_next_step，只留下 AgentStep 转成的 (action, observation) 元组。on_agent_action 回调并不在 yield 时触发，而是在 _perform_agent_action 开头、每个工具执行前各触发一次——多动作时实跑顺序是"动作 a → 工具 a → 动作 b → 工具 b"。

14.3.5 工具执行：_perform_agent_action

# libs/langchain/langchain_classic/agents/agent.py:1380
def _perform_agent_action(
    self, name_to_tool_map, color_mapping,
    agent_action, run_manager=None,
) -> AgentStep:
    if run_manager:
        run_manager.on_agent_action(agent_action, color="green")

    if agent_action.tool in name_to_tool_map:
        tool = name_to_tool_map[agent_action.tool]
        return_direct = tool.return_direct
        color = color_mapping[agent_action.tool]
        tool_run_kwargs = self._action_agent.tool_run_logging_kwargs()
        if return_direct:
            tool_run_kwargs["llm_prefix"] = ""

        observation = tool.run(
            agent_action.tool_input,
            verbose=self.verbose,
            color=color,
            callbacks=run_manager.get_child() if run_manager else None,
            **tool_run_kwargs,
        )
    else:
        # 工具不存在时，使用 InvalidTool 返回友好错误信息
        observation = InvalidTool().run(
            {
                "requested_tool_name": agent_action.tool,
                "available_tool_names": list(name_to_tool_map.keys()),
            },
            ...
        )

    return AgentStep(action=agent_action, observation=observation)

当 Agent 试图调用一个不存在的工具时，不是直接报错，而是通过 InvalidTool 将错误信息作为观察结果返回给 Agent。这使得 Agent 有机会在下一轮推理中自我修正工具选择。

这个容错设计体现了一个重要的 Agent 设计哲学：错误是信息，而非终止条件。在传统编程中，调用一个不存在的函数会立即抛出异常。但在 Agent 系统中，"工具不存在"这个信息如果能被传回给 LLM，LLM 很可能会自我修正 -- 选择正确的工具名，或者用其他方式完成任务。这种"错误即观察"的设计给了 Agent 自我修复的机会。实跑时 observation 是 nope is not a valid tool, try one of [search]. 这样一句提示。

_perform_agent_action 中还有一个细节值得关注：return_direct 标志。当一个工具被标记为 return_direct=True 时，这个工具的输出会直接作为 Agent 的最终答案返回，跳过后续的 LLM 推理。注意 _call 只在本轮恰好一个步骤（len(next_step_output) == 1）时才检查它：多动作 Agent 一轮调了两个 return_direct 工具，循环照常继续（实跑）。这在某些场景下非常有用，比如一个"数据库查询"工具 -- 用户问的就是数据库中的数据，工具返回的就是最终答案，不需要 LLM 再做一轮"润色"。这种优化不仅节省了一次 LLM 调用，还避免了 LLM 可能对工具结果进行不必要的修改或"幻觉"。

14.4 停止条件：三重保护

Agent 执行循环的停止由三个条件共同控制，形成一套完整的安全防护网。

14.4.1 _should_continue 判断

def _should_continue(self, iterations: int, time_elapsed: float) -> bool:
    if self.max_iterations is not None and iterations >= self.max_iterations:
        return False
    return (
        self.max_execution_time is None
        or time_elapsed < self.max_execution_time
    )

这个简洁的方法实现了双重保护：

迭代次数限制：默认 15 次，防止 LLM 陷入无效循环
墙钟时间限制：可选，防止单次执行消耗过多时间

两个条件是"或"关系，任一触发即停止循环。

14.4.2 early_stopping_method

当循环被停止条件终止时，early_stopping_method 决定如何生成最终输出。基础版只支持 "force"，Agent 子类扩展支持 "generate"：

# libs/langchain/langchain_classic/agents/agent.py:920
# Agent 子类的 return_stopped_response
def return_stopped_response(
    self, early_stopping_method, intermediate_steps, **kwargs
):
    if early_stopping_method == "force":
        return AgentFinish(
            {"output": "Agent stopped due to iteration limit or time limit."},
            "",
        )
    if early_stopping_method == "generate":
        # 拼接所有已有步骤
        thoughts = ""
        for action, observation in intermediate_steps:
            thoughts += action.log
            thoughts += f"\n{self.observation_prefix}{observation}\n{self.llm_prefix}"
        thoughts += (
            "\n\nI now need to return a final answer "
            "based on the previous steps:"
        )
        new_inputs = {"agent_scratchpad": thoughts, "stop": self._stop}
        full_inputs = {**kwargs, **new_inputs}
        # 最后一次 LLM 调用
        full_output = self.llm_chain.predict(**full_inputs)
        parsed_output = self.output_parser.parse(full_output)
        if isinstance(parsed_output, AgentFinish):
            return parsed_output
        return AgentFinish({"output": full_output}, full_output)
    # 其他取值：raise ValueError（此处省略）

"generate" 策略更加智能：它将所有已有步骤拼接起来，追加一句"现在必须给出最终答案"的指令，然后做最后一次 LLM 调用。这样即使 Agent 被强制停止，也能给出一个有意义的回答，而不是生硬的错误消息。

但 "generate" 只有已弃用的 Agent 子类（依赖它的 llm_chain）才认。create_react_agent、create_tool_calling_agent 等返回的 Runnable 会被包成 RunnableAgent / RunnableMultiActionAgent，它们沿用基类实现，只认 "force"：设 early_stopping_method="generate" 并触发迭代上限，实跑抛 ValueError: Got unsupported early_stopping_method `generate`。

14.4.3 异步超时

异步版本使用 asyncio_timeout 进行更精确的时间控制：

async def _acall(self, inputs, run_manager=None):
    try:
        async with asyncio_timeout(self.max_execution_time):
            while self._should_continue(iterations, time_elapsed):
                next_step_output = await self._atake_next_step(...)
                # ... 正常循环逻辑
    except (TimeoutError, asyncio.TimeoutError):
        # 超时时也触发早停
        output = self._action_agent.return_stopped_response(
            self.early_stopping_method, intermediate_steps, **inputs,
        )
        return await self._areturn(output, intermediate_steps, run_manager)

flowchart LR
    subgraph "停止条件"
        A["max_iterations<br/>默认 15"] --> D{"_should_continue()"}
        B["max_execution_time<br/>可选"] --> D
        C["asyncio_timeout<br/>异步专用"] --> E["TimeoutError"]
    end
    subgraph "早停策略"
        D -->|超出限制| F{early_stopping_method}
        E --> F
        F -->|"force"| G["返回固定消息"]
        F -->|"generate"| H["再调 LLM 一次<br/>生成最终答案"]
    end
    subgraph "正常结束"
        I[AgentFinish] --> J["_return()"]
        K["return_direct 工具"] --> J
    end

14.5 错误处理与重试

Agent 执行过程中最常见的错误是 LLM 输出无法被解析为有效的 AgentAction。handle_parsing_errors 参数提供了灵活的处理策略。

14.5.1 多态错误处理

handle_parsing_errors: bool | str | Callable[[OutputParserException], str] = False

这个字段的三种类型对应三种策略：

# libs/langchain/langchain_classic/agents/agent.py:1322（异步版在 :1461）
except OutputParserException as e:
    if isinstance(self.handle_parsing_errors, bool):
        raise_error = not self.handle_parsing_errors
    else:
        raise_error = False

    if raise_error:
        raise ValueError(
            "An output parsing error occurred. "
            "In order to pass this error back to the agent and have it try "
            "again, pass `handle_parsing_errors=True` to the AgentExecutor. "
            f"This is the error: {e!s}"
        ) from e

    text = str(e)
    if isinstance(self.handle_parsing_errors, bool):
        if e.send_to_llm:
            observation = str(e.observation)
            text = str(e.llm_output)
        else:
            observation = "Invalid or incomplete response"
    elif isinstance(self.handle_parsing_errors, str):
        observation = self.handle_parsing_errors
    elif callable(self.handle_parsing_errors):
        observation = self.handle_parsing_errors(e)
    # （其余类型抛 ValueError，此处省略）

    output = AgentAction("_Exception", observation, text)
    # 随后用 ExceptionTool().run(output.tool_input, ...) 得到 observation，
    # yield AgentStep(action=output, observation=observation)

参数值	行为
`False`（默认）	直接抛出异常，由调用方处理
`True`	异常带 `send_to_llm=True` 时，用异常的 `observation` 作观察、`llm_output` 作 log；否则观察固定为 `"Invalid or incomplete response"`，log 为 `str(e)`
`"自定义字符串"`	将指定字符串作为观察结果反馈
`callable`	调用自定义函数处理异常，返回值作为观察

True 这一行最容易误解：它并不把异常信息原样当观察。第 7 章 §7.10 讲过 OutputParserException 的 send_to_llm 字段，在 pin 树里读它的只有 AgentExecutor 这两处 except 分支（libs/langchain/langchain_classic/agents/agent.py:1337、:1476）。以 create_react_agent 默认的 ReActSingleInputOutputParser 为例：模型输出 I dunno（缺 Action:）时，解析器抛出的异常带 send_to_llm=True，实跑得到的中间步骤是 _Exception 动作，观察为 Invalid Format: Missing 'Action:' after 'Thought:'，log 为原始输出 I dunno；而输出里同时出现可解析的 Action 和 Final Answer: 时，异常不带 send_to_llm，观察只剩 Invalid or incomplete response，完整错误文本落在 log 里——ReAct 的 scratchpad 会把 log 和观察一起拼回提示，模型仍看得到它。

14.5.2 ExceptionTool -- 错误的工具化

当解析错误被处理时，LangChain 创建一个指向 _Exception 工具的 AgentAction：

class ExceptionTool(BaseTool):
    name: str = "_Exception"
    description: str = "Exception tool"

    def _run(self, query: str, run_manager=None) -> str:
        return query  # 简单地将输入原样返回

通过这种"工具化"的错误处理，解析错误被统一为正常的 Agent 步骤。Agent 在下一轮推理中会看到这个"工具调用"的结果（即错误信息），从而有机会修正输出格式。这个设计极其巧妙：它利用已有的工具调用机制来实现错误恢复，无需引入新的概念。

14.5.3 trim_intermediate_steps -- 上下文管理

随着执行步骤的累积，传给 LLM 的上下文可能会超出窗口限制。trim_intermediate_steps 提供了两种裁剪策略：

# libs/langchain/langchain_classic/agents/agent.py:1718
def _prepare_intermediate_steps(
    self, intermediate_steps
) -> list[tuple[AgentAction, str]]:
    if (
        isinstance(self.trim_intermediate_steps, int)
        and self.trim_intermediate_steps > 0
    ):
        return intermediate_steps[-self.trim_intermediate_steps:]
    if callable(self.trim_intermediate_steps):
        return self.trim_intermediate_steps(intermediate_steps)
    return intermediate_steps

整数值表示只保留最近 N 步；callable 则允许自定义裁剪逻辑，例如按 token 数量裁剪或按重要性排序。

14.6 异步执行与并发工具调用

异步版本的执行循环结构与同步版本基本相同，但有一个关键优化。在 _aiter_next_step 中，多个工具调用使用 asyncio.gather 并发执行：

async def _aiter_next_step(self, ...):
    # ... 获取 actions ...

    # 先 yield 所有动作
    for agent_action in actions:
        yield agent_action

    # 并发执行所有工具
    result = await asyncio.gather(
        *[
            self._aperform_agent_action(
                name_to_tool_map, color_mapping,
                agent_action, run_manager,
            )
            for agent_action in actions
        ],
    )

    for chunk in result:
        yield chunk

这意味着当 MultiActionAgent 一次返回多个工具调用时（如 OpenAI 的 parallel function calling），走 ainvoke / astream 的异步路径会并发执行它们；同步 invoke 的 _iter_next_step 仍是逐个执行。实跑一个一轮返回 3 个动作、每个工具耗时 1 秒的多动作 Agent：ainvoke 约 1.0 秒，invoke 约 3.0 秒。

sequenceDiagram
    participant U as 用户
    participant AE as AgentExecutor
    participant Ag as Agent.plan()
    participant T1 as 工具 A
    participant T2 as 工具 B
    U->>AE: invoke({"input": "..."})
    loop 异步执行循环 (max_iterations=15)
        AE->>Ag: plan(intermediate_steps, **inputs)
        Ag-->>AE: [AgentAction_A, AgentAction_B]
        par 并发执行 asyncio.gather
            AE->>T1: arun(action_A.tool_input)
            AE->>T2: arun(action_B.tool_input)
            T1-->>AE: observation_A
            T2-->>AE: observation_B
        end
        AE->>AE: intermediate_steps.extend(...)
    end
    AE->>Ag: plan(intermediate_steps, **inputs)
    Ag-->>AE: AgentFinish(return_values)
    AE-->>U: {"output": "最终答案"}

14.7 设计决策分析

生成器模式的 _iter_next_step

_iter_next_step 使用 Iterator[AgentFinish | AgentAction | AgentStep] 而非简单地返回一个元组。这种生成器设计使得每个中间产物都可以被即时处理：AgentAction 在 yield 时触发回调，AgentStep 在 yield 时可以被流式消费。这为 AgentExecutor.stream() 方法提供了基础，使得用户可以实时看到 Agent 的每一步决策。

Agent 与 AgentExecutor 的职责分离

Agent 只负责"决策"（plan），AgentExecutor 负责"执行"（工具调用、错误处理、循环控制）。这种分离使得同一个 Agent 可以被不同的执行器驱动（例如将来可能有分布式的 AgentExecutor），也使得执行器可以复用不同的 Agent 实现。

Runnable 接口的统一

通过 validate_runnable_agent 这个 Pydantic 验证器，AgentExecutor 无缝支持直接传入 Runnable 作为 agent。任何 LCEL 管道只要输出类型正确，就可以直接作为 Agent 使用：

# 示意：... 处是把 intermediate_steps 格式化为消息的函数，完整写法见第 15 章 §15.2
agent = (
    RunnablePassthrough.assign(agent_scratchpad=...)
    | prompt
    | llm_with_tools
    | ToolsAgentOutputParser()
)
agent_executor = AgentExecutor(agent=agent, tools=tools)

这种设计大大减轻了学习专有 Agent API 的负担：组装 Agent 用的就是 Runnable 这一套核心抽象，只需额外约定输出是 AgentAction / AgentFinish。

中间步骤的数据格式

intermediate_steps 使用 list[tuple[AgentAction, str]] 而非 list[AgentStep]。这个看似奇怪的选择有其历史原因：早期版本中 AgentStep 尚不存在，tuple 格式已经成为 API 的一部分。元组也有解构方便的好处（for action, observation in intermediate_steps），各 scratchpad 格式化函数都这样写。

14.8 AgentExecutor 的流式输出机制

AgentExecutor 继承自 Chain，而 Chain 实现了 Runnable 接口，因此它天然支持 stream 方法。但 Agent 的流式输出比普通 Chain 复杂得多，因为一次 Agent 执行涉及多轮循环，每轮都会产生不同类型的输出。

AgentExecutor 的 stream 方法覆盖了基类实现（libs/langchain/langchain_classic/agents/agent.py:1732），内部委托给 AgentExecutorIterator，返回一个 Iterator[AddableDict]。在每一步执行中，它会逐步产出包含不同键的字典片段：当 Agent 决定调用工具时，产出 AddableDict(actions=[...], messages=...)；当工具执行完毕后，产出 AddableDict(steps=[...], messages=...)（两者都在 libs/langchain/langchain_classic/agents/agent_iterator.py:212-214）；当 Agent 给出最终答案时，产出带 output 键的字典。

这种设计使得前端应用可以根据字典的键类型做不同的展示处理。收到 actions 时可以显示"正在调用搜索工具..."的状态提示；收到 steps 时可以展示工具返回的结果；收到 output 时可以渲染最终答案。整个过程对用户来说是流畅连贯的，而不是漫长的等待后突然出现完整结果。

对于 astream_events 这个更高级的流式接口，AgentExecutor 能够产出更细粒度的事件。每次模型生成一个 token 时，就会触发一个流式事件（聊天模型是 on_chat_model_stream，文本补全模型是 on_llm_stream）；每次工具开始和结束执行时，也会触发对应的事件。这使得实时追踪 Agent 行为成为可能，前端可以逐字显示 Agent 的"思考过程"。

流式输出的实现依赖于前面讨论的 _iter_next_step 生成器设计。因为 _iter_next_step 逐个 yield AgentAction 和 AgentStep，stream 方法可以即时将它们转换为流式输出片段，无需等待整个步骤完成。用假模型实跑一次"调一次工具再给答案"的 ReAct Agent，stream() 依次产出三个片段，键分别是 actions+messages、steps+messages、output+messages。这个看似简单的生成器模式，实际上是整个流式体验的基础支撑。

14.9 中间步骤的数据表示选择

Agent 系统中一个看似简单但影响深远的设计选择是中间步骤的数据表示方式。在源码中，intermediate_steps 被定义为 list[tuple[AgentAction, str]] 而非 list[AgentStep]。这个选择可以从两个角度理解。

首先是历史兼容性。在 LangChain 的早期版本中，AgentStep 类尚不存在，中间步骤一直使用元组格式；AgentStep 直到 2023-11 才随 Agent 的 stream() / astream() 支持（#12783）加入。当 AgentStep 被引入后，如果将 intermediate_steps 的类型改为 list[AgentStep]，所有现有的 Agent 实现和用户代码都需要修改。保持元组格式意味着零迁移成本。

其次是解构的便利性。在 Python 中，元组可以通过解包直接提取组件（action, observation = step），而对象需要属性访问（step.action, step.observation）。在 Agent 的格式化函数中，这种解包操作被频繁使用，元组格式使得代码更加简洁。

不过，这个类型标注和运行时并不一致。元组第二项标成 str，但 _consume_next_step 只是把 AgentStep 拆成 (a.action, a.observation)，并不做任何转换：工具返回一个字典，return_intermediate_steps=True 拿回的观察就还是那个字典（实跑）。真正把观察变成字符串的是各 Agent 的 scratchpad 格式化函数，例如 ReAct 用的 format_log_to_str 直接做 f-string 拼接。所以读者自己写 trim_intermediate_steps 回调或处理返回的中间步骤时，不能假定观察一定是 str。

这个设计决策体现了一个普遍的工程权衡：简洁性和兼容性通常比表达力更重要，特别是在接口层面。内部实现可以使用更丰富的类型（如 AgentStep），但对外暴露的接口应该尽可能简单和稳定——代价是类型标注只能当作约定，而不是保证。

14.10 Agent 系统的历史演进

理解 Agent 架构的设计决策，需要了解它的历史背景。LangChain 的 Agent 系统经历了三个主要阶段。

第一阶段：基于 Chain 的 Agent

最早的 Agent 实现完全基于 LLMChain。Agent 基类（与 LLMSingleActionAgent、AgentType、initialize_agent 一样，自 0.1.0 起标 @deprecated）内部持有一个 LLMChain，通过 _construct_scratchpad 方法将中间步骤格式化为文本字符串，拼接到提示模板中。AgentType 枚举列举了所有预定义的 Agent 类型，包括 ZERO_SHOT_REACT_DESCRIPTION、CONVERSATIONAL_REACT_DESCRIPTION 等。

这一阶段的设计特点是"类型驱动"：每种 Agent 行为对应一个类。开发者通过 initialize_agent(agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, ...) 选择预定义的行为模式。这种方式简单直接，但灵活性有限 -- 如果想修改提示模板或输出解析器，就需要深入了解特定 Agent 子类的内部实现。

第二阶段：基于 create_xxx_agent 的函数式构建

第二阶段引入了 create_react_agent、create_tool_calling_agent 等构建函数。这些函数返回 Runnable 而非 Agent 子类，利用 LCEL 管道组装而成。开发者可以自由传入自定义的提示模板、输出解析器等组件。

这一阶段的设计转向了"组合驱动"：Agent 不再是一个不可拆分的类，而是由四个阶段（格式化、提示、推理、解析）组成的管道。每个阶段都可以独立替换。AgentExecutor 通过 validate_runnable_agent 自动将 Runnable 包装为 RunnableAgent，实现了向后兼容。

第三阶段：面向 LangGraph 的演进

当前的发展方向是将 Agent 的执行逻辑从 AgentExecutor 迁移到 LangGraph。LangGraph 提供了图状态机的抽象，能够表达更复杂的执行流程（如并行分支、条件跳转、人工审核节点）。在这个方向上，执行循环不再是 _call 里的 while 循环，而是图的遍历。落到 pin 快照上，新入口是 langchain 1.x 的 create_agent（libs/langchain_v1/langchain/agents/factory.py:684），它接收聊天模型与工具，返回 LangGraph 的 CompiledStateGraph，不再经过 AgentExecutor；LangGraph prebuilt 里的同名 create_react_agent 已标 @deprecated 并指向它（见《LangGraph 设计与实现》第 16 章 §16.1.1）。本章讲的 AgentExecutor、RunnableAgent 在 langchain-classic 1.0.3 里没有挂 @deprecated，但 Agent、initialize_agent 等的弃用消息 AGENT_DEPRECATION_WARNING 已明确建议新项目改用 LangGraph；classic 版 create_react_agent 的 docstring 也写明它不适合生产、推荐 create_agent（第 15 章 §15.3 有两个同名函数的对照）。

理解这个演进脉络，就能明白为什么 AgentExecutor 的源码中同时存在旧式的 Agent 类和新式的 RunnableAgent -- 它们是不同阶段的产物，通过适配器模式共存于同一个执行引擎中。这种兼容性设计既是技术债务，也是对现有用户的负责。

14.11 实际应用中的最佳实践

在生产环境中使用 Agent 系统时，有几个关键的实践经验值得分享。

Agent 的工具集设计原则

Agent 的表现很大程度上取决于提供给它的工具集的质量。工具的名称应该清晰明确，描述应该准确详细地说明工具的功能、输入格式和输出格式。以 ReAct Agent 为例，create_react_agent 会把每个工具的名称和描述渲染进提示词的 {tools}，这就是模型选择工具时能看到的全部工具信息。一个好的工具描述应该告诉 Agent 在什么情况下应该使用这个工具、期望什么格式的输入、以及工具返回结果的含义。

工具集的大小也需要控制。太少的工具会限制 Agent 的能力，太多的工具则会增加 Agent 选择困难的概率——具体在多少个工具上开始退化，取决于模型和工具描述的区分度，需要用自己的评测集测出来，没有普适阈值。如果确实需要大量工具，可以考虑使用分层工具 -- 先让 Agent 选择一个工具类别，再在类别内选择具体工具。

另一个重要的实践是为工具设计"防呆"输入。工具应该尽可能接受宽泛的输入格式，在内部进行规范化处理，而不是要求 Agent 给出精确格式的输入。例如，日期查询工具应该同时接受"今天"、"2024-01-15"、"上周三"等多种格式，而不是强制要求 ISO 格式。

合理设置 max_iterations

默认的 15 次迭代对大多数任务来说足够了，但对于需要多步推理的复杂任务，可能需要适当增加。然而，盲目增大这个值是危险的 -- 如果 Agent 陷入了无效循环，更多的迭代只会浪费更多的 API 调用费用。建议的做法是先在开发环境中设为 None（无限制）观察实际需要的步数，然后在生产环境中在观察值基础上留出适当余量。

始终开启 handle_parsing_errors

生产环境中应该始终设置 handle_parsing_errors=True。LLM 的输出格式不总是稳定的，即使使用了严格的提示模板，偶尔也会产生解析器无法处理的输出。将解析错误作为观察结果反馈给 Agent，让它自我修正，比直接抛出异常导致整个流程失败要好得多。如果你需要更精细的错误处理，可以传入一个自定义函数，根据错误类型返回不同的提示信息。

使用 return_intermediate_steps 进行调试

开发和测试阶段，建议设置 return_intermediate_steps=True。这样你可以看到 Agent 的完整思考过程，包括每一步选择了什么工具、传入了什么参数、得到了什么结果。这些信息对于理解 Agent 的行为模式、优化提示模板、改进工具描述都非常有价值。

利用 trim_intermediate_steps 管理上下文

对于长时间运行的 Agent 任务，中间步骤的累积可能导致上下文窗口溢出。使用 trim_intermediate_steps 可以有效控制传给 LLM 的历史长度。一种常见的策略是传入一个自定义函数，根据 token 数量动态裁剪：保留最早的一两步（提供任务背景）和最近的几步（提供最新状态），中间的步骤如果超出预算则移除。

小结

本章深入剖析了 LangChain Agent 系统的完整架构。从三个核心数据模型（AgentAction、AgentFinish、AgentStep）到两种 Agent 基类（单动作和多动作），从 RunnableAgent 的桥接设计到 AgentExecutor 的执行循环，我们看到了一个精心设计的分层架构。

执行循环的核心是 _call -> _take_next_step -> _iter_next_step 的三层调用链，配合 _should_continue 的双重停止保护和 handle_parsing_errors 的多态错误处理，构成了一个既健壮又灵活的执行引擎。异步版本通过 asyncio.gather 实现了并发工具调用，进一步提升了执行效率。

Agent 系统的历史演进 -- 从基于 Chain 的类型驱动模式，到基于 Runnable 的组合驱动模式，再到面向 LangGraph 的图状态机模式 -- 清晰地展现了框架设计的演进方向。而贯穿始终的设计哲学是"决策与执行分离"：Agent 只管 plan，执行器管一切其他事情。这种关注点分离使得两端都可以独立演进，也为下一章的各种 Agent 模式提供了统一的运行时支撑。

最后需要强调的是，Agent 架构的真正价值不在于代码的复杂度，而在于它为大语言模型提供了一个结构化的"行动空间"。通过将工具调用、错误处理、停止条件等问题系统化地解决，Agent 架构让开发者可以专注于两件最重要的事情：设计好工具集和编写好提示模板。框架负责一切机制性的工作，开发者负责一切创意性的工作。