Agent 工程体系 · 第 46/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

LangChain Agent：Model、Tool、Middleware、State 与适用边界

Agent 不是“让模型自由发挥”的聊天接口，而是一个可循环执行的决策程序。它通常重复以下过程：

读取当前状态；
将状态整理成模型上下文；
调用 Model；
判断模型是否请求调用 Tool；
执行 Tool，并将结果写回状态；
再次调用 Model，直到模型给出最终回答或流程被中断。

LangChain 为这个循环提供了较高层的 Agent 抽象；LangGraph 则提供更底层的状态图、持久化、流式执行、人机协同和长时间运行能力。当前 LangChain 文档将 create_agent 定位为常见模型—工具循环的高层入口，而 LangGraph 更适合需要自定义拓扑和精细控制的流程。(docs.langchain.com)

本文中的五个核心术语可以先放在同一张图中理解：

flowchart TD
    U[用户输入] --> S[State]
    S --> MW1[Middleware: before_model]
    MW1 --> C[Model Context]
    C --> M[Model]
    M --> D{是否包含 Tool Call?}
    D -- 否 --> A[最终回答]
    D -- 是 --> MW2[Middleware: wrap_tool_call]
    MW2 --> T[Tool]
    T --> R[Tool Result / ToolMessage]
    R --> S2[State Update]
    S2 --> MW1

其中：

Model 负责生成下一步消息或工具调用请求；
Tool 负责执行真实动作；
Middleware 负责在模型调用、工具调用和 Agent 生命周期节点上拦截或修改数据流；
State 保存一次 Agent 执行所需的短期工作记忆；
Agent 负责把这些部件组织成一个循环。

最重要的边界是：Model 只提出意图，Tool 才产生副作用；Middleware 可以改变流程，但不应该被误认为业务流程本身；State 保存事实和执行上下文，但不等于长期数据库。

一、先明确 Agent 的计算模型

1. Model 不是 Agent

给定状态 $S_t$ ，模型得到一个上下文 $C_t$ ：

C_t = f(S_t, P, T_t, X)

其中：

$S_t$ ：第 $t$ 步的 State；
$P$ ：系统提示词、开发者指令等静态规则；
$T_t$ ：当前允许模型看到的工具集合；
$X$ ：运行时上下文，例如用户身份、租户、数据库连接或请求配置；
$f$ ：上下文工程逻辑，包括消息裁剪、摘要、动态提示词和工具筛选。

模型随后产生消息：

M_t = \operatorname{Model}(C_t)

消息可能是：

普通文本；
一个或多个 Tool Call；
结构化输出；
模型错误或格式错误。

因此，模型本身并不知道工具是否真的执行成功。它只生成类似下面的请求：

{
  "name": "get_weather",
  "arguments": {
    "location": "杭州"
  }
}

这不是天气查询结果，也不是已经执行的函数，而是一个待验证、待授权、待执行的调用意图。

Agent 的下一步必须由程序决定：

S_{t+1} = \begin{cases} \operatorname{append}(S_t, M_t), & \text{没有 Tool Call} \\ \operatorname{append}(S_t, M_t, R_t), & \text{执行 Tool 后} \end{cases}

其中 $R_t$ 是工具结果。

这解释了一个常见误解：把“模型能调用工具”理解为“模型拥有系统权限”。实际上，模型只拥有被暴露给它的工具描述；真正的权限检查、参数校验、资源访问和副作用控制，都必须发生在 Tool 执行边界内。

2. Agent 循环的终止条件

一个最小 Agent 循环可以写成：

while True:
    response = model.invoke(messages)

    if not response.tool_calls:
        return response

    tool_messages = execute_tool_calls(response.tool_calls)
    messages.extend([response, *tool_messages])

它的终止条件是：

\operatorname{tool\_calls}(M_t) = \varnothing

但生产系统通常还需要额外终止条件：

\text{stop} = \text{no tool call} \lor \text{step limit exceeded} \lor \text{timeout} \lor \text{authorization denied} \lor \text{human interruption} \lor \text{fatal error}

如果只依赖模型自行停止，就可能出现：

模型重复调用同一个工具；
工具结果无法满足模型，导致无限循环；
某个失败工具被不断重试；
大量无效模型调用消耗预算。

LangChain 的 Agent 抽象负责常见的模型—工具循环，但当流程不再是“调用模型，执行工具，再调用模型”的简单循环时，应考虑直接使用 LangGraph 的节点和边表达流程。(docs.langchain.com)

二、Model：决策器、上下文消费者和工具调用协议

1. Model 的输入不是“用户问题”，而是上下文

Model 通常接收一组消息：

[
    {"role": "system", "content": "你是一个严谨的助手。"},
    {"role": "user", "content": "杭州现在天气如何？"},
    {
        "role": "assistant",
        "content": "",
        "tool_calls": [
            {
                "name": "get_weather",
                "args": {"location": "杭州"},
                "id": "call_001",
            }
        ],
    },
    {
        "role": "tool",
        "tool_call_id": "call_001",
        "content": "{\"temperature\": 28, \"condition\": \"晴\"}",
    },
]

模型能否正确行动，取决于上下文中是否包含：

清晰的系统指令；
未被截断的关键历史；
工具名称和参数描述；
工具调用结果；
当前用户和权限信息；
输出格式约束。

因此，Agent 的可靠性不只是模型能力问题，也取决于上下文构造：

\text{Agent Reliability} \approx \text{Model Quality} \times \text{Context Correctness} \times \text{Tool Correctness} \times \text{Loop Control}

这不是严格的统计定律，而是一个工程上的因果模型：其中任何一项接近零，整体行为都会明显失效。

2. 工具调用不是普通文本解析

现代模型通常通过结构化 Tool Call 表达调用请求，而不是要求应用程序从自然语言中正则匹配：

我要调用 get_weather，参数是杭州。

结构化调用至少包含：

工具名；
参数对象；
调用 ID；
有时还包含并行调用列表。

调用 ID 很重要，因为模型可能在一次响应中请求多个工具，应用程序需要将每个结果准确关联回对应请求：

\text{ToolMessage.tool\_call\_id} = \text{AIMessage.tool\_calls[i].id}

如果关联错误，模型会看到不属于当前调用的结果，后续推理可能产生看似合理但事实错误的答案。

3. Model 的适用边界

Model 适合：

处理自然语言；
对候选工具进行选择；
从上下文中提取参数；
生成解释性回答；
在多个可行步骤之间进行启发式决策。

Model 不适合直接负责：

最终权限判断；
金额计算；
数据库事务提交；
幂等保证；
合规策略判断；
对外部系统的无条件写入；
需要绝对确定性的路由。

例如，下面这个判断不应仅由模型完成：

如果金额大于 10000 元，就自动批准退款。

正确做法是：

模型可以提出“申请退款”；
Tool 校验订单、用户、金额和订单状态；
确定性代码判断金额阈值；
超过阈值时返回拒绝或触发人工审批；
只有通过授权的调用才执行写操作。

模型的“推理正确”不能替代业务规则的“判定正确”。

三、Tool：把模型意图转换为受控动作

1. Tool 的本质

Tool 是一个由程序执行的、具有明确输入输出契约的函数。它至少包含：

\text{Tool} = (\text{name}, \text{description}, \text{input schema}, \text{executor})

其中：

name：模型调用时使用的标识；
description：提供给模型的用途说明；
input schema：参数结构和类型；
executor：真正执行操作的代码。

LangChain 支持将普通 Python 函数或协程定义为工具；工具可以访问当前 Agent 的 State、运行时上下文和配置，但这些运行时参数可以不暴露给模型。(docs.langchain.com)

一个只读工具可以写成：

from langchain.tools import tool


@tool
def get_weather(location: str) -> str:
    """查询指定城市的当前天气。"""
    if not location.strip():
        raise ValueError("location 不能为空")

    # 示例中使用固定值；生产环境应调用真实天气服务。
    return f"{location}：28°C，晴"

模型看到的是工具名、描述和 location 参数，而不是 Python 函数的全部实现。

2. Tool 的执行过程

一次工具调用至少经过以下步骤：

sequenceDiagram
    participant M as Model
    participant V as Schema Validator
    participant A as Authorization
    participant T as Tool Executor
    participant S as State

    M->>V: tool_call(name, arguments)
    V-->>M: 参数合法 / 参数错误
    V->>A: 请求执行权限
    A-->>V: 允许 / 拒绝
    V->>T: 已授权参数
    T-->>V: 结果或异常
    V->>S: 写入 ToolMessage 与状态更新
    S->>M: 下一轮上下文

这里的关键因果关系是：

模型输出参数，不代表参数可信；
Schema 校验通过，不代表调用有权限；
权限通过，不代表外部服务成功；
外部服务成功，不代表结果适合直接写入 State；
Tool 返回结果后，模型仍可能误解结果，因此必要时需要确定性校验或人工确认。

3. Tool 参数校验与授权必须分层

参数校验回答：

调用参数的形状和类型是否正确？

授权回答：

当前用户是否允许对指定资源执行该动作？

例如：

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime


@dataclass
class RuntimeContext:
    user_id: str


@tool
def get_order(order_id: str, runtime: ToolRuntime[RuntimeContext]) -> str:
    """查询当前用户拥有的订单。"""
    if not order_id.startswith("order_"):
        raise ValueError("非法订单号格式")

    user_id = runtime.context.user_id

    # 生产环境必须在数据库查询条件中同时约束 order_id 和 user_id。
    order = load_order(order_id=order_id, user_id=user_id)

    if order is None:
        raise PermissionError("订单不存在或当前用户无权访问")

    return serialize_order(order)

不能只依赖模型提供 user_id：

# 错误示例：user_id 完全由模型传入
@tool
def get_order(order_id: str, user_id: str) -> str:
    ...

因为模型可以生成任意 user_id。身份应来自认证后的运行时上下文，而不是来自不可信的模型参数。

4. Tool 返回值有两种用途

工具结果通常需要同时服务于两个对象：

提供给模型继续推理；
提供给应用程序记录、展示或后续节点处理。

因此，工具返回值不应只追求“适合模型阅读”。对于重要业务，建议区分：

{
    "ok": True,
    "data": {
        "order_id": "order_123",
        "status": "paid"
    },
    "error": None,
    "meta": {
        "source": "order-service",
        "request_id": "req_789"
    }
}

模型可以看到可读文本，应用程序则依赖稳定字段。若外部 API 的原始错误对象、堆栈信息或凭证被直接拼接进 ToolMessage，可能导致敏感信息泄漏或提示注入。

5. Tool 的副作用边界

可以把工具分成三类：

类型	示例	默认风险
纯计算	汇率换算、日期计算	低
只读访问	查询订单、搜索文档	中
外部写入	发邮件、退款、删除文件	高

副作用越强，越不应直接由模型一次调用完成。高风险工具通常需要：

明确的授权上下文；
参数二次校验；
幂等键；
超时；
重试策略；
审计日志；
人工确认；
可恢复的状态记录。

LangChain 的工具机制可以承载这些逻辑，但不会自动替业务系统设计出正确的授权、幂等和事务语义。这些属于工具执行器和业务服务的责任。

四、Middleware：围绕生命周期插入控制逻辑

1. Middleware 解决什么问题

如果每个工具、每个模型调用都手写日志、重试、超时、敏感信息过滤和动态模型选择，业务代码会迅速被横切逻辑污染。

Middleware 将这些逻辑放在 Agent 生命周期周围。LangChain 当前文档将 Middleware 分为两种：

Node-style hooks：在生命周期节点执行；
Wrap-style hooks：包裹一次模型调用或工具调用。

可用的生命周期位置包括：

before_agent
before_model
after_model
after_agent
wrap_model_call
wrap_tool_call

多个 Middleware 的执行顺序具有类似洋葱模型的结构：before_* 按声明顺序执行，after_* 按逆序执行，wrap_* 逐层嵌套。(docs.langchain.com)

2. Node-style 与 Wrap-style 的区别

Node-style Hook 更适合：

修改 State；
在模型调用前注入上下文；
记录一次模型响应；
做执行前校验；
在 Agent 开始或结束时做审计。

例如：

from typing import Any
from langchain.agents.middleware import before_model, AgentState
from langgraph.runtime import Runtime


@before_model
def inject_policy(
    state: AgentState,
    runtime: Runtime,
) -> dict[str, Any] | None:
    """在每次模型调用前追加确定性策略。"""
    return {
        "system_note": (
            "禁止泄露凭证；涉及外部写入时必须确认授权和目标资源。"
        )
    }

Wrap-style Hook 更适合：

包裹模型或工具调用；
重试；
统一异常处理；
动态选择模型；
统计调用耗时；
根据执行结果生成状态更新。

工具错误处理示例：

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage


@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except PermissionError:
        return ToolMessage(
            content="工具调用被拒绝：当前身份无权访问该资源。",
            tool_call_id=request.tool_call["id"],
        )
    except TimeoutError:
        return ToolMessage(
            content="工具调用超时：请稍后重试，不要假设操作已经成功。",
            tool_call_id=request.tool_call["id"],
        )
    except Exception:
        return ToolMessage(
            content="工具调用失败：输入可能不正确，或后端暂时不可用。",
            tool_call_id=request.tool_call["id"],
        )

这段代码的关键点是：异常被转换成与原 Tool Call 对应的 ToolMessage，模型可以据此决定修正参数、换用其他工具或向用户报告失败。官方文档也采用 wrap_tool_call 将工具异常转成自定义 ToolMessage 的方式。(docs.langchain.com)

但必须注意：把异常转换成 ToolMessage 不等于把操作回滚了。

如果工具已经向支付服务提交请求，随后在序列化响应时抛异常，Middleware 只能告诉模型“调用失败”，却无法证明支付没有发生。对于有副作用的工具，必须在工具内部或下游服务中实现查询状态、幂等和补偿。

3. Middleware 的执行顺序会改变语义

假设定义三个 Middleware：

agent = create_agent(
    model=model,
    tools=[...],
    middleware=[m1, m2, m3],
)

典型执行顺序是：

m1.before_model
m2.before_model
m3.before_model

m1.wrap_model_call(
    m2.wrap_model_call(
        m3.wrap_model_call(model)
    )
)

m3.after_model
m2.after_model
m1.after_model

因此，顺序不是格式问题，而是语义问题。

例如：

如果先做 PII 脱敏，再记录日志，日志不会包含原始身份证号；
如果先记录日志，再脱敏，日志可能已经泄露；
如果重试 Middleware 包住超时 Middleware，重试次数和单次超时的含义不同；
如果授权 Middleware 只包住模型调用而不包住工具调用，它实际上没有保护副作用。

工程上应明确每个 Middleware 的作用域：

身份与授权：必须覆盖工具执行
参数过滤：应在工具执行前
日志脱敏：应早于日志写入
重试：只包裹明确可重试的操作
超时：包裹实际 I/O，而不是只包裹模型决策

4. 动态模型选择

Middleware 可以根据 State 或运行时上下文选择不同模型。例如，短问题使用低成本模型，复杂任务切换到更强模型：

from typing import Callable
from langchain.agents.middleware import (
    wrap_model_call,
    ModelRequest,
    ModelResponse,
)


@wrap_model_call
def select_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    messages = request.state["messages"]

    if len(messages) > 12:
        selected = advanced_model
    else:
        selected = fast_model

    return handler(request.override(model=selected))

这个逻辑只能改变“调用哪个模型”，不能保证复杂任务一定适合高级模型。复杂度判定本身也可能失真，例如一条很短的消息可能包含高风险写操作。因此，动态模型选择应与工具风险、任务类型和预算策略结合，而不应只按消息数量判断。官方文档展示了通过 wrap_model_call 基于当前 State 动态选择模型的方式。(docs.langchain.com)

五、State：Agent 的短期工作记忆和控制面

1. State 不是变量字典那么简单

State 是 Agent 执行过程中跨节点、跨模型调用传递的数据结构。最常见的字段是消息列表：

{
    "messages": [...]
}

也可以包含业务和控制字段：

{
    "messages": [...],
    "user_id": "u_123",
    "cart_id": "cart_456",
    "risk_level": "high",
    "approval_required": True,
    "tool_call_count": 2
}

LangChain 将 Agent 的消息状态视为短期记忆；State 可以扩展自定义字段。当前 LangChain 1.x 文档要求自定义 Agent State 使用 TypedDict，并更推荐通过 Middleware 声明与特定中间件和工具相关的状态。(docs.langchain.com)

2. State 更新不是简单覆盖

对同一个 State 字段，更新规则由 Reducer 决定。

如果字段使用覆盖语义：

S_{t+1}[k] = U_t[k]

如果字段使用追加或合并语义：

S_{t+1}[k] = R(S_t[k], U_t[k])

消息列表通常需要追加，而不是覆盖。否则，模型刚刚产生的 Tool Call 可能被下一步工具结果覆盖，导致消息链断裂。

在 LangGraph 中，一个简单节点可能返回：

def classify(state):
    return {
        "risk_level": "high",
        "messages": [
            {
                "role": "system",
                "content": "该请求需要人工确认。",
            }
        ],
    }

返回值是状态更新，不是完整的新状态。图运行时会根据字段的 Reducer 将它合并到当前 State。LangGraph 的 Command 也可以同时更新 State 和控制下一步路由。(docs.langchain.com)

3. State、Runtime Context、Store 的区别

这三个概念容易混淆。

State：本次会话或本次流程的工作数据

例如：

消息历史；
当前订单号；
当前审批阶段；
已调用工具列表；
当前风险等级。

工具可以通过 ToolRuntime 读取 State：

from langchain.tools import tool, ToolRuntime


@tool
def get_current_user(runtime: ToolRuntime) -> str:
    """返回当前会话中的用户标识。"""
    return runtime.state.get("user_id", "unknown")

runtime 参数不会出现在模型看到的工具输入 Schema 中，适合放置不应由模型伪造的上下文。(docs.langchain.com)

Runtime Context：本次调用的依赖和静态上下文

例如：

当前用户 ID；
数据库连接；
租户信息；
请求追踪 ID；
配置对象。

它更像依赖注入，而不是对话历史。

Store：跨会话的长期数据

例如：

用户偏好；
历史订单摘要；
组织级配置；
跨线程的记忆。

不能因为 State 可以持久化，就把所有长期数据都塞入 State。State 过大将导致：

每轮模型上下文变长；
Checkpoint 变大；
恢复成本升高；
敏感数据扩散到更多日志和快照；
并发更新更难处理。

LangChain 的运行时同时提供 State、Store 和执行配置等不同上下文来源；文档明确区分了短期 State 与长期 Store。(docs.langchain.com)

4. State 的完整变化示例

用户输入：

帮我查询订单 order_123，并告诉我是否已经支付。

初始 State：

{
  "messages": [
    {
      "role": "user",
      "content": "帮我查询订单 order_123，并告诉我是否已经支付。"
    }
  ],
  "user_id": "u_001"
}

模型第一次响应：

{
  "role": "assistant",
  "tool_calls": [
    {
      "id": "call_001",
      "name": "get_order",
      "args": {
        "order_id": "order_123"
      }
    }
  ]
}

Tool 执行后追加：

{
  "role": "tool",
  "tool_call_id": "call_001",
  "content": "{\"order_id\":\"order_123\",\"status\":\"paid\"}"
}

下一轮模型读取完整消息链后，返回：

{
  "role": "assistant",
  "content": "订单 order_123 已支付。"
}

最终 State 至少保留：

{
  "messages": [
    {"role": "user", "...": "..."},
    {"role": "assistant", "tool_calls": ["call_001"]},
    {"role": "tool", "tool_call_id": "call_001", "...": "..."},
    {"role": "assistant", "content": "订单 order_123 已支付。"}
  ],
  "user_id": "u_001"
}

这里不能只保留最终文本，因为中间的 Tool Call 和 Tool Result 是后续恢复、审计和重放所需要的因果链。

六、用 `create_agent` 组合 Model、Tool、Middleware 和 State

下面给出一个简化示例。它需要安装 LangChain、LangGraph 以及对应模型提供商的集成包，并配置模型提供商的 API Key。模型名称应替换为当前提供商实际支持的名称。

from typing import Any
from typing_extensions import NotRequired

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import (
    before_model,
    wrap_tool_call,
)
from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime


class AppState(AgentState):
    user_id: str
    tool_call_count: NotRequired[int]


@tool
def get_weather(location: str) -> str:
    """查询指定城市的当前天气。"""
    if not location.strip():
        raise ValueError("location 不能为空")

    return f"{location}：28°C，晴"


@tool
def get_user_id(runtime: ToolRuntime) -> str:
    """读取当前认证用户 ID。"""
    return runtime.state["user_id"]


@before_model
def count_model_rounds(
    state: AppState,
    runtime,
) -> dict[str, Any] | None:
    current = state.get("tool_call_count", 0)

    if current >= 5:
        return {
            "messages": [
                {
                    "role": "assistant",
                    "content": "已达到工具调用上限，流程停止。",
                }
            ]
        }

    return {"tool_call_count": current + 1}


@wrap_tool_call
def safe_tool_execution(request, handler):
    try:
        return handler(request)
    except ValueError as exc:
        return ToolMessage(
            content=f"参数错误：{exc}",
            tool_call_id=request.tool_call["id"],
        )
    except Exception:
        return ToolMessage(
            content="工具执行失败，请检查输入或稍后重试。",
            tool_call_id=request.tool_call["id"],
        )


agent = create_agent(
    model="替换为实际模型名称",
    tools=[get_weather, get_user_id],
    middleware=[
        count_model_rounds,
        safe_tool_execution,
    ],
    state_schema=AppState,
)

result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "我的用户 ID 是什么？杭州天气如何？",
        }
    ],
    "user_id": "u_001",
})

print(result)

这个示例的执行过程是：

create_agent 注册 Model 和两个 Tool；
Agent 接收用户消息，写入 messages；
before_model 在每次模型调用前增加计数；
Model 决定是否调用 get_user_id 和 get_weather；
wrap_tool_call 统一捕获工具异常；
Tool 结果以 Tool Message 形式写回 State；
Agent 再次调用 Model；
Model 不再请求工具时返回最终回答。

需要区分两种状态字段：

messages 是模型推理所需的对话和工具调用链；
user_id 是安全上下文，不应该由模型生成；
tool_call_count 是控制字段，用于限制循环。

当前文档仍支持在 create_agent 上使用 state_schema，但推荐把与特定工具或 Middleware 绑定的自定义状态放进 Middleware 的状态声明中，以缩小作用域。(docs.langchain.com)

七、何时需要直接使用 LangGraph

create_agent 适合标准 Agent 循环：

START
  ↓
Model
  ↓
有工具调用？
 ├─ 否 → END
 └─ 是
       ↓
     Tool
       ↓
     Model

但以下情况通常已经超出标准循环：

先分类，再选择不同 Agent；
多个 Agent 并行研究；
工具执行后必须经过确定性校验；
某些步骤必须人工审批；
需要失败恢复和断点续跑；
需要显式表达分支、循环和子图；
需要把 Agent 嵌入更大的业务工作流；
需要在不同节点使用不同 State 结构。

LangGraph 是低层编排框架和运行时，可以混合确定性节点与模型驱动节点；LangChain Agent 本身运行在 LangGraph runtime 之上，因此整个 Agent 可以作为更大 StateGraph 中的节点或子图使用。(docs.langchain.com)

1. State、Node、Edge 的关系

一个简化的 LangGraph 工作流：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END


class WorkflowState(TypedDict):
    question: str
    category: str
    answer: str


def classify(state: WorkflowState):
    question = state["question"]

    if "订单" in question:
        return {"category": "order"}

    return {"category": "general"}


def answer_general(state: WorkflowState):
    return {"answer": f"普通问题：{state['question']}"}


def answer_order(state: WorkflowState):
    return {"answer": f"订单问题：{state['question']}"}


def route(state: WorkflowState):
    if state["category"] == "order":
        return "answer_order"
    return "answer_general"


builder = StateGraph(WorkflowState)
builder.add_node("classify", classify)
builder.add_node("answer_general", answer_general)
builder.add_node("answer_order", answer_order)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route)
builder.add_edge("answer_general", END)
builder.add_edge("answer_order", END)

graph = builder.compile()

print(graph.invoke({
    "question": "订单 order_123 是否已支付？"
}))

预期输出：

{
    "question": "订单 order_123 是否已支付？",
    "category": "order",
    "answer": "订单问题：订单 order_123 是否已支付？"
}

这里：

WorkflowState 定义状态结构；
classify、answer_general、answer_order 是 Node；
add_edge 定义确定性边；
add_conditional_edges 根据 State 选择下一节点；
compile 将图编译为可执行对象。

与标准 Agent 循环相比，图的优势在于：路由逻辑不再隐藏在模型输出中，而是直接由程序表达。模型仍然可以作为某个 Node，但不会垄断整个控制流。

2. `Command` 同时更新状态和路由

当一个节点既要修改 State，又要跳转到指定节点时，可以使用 Command：

from langgraph.types import Command


def check_risk(state: WorkflowState) -> Command:
    if "退款" in state["question"]:
        return Command(
            update={"category": "refund"},
            goto="human_review",
        )

    return Command(
        update={"category": "general"},
        goto="answer_general",
    )

Command 可以表达：

update：更新 State；
goto：跳转到节点；
resume：恢复中断；
graph：在子图和父图之间指定目标。

这些能力使“数据更新”和“控制流跳转”可以在同一个节点返回值中保持一致。(docs.langchain.com)

八、Checkpoint 与中断：Agent 不再只是一次函数调用

1. 为什么 Agent 需要 Checkpoint

普通函数调用通常假设：

调用开始 → 执行 → 返回结果

但真实 Agent 可能：

执行数分钟甚至更久；
等待用户确认；
调用多个外部服务；
中途进程崩溃；
需要人工修改状态后继续；
需要恢复到某个历史节点。

Checkpoint 是某个执行时刻的持久化 State 和执行位置。它让系统可以从已有状态继续，而不是从头重新调用模型和工具。

LangGraph 文档将持久化、可恢复执行和人机协同作为核心能力；中断发生后，图可以保存当前状态并等待外部输入。(docs.langchain.com)

2. 中断的因果路径

sequenceDiagram
    participant C as Client
    participant G as Graph
    participant DB as Checkpointer
    participant H as Human

    C->>G: invoke(request, thread_id)
    G->>G: Model 请求高风险 Tool
    G->>DB: 保存当前 State
    G-->>C: interrupt: 需要确认
    C->>H: 展示待审批内容
    H-->>C: approve / reject
    C->>G: Command(resume=decision, thread_id)
    G->>DB: 读取 checkpoint
    G->>G: 继续执行
    G-->>C: 最终结果

中断时必须使用稳定的 thread_id。它相当于指向某条持久化执行线程的游标：复用同一个 thread_id 才能恢复原来的 State；换一个 ID 则会创建新的执行上下文。(docs.langchain.com)

一个典型的人工审批节点：

from langgraph.types import interrupt, Command


def approve_refund(state):
    decision = interrupt({
        "type": "refund_approval",
        "order_id": state["order_id"],
        "amount": state["amount"],
        "message": "是否批准退款？",
    })

    if decision != "approve":
        return Command(
            update={"approval_status": "rejected"},
            goto="finish",
        )

    return Command(
        update={"approval_status": "approved"},
        goto="execute_refund",
    )

恢复时：

graph.invoke(
    Command(resume="approve"),
    config={
        "configurable": {
            "thread_id": "refund-thread-001"
        }
    },
)

中断机制解决的是“暂停并恢复”，不是“自动保证业务安全”。恢复前仍然需要重新确认：

当前用户身份是否仍有效；
订单状态是否发生变化；
审批是否过期；
工具调用是否已经执行过；
幂等键是否仍然有效。

九、并发与故障路径：模型调用成功，不代表流程成功

1. 并行 Tool Call 的问题

模型可能一次请求多个工具：

{
  "tool_calls": [
    {
      "id": "call_001",
      "name": "get_weather",
      "args": {"location": "杭州"}
    },
    {
      "id": "call_002",
      "name": "get_exchange_rate",
      "args": {"currency": "USD"}
    }
  ]
}

如果两个工具互不依赖，可以并行执行：

T_{\text{parallel}} \approx \max(T_1, T_2)

而顺序执行的延迟是：

T_{\text{serial}} = T_1 + T_2

但并行只在以下条件满足时安全：

工具之间没有写入冲突；
工具不依赖对方结果；
外部系统允许并发；
每个调用都有独立的超时和错误处理；
结果仍能按 tool_call_id 正确关联。

如果工具 A 创建资源、工具 B 删除同一资源，那么并行执行可能导致竞态。此时应由图结构或确定性代码表达先后关系，而不是让模型隐式决定。

2. 超时后的不确定状态

考虑以下故障：

1. Agent 调用 create_payment
2. 支付服务已扣款
3. 网络响应超时
4. Tool 抛出 TimeoutError
5. Middleware 返回“调用失败”

此时系统真实状态是：

\text{支付状态} \in \{\text{成功}, \text{失败}, \text{未知}\}

而不是简单的 failed。

错误处理必须区分：

确定失败：服务明确返回未执行；
确定成功：服务明确返回成功；
未知结果：客户端超时、连接中断、响应丢失。

未知结果不能直接重试非幂等操作。正确流程通常是：

生成稳定幂等键；
查询下游服务是否已接受请求；
若不存在，再决定是否重试；
把最终状态写入业务数据库；
让模型看到“成功”“失败”或“处理中”，而不是模糊异常文本。

3. Checkpoint 恢复与副作用重复

Checkpoint 可以恢复图的执行位置，但不能自动撤销外部副作用。

例如：

Tool A 已成功发邮件
进程在写入 State 前崩溃
恢复后重新执行 Tool A
用户收到两封邮件

所以副作用工具需要：

@tool
def send_email(
    recipient: str,
    subject: str,
    body: str,
    idempotency_key: str,
) -> str:
    """
    发送邮件。
    idempotency_key 必须由业务流程确定性生成。
    """
    return mail_service.send_once(
        recipient=recipient,
        subject=subject,
        body=body,
        idempotency_key=idempotency_key,
    )

幂等键不能完全由模型随机生成，因为模型重试时可能生成不同值。更可靠的来源是：

\text{idempotency\_key} = H(\text{thread\_id}, \text{tool\_call\_id}, \text{business\_operation})

其中 $H$ 是稳定哈希函数。

十、常见误解与失败表现

误解一：Agent 会自动理解所有业务规则

失败表现：

用户说“帮我退款”，模型直接调用 refund_order。

问题在于“退款”至少还需要：

订单是否存在；
是否属于当前用户；
是否满足退款时限；
是否已经退款；
金额是否超过人工审批阈值；
是否需要原路退回。

Model 可以识别意图，但这些规则应该由 Tool 和确定性业务服务执行。

误解二：Middleware 可以替代业务流程

Middleware 适合横切控制，例如日志、重试、动态提示词和错误转换。它不适合承载复杂的订单状态机。

如果业务流程是：

申请退款 → 校验资格 → 风控 → 人工审批 → 执行退款 → 对账

将这些步骤全部塞进一个 Middleware 会导致：

状态转换不透明；
恢复点难以定位；
节点级测试困难；
审批和补偿逻辑难以表达；
失败路径与正常路径混在一起。

更适合使用 LangGraph 显式建模节点、边、State 和中断，把 Middleware 用在每个节点或 Agent 的横切控制上。

误解三：State 就是长期记忆

失败表现：

把所有历史对话、工具原始响应和用户资料永久追加到 messages。

后果包括：

上下文无限增长；
每次模型调用成本升高；
历史过长后关键信息被稀释；
敏感信息出现在更多快照中；
Checkpoint 体积不断增大。

短期执行数据放在 State；跨会话数据放在 Store 或业务数据库；模型上下文只保留当前任务真正需要的内容。LangChain 的上下文工程文档也区分了每次模型调用的临时上下文与持久化到 State 的生命周期数据。(docs.langchain.com)

误解四：工具报错返回给模型，就算完成了错误处理

失败表现：

ToolMessage: 操作失败，请重试。

但系统没有记录：

哪个工具失败；
使用了什么参数；
下游请求是否已到达；
是否可以安全重试；
是否已经产生部分副作用；
当前业务状态是什么。

对生产系统而言，错误信封至少应包含内部可观测字段：

{
  "code": "DOWNSTREAM_TIMEOUT",
  "retryable": false,
  "state": "unknown",
  "request_id": "req_789",
  "tool_call_id": "call_001",
  "user_message": "操作状态暂时无法确认，请查询后再决定是否重试。"
}

模型只需要看到适合继续对话的 user_message；日志和监控系统则保存完整诊断信息。

十一、如何判断使用哪一层

可以按控制流复杂度选择。

只需要一次模型调用

用户输入 → Model → 文本输出

直接使用 Model，不必引入 Agent。

适合：

摘要；
改写；
分类；
单次结构化抽取；
固定提示词问答。

需要模型选择和执行工具

Model → Tool → Model → Tool → ...

使用 LangChain Agent。

适合：

搜索和问答；
多工具查询；
简单客服；
低风险自动化；
工具数量有限且循环拓扑稳定的任务。

需要显式流程、持久化或人工介入

分类 → 分支 → Agent → 校验 → 审批 → 执行 → 对账

使用 LangGraph，必要时把 LangChain Agent 作为其中一个节点或子图。

适合：

审批；
订单和支付；
长时间运行任务；
可恢复执行；
多 Agent 编排；
并行分支；
需要审计和明确故障路径的系统。

不应使用 Agent 的场景

以下问题通常更适合普通程序：

输入格式严格固定；
业务规则完全确定；
只有一个已知 API 调用；
失败必须精确可预测；
不允许模型自行决定下一步；
任务涉及高风险副作用但不具备审批、幂等和审计能力。

Agent 的价值在于处理不确定性；如果流程本身完全确定，引入 Agent 只会增加延迟、成本和故障面。

十二、最终边界

Model、Tool、Middleware 和 State 各自承担不同责任：

Model       ：提出下一步意图
Tool        ：执行经过校验和授权的动作
Middleware  ：拦截生命周期，实施横切控制
State       ：保存执行过程中的事实和上下文
LangGraph   ：表达可恢复、可分支、可中断的流程

一个可审计的 Agent 系统应满足以下因果链：

\text{Model Intent} \rightarrow \text{Schema Validation} \rightarrow \text{Authorization} \rightarrow \text{Tool Execution} \rightarrow \text{State Update} \rightarrow \text{Next Decision}

缺少其中任一环节，都可能产生错误边界：

没有 Schema Validation：模型参数可能无法解析；
没有 Authorization：模型可能访问越权资源；
没有 Tool Execution 隔离：模型意图可能直接变成危险副作用；
没有 State Update：下一轮模型看不到真实执行结果；
没有循环控制：Agent 可能无限调用；
没有 Checkpoint：长流程失败后只能从头开始；
没有幂等和未知状态处理：恢复或重试可能重复执行副作用。

因此，LangChain Agent 适合做“受控的不确定性处理器”，而不是业务系统的最终裁判。Model 可以决定“接下来建议做什么”，但能否做、对谁做、做几次、失败后如何恢复，必须由 Tool 执行器、State 结构和 LangGraph 流程共同约束。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。

LangChain Agent：Model、Tool、Middleware、State 与适用边界

一、先明确 Agent 的计算模型

1. Model 不是 Agent

2. Agent 循环的终止条件

二、Model：决策器、上下文消费者和工具调用协议

1. Model 的输入不是“用户问题”，而是上下文

2. 工具调用不是普通文本解析

3. Model 的适用边界

三、Tool：把模型意图转换为受控动作

1. Tool 的本质

2. Tool 的执行过程

3. Tool 参数校验与授权必须分层

4. Tool 返回值有两种用途

5. Tool 的副作用边界

四、Middleware：围绕生命周期插入控制逻辑

1. Middleware 解决什么问题

2. Node-style 与 Wrap-style 的区别

3. Middleware 的执行顺序会改变语义

4. 动态模型选择

五、State：Agent 的短期工作记忆和控制面

1. State 不是变量字典那么简单

2. State 更新不是简单覆盖

3. State、Runtime Context、Store 的区别

State：本次会话或本次流程的工作数据

Runtime Context：本次调用的依赖和静态上下文

Store：跨会话的长期数据

4. State 的完整变化示例

六、用 create_agent 组合 Model、Tool、Middleware 和 State

七、何时需要直接使用 LangGraph

1. State、Node、Edge 的关系

2. Command 同时更新状态和路由

八、Checkpoint 与中断：Agent 不再只是一次函数调用

1. 为什么 Agent 需要 Checkpoint

2. 中断的因果路径

九、并发与故障路径：模型调用成功，不代表流程成功

1. 并行 Tool Call 的问题

2. 超时后的不确定状态

3. Checkpoint 恢复与副作用重复

十、常见误解与失败表现

误解一：Agent 会自动理解所有业务规则

误解二：Middleware 可以替代业务流程

误解三：State 就是长期记忆

误解四：工具报错返回给模型，就算完成了错误处理

十一、如何判断使用哪一层

只需要一次模型调用

需要模型选择和执行工具

需要显式流程、持久化或人工介入

不应使用 Agent 的场景

十二、最终边界

系列导航与关联阅读

官方资料

六、用 `create_agent` 组合 Model、Tool、Middleware 和 State

2. `Command` 同时更新状态和路由