Skip to content

🤖 9.9 Agent 现代架构与 create_agent ​

“Chain 是按固定路线行驶的火车,而 Agent 是拥有 GPS、方向盘和随时变道能力的自动驾驶汽车。”
在 LangChain 1.4 中,Agent 的构建方式迎来了彻底的现代化变革:基于底座模型的 Tool Calling 协议与 LangGraph 运行时,实现高可靠的自主推理与工具调用闭环。


💡 先看一个直观例子:固定铁轨与自动驾驶汽车 ​

1. Chain 链与 Agent 智能体的本质区别 ​

  • Chain(固定流水线):步骤是人类写死的(先翻译 ➔ 再摘要 ➔ 再存库)。无论遇到什么输入,都必须机械执行这三步;
  • Agent(自主决策体):大模型根据用户的具体目标,自主规划“第一步调用什么工具、看结果再决定第二步调用什么工具、何时得出结论并向人类汇报”。

2. 生活比喻:高级私人秘书与他的工作草稿本 ​

  • create_agent(秘书总监):1.4 的标准入口,一条命令就把“大脑(模型)+ 双手(工具)+ 记忆(checkpointer)+ 规则(system_prompt)”装配成一位能自主办事的秘书;
  • LangGraph 状态机(秘书的办公流程):秘书的每一步(思考 → 动手 → 看结果 → 再思考)都由底层状态图自动循环驱动,无需手工写 while 循环;
  • messages 消息流水线(秘书的工作日志):每次调用的结果是一条完整消息流水线 —— HumanMessage(老板吩咐)→ AIMessage.tool_calls(秘书决定叫哪个部门)→ ToolMessage(部门回执)→ AIMessage(最终汇报)。审计只需逐条阅读这条流水线;
  • ReAct 推理循环:
    1. Thought(思考):分析现状与缺少的关键数据;
    2. Action(行动):决定调用哪个具体工具并传递参数;
    3. Observation(观察):读取工具返回的结果,合并进上下文,决定是否继续循环或输出最终答案。

2. 生活比喻:高级私人秘书与他的工作草稿本


🏛️ LangChain 1.4 Agent API 演进 ​

LangChain 官方对 Agent API 进行了大规模规范化:

API 版本使用方式状态与评价
远古时代 (<= 0.1.x)initialize_agent(agent_type="zero-shot-react-description")❌ 已废弃。基于脆弱的纯文本正则匹配,极易解析崩溃。
0.3 过渡期create_tool_calling_agent(llm, tools, prompt) + AgentExecutor⚠️ 已迁至 langchain-classic。基于模型原生 Function Calling 协议,但执行器已随 1.4 移除。
1.4 现代化新接口from langchain.agents import create_agent🌟 当前唯一官方标准。底层直接基于 LangGraph 状态机运行时构建,内置 Tool Calling 循环、错误自愈、Checkpointer 记忆与中间件。

🔑 一句话记忆:AgentExecutor 是 0.3 时代负责“循环执行”的引擎;1.4 把它彻底内化进 create_agent(LangGraph 状态机)。新版不再需要 AgentExecutor、agent_scratchpad 占位符,return_intermediate_steps 也不再是必要参数。


💻 核心实操:构建多功能现代 Tool Calling Agent ​

1. 组装工具集(@tool 或普通函数皆可) ​

python
# code/s09_modern_agent.py
import math
from langchain_core.tools import tool
from s01_model_io import get_chat_model

# 1. 准备多模工具(create_agent 也接受普通 Python 函数)
@tool
def calculate_expression(expression: str) -> str:
    """数学表达式求值计算器。支持加减乘除、乘方、括号运算。"""
    try:
        allowed = {k: v for k, v in math.__dict__.items() if not k.startswith("__")}
        return f"计算结果: {eval(expression, {'__builtins__': {}}, allowed)}"
    except Exception as e:
        return f"计算出错: {e}"

@tool
def query_weather(city: str) -> str:
    """查询指定城市的实时气象与穿衣出行建议。"""
    weather_map = {"上海": "上海今天小雨,气温 22°C ~ 26°C,建议带伞。"}
    return weather_map.get(city, f"未收录城市 {city} 的天气。")

@tool
def currency_converter(amount: float, from_curr: str, to_curr: str) -> str:
    """实时汇率转换工具 (支持 USD, CNY, EUR)。"""
    rates = {"USD": 7.25, "EUR": 7.85, "CNY": 1.0}
    cny = amount * rates[from_curr.upper()]
    target = cny / rates[to_curr.upper()]
    return f"{amount} {from_curr} = {target:.2f} {to_curr}"

2. 一行创建 1.4 标准 Agent(create_agent) ​

python
from langchain.agents import create_agent

tools = [calculate_expression, query_weather, currency_converter]
llm = get_chat_model(temperature=0.1)

# 1.4 标准姿势:无需 AgentExecutor / agent_scratchpad
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="你是一位顶级智能私人助理。涉及天气、汇率和算术时严禁盲猜,必须调用对应工具!",
)

# 2. 以消息列表方式传入复合多任务指令
query = "我打算去上海玩 3 天,查下天气;另外预定每晚 180 美元的酒店,住 3 晚折合人民币多少钱?"
response = agent.invoke({"messages": [("user", query)]})

# 3. 取最终答复(结果是一个“完整消息列表”,最后一条即最终回答)
print("最终回答:\n", response["messages"][-1].content)

3. 推理审计:从 messages 中还原工具调用链 ​

create_agent 的结果本身就是一条完整的消息流水线(用户提问 → AI 发起工具调用 → 工具返回 → AI 最终答复)。逐一打印即可审计每一步推理:

python
print("\n--- 🔍 Agent 思考与工具调用链明细 ---")
for msg in response["messages"]:
    if msg.type == "ai" and msg.tool_calls:
        for tc in msg.tool_calls:
            print(f"🧠 思考:调用工具 [{tc['name']}],参数 {tc['args']}")
    elif msg.type == "tool":
        print(f"⚙️ 工具返回: {msg.content}")
    elif msg.type == "ai":
        print(f"💬 AI: {msg.content}")

4. 流式输出与事件流 ​

python
# 逐 token 流式(打字机效果)
async for chunk in agent.astream({"messages": [("user", query)]}, stream_mode="values"):
    last = chunk["messages"][-1]
    if last.type == "ai" and last.content:
        print(last.content, end="", flush=True)

5. system_prompt 支持 SystemMessage 实例(1.1) ​

create_agent(..., system_prompt=...) 除了字符串,还能直接传 SystemMessage 对象,方便程序化组合系统指令(配合中间件动态修改系统提示时也更内聚):

python
# code/s09_modern_agent.py —— build_modern_agent()
from langchain_core.messages import SystemMessage

system_prompt = SystemMessage(
    content="""你是一位精通多领域的超级智能私人助理。
遇到任何涉及数学算术、实时天气、汇率计算的问题时,请严格调用对应的工具,绝不自行盲猜!"""
)
agent = create_agent(model=llm, tools=tools, system_prompt=system_prompt)

6. 中间件:给 Agent 一键加装生产级可靠性(1.1) ​

配合 9.7 讲到的官方预置中间件,create_agent 可直接装配 ModelRetryMiddleware(弱网/429/5xx 自动重试),不再需要手写重试逻辑:

python
# code/s09_modern_agent.py —— build_modern_agent()
from langchain.agents.middleware import ModelRetryMiddleware

agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt=system_prompt,
    middleware=[ModelRetryMiddleware(max_retries=2)],
)

7. response_format:整个 Agent 的最终答复强制结构化(1.1+) ​

9.4 讲的是“单次 LLM 调用”结构化;create_agent(..., response_format=Pydantic模型) 则让整个 Agent 跑完所有工具调用后的最终答复都强转成该 Schema —— 非常适合“对话结束后自动生成报表/单据”:

python
# code/s09_modern_agent.py —— demo_structured_response()
from pydantic import BaseModel, Field

class FinalAnswer(BaseModel):
    summary: str = Field(description="最终中文总结")
    key_points: list[str] = Field(description="3~5 条关键要点")
    used_tools: list[str] = Field(description="本次调用过的工具名列表")

agent = create_agent(model=llm, tools=[...], system_prompt="...", response_format=FinalAnswer)
res = agent.invoke({"messages": [("user", "请帮我算一下 125 × 38.5 ÷ 1.05² 的结果")]})
print(res["messages"][-1].content)   # 已符合 FinalAnswer 的 JSON

8. v3 流式协议:更细粒度的 Agent 级事件流(1.3) ​

1.3 为 create_agent 接入全新的 astream_events(..., version="v3")(链式场景仍用 9.3 的 version="v2"),事件结构更统一、含 Agent 节点级事件,适合前端实时渲染思考与工具调用过程:

python
# code/s09_modern_agent.py —— demo_stream_v3()
async for event in await agent.astream_events({"messages": [("user", "上海天气如何?")]}, version="v3"):
    if "on_chat_model_stream" in event["event"]:
        chunk = event["data"].get("chunk")
        if chunk is not None and getattr(chunk, "content", None):
            print(chunk.content, end="", flush=True)

⚠️ 注意:CompiledStateGraph.astream_events 返回的是协程,需先 await 拿到异步迭代器再遍历。

9. 中间件进阶:动态工具注册与 HITL 精细决策(1.3) ​

1.3 给中间件新增两项能力,让 Agent 更聚焦、更可控:

  • 动态工具注册:中间件可在运行中按需动态注册/卸载工具(对应 Anthropic “Agent Skills / 渐进式披露”思路),不必在创建时一次列全所有工具——上下文更省、Agent 更聚焦。
  • HITL respond 决策:HumanInTheLoopMiddleware(人类在环)新增 respond 分支决策,可精细控制“哪些工具调用需要人工确认、哪些直接放行”。
python
from langchain.agents.middleware.human_in_the_loop import HumanInTheLoopMiddleware

# 示意:仅对"删除/付款"等高危工具要求人工确认
agent = create_agent(
    model=llm,
    tools=tools,
    middleware=[HumanInTheLoopMiddleware()],
)

10. AgentState 自定义状态:给 Agent 加一本私房账本(state_schema) ​

每个 Agent 内部都维护一份 AgentState(TypedDict),内置 messages 字段(只增不改的完整对话历史)。当你想在运行中额外跟踪 user_id、调用次数、业务计数器时,只需子类化 AgentState 并通过 state_schema= 注入:

python
# code/s09_modern_agent.py —— demo_custom_state()
from langchain.agents import AgentState, create_agent
from typing_extensions import NotRequired

class MyState(AgentState):
    user_id: NotRequired[str]      # NotRequired:可选自定义字段
    call_count: NotRequired[int]

agent = create_agent(model=llm, tools=tools, state_schema=MyState)

# invoke 时把自定义字段作为初始状态一并传入
result = agent.invoke({
    "messages": [("user", "你好,我是谁?")],
    "user_id": "user-123",
    "call_count": 0,
})

🔑 工程价值:这些自定义字段对中间件完全可见——9.11 会讲到,before_model / after_model / wrap_model_call 的 state 参数里都能直接 state["user_id"]、state["call_count"],从而实现"按调用次数限流熔断""按用户等级动态切换模型/工具"等高级控制。AgentState 也是所有中间件钩子的标准签名。

11. 人类在环 HITL:高危操作必须签字画押(完整流程) ​

对"发邮件、删库、付款"等高危工具,HumanInTheLoopMiddleware 能让 Agent 在执行前暂停,等人类批准后才继续。完整姿势 = interrupt_on(声明哪些工具要审批)+ checkpointer(中断后能恢复)+ Command(resume=...)(人工批准/拒绝):

python
# code/s09_modern_agent.py —— demo_hitl()
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model=llm,
    tools=[search, send_email, delete_database],
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"send_email": True, "delete_database": True},  # 高危工具需审批
    )],
    checkpointer=InMemorySaver(),   # 必须有 checkpointer,中断状态才能持久化
)

config = {"configurable": {"thread_id": "user_1"}}
# 第一次调用:Agent 规划到 send_email 时自动暂停,返回中断信息等待人类
result = agent.invoke({"messages": [("user", "帮我给团队发一封会议通知邮件")]}, config=config)

# 人类审阅后,用同一 thread_id 恢复执行(approve 放行 / reject 拒绝)
result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
)

💡 三句话记忆:interrupt_on 是"审批白名单";checkpointer + thread_id 是"中断后的存档点";Command(resume=...) 是"签字后的放行条"。

12. 🤿 deepagents:站在 create_agent 肩膀上的"深度智能体"整机 ​

如果说 create_agent 给你的是一台裸机发动机(模型 + 工具 + 循环),那官方另一开源库 deepagents(pip install deepagents,本书实测 0.7.14)就是整车厂:它把第九章散装的中件间预装成一套"深度工作体",专治"规划不清、上下文爆炸、单兵作战"三大癌症。它的定位类似 Claude Code 这类编码 Agent 的通用版——虚拟文件系统 + 任务规划 + 子智能体委派 + 长期记忆 + 技能库五件套开箱即用。

一个生活比喻:create_agent 是一个能干的店员;deepagents 是一家带后厨、储藏室和分店调度台的总店——店员(主 Agent)可以把大活写进工单(write_todos)、去储藏室(虚拟文件系统)翻资料存档、把专项任务外包给分店(task 子智能体),还能把老顾客的偏好记在门店记忆本(AGENTS.md 式 Memory)里。

五大能力底层实现(全是 9.7/9.11 学过的中间件!)提供的工具
虚拟文件系统FilesystemMiddleware + 可插拔 Backendls / read_file / write_file / edit_file / glob / grep / delete(0.7+)/ execute(沙箱后端)
任务规划TodoListMiddleware(0.7 起默认关闭,按需开启)write_todos
子智能体委派SubAgentMiddleware(内置 general-purpose 通用子代理)task(子代理新上下文、无状态、只交回一份最终报告)
长期记忆MemoryMiddleware(memory=["..."] 指定记忆文件)复用 read_file/edit_file 读写 AGENTS.md 式记忆
技能库SkillsMiddleware(渐进式披露:先看简介,用到再读全文)SKILL.md 标准技能目录

最速上手(与 create_agent 参数几乎同构,零迁移成本):

python
# pip install deepagents   (要求 Python >=3.11,本书实测 0.7.14)
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",                 # 或传入 BaseChatModel 实例
    tools=[my_search_tool],                 # 你的领域工具
    system_prompt="你是一名深度研究员……",
    # subagents=[{"name": "researcher", "description": "…", "system_prompt": "…"}],  # 自定义子代理
    # memory=["/memories/AGENTS.md"],       # 跨会话长期记忆
    # checkpointer=..., interrupt_on=...,   # 与 create_agent 完全同款:存档 & 高危审批
)

result = agent.invoke({"messages": [("user", "调研 X 并写成报告")]})

后端(Backend)是 0.7 版的重头戏——文件系统往哪存,全靠注入:

Backend文件存哪适合场景
StateBackend(默认)LangGraph 状态里,会话结束即散无状态演示 / 隔离测试
FilesystemBackend真实本地磁盘目录本机文件整理、代码任务
StoreBackendLangGraph Store(9.6 讲过)跨会话持久化、多用户隔离
CompositeBackend路由组合:如 /memories/ 走 Store、其余走磁盘生产标配:记忆长期留存 + 工作区落盘
LangSmithSandbox 等沙箱执行环境需要跑 execute shell 命令时

1. 生产组合模式:一个 CompositeBackend 把文件"分家安放"

CompositeBackend(default, routes) 按路径前缀把不同文件路由到不同存储——就像公司前台分诊:员工的临时草稿放工位抽屉(StateBackend,随会话消亡),正式档案放档案室(StoreBackend,跨会话永久留存),项目代码放机柜(FilesystemBackend,落盘):

python
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend, FilesystemBackend
from langgraph.store.memory import InMemoryStore

agent = create_deep_agent(
    model=llm,
    backend=CompositeBackend(
        default=StateBackend(),                                     # 其余路径:会话草稿(临时)
        routes={
            "/memories/": StoreBackend(                              # 记忆:跨会话持久
                namespace=lambda rt: (rt.server_info.user.identity,) # 按用户隔离,Alice/Bob 各看各的
            ),
            "/workspace/": FilesystemBackend(                        # 工作区:真实磁盘
                root_dir="/path/to/project", virtual_mode=True       # virtual_mode=True 才有路径越界防护!
            ),
        },
    ),
    store=InMemoryStore(),   # 注意:Store 传给 create_deep_agent,不是传给 backend
)

⚠️ 安全提示(官方文档原话警告):FilesystemBackend 默认 virtual_mode=False 没有任何路径防护,即使设了 root_dir 也能被路径穿越写出目录——生产务必 virtual_mode=True。

2. 权限系统:FilesystemPermission 三模式

每条规则声明"对哪些路径的读/写操作,采取哪种处置",在调用后端之前拦截。三种模式对应三种管理风格:

mode行为生活比喻
allow放行工牌刷开门禁
deny直接拒绝(结构化错误,不抛异常)红线区域,刷卡也不开
interrupt暂停等人批准(衔接 9.9 的 HITL 流程)要动档案室?先找领导签字
python
from deepagents import FilesystemPermission

agent = create_deep_agent(
    ...,
    permissions=[
        FilesystemPermission(operations=["write"], paths=["/policies/**"], mode="deny"),
        FilesystemPermission(operations=["write"], paths=["/memories/AGENTS.md"], mode="interrupt"),
    ],
)

3. 子代理(SubAgent):task 工具的"外包单"怎么写

主 Agent 通过 task(subagent_type, description) 把活外包,中间过程发生在子代理的新上下文里,只有最终报告回到主 Agent——这就是官方说的"上下文隔离(context quarantine)":脏活累活的几千条中间消息,不会污染主管的视野。SubAgent 字典的 11 个字段里,工程上最关键的是这几个(0.7.14 本地实测):

python
researcher = {
    "name": "research-agent",            # 必填:task(subagent_type=...) 按名派单
    "description": "用于深度调研问题",    # 必填:主 Agent 靠它决定"何时外包"
    "system_prompt": "你是资深研究员……",  # isolated 模式必填(绝不继承父提示)
    "tools": [internet_search],          # 不填=继承父工具;填了=完全替换
    "model": "openai:gpt-5.5",           # 不填=继承父模型;可给子代理单独配模型
    "response_format": Findings,         # 可选:Pydantic 结构化返回(对接 9.4)
    "mode": "isolated",                  # 默认 isolated;或 "fork"(0.7.13+ beta)
}

agent = create_deep_agent(model=llm, subagents=[researcher])

两种模式的取舍是面试高频题:

模式子代理看到什么适合
isolated(默认)只看到任务描述,全新开局独立专项:调研、翻译、批量处理
fork(beta)继承父对话全部历史与系统提示,"接着说"续写父任务:父刚扫完全库,fork 接着写评审意见

💡 还有一个永不下岗的 general-purpose 通用子代理(自带文件系统工具)——不派单也能用;要封装现成的 LangGraph 图当子代理,用 CompiledSubAgent(name=..., description=..., runnable=graph.compile())。别找"删除内置子代理"的参数:官方明确 excluded_middleware 里列 SubAgentMiddleware 会直接抛 ValueError。

4. 长期记忆:memory=["..."] + AGENTS.md 文件

deepagents 的记忆是文件化的——记忆就是一个 Markdown 文件,启动时经 MemoryMiddleware 注入系统提示,Agent 学到新东西就自己调 edit_file 更新它(类 Claude Code 的 CLAUDE.md 机制):

python
agent = create_deep_agent(
    model=llm,
    memory=["/memories/AGENTS.md"],   # 声明记忆文件路径
    backend=CompositeBackend(default=StateBackend(),
                             routes={"/memories/": StoreBackend(namespace=...)}),
)

三种作用域通过 Store 的 namespace 设计实现(对接 9.6 的"记忆三数据源"):namespace=(assistant_id,) Agent 级(一个共享人格)、namespace=(user_identity,) 用户级(Alice/Bob 记忆互不可见)、组织级建议设为只读(防共享记忆被注入恶意指令——官方的安全建议,与 9.12 的 Prompt 注入防线呼应)。

🔑 学习路径闭环:deepagents 没有任何"新魔法"——文件系统是 FilesystemMiddleware、摘要瘦身是 SummarizationMiddleware、子代理是 SubAgentMiddleware、记忆注入是 MemoryMiddleware,全部构建在第九章的 AgentMiddleware 体系之上。看懂 9.7~9.11,你就具备了给 deepagents 这类"整机"拆机改造的能力(源码仓库:github.com/langchain-ai/deepagents)。

💡 旧代码迁移对照(0.3 → 1.4):

0.3 写法1.4 写法
from langchain.agents import create_tool_calling_agent, AgentExecutorfrom langchain.agents import create_agent
agent = create_tool_calling_agent(llm, tools, prompt)agent = create_agent(model=llm, tools=tools, system_prompt=...)
agent_executor = AgentExecutor(agent=agent, tools=tools, ...)不再需要;create_agent 即完整可调用对象
agent_executor.invoke({"input": query})agent.invoke({"messages": [("user", query)]})
response["output"]response["messages"][-1].content
response["intermediate_steps"]遍历 response["messages"] 中的 tool_calls / tool 消息
MessagesPlaceholder(variable_name="agent_scratchpad")由运行时自动维护,无需占位符

📚 权威官方资料直达 ​


Agent 循环需要明确退出路径 ​

create_agent 帮我们组装循环,但任务边界仍由开发者定义。最大步数、工具连续失败、用户中断、等待审批和达到验收条件,都应有对应的退出状态。否则系统可能不断调用工具,或在尚未得到结果时提前返回。

调试时不要只看最终消息。应检查模型为何选择某个工具、参数从哪里来、工具返回了什么、状态怎样变化。涉及写操作时,再增加人类在环和可恢复设计,确保中断后不会留下无法判断的半完成状态。

🎯 本节小结与思考 ​

  1. 核心收获:掌握了 LangChain 1.4 标准 Agent 入口 create_agent(model + tools + system_prompt + 可选 checkpointer / middleware / response_format / state_schema),学会了通过 messages 消息流水线进行工具调用链推理审计,掌握了 AgentState 自定义状态与 HITL 人类在环的完整审批流程,认识了基于同一中间件体系的"深度智能体整机" deepagents(虚拟文件系统 / 任务规划 / 子代理 / 记忆 / 技能五件套),以及 0.3 → 1.4 的完整迁移对照。
  2. 下一步探索:现在我们已经掌握了 LangChain 1.4 的全部核心零件!下一节我们将继续深入上下文工程与动态上下文注入——如何给 Agent 在正确时机递上正确的上下文。