智能体编排框架横评（1）：LangChain 1.x

智能体编排框架横评 第 1 站 · 上一站：总览 · 下一站：LangGraph · 全程：总览

LangChain 是什么

LangChain 是一个用来开发大模型应用的 Python 库（也有 JS 版）。它做两件事：

1. 统一各家模型的接口。 OpenAI、Anthropic、Google 等厂商的 API 格式各不相同，LangChain 把它们包成同一套调用方式，换模型时业务代码基本不用改。
2. 提供一个现成的 Agent。 用 create_agent 传入模型和工具，就得到一个会自己决定调哪个工具、调完再看结果、直到能回答为止的 Agent。

create_agent 内部跑的是一个固定的循环：

1. 把对话发给模型；
2. 模型如果要求调用工具，就执行工具，把结果加进对话，回到第 1 步；
3. 模型不再要求调用工具，就把它的回答返回给你。

这个循环你不能改形状，但可以通过中间件在循环的固定位置插入自己的逻辑，比如调用工具前先让人确认、对话太长时自动摘要。

LangChain 和下一站 LangGraph 的关系：create_agent 编译出来就是一张 LangGraph 图，记忆、暂停、恢复这些能力都来自 LangGraph。只需要"模型加工具"时用 LangChain；需要自己规定流程的分支和并行时，到 智能体编排框架横评（2）：LangGraph 直接写图。

安装

pip install -U langchain langchain-openai

langchain 是主包，langchain-openai 是 OpenAI 模型的集成包。用别家模型就装对应的包，比如 langchain-anthropic、langchain-google-genai。

然后把 API key 设成环境变量：

export OPENAI_API_KEY="sk-..."

本文的代码按 LangChain 1.x 的 Python 官方文档写，是示意代码，没有运行过。

第一个 Agent

写一个能查报销制度的 Agent。

first_agent.py 文件代码：

from langchain.agents import create_agent

def search_policy(query: str) -> str:
    """查报销制度条款"""
    return "市内打车单次标准 80 元"          # 示意，真实情况查数据库或知识库

agent = create_agent(
    model="openai:gpt-5.5",                # "厂商:模型名"，可以换成别的模型
    tools=[search_policy],
    system_prompt="你是公司的报销助手。",
)

result = agent.invoke({"messages": [{"role": "user", "content": "打车报销标准是多少？"}]})
print(result["messages"][-1].content)

输出结果类似：

根据公司制度，市内打车单次报销标准是 80 元。

这次运行里，模型先发现需要查制度，调用了 search_policy，拿到"80 元"后再组织成回答。result["messages"] 里保存了完整过程：用户的问题、模型的工具调用、工具结果、模型的最终回答，最后一条就是回答。

基本概念

模型

model 参数可以直接写字符串 "厂商:模型名"，也可以先创建一个模型对象再传进去，方便设置温度等参数：

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", temperature=0)
agent = create_agent(model=model, tools=[search_policy])

模型对象也能单独使用，不经过 Agent：

print(model.invoke("用一句话介绍你自己").content)

工具

工具就是 Python 函数。模型靠两样东西决定什么时候调用、怎么传参数：函数的 docstring（当作工具说明）和参数的类型提示（当作参数格式）。所以这两样都要写清楚。

可以直接传函数，也可以用 @tool 装饰器，后者能自定义工具名和说明：

from langchain.tools import tool

@tool
def get_invoice(invoice_id: str) -> dict:
    """按发票号查金额和报销标准"""
    return {"id": invoice_id, "amount": 118.5, "limit": 80}

print(get_invoice.name)          # get_invoice
print(get_invoice.description)   # 按发票号查金额和报销标准

消息和状态

Agent 的输入输出都是一个字典，核心字段是 messages，也就是对话记录。消息有几种角色：

角色	类	内容
用户	HumanMessage	用户说的话
模型	AIMessage	模型的回答，或者它要调用的工具（在 .tool_calls 里）
工具	ToolMessage	工具的执行结果，用 tool_call_id 对应到是哪一次调用
系统	SystemMessage	系统提示词

传入时可以写成 {"role": "user", "content": "..."} 这样的字典，LangChain 会自动转换。想看 Agent 一路做了什么，把消息挨个打印出来：

for m in result["messages"]:
    m.pretty_print()

结构化输出

有时你要的不是一段文字，而是一个能被程序直接使用的结果。用 response_format 传一个 Pydantic 模型：

from pydantic import BaseModel

class Verdict(BaseModel):
    reimbursable: bool     # 能不能报
    over_amount: float     # 超标多少元
    reason: str

agent = create_agent("openai:gpt-5.5", tools=[search_policy, get_invoice], response_format=Verdict)
r = agent.invoke({"messages": [{"role": "user", "content": "INV-102 能报吗？"}]})
print(r["structured_response"])

输出结果类似：

reimbursable=True over_amount=38.5 reason='超出标准 38.5 元，需要说明原因后提交'

模型支持原生结构化输出时，LangChain 直接用模型的功能；不支持时，它把格式包装成一个特殊工具让模型调用。两种情况你的写法一样。

记忆：checkpointer 和 thread_id

默认情况下每次 invoke 都是全新的对话，Agent 不记得上一次说了什么。要让它记住，编译时传一个 checkpointer，调用时用 thread_id 区分不同的对话：

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent("openai:gpt-5.5", tools=[search_policy], checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "emp-001"}}

agent.invoke({"messages": [{"role": "user", "content": "我叫小王"}]}, cfg)
r = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, cfg)
print(r["messages"][-1].content)   # 你叫小王

InMemorySaver 存在内存里，进程退出就没了。上线时换成存数据库的版本，比如 PostgresSaver。checkpointer 也是后面"暂停等人确认"的前提。

跨对话的长期数据（比如员工的偏好）用 store= 参数，见 智能体编排框架横评（2）：LangGraph 的 Store 一节。

中间件

中间件是挂在 Agent 循环上的钩子。LangChain 内置了一批常用的：

中间件	作用
HumanInTheLoopMiddleware	指定的工具执行前暂停，等人批准
SummarizationMiddleware	对话太长时把旧消息压缩成摘要
ModelCallLimitMiddleware	限制调用模型的次数，防止死循环
ToolCallLimitMiddleware	限制调用工具的次数
ModelRetryMiddleware / ToolRetryMiddleware	模型或工具出错时自动重试
PIIMiddleware	检测并处理手机号、邮箱这类个人信息

用法是把它们放进 middleware 列表：

from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    "openai:gpt-5.5",
    tools=[search_policy],
    middleware=[ModelCallLimitMiddleware(run_limit=10)],   # 一次运行最多调 10 次模型
)

也可以自己写。最简单的是用装饰器，下面这个在每次调用模型前打印消息条数：

from langchain.agents.middleware import before_model

@before_model
def log_size(state, runtime):
    print(f"调用模型前，共 {len(state['messages'])} 条消息")
    return None            # 返回字典会合并进状态，返回 None 表示不改

agent = create_agent("openai:gpt-5.5", tools=[search_policy], middleware=[log_size])

钩子一共有六个位置：

• before_agent / after_agent：一次运行开始前、结束后；
• before_model / after_model：每次调用模型前、后；
• wrap_model_call / wrap_tool_call：把一次模型调用或工具调用包起来，可以在前后加逻辑、重试，或者干脆不调用。

动手：一步步写报销 Agent

场景见 主线：一个报销请求走完整个 Agent 系统：员工问"上周打车超标还能报吗，能就帮我提单"。三个工具里只有 create_ticket 会真正改数据，所以只有它需要员工确认。

第 1 步：准备三个工具

from langchain.tools import tool

@tool
def search_policy(query: str) -> str:
    """查报销制度条款"""
    return "市内打车单次标准 80 元"

@tool
def get_invoice(invoice_id: str) -> dict:
    """按发票号查金额和报销标准"""
    return {"id": invoice_id, "amount": 118.5, "limit": 80}

@tool
def create_ticket(invoice_id: str, reason: str) -> str:
    """提交报销单，会真正建单"""
    return "R-55821"

第 2 步：创建 Agent，给建单加确认

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    "openai:gpt-5.5",
    tools=[search_policy, get_invoice, create_ticket],
    system_prompt="先查制度和发票，再判断能不能报。",
    middleware=[HumanInTheLoopMiddleware(interrupt_on={
        "create_ticket": {"allowed_decisions": ["approve", "reject"]},
    })],
    checkpointer=InMemorySaver(),      # 暂停要靠它保存状态，必须有
)

interrupt_on 列出要拦的工具，没列出的工具照常执行。allowed_decisions 规定员工能做哪些选择，这里只给"批准"和"拒绝"，不允许员工改金额。

为什么不在提示词里写"建单前先问员工"？提示词只是建议，模型可能不照做。中间件是在代码里拦截的：模型提出调用 create_ticket 之后、工具真正执行之前，Agent 一定会停下来。

第 3 步：运行，停在建单前

cfg = {"configurable": {"thread_id": "emp-001"}}
r = agent.invoke(
    {"messages": [{"role": "user", "content": "上周打车超标还能报吗，能就帮我提单"}]},
    config=cfg, version="v2",
)
print(r.interrupts)

version="v2" 让返回值带一个 .interrupts 属性。模型查完制度和发票、决定建单时，Agent 暂停，.interrupts 里是这次待批准的工具调用和参数，把它展示给员工。

第 4 步：员工确认后恢复

from langgraph.types import Command

agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=cfg, version="v2")

用同一个 thread_id 恢复，Agent 执行 create_ticket，再让模型组织回复。恢复可以发生在几分钟之后，也可以在另一个进程里，只要 checkpointer 用的是持久化的存储。

第 5 步：员工拒绝

agent.invoke(
    Command(resume={"decisions": [{"type": "reject", "message": "这张发票我自己再核对一下"}]}),
    config=cfg, version="v2",
)

拒绝时工具不执行，模型会收到拒绝原因，再决定怎么回复员工。

下面这张图画的是 create_agent 的一圈循环和各个钩子的位置。点开 HumanInTheLoopMiddleware 可以看到建单调用怎么被拦下、员工的决定怎么送回循环。

进阶

人在回路的更多选项

allowed_decisions 一共有四种：

决定	效果
approve	按原参数执行
edit	改参数后执行
reject	不执行，把原因告诉模型
respond	不执行，把员工写的一段文字当成工具结果交给模型

respond 要小心：模型会把那段文字当作工具成功执行的结果。所以不能用 respond 来拒绝建单，否则模型会以为单子已经建好了。

还有两个常用参数：

• when：一个判断函数，按参数决定要不要拦，比如金额超过 500 元才要审批。恢复时它会重新执行一次，所以结果必须固定，不能依赖时间、随机数这类会变的东西。
• interrupt_mode="per_call"：模型一次提出多个待审批的调用时，每个调用单独暂停、单独批复。默认是 batched，一次批复全部。

中间件的执行顺序

middleware 列表里的顺序有意义：

• before_* 按列表顺序执行；
• after_* 按列表倒序执行；
• wrap_* 层层嵌套，列表第一个在最外层，包住后面所有的。

官方建议 HumanInTheLoopMiddleware 排在 ToolRetryMiddleware 之类的工具中间件前面，否则会出现先重试、后审批的错位。

中间件是跑在 LangGraph 图里的，所以暂停恢复时，排在人在回路前面的中间件会再执行一遍。这些中间件在调用下一层之前不要有副作用。

运行时上下文

用户 ID、数据库连接这类不属于对话、但工具要用的东西，不要塞进消息里，用 context 传：

from dataclasses import dataclass

@dataclass
class Ctx:
    user_id: str

agent = create_agent("openai:gpt-5.5", tools=[...], context_schema=Ctx)
agent.invoke({"messages": [...]}, context=Ctx(user_id="emp-001"))

工具和中间件可以通过 runtime 读到它。

多个 Agent 怎么组合

官方文档列了五种模式：

• Subagents：主 Agent 把子 Agent 包成工具来调用，最常用；
• Handoffs：通过工具调用切换当前由哪个 Agent 回答；
• Skills：按需加载额外的提示词和知识；
• Router：先分类，再分发给不同的 Agent；
• Custom workflow：用 LangGraph 自己写流程。

官方也提醒，很多场景其实一个 Agent 就够了。和其他框架的比较见 智能体编排框架横评（4）：多智能体框架。

观察 Agent 在做什么

接 LangSmith 只需设两个环境变量：

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="..."

之后每次运行的模型调用、工具调用、耗时都能在 LangSmith 网页上看到。不想用 LangSmith，可以装 langsmith[otel] 走 OpenTelemetry，送到自己的追踪系统；Langfuse 这类开源工具通过 callbacks 接入，写法是 config={"callbacks": [CallbackHandler()]}。该看哪些信号见 可观测性：追踪一条任务的代价。

接 MCP 工具

MCP 是一种让工具和 Agent 解耦的协议，见 技能、MCP 与集成。LangChain 1.4 起内置了 MCP 支持，装 pip install "langchain[mcp]" 后用 from langchain.mcp import MCPAdapter，它把 MCP 服务器上的工具转成普通的 LangChain 工具，再传给 create_agent。这部分目前还是 beta，导入时会有警告。

常见问题

模型不调用工具，或者参数传错。 先检查 docstring 和类型提示。工具说明写得含糊，模型就不知道什么时候该用。

Agent 停不下来，一直调工具。 加 ModelCallLimitMiddleware 或 ToolCallLimitMiddleware 设上限，它们默认都不限制次数。

工具抛异常，整个运行就中断了。 想让模型看到错误信息自己换个办法，加 ToolErrorMiddleware，它把异常转成一条错误 ToolMessage 交给模型。

人在回路不生效。 检查有没有传 checkpointer，恢复时 thread_id 是否和第一次一样。

审批之后单子建了两张。 审批只管"要不要执行"，不管"执行了几次"。重试、恢复都可能让工具再跑一次，create_ticket 自己要带幂等键，见 可靠执行：状态、幂等与恢复。

照着旧教程写，报 AgentExecutor 找不到。 见下面的版本说明。

版本说明

LangChain 1.0 在 2025-10-22 发布，本文按 2026-11 时的 1.4 系列文档写。1.0 是一次大调整，网上很多教程还是 0.x 的写法，对照时注意：

• 旧的各种 chain 和 AgentExecutor、initialize_agent 都移到了 langchain-classic 包，旧代码装这个包、改 import 还能跑。新代码统一用 create_agent。
• LangGraph 里原来的 create_react_agent 就是现在的 create_agent，参数 prompt= 改名 system_prompt=，pre_model_hook 这类钩子改成了中间件。
• 自定义状态只支持 TypedDict，Python 最低 3.10。
• MCP 以前要装独立的 langchain-mcp-adapters，1.4 起并进了主包。

JS 版的用法基本一致，名字改成驼峰（createAgent、humanInTheLoopMiddleware），schema 用 zod。

和手写循环的对照

跟读 learn-claude-code（1）：循环内核 里 s01 的循环只有一个出口：模型这一轮没有要求调工具就停。create_agent 的循环骨架和它一样，差别在骨架外面。

手写 harness（s01 到 s04）	LangChain 1.x
while 循环，没有 tool_use 就停	create_agent 的 model 节点和 tools 节点
tool_use 和 tool_result 按 id 配对	AIMessage.tool_calls 和 ToolMessage 按 tool_call_id 配对，框架代做
工具出错返回字符串给模型	需要加 ToolErrorMiddleware
UserPromptSubmit 钩子	before_agent
PreToolUse 加三道闸门	wrap_tool_call 前半段，或 HumanInTheLoopMiddleware
PostToolUse 只能看	wrap_tool_call 后半段，可以改写结果
Stop 钩子否决结束	after_model 返回 jump_to
压缩	SummarizationMiddleware
没有检查点	checkpointer 加 thread_id，来自 LangGraph

几个值得对照的地方：

钩子的组合方式不同。 s04 的钩子是平铺列表，遇到第一个有返回值的就停。LangChain 的 before_*、after_* 返回字典合并进状态，不会提前停；wrap_* 是一层包一层，调用下一层之前相当于 PreToolUse，之后相当于 PostToolUse，不调用就等于拦截。

Stop 的方向相反。 s04 的 Stop 钩子在模型想结束时触发，可以否决结束，让它继续干活。LangChain 的 after_model 加 jump_to 常用来提前跳到结束，也可以跳回 model 重来，用后者能做出类似 s04 否决结束的效果。目标闸门见 固定编排与目标闸门：Workflow 运行时和 Goal Loop。

权限只覆盖一部分。 s03 有三道闸门：硬拒绝、规则、问用户。人在回路中间件只对应第三道，前两道要自己写成 wrap_tool_call 中间件，或者用 when 表达规则。

业务判断仍然在你手上。 工具说明写得好不好，哪个工具算有副作用，审批界面展示什么，超标判断写在工具里还是交给模型，框架都不管。工具调用的标准形态见 工具调用。

自测

create_agent 内部的循环是怎么走的？什么时候结束？

参考：把对话发给模型；模型要求调工具就执行工具、把结果加进对话再发给模型；模型不再要求调工具时，把它的回答返回。

模型靠什么决定调用哪个工具、传什么参数？

参考：工具名、docstring（工具说明）和参数的类型提示。这几样写不清楚，模型就容易不调或者调错。

想让 Agent 记住上一轮对话，要做哪两件事？

参考：创建时传 checkpointer（比如 InMemorySaver），调用时在 config 里给同一个 thread_id。

报销例子里，为什么建单前的确认要交给 HumanInTheLoopMiddleware，而不是写在 system_prompt 里？为什么不开 respond？

参考：提示词只是建议，模型可能直接调用 create_ticket；中间件在模型提出调用之后、工具执行之前暂停，拦截发生在代码里。respond 的文字会被当成工具成功的结果交给模型，拿它拒绝建单，模型会以为单子已经建好了。

中间件列表是 [ToolRetryMiddleware(...), HumanInTheLoopMiddleware(...)]，会出什么问题？

参考：wrap_* 按列表嵌套，第一个在最外层，于是重试包在审批外面，和官方建议的顺序相反。应该把人在回路排在前面。另外恢复时排在人在回路前面的中间件会再执行一遍，它们在调用下一层之前不能有副作用。

照着 0.x 的教程写了 AgentExecutor，在 1.x 里怎么办？

参考：装 langchain-classic 并改 import 可以继续用；新代码改用 langchain.agents.create_agent，中间件等新功能只在它上面。

参考资料

以下链接均查阅于 2026-11。

• LangChain overview
• Quickstart
• Agents
• Models
• Tools
• Messages
• Structured output
• Short-term memory
• Middleware overview
• Built-in middleware
• Custom middleware
• Human-in-the-loop
• Multi-agent
• MCP
• Observability
• What's new in LangChain v1
• LangChain v1 migration guide
• Langfuse LangChain integration

连接

• 智能体编排框架横评：同一个报销 Agent 换六种写法：系列总览，本站是第 1 站。
• 智能体编排框架横评（2）：LangGraph：下一站，create_agent 的记忆、暂停、恢复都来自 LangGraph，要自己规定流程也到那里写图。
• 智能体编排框架横评（4）：多智能体框架：LangChain 的 subagents、handoffs 等模式和专门的多智能体框架放在一起比。
• Agent 核心：面试版的 LangChain 与 LangGraph 问答。
• 主线：一个报销请求走完整个 Agent 系统：报销例子的场景和示意数字来源。
• 跟读 learn-claude-code（1）：循环内核：手写循环和钩子，本站"和手写循环的对照"的另一边。
• 固定编排与目标闸门：Workflow 运行时和 Goal Loop：Stop 否决和目标闸门，对应 after_model 跳回模型的用法。
• 技能、MCP 与集成：手写 harness 里 MCP 怎么接。
• 人在回路：审批该展示什么、什么操作该拦，本站只讲中间件怎么挂。
• 工具调用：tool_calls 和 ToolMessage 的配对就是工具调用的标准形态。
• 可靠执行：状态、幂等与恢复：步数上限、错误交回模型、副作用工具的幂等，框架只给挂载点。
• 可观测性：追踪一条任务的代价：LangSmith、OTel、callbacks 之外，观测要看哪些信号。
• 记忆：checkpointer 管短期记忆，store 管长期记忆。