01. AgentExecutor 源码解析

在 LangChain 中，无论是什么类型的 Agent（内置封装），都必须通过 AgentExecutor 来创建执行者才可以运行具有 循环 + 工具执行 的智能体，在 智能体执行者 的底层，实际操作是调用 Agent 智能体，执行它选择的操作/工具，将操作输出传递回 Agent，然后重复，伪代码如下

next_action = agent.get_action(...)

while next_action != AgentFinish:

    observation = run(next_action)

    next_action = agent.get_action(..., next_action, observation)

return next_action

简易后的运行流程如下：

其核心代码是 AgentExecutor 中的 _call() 函数，核心代码如下：

def _call(

    self,

    inputs: Dict[str, str],

    run_manager: Optional[CallbackManagerForChainRun] = None,

) -> Dict[str, Any]:

    # 将工具列表组装成字典便于查询

    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"]

    )

    # 定义变量存储中间步骤，类型为元组列表，第1个值为智能体行动，第2个值为行动/工具结果(观察)

    intermediate_steps: List[Tuple[AgentAction, str]] = []

    # 定义变量记录迭代次数+开始时间

    iterations = 0

    time_elapsed = 0.0

    start_time = time.time()

    # 执行循环知道不能遍历

    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,

        )

        # 检测下一步输出是否为结束标记，如果是则直接返回

        if isinstance(next_step_output, AgentFinish):

            return self._return(

                next_step_output, intermediate_steps, run_manager=run_manager

            )

        # 将数据添加到中间步骤

        intermediate_steps.extend(next_step_output)

        # 检测是否只有一个步骤，并且该工具需要直接返回(return_direct=True)

        if len(next_step_output) == 1:

            next_step_action = next_step_output[0]

            # 检测是否直接返回

            tool_return = self._get_tool_return(next_step_action)

            if tool_return is not None:

                return self._return(

                    tool_return, intermediate_steps, run_manager=run_manager

                )

        # 更新迭代次数+时间间隔

        iterations += 1

        time_elapsed = time.time() - start_time

    # 超过时间限制或者迭代次数则获取停止响应输出并返回最后内容

    output = self.agent.return_stopped_response(

        self.early_stopping_method, intermediate_steps, **inputs

    )

return self._return(output, intermediate_steps, run_manager=run_manager)

02. 传统 Agent 组件的缺陷

LangChain 封装的 Agent 和 Agent执行者 虽然解决了 LCEL 表达式创建的单链应用没法执行 循环步骤 的问题，对于一些简单类型的 Agent 智能体创建已经足够使用，但是仍然存在不少缺陷。

假设我们要创建一个 Agent 应用，该 Agent 可以处理数学和物理问题，并由两个不同的 LLM 负责不同的模块，根据用户的提问使用不同的 LLM + 工具列表 + 线路来回答用户的问题，传统的 Agent 就无能为力了。

其实除了上述的问题，传统 Agent 还存在不少缺陷，如下：

1. 只有 循环步骤 并没有 条件步骤，一个 Agent 应用只能一条路走到黑，不能执行不同的路由；
2. 没法亦或者很难将多个 Agent 融合起来相互协作；
3. 因对 Prompt 与输出解析器的过度封装，导致要修改 Agent 内部的方案变得异常困难；
4. 无论是 Agent 还是 AgentExecutor，因其 黑盒机制，无法在执行的过程中进行额外的干预；
5. 想对 Agent 进行扩展或者动态切换 LLM 难度非常大，例如 添加记忆、切换LLM 等；

这也是 LangChain 旧版本被人诟病的原因，包括 2023 年底在 LLM 领域很火的一篇文章《为什么我放弃了 LangChain？》，被传播了一轮又一轮，链接：https://www.jiqizhixin.com/articles/2024-06-24-10，大家一直在放弃和使用中挣扎。

这是因为在 LangChain 设计的早期（v0.1版本之前），早期的传统链（非LCEL表达式）、传统 Agent 智能体的过度封装，并且层层抽象，让代码变得非常复杂（就像一个组件同时拥有 N 种调用方法一样）。

不过这个弊端随着 LCEL 表达式和今年年初的 LangGraph 发布，普遍都被解决了，更易用与统一的接口（LCEL 表达式统一 invoke、stream、batch 接口），控制性更高、支持循环/逻辑的 LangGraph 图结构程序、实时监控 LLM 的 LangSmith 平台，让 LLM/Agent 应用的开发变得超级简单，甚至串联数百个节点、上百种工具与逻辑路线的 Agent 工作流也成为了现实。

但是对于 传统Agent组件 来讲，我们还是需要去掌握，因为在复杂度较小的情况下，该思路还是非常值得借鉴的，但对于一些复杂应用就无能为力了，下一章我们会来学习 LangChain 的最后一块拼图——LangGraph，掌握利用 LangGraph 构建复杂应用的技巧及底层运行原理，并且在 LLMOps 项目中，将 LangGraph 与知识库、工具、审核、多用户/多应用、多模态输入输出等功能结合起来，在实战中感受 LangGraph + LECL 表达式带来的酣畅淋漓的体验。