Agent 框架底层原理深度解析：LangChain、LangGraph 与 Eino 的架构设计与工程实践#

引言：从概念到工程的必然演进

在上一篇文章中，我们系统梳理了 AI Agent 的核心工作模式——从 ReAct 到多智能体协作。然而，理解了模式与真正能在生产环境中落地之间，横亘着一道工程鸿沟：如何用代码将模式转化为可运行的、稳定的、可观测的系统？

这正是框架存在的意义。LangChain、LangGraph、Eino——给出了不同的工程答案。LangChain 以"组件化"著称，LangGraph 以"图状态机"见长，Eino 则面向 Go 生态提供了工程化的编排体系。本文将从源码级视角深入这三者的底层原理，同时揭示这些设计背后的工程逻辑——为什么这么设计？解决了什么问题？付出了什么代价？

LangChain：从组件库到 Agent 引擎#

核心设计哲学：一切皆组件#

LangChain 的设计目标可以概括为：将 AI 应用开发抽象为三个基本组件——Models、Prompts、Chains，通过模块化的方式让开发者能够灵活构建、调试和扩展 AI 应用。

这一设计背后是一个清晰的工程判断：大语言模型的能力再强，也只是 AI 应用的一个"零部件"。真正让应用跑起来的，是围绕模型构建的提示词管理、工具调用、记忆存储、任务编排等一系列基础设施。LangChain 将这些能力全部组件化，开发者可以像搭积木一样自由组合。

工程逻辑：LangChain 选择组件化而非提供一体化解决方案，其根源在于 AI 应用开发的"长尾"特性——每个业务场景对模型、提示词、工具的需求各不相同。组件化牺牲了一定的端到端便利性，换取了极高的灵活性和可扩展性。这也是 LangChain 能成为"AI 开发标准库"的根本原因。

Agent 的底层本质：一张只有两个节点的图#

这是理解 LangChain Agent 源码最核心的一点：LangChain 1.x 版本的 Agent 底层实际上是一张由 LangGraph 驱动的有向状态图（StateGraph）。create_agent 的源码第一行导入就是 from langgraph.graph.state import StateGraph——LangChain 在这里只是对 LangGraph 做了上层封装。

这张图只有两个节点：

# create_agent 的核心逻辑（源码简化）
def create_agent(model, tools, ...):
    # 创建状态图，状态里只有一个核心字段：messages（所有对话历史）
    graph = StateGraph(AgentState)

    # 添加 model 节点：调用 LLM
    graph.add_node("model", model_node)

    # 添加 tools 节点：用 LangGraph 内置的 ToolNode 封装所有工具
    tool_node = ToolNode(tools)
    graph.add_node("tools", tool_node)

    # 入口：先调用 model
    graph.add_edge(START, "model")

    # 条件边：model → tools（有 tool_calls 时）或 model → END（无 tool_calls 时）
    graph.add_conditional_edges(
        "model",
        _make_model_to_tools_edge(...),  # 判断条件：tool_calls 是否为空
        {"tools": "tools", END: END}
    )

    # tools → model（工具执行完，回 LLM 看结果）
    graph.add_edge("tools", "model")

    return graph.compile()

这张图的执行逻辑极其简单清晰：

  • model 节点：调用 LLM 时，将所有工具的 JSON schema 作为参数传递给模型。模型根据对话历史决定是直接回答（无 tool_calls）还是调用某个工具（返回 tool_calls）
  • tools 节点：拿到 LLM 返回的 tool_calls 后，根据函数名在注册表中找到对应的真实函数，执行它，将返回值包装成 ToolMessage
  • 条件边：检查 AIMessage.tool_calls 是否为空——非空就去 tools 节点，为空就结束

图引擎在这个循环里来回跳转，直到 LLM 决定不再调用工具为止。这就是 LangChain Agent 的全部机制。

工程逻辑：为什么 LangChain 最终选择用 LangGraph 重写 Agent，而不是继续维护 AgentExecutor？核心原因是 AgentExecutor 的 while 循环对循环控制、状态管理、错误恢复的支持天然不足。当 Agent 需要支持"暂停→人工审批→继续"这样的 HITL 流程时，while 循环无法做状态快照和恢复。而 LangGraph 的 Pregel 模型天然支持在每一步打检查点，这正是 LangChain 团队将 Agent 底层切换为 LangGraph 的工程动机。

Prompt 构建：Agent 行为控制的"软约束"#

LangChain Agent 的 Prompt 构建是控制其行为的关键环节。从源码可以看到，Agent 的 Prompt 主要由三部分构成：

# ZeroShotAgent 的 from_llm_and_tools 方法中
@classmethod
def from_llm_and_tools(cls, llm, tools, ...):
    prompt = cls.create_prompt(
        tools=tools,
        format_instructions=format_instructions,  # 格式指令：告诉 LLM 如何输出工具调用
        input_variables=input_variables,
    )
    llm_chain = LLMChain(llm=llm, prompt=prompt, ...)
    return cls(llm_chain=llm_chain, allowed_tools=tool_names, ...)

其中关键的格式指令（FORMAT_INSTRUCTIONS）定义了 LLM 输出工具调用的格式规范——通常要求以 JSON 格式输出 action 和 action_input 字段。

工程逻辑：这里揭示了一个深刻的设计约束——Agent 的行为并非由代码硬编码控制，而是通过精心设计的 Prompt 向 LLM 传递"行为规范"。这是一种"软约束"设计：框架无法强制 LLM 按照预期输出格式响应，只能通过 Prompt 工程提高格式遵循的概率，并在解析失败时兜底。这种设计是 LangChain 这类框架的根本性局限，也是 LangGraph 试图通过"状态机"来部分解决的工程问题——将"控制流决策"从 Prompt 中抽离，转移到代码层面。

AgentExecutor：循环执行的"运行时"（早期架构）#

在 LangChain 1.x 的早期架构中，AgentExecutor 是驱动 Agent 运行的引擎，其核心是一个 while 循环：

# AgentExecutor 的核心循环逻辑（概念示意）
def _call(self, inputs):
    # 构建初始 prompt，包含工具描述和当前任务
    # 进入迭代循环
    while iterations < max_iterations:
        # 1. 获取下一步行动：调用 LLM，解析出 action
        action = self._get_next_action(inputs)

        # 2. 如果是 AgentFinish，说明 LLM 决定直接回答，结束循环
        if isinstance(action, AgentFinish):
            return action

        # 3. 执行工具：从 name_to_tool_map 中查找并运行
        observation = self._execute_tool(action)

        # 4. 将观察结果追加到"scratchpad"（思考草稿区）
        #    下一轮 LLM 调用会看到这个历史
        inputs["agent_scratchpad"] += f"\n{action.log}\n{observation}"

这个循环有两个关键限制机制：

  • max_iterations：防止无限循环
  • max_execution_time：防止任务无休止执行

工程逻辑：AgentExecutor 的核心设计是一个显式的控制循环，而非事件驱动。这意味着：

  • 优点：执行流程完全可预测，每一步做什么都清晰可见
  • 缺点：无法在循环中间"暂停"并等待外部输入（如人工审批），也无法在崩溃后从断点恢复

这也是 LangChain 团队在后续版本中放弃 AgentExecutor，全面转向 LangGraph 的根本原因。

LangGraph：从"链"到"图"的架构升维#

为什么需要 LangGraph？#

LangChain 的 Chain 基于 DAG，天然适合线性任务流，但 Agent 需要的能力是循环（Loop）、条件回退、分支和状态持久化。用 LangChain 的旧式 AgentExecutor 去实现这些逻辑，往往导致代码难以维护和调试。

LangGraph 的底层运行时设计，直接回答了这一问题。

工程逻辑：LangGraph 并非 LangChain 的替代品，而是其核心能力的升维——从"链"进化到"图"。两者的关系是：LangGraph 是骨架，LangChain 是肌肉。通常的做法是用 LangGraph 定义智能体的整体流程（图结构），而在图的节点内部使用 LangChain 的组件来调用模型或处理文档。

核心运行时：Pregel 模型的工程化#

LangGraph 的运行时以 Google 的 Pregel 算法命名——这是一种描述使用图进行大规模并行计算的有效方法。选择 Pregel 作为基础模型，背后有深刻的工程考量：BSP（Bulk Synchronous Parallel，批量同步并行）模型天然支持状态快照，而状态快照是实现 HITL 和容错恢复的基础设施。

在 LangGraph 的 Pregel 实现中，执行被组织成多个步骤，每个步骤包含三个阶段：

  1. Plan（计划）：确定在此步骤中要执行哪些 actor（节点）。第一步选择订阅 input channels 的 actor；后续步骤选择订阅了上一步被更新 channels 的 actor
  2. Execution（执行）：并行执行所有选定的 actor，直到全部完成、发生故障或达到超时。在此阶段，channels 的更新对 actor 不可见——需要等到下一个步骤才能读取
  3. Update（更新）：使用 actor 在此步骤中写入的值更新 channels
# Pregel 执行模型的核心结构（概念示意）
class Pregel:
    def invoke(self, inputs):
        # 将输入写入 input channels
        # 开始 Step 循环
        while True:
            # Plan: 找出订阅了被更新 channels 的所有 actor
            actors = self._plan_next_actors()
            if not actors:
                break
            # Execution: 并行执行所有 actor
            results = self._execute_parallel(actors)
            # Update: 将结果写入 channels
            self._update_channels(results)
        # 返回 output channels 的值
        return self._read_output_channels()

工程逻辑：Pregel 模型的三个关键工程优势：

  • 天然支持并行：同一 Step 内的所有 actor 可以并行执行，大幅提升吞吐量
  • Step 间天然同步：每个 Step 结束时所有 channels 更新可见，没有竞态条件
  • Step 级可检查点：在 Step 边界做状态快照，是实现容错和 HITL 的基础

代价则是延迟增加——并行 actor 中最慢的那个决定了整个 Step 的完成时间（同步屏障），以及状态序列化带来的额外开销。

Channels：状态流转的核心抽象#

LangGraph 中，actor（节点）之间通过 Channels 进行通信。每个 channel 有一个值类型、更新类型和更新函数：

Channel 类型行为适用场景
LastValue存储发送到 channel 的最后一个值默认 channel，适用于输入输出值，或一步到下一步的数据传递
Topic可配置的 PubSub Topic多个 actor 间发送多个值，或在多步中累积输出，可去重
BinaryOperatorAggregate通过二元运算符累积更新（如累加求和）多步聚合计算

当使用 StateGraph API 时，开发者定义的图结构会自动被编译为 Pregel 实例——这意味着 graph.compile() 返回的对象本质上就是一个 Pregel 运行时。

工程逻辑：LangGraph 的 Channel 机制回答了一个关键问题——图的节点之间如何共享和传递数据？

LangChain 的旧式方案是通过一个中心化的"状态字典"传递所有数据，所有节点共享同一个可变对象。这在并发场景下会带来竞态条件，在长周期运行中则会导致"状态膨胀"（所有历史数据堆在一起，越来越大）。

LangGraph 通过类型化的 Channels 来规范状态流转：

  • 每个 Channel 定义了写入什么类型、怎么合并写入、谁可以订阅
  • 这实际上是在编译期定义了一套数据流协议，降低了运行时出错的可能性
  • 同时，Channels 的更新是增量的（只记录变化），而非全量快照，这为检查点机制提供了效率基础

HITL 源码拆解#

LangGraph 实现 HITL（Human-in-the-Loop）的核心是三件套：interrupt() + checkpointer + Command(resume=...)。

interrupt()：一个会"重放"的暂停#

# interrupt 的实现核心逻辑（源码简化）
def interrupt(value: Any) -> Any:
    scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
    idx = scratchpad.interrupt_counter()  # 本节点第几个 interrupt
    # ... 将中断信息（节点名、中断值、状态等）保存到检查点
    # 抛出特殊异常，让图引擎停摆
    raise GraphInterrupt(value)

当一个节点内部调用 interrupt(value) 时，LangGraph 会：

  1. 记录当前节点、中断值、执行状态
  2. 通过 checkpointer 将完整状态（消息历史、变量值、当前节点）持久化到存储中
  3. 抛出 GraphInterrupt 异常，让整个图停摆

恢复的执行路径：#

# 从外部恢复执行
# 人给出决策后，用 Command(resume=...) 喂回去
for event in graph.stream(
    Command(resume={"action": "approve", "value": "some_data"}),
    config={"configurable": {"thread_id": thread_id}}
):
    # 图从检查点恢复，并继续执行

恢复时，框架从检查点读出状态，从当初调用 interrupt 的那一行代码之后继续执行，并将 resume 的数据作为 interrupt() 的返回值注入。

这也解释了为什么 LangGraph 强制要求配置 checkpointer 才能使用 HITL——“没有存档，就没有读档”。

工程逻辑：HITL 实现方式的选择反映了深层的工程权衡：

LangGraph 选择用异常 + 检查点来实现暂停，而非用异步事件或协程。这意味着：

  • 优点：实现极其简单——interrupt() 不改变任何控制流，只是抛出一个异常。所有上层代码不用感知"暂停"的存在，只需要在异常处理链的顶层捕获 GraphInterrupt 即可
  • 代价：暂停必须发生在 Step 边界，不能在任何位置暂停。这限制了暂停的粒度

这一选择背后的工程判断是：Agent 应用需要的是"粗粒度暂停"（在一个步骤结束时暂停等待人工输入），而非"细粒度暂停"（在任意代码行暂停）。前者简单可靠，后者则引入了巨大的复杂性（需要类似 Erlang 的进程快照机制）。

Eino：Go 生态的工程化 Agent 框架源码解析#

设计哲学#

Eino（字节跳动开源）的核心设计哲学之一，是在 Go 的强类型系统之上构建 Agent 框架。与 Python 框架的"运行时类型检查"不同，Eino 在编译期做类型校验。

// 编译期类型校验：validateFieldMapping 在编译期检查映射合法性
func validateFieldMapping(predecessorType, successorType reflect.Type, mappings []*FieldMapping) error {
    // 检查类型是否可赋值
    // 不匹配则直接返回编译错误，而非运行时 panic
}

这意味着在 Eino 中编排 Graph 时，类型错误会在编译阶段暴露，而非等到线上运行时崩溃。

工程逻辑：Eino 选择编译期类型安全的代价和收益分别是：

  • 收益：线上 0 类型 panic，在 LLM 应用这个"天生不确定"的领域，框架层面最大程度消除不确定因素，让开发者只处理"LLM 输出不确定"这一个不确定性来源
  • 代价：框架实现较复杂——需要使用泛型和反射来构建类型安全的图编排 API，同时限制了某些动态能力

Callback 系统：基于 Context 的"事件总线"#

Eino 的 callback 系统把 handler 列表藏在 context 里，随 context 流经每个节点时触发——没有全局事件总线、没有注册表、没有反射，只有 context。

// manager 结构：handler 随 context 流动
type manager struct {
    globalHandlers []Handler  // 进程级全局 handlers
    handlers       []Handler  // 此次调用注册的 handlers
    runInfo        *RunInfo   // 当前组件的身份信息（类型、名字、组件种类）
}

func ctxWithManager(ctx context.Context, m *manager) context.Context {
    return context.WithValue(ctx, CtxManagerKey{}, m)
}

func managerFromCtx(ctx context.Context) (*manager, bool) {
    v := ctx.Value(CtxManagerKey{})
    m, ok := v.(*manager)
    if ok && m != nil {
        n := *m        // ← 关键：返回副本，不共享指针
        return &n, true
    }
    return nil, false
}

关键设计：managerFromCtx 返回的是 *m 的副本（n := *m），不是指向原结构体的指针。每个节点读到的是独立的 manager 实例——Graph 并发执行时不会有数据竞争。

两条注册路径：

注册方式作用域适用场景
AppendGlobalHandlers进程级，所有节点生效OpenTelemetry 链路追踪、全局 Token 计量
CtxWithHandlers本次调用，通过 context 传播每个请求独立的审计日志、会话追踪

工程逻辑：Eino 的 Callback 设计体现了 Go 生态的工程思维：

  • 不依赖全局状态（没有全局事件总线），所有观测数据随请求流动
  • 用副本隔离避免并发竞争，而非用锁
  • 用 Context 传播而非依赖注入，保持代码简洁

这种设计的基础是 Go 的 Context 机制——虽然不如 Python 的异步上下文管理器灵活，但在高并发场景下更可靠。

组件调用机制：CallOption 的双层设计#

Eino 的 CallOption 设计区分了两个层级：

// Option 结构：同时承载抽象层和实现层的选项
type Option struct {
    // 抽象层选项：如 WithTemperature（由组件抽象接口统一定义）
    apply func(opts *Options)

    // 实现层选项：特定实现的私有配置
    implSpecificOptFn any
}

这种设计既保证了对所有实现（OpenAI、Claude、Ollama 等）的统一配置接口（如 WithTemperature），又为特定实现保留了私有的扩展能力——一个实现可以定义自己特有的 CallOption，而不影响其他实现。

工程逻辑：CallOption 设计解决的是"统一接口 vs. 实现差异"的矛盾。

在 Python 中，这种矛盾通常通过 kwargs 透传或运行时类型判断来解决——既不安全也不优雅。Eino 通过 Option 结构承载两个层级的配置，在编译期保证类型安全，在运行时灵活扩展。这是 Go 强类型系统 + 接口组合能力的体现。

分支控制与状态管理#

Eino 的 Graph 支持单选分支、多选分支和流式分支：

// 单选分支：根据条件返回下一个节点 key
branch := NewGraphBranch(
    func(ctx context.Context, in T) (string, error) {
        return "nodeA", nil
    },
    map[string]bool{"nodeA": true, "nodeB": true}
)

// 多选分支：并行执行多个分支
multiBranch := NewGraphMultiBranch(
    func(ctx context.Context, in T) (map[string]bool, error) {
        return map[string]bool{"nodeA": true, "nodeB": false}
    }
)

// 流式分支：只读第一个 chunk 就做决策
streamBranch := NewStreamGraphBranch(
    func(ctx context.Context, sr *schema.StreamReader[T]) (string, error) {
        first, _ := sr.Recv()
        return decideByFirst(first), nil
    }
)

状态管理方面，Eino 支持 WithGenLocalState 为每个请求创建独立的局部状态，以及 StatePreHandler 在节点执行前对状态进行预处理。

工程逻辑：Eino 的分支设计区分了三种决策场景：

  • 单选：最常见的 if-else，决策后只走一条路
  • 多选：并行处理，适用于需要同时走多条路径的场景
  • 流式分支：不需要等待完整输入就能做决策——这在流式处理场景下能显著降低延迟

这种细粒度的分支设计表明，Eino 不仅仅是为"Agent 对话"设计的，而是为更广泛的流式数据处理场景设计的。

Eino 的 HITL 实现：显式定向恢复#

与 LangGraph 类似，Eino 也原生支持 HITL，但采用显式定向恢复策略：

// 中断信息提取
composeInfo, ok := compose.ExtractInterruptInfo(err)
// interruptContexts 中包含了所有中断点的 ID 和数据

// 定向恢复：通过 ResumeWithData 指定中断 ID 和恢复数据
resumeCtx := compose.ResumeWithData(context.Background(), interruptID, "用户提供的恢复数据")

// 或批量恢复多个中断点
resumeCtx := compose.BatchResumeWithData(ctx, map[string]any{
    "interrupt-id-1": "数据1",
    "interrupt-id-2": 42,  // 任意类型
})

被恢复的组件内部可以通过 GetInterruptState 检查是否有被中断的历史状态，通过 GetResumeContext 获取用户提供的恢复数据。

工程逻辑：Eino 的 HITL 实现与 LangGraph 的差异值得深入分析：

维度LangGraphEino
中断机制interrupt() 抛异常ExtractInterruptInfo 捕获错误
恢复方式Command(resume=...)ResumeWithData(中断ID, 数据)
语义“继续被中断的图”“向指定中断点注入恢复数据”
适用场景单中断点恢复多中断点批量恢复

Eino 选择显式定向恢复而非 LangGraph 的隐式恢复，根本原因是 Go 的错误处理哲学——Go 不鼓励使用异常控制流，而是通过显式返回 error 来处理异常。Eino 的 ExtractInterruptInfo 实际上是从返回的 error 中提取中断信息，再通过 ResumeWithData 显式指定恢复目标。这种方式在 Go 生态中更符合习惯，但相比 LangGraph 的"从 interrupt 那一行继续执行"，Eino 需要开发者更明确地管理中断 ID 和恢复数据。

三框架源码级横向对比#

维度LangChainLangGraphEino
核心运行时AgentExecutor（while 循环）→ v1 迁移至 LangGraphPregel 状态图引擎（BSP 模型）Chain/Graph/Workflow 三层编排
节点执行顺序执行并行执行（Step 内并行，Step 间同步）支持并行与分支
状态管理隐式（Memory 组件，可变状态）显式（State Schema + Channel + Checkpointer）显式（GenLocalState + StatePreHandler）
HITL 实现有限支持interrupt + checkpointer + Command(resume)（异常驱动）ExtractInterruptInfo + ResumeWithData（显式错误处理）
类型安全运行时检查（Python 动态特性）运行时检查编译期检查（Go 泛型+反射）
回调机制CallbackManager（全局注册）继承自 LangChainContext 传播 + 副本隔离 + 两条注册路径
核心语言PythonPythonGo

工程启示：从源码看设计取舍#

Agent 到底需要多复杂的图？#

从源码层面看，LangChain Agent 的本质是两个节点（model ↔ tools）的循环。这个事实揭示了一个关键工程判断：大多数业务场景下的 Agent，并不需要复杂的图结构。

LangGraph 引入了丰富的图能力（条件边、循环、并行、子图），但代价是性能开销（基准测试显示 LangGraph 平均延迟 10,155ms vs LangChain 6,046ms，慢约 68%）和认知复杂度。在决定引入 LangGraph 之前，问自己一个问题：我的 Agent 真的需要暂停恢复、人工审批、并行执行吗？ 如果答案是否定的，LangChain 的 create_agent 已经足够。

状态持久化的"物理基础"#

LangGraph 的 interrupt+checkpointer 和 Eino 的 ExtractInterruptInfo+ResumeWithData，核心都是围绕状态快照 + 定向恢复展开。

从工程角度理解，HITL 需要两个基础设施：

  1. 序列化机制：能将图运行的完整状态（消息历史、变量值、当前节点位置）序列化到存储中
  2. 恢复机制：能从存储中读取状态，在当初中断的位置继续执行

LangGraph 通过 Pregel 模型的 Step 边界天然提供序列化点；Eino 则通过显式的中断 ID 管理来实现类似能力。两者殊途同归，都说明了同一个工程事实：没有持久化的状态管理，就没有真正的 HITL。

类型安全的代价#

Eino 在编译期校验类型，换来了"线上 0 类型 panic"的确定性；LangGraph 在运行时检查类型，换来了 Python 生态的动态灵活性和快速迭代能力。

从工程选型角度看，这不是"哪个更好"的问题，而是你的团队更看重什么：

  • Python + LangChain/LangGraph：快速原型，灵活迭代，但运行时可能因类型错误崩溃
  • Go + Eino：编译期安全，生产稳定，但开发周期更长，框架限制更多

回调机制的两种实现范式#

LangChain/LangGraph 的回调机制基于全局注册 + 事件分发，Eino 的回调机制基于 Context 传播 + 副本隔离。

两种范式的工程差异：

  • 全局注册：部署简单，但存在"回调地狱"风险（多个回调互相干扰），且并发场景下需要额外处理
  • Context 传播：每个请求独立，天然隔离，但需要开发者显式传递 Context，代码略显冗余

LangChain 的选择更适合"开箱即用"的场景，Eino 的选择更适合"大规模并发"的场景。

结语#

LangChain、LangGraph 和 Eino 各自回答了 Agent 框架设计中的不同问题。从源码层面看，它们最终都在解决同一个核心矛盾：如何将 LLM 这个"不可靠的推理引擎"通过工程的约束与补偿，变成"可靠的系统组件"。

  • LangChain 通过 Prompt 工程和循环运行时实现了"软约束"，降低了 AI 应用的门槛
  • LangGraph 通过 Pregel 模型和检查点机制实现了"状态化"执行，为复杂 Agent 提供了运行时保障
  • Eino 则通过 Go 的类型系统和 Context 传播机制实现了"编译期安全"，将 Go 的工程确定性带入了 AI 领域

理解这些框架的源码实现，本质上是在理解：每个设计决策都对应一个工程问题，每个工程问题都有其代价和收益。

Backward Agent核心模式剖析 DDD领域驱动 Forward
你又来啦！好开心 🐟