lang

1. 本章目标 #

第 2、9 章里的 create_agent，默认只有一个 Agent：一份系统提示、一套工具、一轮「模型 ↔ 工具」循环。工具越来越多、业务越来越深时，很多人第一反应就是「上多 Agent」。官方文档一开头却提醒你别急：

不是每件复杂任务都需要多 Agent。提示写对、工具选对（必要时再动态加载），一个单 Agent 往往就够了。

这句话要当真。多 Agent 换来的不是「更聪明」，而是三种工程上的能力：

你真正缺的	只用单 Agent 时	多 Agent 能帮你做什么
管住上下文	所有领域文档塞进同一份提示，窗口爆、费用高、选型乱	每次只让相关的那位专家看见自己的资料
分团队维护	日历、邮箱、CRM 的提示和工具挤在同一个文件里	每个专家独立演进，主 Agent 只认工具名
并行干活	一个模型把所有子任务串行想完	多个专家同时跑，主 Agent / 路由器负责汇总

这和「再叠一个人格」不是一回事。换人格，改系统提示就行；隔离上下文、隔离代码边界、隔离执行，才需要专门的模式。

本章对照官方 Multi-agent，用四种常用模式把机制亲手跑通：

先判断该不该拆；再按场景选子 Agent（Subagents）、交接（Handoffs）、技能（Skills）或路由器（Router）；

学完你应能：

说清「工具太多 / 领域文档太长 / 必须按步骤解锁」这三类问题，各自适合哪种模式
把一个 create_agent 包成工具，交给另一个 Agent 调用（子 Agent）
用工具返回 Command 改 current_step，再用中间件换提示和工具（交接）
用 load_skill 做渐进披露：系统提示只放目录，正文用到再加载（技能）
用结构化输出做一次分类，再分发给彼此看不见的专家（路由器）
看懂官方那张「买咖啡 / 再买一次 / 对比三门语言」的调用次数表，并能自己数一遍
加载技能时按状态解锁工具（先预注册，再过滤）
用第 23 章的 Send，把多个专家并行扇出再合成
用「启动 / 查状态 / 取结果」三件套，把长任务丢到后台
说清 Deep Agents 和本章手写模式的对应关系，而不是把它当成另一套 API

参考文档：

官方还提到一个更高层的现成架子 Deep Agents（自带子 Agent、技能、规划、虚拟文件系统）。它建立在本章这些模式之上。入门先把手写的四种跑懂，再决定要不要换成现成架子。

先预告几条和直觉相反的实测结论：

你可能以为	实际情况	见
两个工具就该拆成两个 Agent	单 Agent 一次并行调两个工具，只要 2 次模型调用	§3、§5.1
包成子 Agent 更「专业」，调用次数差不多	两个带内部工具的子 Agent，实测合计 6 次	§5.2
子 Agent 会记住上次用户说了什么	默认每次从空白上下文起步，记忆在主 Agent 这边	§5.4
`current_step: str = "triage"` 会自动有默认值	`AgentState` 是 TypedDict，注解里的默认值不会写进运行时；要用 `or "triage"`	§6.3
专家提示写了「只谈本语言」，它就真的不知道别的语言	隔离的是这次请求里塞进去的文档，不是模型预训练知识	§8.2

2. 什么时候才需要拆 #

2.1. 官方说的三种动机 #

打开 Multi-agent 索引页，「为什么要多 Agent」只列了三条：

管住上下文
假如上下文无限、延迟为零，你可以把全部制度、schema、API 说明倒进一份提示。实际上窗口有限、按 token 计费，所以只能按需露出相关知识。
分团队开发
日历组、邮箱组、CRM 组各自维护自己的 Agent，主系统只通过工具名把它们拼起来。
并行
子任务彼此独立时，同时跑几个专家，往往比一个模型串行想完更快。

三条里哪一条都不缺，就优先加工具、改提示，或用第 9 章的动态过滤。不要一上来就上多 Agent。

2.2. 三种痛点，三种拆法 #

痛点	典型症状	优先试
工具太多，选型乱飘	20 个工具挂在同一个 Agent 上，该查日历时去搜邮件	拆子 Agent，按领域切开
领域文档太长	每条业务线都有几千 token 的 schema / 制度	技能（按需加载）或子 Agent（每次只带一份）
必须按顺序解锁	没确认在保就不能走退款；没收集工单号就不能查库	交接：用状态把下一步工具藏起来
一次问题打到多个垂直领域	「对比 Python / JS / Rust」这种要并行再汇总	路由器或子 Agent

多 Agent 设计的关键是 上下文工程——决定每一位看见什么。质量不取决于你堆了几个「人格」，而取决于：相关资料有没有送到、无关资料有没有被挡在外面。

3. 五种模式一张表 #

官方把多 Agent 收成五种。前四种是常见模式，第五种是「自己画图」。

模式	谁在跟用户说话	专家之间怎么协作	状态
子 Agent（Subagents）	只有主 Agent	主 Agent 把专家当工具调用，结果回到主 Agent	子 Agent 默认无状态
交接（Handoffs）	当前激活的那一位（可以换人）	工具改一个状态变量，行为跟着变	跨轮持久
技能（Skills）	始终同一个 Agent	不换人，只按需 `load_skill` 把说明书灌进对话	技能正文留在消息历史里
路由器（Router）	路由器自己通常不聊天	先分类，再派一个或多个专家，最后合成	默认无状态
自定义工作流	由你的图决定	确定性节点和 Agent 节点混排	由图的状态说了算

能力对照

模式	分团队维护	并行	连续多跳	专家直接跟用户聊
子 Agent	最合适	最合适	最合适	弱（结果先回主 Agent）
交接	弱	弱（基本串行）	最合适	最合适
技能	最合适	中	最合适	最合适（还是同一个人）
路由器	中	最合适	弱（一次分类，不是多轮编排）	中

可以混用。例如主 Agent（子 Agent 模式）调到的某个工具内部，其实是一张 LangGraph；某个子 Agent 自己再用技能加载文档。

flowchart LR U[用户] --> M[主 Agent] M --> A[日历子 Agent] M --> B[邮箱子 Agent] A --> M B --> M M --> U

上图是子 Agent：所有路由都经过主 Agent。交接则是同一条对话里换配置；技能是同一个 Agent 去「翻说明书」；路由器是分类器先做一次选择题。

4. 怎么选：先看任务形态，再看调用次数 #

官方用三个场景，把四种模式的模型调用次数和 token 放在一张表里。

4.1. 一次请求：「帮我买咖啡」 #

只有一个领域，做完就回用户。

模式	大约几次模型调用	更合适？
子 Agent	4	多付一次：结果要回到主 Agent 再组织语言
交接 / 技能 / 路由器	3	少一跳

交接、技能、路由器更省。子 Agent 多出来的那一次，换来的是「中心控制」：由主 Agent 决定何时调用、如何汇总。

4.2. 同一会话里再说一次：「再买一杯」 #

模式	第二轮	两轮合计	原因
子 Agent	还是 4	8	子 Agent 无状态，每次走完全流程
交接	2	5	咖啡 Agent 还在激活，不必再交接
技能	2	5	说明书已经在历史里，不必再 `load_skill`
路由器	3	6	每次重新分类

有状态的模式（交接、技能）在重复请求上大约能省一半。子 Agent 每次费用稳定，换来的是强隔离。

4.3. 跨领域：「对比 Python、JS、Rust」 #

每个语言专家带着约 2000 token 的文档。

模式	大约调用	大约 token	更合适？
子 Agent	5	~9K	并行 + 隔离，每位只看见自己那 2K
交接	7+	~14K+	串行交接，历史越积越长
技能	3	~15K	调用少，但加载后后续每一跳都带着全部技能正文
路由器	5	~9K	和子 Agent 类似，分类是单独一跳

跨领域要并行、文档又长时，选子 Agent 或路由器。交接在这里最亏。技能调用次数最少，但 token 可能最高。

4.4. 选型口诀 #

你要优化的	优先
单次简单请求、重复请求	交接或技能
并行、大段领域文档	子 Agent 或路由器
任务简单、只要按需加载说明	技能
专家必须直接跟用户多轮聊，并且步骤有硬约束	交接
夹着鉴权 / 检索 / 校验这种确定步骤	自定义工作流

§5.2 会亲手数一遍子 Agent 的调用次数，用来对照这张表。

5. 子 Agent：主 Agent 把专家当工具 #

这是最推荐先学的模式，也叫监督者模式（supervisor）。

主 Agent 负责跟用户说话、决定叫谁、如何汇总。子 Agent 被包成普通工具：主 Agent 一旦调用 ask_calendar(...)，框架就去 invoke 那个子 Agent，把它的最后一句当成 ToolMessage 回灌。子 Agent 不直接回用户。

和路由器的差别：监督者本身是完整 Agent，多轮对话、多次调用专家都由它动态决定。路由器通常只是请求前的一次分类。

5.1. single.py：对照，两个工具时先别拆 #

同一句用户问题——「明天下午有会吗？邮箱里有没有关于预算的邮件？」——先用第 2 章那种单 Agent，日历和邮箱都是普通 @tool。

# 标准库：定位项目根目录的 .env
from pathlib import Path

# find_dotenv 会沿目录往上找 .env，脚本放在 _v37/ 也能加载到仓库根目录的密钥
from dotenv import find_dotenv, load_dotenv
# 创建 Agent 的官方入口，和第 2、9 章相同
from langchain.agents import create_agent
# 第 8 章的工具装饰器
from langchain.tools import tool

# 覆盖已有环境变量，保证用的是仓库根目录 .env 里的 DEEPSEEK_API_KEY
load_dotenv(find_dotenv(filename=".env", usecwd=True) or Path(__file__).resolve().parents[1] / ".env", override=True)


# 日历工具：用字典模拟，避免本章再接真实日历 API
@tool
def list_meetings(day: str) -> str:
    """按日期查询会议。day 用「今天」「明天」或 YYYY-MM-DD。"""
    # 写死两天的日程，方便对照轨迹
    table = {
        "今天": "无会议",
        "明天": "14:00-15:00 产品周会；16:00-16:30 与设计一对一",
    }
    # 查不到就明确说没有，不要让模型去编
    return table.get(day, f"{day} 无会议记录")


# 邮箱工具：同样用固定字符串模拟搜索结果
@tool
def search_inbox(keyword: str) -> str:
    """按关键词搜索收件箱，返回匹配邮件的主题和摘要。"""
    # 关键词里带「预算」才命中，其它返回空
    if "预算" in keyword:
        return "主题：Q3 预算评审；发件人：财务部；摘要：请于周五前提交部门预算草案。"
    return f"收件箱里没有与「{keyword}」匹配的邮件。"


def main() -> None:
    # 只有两个工具时，官方建议先用单 Agent，不必上多 Agent
    agent = create_agent(
        # 模型字符串和前几章保持一致，方便对照
        model="deepseek:deepseek-v4-flash",
        # 日历和邮箱都直接挂在这一个 Agent 上
        tools=[list_meetings, search_inbox],
        # 系统提示把路由说清楚，避免模型心算日程
        system_prompt=(
            "你是简洁的中文助理。查会议必须调用 list_meetings，"
            "查邮件必须调用 search_inbox，不要编造。"
        ),
    )

    # 一句里同时问会议和邮件，观察它会不会并行调两个工具
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "明天下午有会吗？邮箱里有没有关于预算的邮件？",
                }
            ]
        }
    )

    # 把完整轨迹打出来：哪一步调了哪个工具
    for i, message in enumerate(result["messages"]):
        # 消息类型名：HumanMessage / AIMessage / ToolMessage
        kind = type(message).__name__
        # 若模型发起了工具调用，把名字和参数附在后面
        calls = ""
        if getattr(message, "tool_calls", None):
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        # 正文可能较长，截断便于阅读
        content = message.content
        if isinstance(content, str) and len(content) > 240:
            content = content[:240] + "..."
        print(f"[{i}] {kind}{calls}: {content!r}")


if __name__ == "__main__":
    main()

PowerShell：

Set-Location d:\forever\docs\langrag
.venv\Scripts\python.exe -X utf8 _v37\1_single.py

输出：

[0] HumanMessage: '明天下午有会吗？邮箱里有没有关于预算的邮件？'
[1] AIMessage [('list_meetings', {'day': '明天'}), ('search_inbox', {'keyword': '预算'})]: ''
[2] ToolMessage: '14:00-15:00 产品周会；16:00-16:30 与设计一对一'
[3] ToolMessage: '主题：Q3 预算评审；发件人：财务部；摘要：请于周五前提交部门预算草案。'
[4] AIMessage: '**明天下午的会议：**\n- 14:00–15:00 产品周会\n- 16:00–16:30 与设计一对一\n\n**邮箱中的预算邮件：**\n- 主题：Q3 预算评审（来自财务部），摘要：请于周五前提交部门预算草案。\n\n需要我帮你查看会议详情或起草回复邮件吗？'

读这段轨迹：

一条 AIMessage 里并行发出了两个 tool_calls（第 8、9 章讲过）。
模型调用次数是 2：一次决定调工具，一次看结果写回复。
两个工具的实现都在当前进程里，没有隔离，也没有额外费用。

所以：工具少、文档短、没有分团队需求时，停在这里就好。 下面的拆分是为了对照「多付了几次模型调用、换来了什么」。

5.2. subagents.py：一个专家一个工具 #

把日历和邮箱各自做成 create_agent，再 @tool 包一层。主 Agent 只看见 ask_calendar / ask_email，看不见 list_meetings / search_inbox。这就是官方说的「每个专家一个工具」（tool per agent）。

为了对照 §4 的次数表，给主 Agent 和两个子 Agent 都挂上计数中间件。wrap_model_call 按「每次调模型」执行，不是按每次 invoke（第 9 章 §2.1）。

# 标准库：定位项目根目录的 .env
from pathlib import Path

# find_dotenv 会沿目录往上找 .env
from dotenv import find_dotenv, load_dotenv
# 创建 Agent；wrap_model_call 用来数这次一共调了几次模型
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest
# 第 8 章的工具装饰器
from langchain.tools import tool

# 覆盖已有环境变量
load_dotenv(find_dotenv(filename=".env", usecwd=True) or Path(__file__).resolve().parents[1] / ".env", override=True)

# 用来累计主 Agent 和两个子 Agent 各自的模型调用次数
CALLS = {"main": 0, "calendar": 0, "email": 0}


# 给指定桶加一的中间件工厂：每次真正调模型前打印一行
def make_counter(bucket: str):
    # wrap_model_call 按「每次调模型」执行，不是按每次 invoke
    @wrap_model_call
    def count_calls(request: ModelRequest, handler):
        # 对应桶加一
        CALLS[bucket] += 1
        # 打出来方便对照官方「子 Agent 一次请求大约 4 次模型调用」
        print(f"[call] {bucket} #{CALLS[bucket]}")
        # 必须把控制权交还，否则链路会断
        return handler(request)

    # 装饰器可能用函数名做标识，三个计数器要名字不同
    count_calls.__name__ = f"count_{bucket}"
    return count_calls


# ---------- 日历子 Agent 自己的工具 ----------
@tool
def list_meetings(day: str) -> str:
    """按日期查询会议。day 用「今天」「明天」或 YYYY-MM-DD。"""
    table = {
        "今天": "无会议",
        "明天": "14:00-15:00 产品周会；16:00-16:30 与设计一对一",
    }
    return table.get(day, f"{day} 无会议记录")


# ---------- 邮箱子 Agent 自己的工具 ----------
@tool
def search_inbox(keyword: str) -> str:
    """按关键词搜索收件箱，返回匹配邮件的主题和摘要。"""
    if "预算" in keyword:
        return "主题：Q3 预算评审；发件人：财务部；摘要：请于周五前提交部门预算草案。"
    return f"收件箱里没有与「{keyword}」匹配的邮件。"


# 日历专家：只看见会议工具，系统提示也只谈日程
calendar_agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[list_meetings],
    system_prompt="你是日历助手。查会议必须调用 list_meetings。用一两句话报告结果。",
    middleware=[make_counter("calendar")],
)

# 邮箱专家：只看见搜索工具
email_agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[search_inbox],
    system_prompt="你是邮箱助手。搜索必须调用 search_inbox。用一两句话报告结果。",
    middleware=[make_counter("email")],
)


def last_text(result: dict) -> str:
    """取出子 Agent 最后一条消息的正文，交给主 Agent 当工具返回值。"""
    content = result["messages"][-1].content
    # 正常情况是 str；防御一下 content block 列表
    if isinstance(content, str):
        return content
    return str(content)


# 把日历子 Agent 包成主 Agent 可调用的工具
@tool("ask_calendar", description="查询日程和会议。传入一句自然语言，例如「明天下午有哪些会」。")
def ask_calendar(query: str) -> str:
    # 子 Agent 每次从空白上下文起步，只看见这一句 query
    result = calendar_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return last_text(result)


# 把邮箱子 Agent 同样包成工具
@tool("ask_email", description="搜索收件箱。传入一句自然语言，例如「找预算相关的邮件」。")
def ask_email(query: str) -> str:
    result = email_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return last_text(result)


def main() -> None:
    # 主 Agent 不直接碰 list_meetings / search_inbox，只调度两个专家工具
    main_agent = create_agent(
        model="deepseek:deepseek-v4-flash",
        tools=[ask_calendar, ask_email],
        system_prompt=(
            "你是简洁的中文助理。"
            "查会议必须调用 ask_calendar，查邮件必须调用 ask_email。"
            "可以一次并行调用多个专家。最后用中文汇总。"
        ),
        middleware=[make_counter("main")],
    )

    result = main_agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "明天下午有会吗？邮箱里有没有关于预算的邮件？",
                }
            ]
        }
    )

    print("--- 轨迹 ---")
    for i, message in enumerate(result["messages"]):
        kind = type(message).__name__
        calls = ""
        if getattr(message, "tool_calls", None):
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        content = message.content
        if isinstance(content, str) and len(content) > 240:
            content = content[:240] + "..."
        print(f"[{i}] {kind}{calls}: {content!r}")

    # 把三个桶加起来，对照官方「单次请求大约 4 次」
    total = sum(CALLS.values())
    print(f"--- 模型调用次数 {CALLS} 合计 {total} ---")


if __name__ == "__main__":
    main()

输出（计数行在轨迹前面，因为子 Agent 在工具执行时就已经调模型了）：

[call] main #1
[call] calendar #1
[call] email #1
[call] email #2
[call] calendar #2
[call] main #2
--- 轨迹 ---
[0] HumanMessage: '明天下午有会吗？邮箱里有没有关于预算的邮件？'
[1] AIMessage [('ask_calendar', {'query': '明天下午有哪些会议'}), ('ask_email', {'query': '关于预算的邮件'})]: ''
[2] ToolMessage: '明天下午有两场会议：14:00-15:00 产品周会，16:00-16:30 与设计一对一。'
[3] ToolMessage: '找到一封关于预算的邮件：财务部发来的《Q3 预算评审》，要求周五前提交部门预算草案。'
[4] AIMessage: '为您汇总如下：\n\n**📅 明天下午的会议（2场）**\n- 14:00-15:00 产品周会\n- 16:00-16:30 与设计一对一\n\n**📧 预算相关邮件**\n- 财务部发来《Q3 预算评审》，要求**周五前**提交部门预算草案。\n\n需要我帮您进一步查看会议详情或邮件内容吗？'
--- 模型调用次数 {'main': 2, 'calendar': 2, 'email': 2} 合计 6 ---

主 Agent 的轨迹里没有 list_meetings。它只看见两个专家工具，以及专家已经组织好的中文结论。日历子 Agent 内部那次「决定调 list_meetings → 再写一句话」被压成了一条 ToolMessage。

次数为什么是 6？

本例是两个子 Agent，每个内部都有「调工具 + 写回执」：

主 Agent 并行叫两个专家      1
日历专家调工具 + 写回执      2
邮箱专家调工具 + 写回执      2
主 Agent 汇总                1
合计                         6

对照 §5.1 的 2 次：同样的用户问题，拆成子 Agent 多付了 4 次。换来的是：

日历专家的提示和工具，邮箱专家看不见；反过来也一样
主对话不会被子 Agent 的内部 tool_calls 撑爆
两个团队可以各自改自己的 create_agent，主 Agent 只依赖工具名和描述

工具少的时候这笔账不划算。领域提示变成几千 token、或者专家开始自己多跳工具时，隔离才值回票价。

5.3. dispatch.py：统一调度工具 task #

专家一多，主 Agent 为每个人各挂一个工具会变得啰嗦，而且每加一个专家都要改 tools=[...]。官方的另一种接法是统一调度：只暴露一个 task(agent_name, description)，用注册表按名字取出子 Agent。

agent_name 做成 Enum，模型填参时只能选名单里的值，比纯字符串稳。

# 标准库：枚举，用来把可调度的专家名写进工具 schema
from enum import Enum
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool

load_dotenv(find_dotenv(filename=".env", usecwd=True) or Path(__file__).resolve().parents[1] / ".env", override=True)


@tool
def list_meetings(day: str) -> str:
    """按日期查询会议。day 用「今天」「明天」或 YYYY-MM-DD。"""
    table = {
        "今天": "无会议",
        "明天": "14:00-15:00 产品周会；16:00-16:30 与设计一对一",
    }
    return table.get(day, f"{day} 无会议记录")


@tool
def search_inbox(keyword: str) -> str:
    """按关键词搜索收件箱。"""
    if "预算" in keyword:
        return "主题：Q3 预算评审；发件人：财务部；摘要：请于周五前提交部门预算草案。"
    return f"收件箱里没有与「{keyword}」匹配的邮件。"


# 两个专家仍然是独立的 create_agent，只是主 Agent 不再为每个专家各挂一个工具
calendar_agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[list_meetings],
    system_prompt="你是日历助手。查会议必须调用 list_meetings。用一两句话报告结果。",
)

email_agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[search_inbox],
    system_prompt="你是邮箱助手。搜索必须调用 search_inbox。用一两句话报告结果。",
)

# 注册表：名字 → 子 Agent。以后加专家只改这里和 Enum
SUBAGENTS = {
    "calendar": calendar_agent,
    "email": email_agent,
}


# 把可选专家名做成枚举，模型填参时只能选这两个值
class AgentName(str, Enum):
    calendar = "calendar"
    email = "email"


@tool
def task(agent_name: AgentName, description: str) -> str:
    """把一项独立任务交给指定专家。

    可选专家：
    - calendar：查日程和会议
    - email：搜索收件箱
    """
    # 按名字取出子 Agent；Enum 转 str 后就是注册表的键
    worker = SUBAGENTS[agent_name.value]
    # 子 Agent 仍然从空白上下文起步，只看见 description
    result = worker.invoke({"messages": [{"role": "user", "content": description}]})
    content = result["messages"][-1].content
    return content if isinstance(content, str) else str(content)


def main() -> None:
    # 主 Agent 只看见一个 task 工具，靠参数区分专家
    main_agent = create_agent(
        model="deepseek:deepseek-v4-flash",
        tools=[task],
        system_prompt=(
            "你是简洁的中文助理，通过 task 把工作交给专家。"
            "查会议用 agent_name=calendar，查邮件用 agent_name=email。"
            "可以一次并行调用多次 task。最后用中文汇总。"
        ),
    )

    result = main_agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "明天有哪些会？邮箱里有没有预算相关邮件？",
                }
            ]
        }
    )

    for i, message in enumerate(result["messages"]):
        kind = type(message).__name__
        calls = ""
        if getattr(message, "tool_calls", None):
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        content = message.content
        if isinstance(content, str) and len(content) > 240:
            content = content[:240] + "..."
        print(f"[{i}] {kind}{calls}: {content!r}")


if __name__ == "__main__":
    main()

输出：

[0] HumanMessage: '明天有哪些会？邮箱里有没有预算相关邮件？'
[1] AIMessage [('task', {'agent_name': 'calendar', 'description': '查询明天的所有会议安排'}), ('task', {'agent_name': 'email', 'description': '搜索收件箱中与预算相关的邮件'})]: ''
[2] ToolMessage: '明天共有2个会议：14:00-15:00 产品周会，16:00-16:30 与设计一对一。'
[3] ToolMessage: '找到一封与预算相关的邮件：主题为"Q3 预算评审"，发件人是财务部，要求周五前提交部门预算草案。'
[4] AIMessage: '为您汇总如下：\n\n**📅 明天的会议（共2个）**\n- 14:00–15:00：产品周会\n- 16:00–16:30：与设计一对一\n\n**📧 预算相关邮件**\n- 主题：**Q3 预算评审**\n- 发件人：财务部\n- 内容：要求**周五前**提交部门预算草案\n\n需要我帮您安排提醒或起草回复吗？'

一条 AIMessage 里两次 task，参数不同。这和第 8 章「一条消息多个 tool_calls」是同一件事，只是被调到的函数内部又跑了一个 Agent。

两种挂法怎么选：

	一个专家一个工具	统一 `task`
适合	专家少、每个的入参要定制	专家多、团队分散、约定大于配置
主 Agent 看见的	`ask_calendar` / `ask_email` 各自的描述	一份名单（系统提示或 Enum）
加新专家	改 `tools=[...]`	改注册表 + Enum

入门用「一个专家一个工具」，描述可以写得很具体。专家过了大约十个，再收成 task。

5.4. 设计时就要想清楚的几件事 #

官方子 Agent 页面还列了不少生产向选项（后台任务、job id、子 Agent 规格发现）。

（1）子 Agent 默认无状态。
ask_calendar 每次都 invoke({"messages": [这一句 query]})，不会自动带上用户昨天说的「我在北京」。需要历史时，由主 Agent 在 query 里写清楚，或者你显式把截断后的消息传进去。这是隔离的代价，也是隔离的收益。

（2）回给主 Agent 的只应是结论。
last_text 只取最后一条。不要把子 Agent 的完整 messages（含内部 tool_calls）回灌给主 Agent，否则主对话会迅速膨胀，模型还可能被内部推理带跑。需要过程时，在 ToolMessage 里写一段摘要。

（3）默认同步等待，入门也该用这个。
主 Agent 调用专家工具时会堵住，等子 Agent 跑完。官方另有「丢到后台、返回 job_id」的异步任务模式，那是另一套作业系统，本章不写。Python 的 async/await 和这里说的「后台任务」不是一回事。

（4）描述就是路由。
主 Agent 靠工具名和 description 决定叫谁。写「查询日程和会议」比写「日历 Agent」有用得多。这和第 8 章「模型只能看见名 / 描述 / schema」完全一致。

6. 交接：同一条对话里换配置 #

「交接」（Handoffs）这个词来自 OpenAI 的 Agents SDK：用一次工具调用表示「把控制权交给另一位」。LangChain 把它泛化成：

工具更新一个会跨轮保留的状态变量（例如 current_step），系统读这个变量，换成另一套系统提示和工具。

用户始终在跟「当前这一位」说话。和子 Agent 相反：专家直接回用户，不必每次绕回主 Agent。

适合：售后必须先问在不在保、再给方案；客服必须先拿到订单号、才能查库。步骤是硬约束，单靠提示词写「请先问保修」不够，要把下一步的工具藏起来。

官方给了两条实现路：

	单 Agent + 中间件	多 Agent 子图 + `Command.PARENT`
本质	还是一个 `create_agent`，按状态换配置	每个专家是图上的一个节点
消息怎么流	同一条 `messages`，自然连续	你必须自己决定传哪些消息，配错就历史残缺
入门	走这条	专家本身已经是复杂图时再上

下面只做第一条。第二条的消息配对（AIMessage + ToolMessage 必须一起交给下一个节点）坑很多，官方自己也说「大多数场景用中间件就够」。

6.1. 前置：工具如何改状态 #

第 11 章 §7.4.4 已经用过：工具返回 Command，在 update 里同时写槽位和一条配对的 ToolMessage。漏掉 ToolMessage，下一跳模型会看到一次没有结果的 tool_call，历史就是坏的。

交接只是把那个槽位命名成 current_step，再用第 9 章的 wrap_model_call 按槽位过滤工具、覆盖系统提示。

用户说话
    → 中间件读 current_step（没有则当分流）
    → 只把这一步的工具和提示送给模型
    → 模型若调用 record_warranty_status
    → Command 写入 current_step="specialist"
    → 同一轮 Agent 循环继续，下一次调模型已经是专家配置

跨 invoke 还要记住步骤，必须加 checkpointer，并且两次传入同一个 thread_id（第 11 章）。

6.2. handoffs.py #

售后两步：分流只允许记录保修；专家只允许给方案。用户第一句只说「坏了」，第二句才补「在保、屏幕碎了」。

# 标准库：标注 state 里可选字段
from typing import NotRequired
from pathlib import Path

from dotenv import find_dotenv, load_dotenv

# AgentState 是 create_agent 默认状态；扩展后才能放下 current_step
from langchain.agents import AgentState, create_agent

# wrap_model_call 在每次调模型前按当前步骤换提示和工具
from langchain.agents.middleware import wrap_model_call, ModelRequest

# ToolMessage 必须和模型发出的 tool_call 配对
from langchain.messages import ToolMessage

# ToolRuntime 由框架注入，用来取 tool_call_id 和当前 state
from langchain.tools import tool, ToolRuntime

# 内存 checkpointer：让 current_step 跨多轮用户消息还在
from langgraph.checkpoint.memory import InMemorySaver

# 工具返回 Command 才能同时写 state 和补 ToolMessage
from langgraph.types import Command

load_dotenv(
    find_dotenv(filename=".env", usecwd=True)
    or Path(__file__).resolve().parents[1] / ".env",
    override=True,
)


# 在默认 AgentState 上增加两个槽位：当前步骤、已记录的保修状态
class SupportState(AgentState):
    # 当前处于哪一步；第一轮可能还没有这个键，中间件里要给默认值
    current_step: NotRequired[str]
    # 用户告知的保修状态
    warranty_status: NotRequired[str]


@tool
def record_warranty_status(status: str, runtime: ToolRuntime) -> Command:
    """记录保修状态并进入下一步。status 只能是 in_warranty 或 out_of_warranty。"""
    # 归一化，避免模型偶尔带空格
    key = status.strip()
    # 返回 Command 而不是普通字符串，才能改 current_step
    return Command(
        update={
            # 写入保修槽位
            "warranty_status": key,
            # 切到专家步骤，下一轮调模型时中间件会换配置
            "current_step": "specialist",
            # 必须补上与本次 tool_call 配对的 ToolMessage，否则对话历史残缺
            "messages": [
                ToolMessage(
                    content=f"已记录保修状态：{key}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )


@tool
def provide_solution(issue: str, runtime: ToolRuntime) -> str:
    """根据已记录的保修状态给出处理方案。issue 是用户描述的故障。"""
    # 从 state 读槽位，不要让模型把保修状态再口述一遍当参数
    warranty = runtime.state.get("warranty_status") or "未知"
    if warranty == "in_warranty":
        return f"保内处理「{issue}」：带购机凭证到授权点免费维修，周期 3-5 个工作日。"
    if warranty == "out_of_warranty":
        return f"过保处理「{issue}」：走付费维修，屏幕类参考价 800-1500 元。"
    return f"尚未记录保修状态，不能给方案。当前 issue={issue}"


# 按当前步骤决定系统提示和可见工具
@wrap_model_call
def apply_step_config(request: ModelRequest, handler):
    # 第一轮还没写过槽位时，当成分流步骤
    step = request.state.get("current_step") or "routing"
    # 已记录的保修状态，专家步骤的提示里会用到
    warranty = request.state.get("warranty_status") or "尚未记录"
    # 打印出来，跑多轮时能看见配置确实换了
    print(f"[middleware] current_step={step!r} warranty={warranty!r}")

    if step == "routing":
        # 分流阶段只允许记录保修，不允许直接给方案
        system_message = (
            "你是售后分流助手。还没记录保修状态时，先问用户设备是否在保。"
            "用户明确说了在保或过保之后，必须调用 record_warranty_status，"
            "status 填 in_warranty 或 out_of_warranty。不要猜测，不要给维修方案。"
        )
        tools = [record_warranty_status]
    else:
        # 专家阶段只给方案工具
        system_message = (
            f"你是售后专家。当前保修状态：{warranty}。"
            "用户描述故障后必须调用 provide_solution，不要自己编流程。"
            "工具返回后再用一两句中文转告用户。"
        )
        tools = [provide_solution]

    # 换提示、换可见工具；
    return handler(request.override(system_message=system_message, tools=tools))


def print_trace(result: dict, title: str) -> None:
    print(f"=== {title} ===")
    for i, message in enumerate(result["messages"]):
        kind = type(message).__name__
        calls = ""
        if getattr(message, "tool_calls", None):
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        content = message.content
        if isinstance(content, str) and len(content) > 200:
            content = content[:200] + "..."
        print(f"[{i}] {kind}{calls}: {content!r}")
    print(
        f"state current_step={result.get('current_step')!r} warranty={result.get('warranty_status')!r}"
    )


def main() -> None:
    # checkpointer 让 current_step 跨 invoke 还在；忘了它，第二轮会重新从 routing 开始
    checkpointer = InMemorySaver()
    agent = create_agent(
        model="deepseek:deepseek-v4-flash",
        # 两个工具都要预注册，真正暴露哪几个由中间件过滤
        tools=[record_warranty_status, provide_solution],
        state_schema=SupportState,
        middleware=[apply_step_config],
        checkpointer=checkpointer,
    )
    # 同一条会话线程，两轮用户消息共享 state
    config = {"configurable": {"thread_id": "handoff-demo"}}

    # 第一轮：用户只说坏了，还没说在不在保 → 期望只提问、不调专家工具
    r1 = agent.invoke(
        {"messages": [{"role": "user", "content": "我的手机坏了。"}]},
        config,
    )
    print_trace(r1, "第 1 轮")

    # 第二轮：补上在保和故障 → 期望先 record 再 provide_solution
    r2 = agent.invoke(
        {"messages": [{"role": "user", "content": "还在保修期内，屏幕碎了。"}]},
        config,
    )
    print_trace(r2, "第 2 轮")


if __name__ == "__main__":
    main()
# 标准库：标注 state 里可选字段
from typing import NotRequired
from pathlib import Path

from dotenv import find_dotenv, load_dotenv

# AgentState 是 create_agent 默认状态；扩展后才能放下 current_step
from langchain.agents import AgentState, create_agent

# wrap_model_call 在每次调模型前按当前步骤换提示和工具
from langchain.agents.middleware import wrap_model_call, ModelRequest

# ToolMessage 必须和模型发出的 tool_call 配对
from langchain.messages import ToolMessage

# ToolRuntime 由框架注入，用来取 tool_call_id 和当前 state
from langchain.tools import tool, ToolRuntime

# 内存 checkpointer：让 current_step 跨多轮用户消息还在
from langgraph.checkpoint.memory import InMemorySaver

# 工具返回 Command 才能同时写 state 和补 ToolMessage
from langgraph.types import Command

load_dotenv(
    find_dotenv(filename=".env", usecwd=True)
    or Path(__file__).resolve().parents[1] / ".env",
    override=True,
)


# 在默认 AgentState 上增加两个槽位：当前步骤、已记录的保修状态
class SupportState(AgentState):
    # 当前处于哪一步；第一轮可能还没有这个键，中间件里要给默认值
    current_step: NotRequired[str]
    # 用户告知的保修状态
    warranty_status: NotRequired[str]


@tool
def record_warranty_status(status: str, runtime: ToolRuntime) -> Command:
    """记录保修状态并进入下一步。status 只能是 in_warranty 或 out_of_warranty。"""
    # 归一化，避免模型偶尔带空格
    key = status.strip()
    # 返回 Command 而不是普通字符串，才能改 current_step
    return Command(
        update={
            # 写入保修槽位
            "warranty_status": key,
            # 切到专家步骤，下一轮调模型时中间件会换配置
            "current_step": "specialist",
            # 必须补上与本次 tool_call 配对的 ToolMessage，否则对话历史残缺
            "messages": [
                ToolMessage(
                    content=f"已记录保修状态：{key}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )


@tool
def provide_solution(issue: str, runtime: ToolRuntime) -> str:
    """根据已记录的保修状态给出处理方案。issue 是用户描述的故障。"""
    # 从 state 读槽位，不要让模型把保修状态再口述一遍当参数
    warranty = runtime.state.get("warranty_status") or "未知"
    if warranty == "in_warranty":
        return f"保内处理「{issue}」：带购机凭证到授权点免费维修，周期 3-5 个工作日。"
    if warranty == "out_of_warranty":
        return f"过保处理「{issue}」：走付费维修，屏幕类参考价 800-1500 元。"
    return f"尚未记录保修状态，不能给方案。当前 issue={issue}"


# 按当前步骤决定系统提示和可见工具
@wrap_model_call
def apply_step_config(request: ModelRequest, handler):
    # 第一轮还没写过槽位时，当成分流步骤
    step = request.state.get("current_step") or "routing"
    # 已记录的保修状态，专家步骤的提示里会用到
    warranty = request.state.get("warranty_status") or "尚未记录"
    # 打印出来，跑多轮时能看见配置确实换了
    print(f"[middleware] current_step={step!r} warranty={warranty!r}")

    if step == "routing":
        # 分流阶段只允许记录保修，不允许直接给方案
        system_message = (
            "你是售后分流助手。还没记录保修状态时，先问用户设备是否在保。"
            "用户明确说了在保或过保之后，必须调用 record_warranty_status，"
            "status 填 in_warranty 或 out_of_warranty。不要猜测，不要给维修方案。"
        )
        tools = [record_warranty_status]
    else:
        # 专家阶段只给方案工具
        system_message = (
            f"你是售后专家。当前保修状态：{warranty}。"
            "用户描述故障后必须调用 provide_solution，不要自己编流程。"
            "工具返回后再用一两句中文转告用户。"
        )
        tools = [provide_solution]

    # 换提示、换可见工具；
    return handler(request.override(system_message=system_message, tools=tools))


def print_trace(result: dict, title: str) -> None:
    print(f"=== {title} ===")
    for i, message in enumerate(result["messages"]):
        kind = type(message).__name__
        calls = ""
        if getattr(message, "tool_calls", None):
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        content = message.content
        if isinstance(content, str) and len(content) > 200:
            content = content[:200] + "..."
        print(f"[{i}] {kind}{calls}: {content!r}")
    print(
        f"state current_step={result.get('current_step')!r} warranty={result.get('warranty_status')!r}"
    )


def main() -> None:
    # checkpointer 让 current_step 跨 invoke 还在；忘了它，第二轮会重新从 routing 开始
    checkpointer = InMemorySaver()
    agent = create_agent(
        model="deepseek:deepseek-v4-flash",
        # 两个工具都要预注册，真正暴露哪几个由中间件过滤
        tools=[record_warranty_status, provide_solution],
        state_schema=SupportState,
        middleware=[apply_step_config],
        checkpointer=checkpointer,
    )
    # 同一条会话线程，两轮用户消息共享 state
    config = {"configurable": {"thread_id": "handoff-demo"}}

    # 第一轮：用户只说坏了，还没说在不在保 → 期望只提问、不调专家工具
    r1 = agent.invoke(
        {"messages": [{"role": "user", "content": "我的手机坏了。"}]},
        config,
    )
    print_trace(r1, "第 1 轮")

    # 第二轮：补上在保和故障 → 期望先 record 再 provide_solution
    r2 = agent.invoke(
        {"messages": [{"role": "user", "content": "还在保修期内，屏幕碎了。"}]},
        config,
    )
    print_trace(r2, "第 2 轮")


if __name__ == "__main__":
    main()

输出：

[middleware] current_step='triage' warranty='尚未记录'
=== 第 1 轮 ===
[0] HumanMessage: '我的手机坏了。'
[1] AIMessage: '您好！很抱歉听到您的手机出了问题。\n\n在为您处理之前，我需要先确认一下：请问您的手机目前**是否还在保修期内**呢？'
state current_step=None warranty=None
[middleware] current_step='triage' warranty='尚未记录'
[middleware] current_step='specialist' warranty='in_warranty'
[middleware] current_step='specialist' warranty='in_warranty'
=== 第 2 轮 ===
[0] HumanMessage: '我的手机坏了。'
[1] AIMessage: '您好！很抱歉听到您的手机出了问题。\n\n在为您处理之前，我需要先确认一下：请问您的手机目前**是否还在保修期内**呢？'
[2] HumanMessage: '还在保修期内，屏幕碎了。'
[3] AIMessage [('record_warranty_status', {'status': 'in_warranty'})]: '好的，您确认设备在保修期内。我先为您记录保修状态。'
[4] ToolMessage: '已记录保修状态：in_warranty'
[5] AIMessage [('provide_solution', {'issue': '屏幕碎了'})]: '保修状态已为您记录。接下来我根据您描述的故障为您查询处理方案。'
[6] ToolMessage: '保内处理「屏幕碎了」：带购机凭证到授权点免费维修，周期 3-5 个工作日。'
[7] AIMessage: '已为您查询到处理方案：您的手机在保修期内，屏幕碎裂可以享受**免费维修**。请携带购机凭证前往授权维修点，维修周期大约为 **3-5 个工作日**。请问还有其他需要帮助的吗？'
state current_step='specialist' warranty='in_warranty'

按行读第二轮的中间件日志：

第二轮开始仍是分流——checkpointer 把第一轮「还没写过槽位」记住了。
模型调用 record_warranty_status 之后，同一轮里下一次调模型已经是专家。交接发生在 Agent 循环内部，用户没有再发一条消息。
专家步骤里 provide_solution 从 runtime.state 读保修状态，不靠模型把「在保」再填一遍参数。这是硬约束：过保和在保走不同返回值，不是提示词里「请酌情」。

第一轮 current_step=None 是正常的：TypedDict 的 NotRequired 不会在运行时填 "triage"。真正的默认值是中间件里那句 or "triage"。不要写成 current_step: str = "triage" 就以为第一轮状态里已经有这个键。

6.3. 入门不要上多 Agent 子图 #

另一条官方示例是：销售 Agent 和售后 Agent 各是 StateGraph 上的节点，交接工具返回 Command(goto="sales_agent", graph=Command.PARENT, update={...})。

那样做时，你必须亲手决定传给下一位的 messages 是什么。官方要求至少带上：

触发交接的那条带 tool_calls 的 AIMessage
一条配对的 ToolMessage

少一条，下一位看到的就是残缺历史。把子 Agent 内部全部消息都塞过去，又容易把下一位带跑，并且浪费 token。这已经是第 25 章「父子图状态边界」的难度。

判断口诀：

只是换提示和工具 → 用中间件。专家内部已经是一张带检索 / 反思的图 → 再考虑子图交接。

7. 技能：不换人，按需翻说明书 #

技能模式里，从头到尾都是同一个 Agent。它不把控制权交给别人，只是在需要时调用 load_skill，把一份长说明书变成 ToolMessage 留在对话里。

这叫渐进披露（progressive disclosure）：系统提示只放技能目录（名字 + 一句话），正文等真正用到再加载。和把全部制度一次性写进 system_prompt 相比，闲聊、问天气时不会先烧掉几千 token。

概念上接近第 15 章的 RAG：技能是一次检索单元。差别是入门实现往往按名字精确取，不是向量相似度。技能多到按名字猜不准时，再把目录检索升级成 RAG。

和子 Agent 的差别：技能不隔离执行、不隔离对话，只隔离「先看见什么」。控制权始终在一个人手里，所以可以直接跟用户多轮聊，重复问题也不必再加载（正文已经在历史里）。代价是：加载之后，后续每一次模型调用都会再次吃到那段正文。跨很多领域时 token 会堆起来——这就是 §4.3 里技能调用少、token 反而高的原因。

7.1. skills.py #

两份制度：报销、差旅预订。系统提示只点名，数字和截止日期必须先 load_skill。

# 从 pathlib 导入 Path，用于拼接和定位项目中的 .env 文件路径
from pathlib import Path

# 从 dotenv 导入查找与加载环境变量的工具函数
from dotenv import find_dotenv, load_dotenv
# 从 langchain.agents 导入创建_agent，用于创建可调用工具的 Agent
from langchain.agents import create_agent
# 从 langchain.tools 导入 tool 装饰器，把普通函数注册成 Agent 工具
from langchain.tools import tool

# 加载环境变量：优先找当前工作目录附近的 .env，找不到再退回上级目录
load_dotenv(
    # 先按文件名在当前工作目录向上查找 .env
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则用本文件上一级目录下的 .env 作为备选路径
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖已存在的同名环境变量
    override=True,
)

# 技能目录：系统提示里只放名字和一句话简介，正文按需加载
# 真实项目里可以换成读 SKILL.md / 数据库 / 对象存储
# 定义技能名到完整说明书正文的字典，供 load_skill 按需取出
SKILLS = {
    # 报销技能的完整说明书（Markdown 文本）
    "expense_report": """# 报销技能

适用：差旅报销、发票、截止日期。

硬规则：
- 差旅报销必须在返程后 7 个自然日内提交
- 单笔餐补上限 80 元，超出部分自理
- 必须附电子发票，缺发票不能报
- 市内交通按实报，出租车需注明事由
""",
    # 差旅预订技能的完整说明书（Markdown 文本）
    "travel_booking": """# 差旅预订技能

适用：订票、酒店、交通等级。

硬规则：
- 高铁优先二等座；机票需提前 3 天、经济舱
- 酒店上限 450 元/晚，含早
- 未经批准不得改签到商务座或头等舱
- 同一城市停留超过 3 晚要走差旅申请
""",
}


# 用 @tool 把下面的函数注册成 Agent 可调用的工具
@tool
# 定义按技能名加载完整说明书的工具函数，返回字符串给模型
def load_skill(skill_name: str) -> str:
    # 工具说明：告诉模型何时调用、有哪些可选技能名
    """加载一项技能的完整说明书。

    可选技能：
    - expense_report：报销截止日期、餐补、发票
    - travel_booking：订票、酒店、交通等级
    """
    # 按名字取出正文；找不到时把可选名单回给模型，让它改名重试
    # 从 SKILLS 字典里按 skill_name 取值，没有则得到 None
    body = SKILLS.get(skill_name)
    # 若技能名不存在，进入错误分支，提示可选名单
    if body is None:
        # 把所有可用技能名用逗号拼成字符串
        available = ", ".join(SKILLS)
        # 返回找不到技能的提示，并附上可选名单
        return f"没有名为 {skill_name} 的技能。可选：{available}"
    # 找到技能后，返回「已加载」前缀加上完整说明书正文
    return f"已加载技能 {skill_name}\n\n{body}"


# 定义程序入口：创建 Agent、发起一次提问并打印消息轨迹
def main() -> None:
    # 仍然是同一个 Agent，控制权不交接；只是按需把长说明书灌进对话
    # 创建带 load_skill 工具的行政助手 Agent
    agent = create_agent(
        # 指定使用的聊天模型
        model="deepseek:deepseek-v4-flash",
        # 只挂载按需加载技能说明书这一个工具
        tools=[load_skill],
        # 系统提示：说明角色、技能清单，并强制先加载再答数字规则
        system_prompt=(
            # 角色设定：简洁的中文行政助手
            "你是简洁的中文行政助手。"
            # 告知模型有两项可加载技能及其用途
            "你有两项技能：expense_report（报销）、travel_booking（差旅预订）。"
            # 硬性要求：涉及数字/截止日期/等级时必须先调用 load_skill
            "涉及具体数字、截止日期或等级时，必须先 load_skill 再回答，不要凭记忆编制度。"
            # 加载后必须严格按说明书中的硬规则作答
            "加载后严格按说明书里的硬规则作答。"
        ),
    )

    # 调用 Agent，传入一条用户关于报销截止日期与餐补的问题
    result = agent.invoke(
        # 输入状态：只包含 messages 列表
        {
            # 对话消息列表，这里只有一条用户消息
            "messages": [
                {
                    # 消息角色为用户
                    "role": "user",
                    # 用户问题：报销最晚几天交、餐补上限多少
                    "content": "出差回来最晚几天内必须交报销？餐补上限多少？",
                }
            ]
        }
    )

    # 遍历结果中的每条消息，打印类型、工具调用与内容摘要
    for i, message in enumerate(result["messages"]):
        # 取出消息类名，例如 HumanMessage / AIMessage / ToolMessage
        kind = type(message).__name__
        # 默认没有工具调用信息字符串
        calls = ""
        # 若消息上带有 tool_calls，则格式化成「工具名 + 参数」列表
        if getattr(message, "tool_calls", None):
            # 把每次工具调用的 name 与 args 拼进 calls 字符串
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        # 取出消息正文内容
        content = message.content
        # 若正文是很长的字符串，截断到 400 字符并加省略号，避免刷屏
        if isinstance(content, str) and len(content) > 400:
            # 截断过长内容并追加省略号
            content = content[:400] + "..."
        # 按「序号、类型、工具调用、内容」格式打印一条轨迹
        print(f"[{i}] {kind}{calls}: {content!r}")


# 仅当本文件被直接运行时才执行 main，被 import 时不自动跑
if __name__ == "__main__":
    # 启动演示：创建 Agent 并打印一次问答轨迹
    main()

输出：

[0] HumanMessage: '出差回来最晚几天内必须交报销？餐补上限多少？'
[1] AIMessage [('load_skill', {'skill_name': 'expense_report'})]: ''
[2] ToolMessage: '已加载技能 expense_report\n\n# 报销技能\n\n适用：差旅报销、发票、截止日期。\n\n硬规则：\n- 差旅报销必须在返程后 7 个自然日内提交\n- 单笔餐补上限 80 元，超出部分自理\n- 必须附电子发票，缺发票不能报\n- 市内交通按实报，出租车需注明事由\n'
[3] AIMessage: '根据公司报销规定：\n\n- **报销截止**：差旅报销必须在**返程后 7 个自然日内**提交，逾期将无法受理。\n- **餐补上限**：单笔餐补**上限 80 元**，超出部分需自理。\n\n另外提醒：报销必须附电子发票，缺发票不能报；市内出租车需注明事由。'

模型没有去加载 travel_booking。这就是渐进披露：问报销就不把订票规则塞进窗口。

真实项目里，技能正文往往来自目录里的 SKILL.md（YAML 头写 name / description，下面写说明）。Deep Agents 的 SkillsMiddleware 就是按这个约定扫盘。本章用手写字典，是为了看清「目录在提示里、正文当工具结果」这一跳，不把文件系统协议绑进来。

官方还演示过「加载技能的同时注册新工具」（比如加载 database_admin 之后才出现 backup 工具）。那是交接里的 wrap_model_call 过滤 / 增补，和第 9 章动态工具的组合，入门不必一上来做。

8. 路由器：先分类，再分发 #

路由器把「决定去哪」从对话循环里抽出来，变成请求前的一次分类。分类结果是「要问哪些专家」，然后分别调用，最后再合成一段话。

和子 Agent 的差别：

	子 Agent	路由器
谁决定叫谁	主 Agent 在对话里动态决定，可以多跳	分类器一次定案
对话记忆	主 Agent 带着	默认每次独立；多轮通常外面再包一个聊天 Agent
合成	主 Agent 的下一跳模型调用	你自己写的合成步骤
适合	流程事先说不清	垂直领域边界清楚

分类可以用规则（关键词、正则），也可以用第 6 章的 with_structured_output。下面用结构化输出，因为「要不要问 Python 专家」必须是布尔值，不能是一段散文。

8.1. DeepSeek 的一个前置坑 #

deepseek-v4 默认开 thinking。with_structured_output 内部会走强制 tool_choice，thinking 模式下会 400：Thinking mode does not support this tool_choice。第 6 章已经写过对策：初始化模型时关掉 thinking。

model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    extra_body={"thinking": {"type": "disabled"}},
)

专家用 create_agent 普通聊天不受这个限制；只有分类这一跳需要关。

8.2. router.py #

问题同时涉及 Python 和 Rust。分类器应把两个布尔都打成 True，然后两个专家各自只带着自己的那份「假文档」回答，最后由一次普通聊天合成。

# 从 pathlib 导入 Path，用于处理文件路径
from pathlib import Path

# 从 dotenv 导入查找与加载环境变量的函数
from dotenv import find_dotenv, load_dotenv

# 第 6 章的结构化输出：路由这一步要的是稳定字段，不是一段散文
# 从 langchain 导入初始化聊天模型的函数
from langchain.chat_models import init_chat_model
# 从 langchain 导入创建智能体的函数
from langchain.agents import create_agent
# 从 pydantic 导入数据模型基类与字段定义
from pydantic import BaseModel, Field

# 加载环境变量：优先找当前工作目录下的 .env，否则用上级目录的 .env，并覆盖已有变量
load_dotenv(
    # 在当前工作目录查找名为 .env 的文件
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则使用本文件上两级目录中的 .env
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖已存在的环境变量
    override=True,
)


# 路由结果：哪几个垂直领域要被问到。可以同时为 True，表示并行咨询
# 定义路由决策的结构化输出模型
class RouteDecision(BaseModel):
    # 类文档字符串：说明这是一次分类结果
    """一次分类结果。"""

    # 布尔字段：问题是否涉及用 Python 做 Web / 后端
    python: bool = Field(description="问题是否涉及 Python 做 Web / 后端")
    # 布尔字段：问题是否涉及用 Rust 做 Web / 后端
    rust: bool = Field(description="问题是否涉及 Rust 做 Web / 后端")


# 每个专家是独立 Agent，系统提示里塞该领域的「假文档」
# 演示用几十个 token；真实场景可能是几千 token 的内部文档
# 创建 Python Web 专家智能体
python_agent = create_agent(
    # 指定使用的模型名称
    model="deepseek:deepseek-v4-flash",
    # 本演示不挂工具，传空列表
    tools=[],
    # 系统提示：限定只谈 Python，并附上简短「假文档」
    system_prompt=(
        # 角色与约束说明
        "你是 Python Web 专家。只根据下面资料谈 Python，不要评价其它语言。\n"
        # 资料前半：生态与招聘、同步异步能力
        "资料：Django / FastAPI 生态成熟，招聘容易，同步和异步都能做；"
        # 资料后半：GIL 与 IO 密集场景，并要求三句以内中文回答
        "GIL 让 CPU 密集场景吃亏，但 Web IO 密集通常不是瓶颈。"
        "用三句以内中文回答。"
    ),
)

# 创建 Rust Web 专家智能体
rust_agent = create_agent(
    # 指定使用的模型名称
    model="deepseek:deepseek-v4-flash",
    # 本演示不挂工具，传空列表
    tools=[],
    # 系统提示：限定只谈 Rust，并附上简短「假文档」
    system_prompt=(
        # 角色与约束说明
        "你是 Rust Web 专家。只根据下面资料谈 Rust，不要评价其它语言。\n"
        # 资料前半：框架性能、内存安全、编译与招聘
        "资料：axum / actix-web 性能好、内存安全；编译慢，招聘比主流动态语言难，"
        # 资料后半：适用场景与成本，并要求三句以内中文回答
        "适合网关、高并发中间件，业务后台求快上线时成本偏高。"
        "用三句以内中文回答。"
    ),
)


# 定义辅助函数：从智能体返回结果中取出最后一条消息文本
def last_text(result: dict) -> str:
    # 取出 messages 列表最后一条消息的 content
    content = result["messages"][-1].content
    # 若已是字符串则直接返回，否则转成字符串
    return content if isinstance(content, str) else str(content)


# 定义主函数：完成路由、专家咨询与意见合成
def main() -> None:
    # 路由器用和专家相同的模型，但只做分类，不聊天
    # deepseek-v4 默认 thinking；with_structured_output 会强制 tool_choice，必须关掉 thinking
    # 初始化路由用的聊天模型，并关闭 thinking
    router_llm = init_chat_model(
        # 模型标识
        "deepseek:deepseek-v4-flash",
        # 通过 extra_body 禁用 thinking，避免与结构化输出冲突
        extra_body={"thinking": {"type": "disabled"}},
    )
    # 将路由模型绑定为输出 RouteDecision 结构
    classifier = router_llm.with_structured_output(RouteDecision)

    # 待咨询的用户问题
    question = "对比 Python 和 Rust 做 Web 开发，哪个更适合创业公司先上线？"
    # 这一步是一次模型调用：只产出 python / rust 两个布尔值
    # 调用分类器，判断需要咨询哪些专家
    decision = classifier.invoke(f"判断用户问题需要咨询哪些专家。问题：{question}")
    # 打印路由得到的布尔结果
    print(f"路由结果: python={decision.python} rust={decision.rust}")

    # 按分类结果分别调用专家；彼此看不到对方的长提示，这就是上下文隔离
    # 用于收集各专家意见的列表
    notes = []
    # 若路由判定需要 Python 专家
    if decision.python:
        # 调用 Python 专家智能体，传入用户问题
        py = python_agent.invoke({"messages": [{"role": "user", "content": question}]})
        # 将专家名称与回答文本加入 notes
        notes.append(("Python 专家", last_text(py)))
        # 打印 Python 专家的回答
        print(f"Python 专家: {last_text(py)}")
    # 若路由判定需要 Rust 专家
    if decision.rust:
        # 调用 Rust 专家智能体，传入用户问题
        rs = rust_agent.invoke({"messages": [{"role": "user", "content": question}]})
        # 将专家名称与回答文本加入 notes
        notes.append(("Rust 专家", last_text(rs)))
        # 打印 Rust 专家的回答
        print(f"Rust 专家: {last_text(rs)}")

    # 汇总也是一次模型调用。路由器的「合成」发生在这里，而不是回到某个主 Agent 循环
    # 把各位专家意见拼成一段文本
    blob = "\n\n".join(f"{name}：{text}" for name, text in notes)
    # 调用路由模型做最终合成
    final = router_llm.invoke(
        [
            # 系统消息：要求只基于专家意见合成中文对比
            {
                # 消息角色为 system
                "role": "system",
                # 合成约束说明
                "content": "把各位专家意见合成一段中文对比。不要编造专家没说的内容。",
            },
            # 用户消息：附上原问题与专家意见汇总
            {
                # 消息角色为 user
                "role": "user",
                # 内容为问题加专家意见 blob
                "content": f"问题：{question}\n\n{blob}",
            },
        ]
    )
    # 打印最终合成回答
    print(f"合成回答: {final.content}")


# 脚本直接运行时的入口判断
if __name__ == "__main__":
    # 执行主流程
    main()

输出：

路由结果: python=True rust=True
Python 专家: 创业公司先上线选 Python 更合适，生态成熟、招聘容易，能快速搭建产品。Web 属于 IO 密集场景，Python 的 GIL 影响不大，同步异步都够用。先验证业务再优化，Python 是稳妥的起点。
Rust 专家: 创业公司先上线时，Rust 编译慢、招聘难，业务后台快速迭代成本偏高，不适合抢时间。它更适合网关、高并发中间件这类对性能和内存安全要求高的场景。若团队已有 Rust 高手且业务偏基础设施，则可考虑。
合成回答: 综合两位专家的意见，创业公司先上线时，Python 是更稳妥的选择，因其生态成熟、招聘容易，且Web场景为IO密集，Python的GIL影响有限，同步异步均能满足快速搭建产品的需求；而Rust因编译慢、招聘难，在业务快速迭代时成本偏高，更适合网关、高并发中间件等对性能和内存安全要求高的场景，仅在团队已有Rust高手且业务偏基础设施时才可考虑。

调用结构是 1（分类）+ 2（两位专家）+ 1（合成）= 4。专家彼此看不到对方那份资料。

这里有一条容易误会的边界：隔离的是这次请求里你塞进去的文档，不是模型的预训练知识。 即使提示写了「不要评价其它语言」，Rust 专家仍然可能根据常识谈到「创业公司求快」。它没看见 Python 专家那份 Django 资料，但训练数据里本来就有 Python。文档越长、越私有，隔离的价值越大；短提示的「假专家」更多是在演示结构。

本例两个专家是顺序 invoke 的，代码更好读。它们互不依赖，可以改成并行（线程池，或第 23 章的 Send 扇出）。

多轮对话不要让路由器自己背历史。官方建议：外面一个带记忆的聊天 Agent，把「无状态路由器」包成工具。否则分类器一会儿派 Python、一会儿派 Rust，用户会觉得人格在跳。

9. 自定义工作流：确定性步骤和 Agent 混排 #

当标准模式套不上——必须先检索再回答、先鉴权再让模型说话、中间夹着规则校验——就自己画图。官方把这叫做自定义工作流（Custom workflow）：用 LangGraph 排节点，节点里可以是普通函数，也可以是整个 create_agent。

这和第 25 章「确定性节点 + Agent 节点」是同一件事。本章只给一个最小模式，证明「Agent 可以当图上的一个节点」；父子状态对不齐会静默丢结果、messages 空值兜底仍会烧 token，那些坑到第 25 章再看。

9.1. workflow.py #

三步里去掉「改写查询」，只留检索（确定性）和 Agent（概率性）。

# 从 pathlib 导入 Path，用于处理文件路径
from pathlib import Path
# 从 typing 导入 TypedDict，用于定义带类型的字典状态
from typing import TypedDict

# 从 dotenv 导入 find_dotenv 与 load_dotenv，用于查找并加载环境变量文件
from dotenv import find_dotenv, load_dotenv
# 从 langchain.agents 导入 create_agent，用于创建 Agent
from langchain.agents import create_agent
# 从 langgraph.graph 导入 StateGraph、START、END，用于构建状态图工作流
from langgraph.graph import StateGraph, START, END

# 加载环境变量：优先在当前工作目录找 .env，找不到则用上级目录的 .env
load_dotenv(
    # 在当前工作目录查找名为 .env 的文件
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则回退到本文件上一级目录下的 .env
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖已有环境变量
    override=True,
)


# 图的状态：问题、检索到的资料、最终回答。没有 messages 键也没关系，
# Agent 节点会自己构造一份临时 messages 再把结论写回 answer
# 定义工作流状态类型，包含问题、文档列表与最终回答
class State(TypedDict):
    # 用户提出的问题字符串
    question: str
    # 检索得到的资料字符串列表
    documents: list[str]
    # Agent 生成的最终回答字符串
    answer: str


# 迷你知识库：演示「确定性检索节点」，不引入向量库
# 用列表模拟知识库，存放几条篮球相关文本
KB = [
    # 第一条：2024 总决赛结果与冠军信息
    "2024 总决赛：甲队 4-2 击败乙队，甲队夺得冠军。",
    # 第二条：甲队主力前锋赛季场均数据
    "甲队主力前锋赛季场均 22.4 分 8.1 篮板。",
    # 第三条：乙队后卫赛季助攻与决赛 G5 得分
    "乙队后卫赛季场均 6.8 次助攻，决赛 G5 砍下 31 分。",
]


# 定义确定性检索节点函数，输入状态，返回要更新的字段字典
def retrieve(state: State) -> dict:
    # 函数文档字符串：说明这是按关键词筛资料、不调模型的确定性节点
    """确定性节点：按关键词从 KB 里筛资料，不调模型。"""
    # 从状态中取出当前问题文本
    q = state["question"]
    # 问冠军 / 决赛就返回相关条目，否则退回全部让 Agent 自己说资料不够
    # 若问题里包含「冠」或「决赛」，则按关键词筛选命中条目
    if "冠" in q or "决赛" in q:
        # 保留 KB 中含「决赛」或「冠军」的行作为命中结果
        hits = [row for row in KB if "决赛" in row or "冠军" in row]
    # 否则不筛选，把整个知识库都交给后续 Agent
    else:
        # 命中结果直接等于全部知识库
        hits = KB
    # 打印本次检索命中了多少条资料
    print(f"[retrieve] 命中 {len(hits)} 条")
    # 把命中资料写回状态的 documents 字段
    return {"documents": hits}


# 回答节点里用的 Agent：可以挂工具，这里刻意不挂，强迫它只用检索结果
# 创建 Agent：指定模型、空工具列表与系统提示词
agent = create_agent(
    # 使用 deepseek-v4-flash 模型
    model="deepseek:deepseek-v4-flash",
    # 不挂任何工具，迫使模型只依据消息里的资料作答
    tools=[],
    # 系统提示：要求简洁中文回答，且只能用提供的资料
    system_prompt="你是简洁的中文助手。只用用户消息里提供的资料回答，资料没有的就说不知道。",
)


# 定义 Agent 节点函数：把检索结果拼进用户消息后调用 Agent
def call_agent(state: State) -> dict:
    # 函数文档字符串：说明本节点如何组装消息并调用 create_agent
    """Agent 节点：把检索结果拼进用户消息，再调用 create_agent。"""
    # 把 documents 列表用换行拼成一段上下文文本
    context = "\n".join(state["documents"])
    # 调用 Agent，传入临时构造的 messages
    result = agent.invoke(
        {
            # messages 键：Agent 期望的对话消息列表
            "messages": [
                {
                    # 消息角色为用户
                    "role": "user",
                    # 内容为「资料 + 问题」拼接而成的提示
                    "content": f"资料：\n{context}\n\n问题：{state['question']}",
                }
            ]
        }
    )
    # 取出 Agent 返回的最后一条消息内容
    content = result["messages"][-1].content
    # 若内容已是字符串则直接用，否则转成字符串
    answer = content if isinstance(content, str) else str(content)
    # 把最终回答写回状态的 answer 字段
    return {"answer": answer}


# 定义主函数：组装图、编译并执行一次示例调用
def main() -> None:
    # 顺序：START → 检索（确定性）→ Agent（概率性）→ END
    # 用 State 类型创建状态图构建器
    builder = StateGraph(State)
    # 注册名为 retrieve 的检索节点
    builder.add_node("retrieve", retrieve)
    # 注册名为 agent 的 Agent 节点
    builder.add_node("agent", call_agent)
    # 从 START 连到检索节点
    builder.add_edge(START, "retrieve")
    # 从检索节点连到 Agent 节点
    builder.add_edge("retrieve", "agent")
    # 从 Agent 节点连到 END
    builder.add_edge("agent", END)
    # 编译状态图，得到可执行的工作流
    workflow = builder.compile()

    # 用示例问题调用工作流，得到最终状态
    result = workflow.invoke({"question": "2024 总冠军是谁？比分多少？"})
    # 打印工作流返回的 answer 字段
    print(f"answer: {result['answer']}")


# 脚本直接运行时的入口判断
if __name__ == "__main__":
    # 执行主流程
    main()

输出：

[retrieve] 命中 2 条
answer: 2024 总冠军是甲队，比分 4-2。

retrieve 一次模型都没调。这就是「该保证执行的交给图」：换模型、换提示，检索节点都会跑。Agent 只负责在已取出的两行资料上组织语言。

官方完整示例还会加一步「用结构化输出改写查询再检索」，并给 Agent 挂「查新闻」工具。那是第 15 章 RAG + 第 6 章结构化输出 + 第 25 章混排的组合题，不是新的多 Agent API。

没读过 LangGraph 的话，这一节先当作地图：标准四种模式不够用时，下一站是第 19 章。

10. 四种模式上的延伸 #

前九节已经把四种模式的最小闭环跑通。官方页上还留了几块「同一模式上的加料」——不是第五种模式，缺了前面的对应节会看不懂：

加料	建立在	解决什么
加载技能时解锁工具	技能 + 交接的状态	说明书和动手的工具一起按需出现
`Send` 并行扇出	路由器 + 第 23 章	多个专家真正同时跑，而不是 `if` 里一个接一个 `invoke`
后台三件套	子 Agent	长任务不堵住和用户的对话
Deep Agents	以上全部的现成胶水	不想自己维护监督者 / 技能 / 规划时换架子

没读过第 23 章的，可以先跳过 §10.2；后台任务和 Deep Agents 不依赖图 API。

10.1. skill_tools.py：加载技能时动态挂工具 #

§7 的 load_skill 只把说明书变成 ToolMessage。官方 Skills · Extending 还写了一句：加载 database_admin 时，可以同时把 backup / restore 注册进当前工具列表。

这不是新 API，是 §6 和 §7 的组合：

load_skill 返回 Command，把技能名写进 loaded_skills
wrap_model_call 读这个槽位，决定本次模型能看见哪些工具
工具必须预先传给 create_agent(tools=...)，中间件只做过滤。第 9 章测过：只 override 一个没预注册的工具，连「你好」都会在调模型前报错

第 1 次调模型：只看见 load_skill
    → 模型调用 load_skill("db_admin")
    → Command 写入 loaded_skills=["db_admin"]
第 2 次调模型：看见 load_skill + list_backups + run_backup
    → 按说明书先 list_backups，再 run_backup

# 从 pathlib 导入 Path，用于处理文件路径
from pathlib import Path

# 从 typing 导入 NotRequired，用于标记状态字段为可选
from typing import NotRequired

# 从 dotenv 导入 find_dotenv 与 load_dotenv，用于查找并加载环境变量文件
from dotenv import find_dotenv, load_dotenv

# 从 langchain.agents 导入 AgentState 与 create_agent，用于定义状态与创建 Agent
from langchain.agents import AgentState, create_agent

# 从 langchain.agents.middleware 导入 wrap_model_call 与 ModelRequest，用于包装模型调用中间件
from langchain.agents.middleware import wrap_model_call, ModelRequest

# 从 langchain.messages 导入 ToolMessage，用于构造工具返回消息
from langchain.messages import ToolMessage

# 从 langchain.tools 导入 tool 与 ToolRuntime，用于声明工具并访问运行时状态
from langchain.tools import tool, ToolRuntime

# 从 langgraph.types 导入 Command，用于在工具中同时更新状态与消息
from langgraph.types import Command

# 加载环境变量：优先在当前工作目录找 .env，找不到则用上级目录的 .env
load_dotenv(
    # 在当前工作目录查找名为 .env 的文件
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则回退到本文件上一级目录下的 .env
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖已有环境变量
    override=True,
)


# 用槽位记住已经加载过哪些技能；下一跳调模型时中间件会据此挂工具
# 定义技能状态类，继承 AgentState，增加已加载技能列表字段
class SkillState(AgentState):
    # 已加载技能名称列表，标记为可选字段
    loaded_skills: NotRequired[list[str]]


# 技能正文：除了说明书，还声明「加载后会多出哪些工具」
# 定义技能字典，键为技能名，值为技能说明书正文
SKILLS = {
    # 数据库管理员技能的说明书文本
    "db_admin": """# 数据库管理员技能

适用：备份、查看备份列表。不要在没加载本技能时操作数据库。

加载后你可以使用：
- list_backups：查看已有备份
- run_backup：对指定库做一次备份

硬规则：备份前先 list_backups 看有没有同名文件；run_backup 的 db_name 只允许 demo 或 orders。
""",
}


# 将函数注册为可供 Agent 调用的工具
@tool
# 定义列出备份文件的工具函数，返回字符串结果
def list_backups() -> str:
    # 工具文档字符串：说明该工具用于列出当前已有的数据库备份文件
    """列出当前已有的数据库备份文件。"""
    # 返回模拟的现有备份文件列表文本
    return "现有备份：demo_20260820.sql"


# 将函数注册为可供 Agent 调用的工具
@tool
# 定义执行数据库备份的工具函数，接收数据库名并返回字符串结果
def run_backup(db_name: str) -> str:
    # 工具文档字符串：说明该工具用于备份指定数据库，且 db_name 受限
    """对指定数据库做一次备份。db_name 只允许 demo 或 orders。"""
    # 定义允许备份的数据库名称集合
    allowed = {"demo", "orders"}
    # 若传入的数据库名不在允许集合中，则返回错误提示
    if db_name not in allowed:
        # 返回不允许备份的错误信息，并列出可选数据库名
        return f"错误：不允许备份 {db_name}，可选 {sorted(allowed)}"
    # 返回备份成功的模拟结果，包含生成的备份文件名
    return f"已备份 {db_name}，文件 {db_name}_20260821.sql"


# 将函数注册为可供 Agent 调用的工具
@tool
# 定义加载技能的工具函数，接收技能名与运行时，返回 Command 以更新状态
def load_skill(skill_name: str, runtime: ToolRuntime) -> Command:
    # 工具文档字符串：说明该工具用于加载技能说明书并解锁对应工具
    """加载一项技能的说明书，并解锁该技能对应的工具。

    可选技能：
    - db_admin：数据库备份与查看备份列表
    """
    # 根据技能名从 SKILLS 字典中取出说明书正文
    body = SKILLS.get(skill_name)
    # 若技能名不存在，则构造错误提示并保持已加载列表不变
    if body is None:
        # 把现有技能名拼成逗号分隔字符串，便于提示可选技能
        available = ", ".join(SKILLS)
        # 构造找不到技能时的返回文本
        text = f"没有名为 {skill_name} 的技能。可选：{available}"
        # 从运行时状态读取已加载技能列表，若不存在则用空列表
        loaded = list(runtime.state.get("loaded_skills") or [])
    # 若技能名存在，则返回说明书并更新已加载技能列表
    else:
        # 构造加载成功的返回文本，附上技能说明书正文
        text = f"已加载技能 {skill_name}\n\n{body}"
        # 从运行时状态读取已加载技能列表，若不存在则用空列表
        loaded = list(runtime.state.get("loaded_skills") or [])
        # 若该技能尚未记录在已加载列表中，则追加进去
        if skill_name not in loaded:
            # 把当前技能名加入已加载技能列表
            loaded.append(skill_name)
    # 返回 Command，同时更新 loaded_skills 状态与工具消息
    return Command(
        # 指定要写回图状态的更新内容
        update={
            # 写回更新后的已加载技能列表
            "loaded_skills": loaded,
            # 写回一条 ToolMessage，内容为加载结果文本，并绑定本次工具调用 id
            "messages": [ToolMessage(content=text, tool_call_id=runtime.tool_call_id)],
        }
    )


# 用 wrap_model_call 装饰器把函数注册为模型调用中间件
@wrap_model_call
# 定义中间件：根据已加载技能动态决定本轮暴露给模型的工具列表
def expose_skill_tools(request: ModelRequest, handler):
    # 从请求状态中读取已加载技能列表，若不存在则用空列表
    loaded = list(request.state.get("loaded_skills") or [])
    # 默认只暴露 load_skill 工具，避免未加载技能时直接调用业务工具
    tools = [load_skill]
    # 若已加载 db_admin 技能，则额外挂上备份相关工具
    if "db_admin" in loaded:
        # 把 load_skill 与数据库备份相关工具一并暴露给模型
        tools = [load_skill, list_backups, run_backup]
    # 打印本轮中间件看到的已加载技能与最终工具名，便于调试
    print(f"[middleware] loaded={loaded} tools={[t.name for t in tools]}")
    # 用覆盖后的工具列表继续调用后续处理器
    return handler(request.override(tools=tools))


# 定义脚本主函数，创建 Agent 并演示一次备份请求
def main() -> None:
    # 创建带技能加载中间件的 Agent
    agent = create_agent(
        # 指定使用的聊天模型
        model="deepseek:deepseek-v4-flash",
        # 注册全部候选工具；实际可见性由中间件按 loaded_skills 裁剪
        tools=[load_skill, list_backups, run_backup],
        # 使用带 loaded_skills 字段的自定义状态 schema
        state_schema=SkillState,
        # 挂上按技能动态暴露工具的中间件
        middleware=[expose_skill_tools],
        # 系统提示：要求先加载技能再调用对应工具，且不要编造结果
        system_prompt=(
            # 系统提示第一段：设定助手人设
            "你是简洁的中文助手。"
            # 系统提示第二段：说明必须先加载 db_admin 技能再操作备份
            "你有技能 db_admin（数据库备份）。涉及备份或查看备份时，"
            # 系统提示第三段：强调加载后再调工具，禁止编造备份结果
            "必须先 load_skill('db_admin')，加载后再调用对应工具，不要编造备份结果。"
        ),
    )
    # 调用 Agent，传入用户备份 demo 库的请求
    result = agent.invoke(
        # 构造初始消息状态：一条用户请求备份 demo 库的消息
        {"messages": [{"role": "user", "content": "帮我把 demo 库备份一下。"}]}
    )
    # 遍历结果中的全部消息，按序号打印类型、工具调用与内容
    for i, message in enumerate(result["messages"]):
        # 取出当前消息的类型名称，便于区分 Human/AI/Tool 等
        kind = type(message).__name__
        # 初始化工具调用摘要字符串为空
        calls = ""
        # 若消息带有 tool_calls 属性且非空，则格式化工具名与参数
        if getattr(message, "tool_calls", None):
            # 把每次工具调用整理成 (名称, 参数) 列表字符串，前面加空格
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        # 取出消息正文内容
        content = message.content
        # 若正文是过长字符串，则截断到 280 字符并加省略号，避免刷屏
        if isinstance(content, str) and len(content) > 280:
            # 截断过长内容并追加省略号
            content = content[:280] + "..."
        # 打印当前消息的序号、类型、工具调用摘要与正文
        print(f"[{i}] {kind}{calls}: {content!r}")
    # 打印最终状态中的已加载技能列表，确认技能是否被写入状态
    print(f"state loaded_skills={result.get('loaded_skills')!r}")


# 脚本直接运行时的入口判断
if __name__ == "__main__":
    # 执行主流程
    main()

输出：

[middleware] loaded=[] tools=['load_skill']
[middleware] loaded=['db_admin'] tools=['load_skill', 'list_backups', 'run_backup']
[middleware] loaded=['db_admin'] tools=['load_skill', 'list_backups', 'run_backup']
[middleware] loaded=['db_admin'] tools=['load_skill', 'list_backups', 'run_backup']
[0] HumanMessage: '帮我把 demo 库备份一下。'
[1] AIMessage [('load_skill', {'skill_name': 'db_admin'})]: ''
[2] ToolMessage: '已加载技能 db_admin\n\n# 数据库管理员技能\n\n适用：备份、查看备份列表。不要在没加载本技能时操作数据库。\n\n加载后你可以使用：\n- list_backups：查看已有备份\n- run_backup：对指定库做一次备份\n\n硬规则：备份前先 list_backups 看有没有同名文件；run_backup 的 db_name 只允许 demo 或 orders。\n'
[3] AIMessage [('list_backups', {})]: ''
[4] ToolMessage: '现有备份：demo_20260820.sql'
[5] AIMessage [('run_backup', {'db_name': 'demo'})]: '当前已有备份文件 demo_20260820.sql。我现在对 demo 库执行一次新备份。'
[6] ToolMessage: '已备份 demo，文件 demo_20260821.sql'
[7] AIMessage: '已完成 demo 库备份，生成文件：**demo_20260821.sql**。'
state loaded_skills=['db_admin']

第一行中间件证明：没加载时模型看不见 run_backup，不是「看见了但提示词不让用」。加载之后它还按说明书先 list_backups 再备份——硬规则写在技能正文里，工具列表负责「能不能调」，提示词负责「调的顺序」。

官方还提到层级技能（加载 data_science 之后才出现 pandas_expert 这种子技能名）。做法一样：loaded_skills 里多一个名字，系统提示或 load_skill 的 docstring 再露出下一层目录。入门有一层解锁就够。

10.2. send.py：路由器用 Send 并行扇出 #

§8 的路由器是 Python if 里顺序 invoke 两个专家。专家互不依赖时，这是在浪费等待。第 23 章的 Send 就是为这种「运行时才知道要扇出几份、每份输入还不一样」准备的。

和 §8 的差别只有调度方式：分类结果仍是那两个布尔值；专家仍是各自的 create_agent；合成仍是一次普通聊天。变的是图负责同时把 question + name 投给多份 expert 节点。

第 23 章三条规矩这里原样用：

Send("expert", {...}) 的第二个参数是专家节点看到的全部状态，主图的 notes 它看不见
专家写回 {"notes": [一段话]}，主状态的 notes 必须挂 operator.add，否则后结束的那份会盖掉先结束的
多份 expert 跑完之后，synthesize 只会走一次，读到的是已经累加好的 notes

# 从 operator 导入，用于 notes 字段的列表累加合并
import operator
# 从 pathlib 导入 Path，用于处理文件路径
from pathlib import Path
# 从 typing 导入 Annotated 与 TypedDict，用于带注解的类型字典状态
from typing import Annotated, TypedDict

# 从 dotenv 导入 find_dotenv 与 load_dotenv，用于查找并加载环境变量文件
from dotenv import find_dotenv, load_dotenv
# 从 langchain.agents 导入 create_agent，用于创建专家 Agent
from langchain.agents import create_agent
# 从 langchain.chat_models 导入 init_chat_model，用于初始化聊天模型
from langchain.chat_models import init_chat_model
# 从 langgraph.graph 导入 START、END、StateGraph，用于构建状态图工作流
from langgraph.graph import START, END, StateGraph
# 从 langgraph.types 导入 Send，用于并行扇出多个专家节点
from langgraph.types import Send
# 从 pydantic 导入 BaseModel 与 Field，用于定义结构化路由结果
from pydantic import BaseModel, Field

# 加载环境变量：优先在当前工作目录找 .env，找不到则用上级目录的 .env
load_dotenv(
    # 在当前工作目录查找名为 .env 的文件
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则回退到本文件上一级目录下的 .env
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖已有环境变量
    override=True,
)


# 定义路由分类的结构化输出模型
class RouteDecision(BaseModel):
    # 类文档字符串：说明这是一次分类结果
    """一次分类结果。"""

    # 布尔字段：问题是否涉及用 Python 做 Web / 后端
    python: bool = Field(description="问题是否涉及 Python 做 Web / 后端")
    # 布尔字段：问题是否涉及用 Rust 做 Web / 后端
    rust: bool = Field(description="问题是否涉及 Rust 做 Web / 后端")


# 定义整张图的状态类型
class GraphState(TypedDict):
    # 用户提出的问题字符串
    question: str
    # 分类器选出的专家名称列表
    agents: list[str]
    # 各位专家返回的意见列表，用 operator.add 做并行合并
    notes: Annotated[list[str], operator.add]
    # 最终合成后的回答字符串
    answer: str


# 定义扇出给单个专家节点的输入状态类型
class ExpertInput(TypedDict):
    # 用户原问题，原样传给专家
    question: str
    # 当前要调用的专家名称，如 python 或 rust
    name: str


# 创建 Python Web 专家 Agent
python_agent = create_agent(
    # 使用 deepseek-v4-flash 模型
    model="deepseek:deepseek-v4-flash",
    # 不挂任何工具，迫使模型只依据系统提示里的资料作答
    tools=[],
    # 系统提示：限定只谈 Python，并给出资料与回答长度约束
    system_prompt=(
        # 角色与边界：只谈 Python，不评价其它语言
        "你是 Python Web 专家。只根据下面资料谈 Python，不要评价其它语言。\n"
        # 资料：生态、招聘与同步异步能力
        "资料：Django / FastAPI 生态成熟，招聘容易，同步和异步都能做；"
        # 资料：GIL 与 Web IO 场景说明
        "GIL 让 CPU 密集场景吃亏，但 Web IO 密集通常不是瓶颈。"
        # 回答格式：三句以内中文
        "用三句以内中文回答。"
    ),
)

# 创建 Rust Web 专家 Agent
rust_agent = create_agent(
    # 使用 deepseek-v4-flash 模型
    model="deepseek:deepseek-v4-flash",
    # 不挂任何工具，迫使模型只依据系统提示里的资料作答
    tools=[],
    # 系统提示：限定只谈 Rust，并给出资料与回答长度约束
    system_prompt=(
        # 角色与边界：只谈 Rust，不评价其它语言
        "你是 Rust Web 专家。只根据下面资料谈 Rust，不要评价其它语言。\n"
        # 资料：框架性能、内存安全、编译与招聘成本
        "资料：axum / actix-web 性能好、内存安全；编译慢，招聘比主流动态语言难，"
        # 资料：适用场景与创业公司成本说明
        "适合网关、高并发中间件，业务后台求快上线时成本偏高。"
        # 回答格式：三句以内中文
        "用三句以内中文回答。"
    ),
)

# 专家名称到 Agent 实例的映射表，供 expert 节点按名查找
SPECIALISTS = {
    # Python 专家对应上面创建的 python_agent
    "python": python_agent,
    # Rust 专家对应上面创建的 rust_agent
    "rust": rust_agent,
}


# 从 Agent 调用结果里取出最后一条消息的文本内容
def last_text(result: dict) -> str:
    # 取 messages 列表最后一条消息的 content 字段
    content = result["messages"][-1].content
    # 若已是字符串则直接返回，否则转成字符串
    return content if isinstance(content, str) else str(content)


# 分类节点：用结构化输出判断需要咨询哪些专家
def classify(state: GraphState) -> dict:
    # 初始化路由用的聊天模型，并关闭 thinking
    router_llm = init_chat_model(
        # 指定模型名称
        "deepseek:deepseek-v4-flash",
        # 通过 extra_body 关闭模型思考过程
        extra_body={"thinking": {"type": "disabled"}},
    )
    # 用 RouteDecision 做结构化输出，对用户问题做专家分类
    decision = router_llm.with_structured_output(RouteDecision).invoke(
        # 把当前问题拼进分类提示
        f"判断用户问题需要咨询哪些专家。问题：{state['question']}"
    )
    # 准备存放命中专家名称的列表
    names = []
    # 若分类结果认为需要 Python 专家，则加入列表
    if decision.python:
        # 追加 python 专家名
        names.append("python")
    # 若分类结果认为需要 Rust 专家，则加入列表
    if decision.rust:
        # 追加 rust 专家名
        names.append("rust")
    # 打印本次分类选出的专家列表
    print(f"[classify] agents={names}")
    # 把专家名称列表写回状态的 agents 字段
    return {"agents": names}


# 条件边函数：按 agents 列表并行扇出 Send，或无人可选时直接结束
def fanout(state: GraphState):
    # 取出分类结果；若缺失则当作空列表
    names = state.get("agents") or []
    # 若没有专家可派，则走到 END
    if not names:
        # 返回图的结束标记
        return END
    # 为每个专家名构造一条 Send，并行进入 expert 节点
    return [
        # Send 目标节点为 expert，并带上问题与专家名
        Send("expert", {"question": state["question"], "name": name}) for name in names
    ]


# 专家节点：按名称调用对应 Agent，并把意见写入 notes
def expert(state: ExpertInput) -> dict:
    # 从输入状态取出当前专家名称
    name = state["name"]
    # 打印专家开始执行的日志
    print(f"[expert start] {name}")
    # 从映射表取出对应的专家 Agent
    worker = SPECIALISTS[name]
    # 调用专家 Agent，把用户问题作为一条 user 消息传入
    result = worker.invoke(
        # 构造仅含用户问题的 messages 输入
        {"messages": [{"role": "user", "content": state["question"]}]}
    )
    # 把专家名与回答文本拼成一条笔记
    note = f"{name}：{last_text(result)}"
    # 打印专家执行完成的日志
    print(f"[expert done] {name}")
    # 以列表形式返回 notes，便于 Annotated 的 operator.add 合并
    return {"notes": [note]}


# 合成节点：汇总各位专家意见，生成最终中文对比回答
def synthesize(state: GraphState) -> dict:
    # 把 notes 用空行拼接成一段文本；缺失时当作空列表
    blob = "\n\n".join(state.get("notes") or [])
    # 初始化用于最终合成的聊天模型，并关闭 thinking
    llm = init_chat_model(
        # 指定模型名称
        "deepseek:deepseek-v4-flash",
        # 通过 extra_body 关闭模型思考过程
        extra_body={"thinking": {"type": "disabled"}},
    )
    # 调用模型，基于专家原文做合成
    final = llm.invoke(
        [
            # 系统消息：约束只能基于已有专家原文合成
            {
                # 消息角色为 system
                "role": "system",
                # 合成约束：人数与内容都不能编造
                "content": (
                    # 动态写入当前专家人数
                    f"下面只有 {len(state.get('notes') or [])} 位专家的原文。"
                    # 要求合成中文对比，且不增加人数、不编造内容
                    "据此合成一段中文对比。不要增加人数，不要编造原文没有的内容。"
                ),
            },
            # 用户消息：附上原问题与专家意见汇总
            {
                # 消息角色为 user
                "role": "user",
                # 内容为问题加专家意见 blob
                "content": f"问题：{state['question']}\n\n{blob}",
            },
        ]
    )
    # 取出模型返回的 content 字段
    content = final.content
    # 若已是字符串则直接用，否则转成字符串作为最终回答
    answer = content if isinstance(content, str) else str(content)
    # 把最终回答写回状态的 answer 字段
    return {"answer": answer}


# 主流程：搭图、编译并调用工作流
def main() -> None:
    # 用 GraphState 创建状态图构建器
    builder = StateGraph(GraphState)
    # 注册分类节点
    builder.add_node("classify", classify)
    # 注册专家节点（可被多条 Send 并行进入）
    builder.add_node("expert", expert)
    # 注册合成节点
    builder.add_node("synthesize", synthesize)
    # 从 START 连到分类节点
    builder.add_edge(START, "classify")
    # 分类后按 fanout 条件边扇出到 expert，或无人可选时到 END
    builder.add_conditional_edges("classify", fanout, ["expert", END])
    # 专家节点结束后进入合成节点
    builder.add_edge("expert", "synthesize")
    # 合成节点结束后进入 END
    builder.add_edge("synthesize", END)
    # 编译状态图，得到可执行工作流
    workflow = builder.compile()

    # 用示例问题调用工作流，并初始化各状态字段
    result = workflow.invoke(
        {
            # 示例问题：对比 Python 与 Rust 做 Web 开发的创业公司选型
            "question": "对比 Python 和 Rust 做 Web 开发，哪个更适合创业公司先上线？",
            # 初始专家列表为空，由 classify 写入
            "agents": [],
            # 初始笔记列表为空，由各 expert 累加写入
            "notes": [],
            # 初始回答为空，由 synthesize 写入
            "answer": "",
        }
    )
    # 打印笔记分隔标题
    print("--- notes ---")
    # 逐条打印各位专家的笔记
    for note in result["notes"]:
        # 打印单条专家意见
        print(note)
    # 打印最终回答分隔标题
    print("--- answer ---")
    # 打印合成后的最终回答
    print(result["answer"])


# 脚本直接运行时的入口判断
if __name__ == "__main__":
    # 执行主流程
    main()

输出（专家完成顺序不必和 start 一致，这正是并行的证据）：

[classify] agents=['python', 'rust']
[expert start] python
[expert start] rust
[expert done] rust
[expert done] python
--- notes ---
python：Python 的 Django/FastAPI 生态成熟，招聘容易，能快速搭建同步或异步服务，适合创业公司抢时间上线。  
Web 场景以 IO 密集为主，GIL 影响有限，通常不会成为瓶颈。  
因此，优先选择 Python 能降低初期团队组建和开发成本。
rust：创业公司先上线求快，Rust 编译慢、招聘难，业务后台成本偏高；它更适合网关、高并发中间件等基础设施场景。
--- answer ---
在对比 Python 和 Rust 做 Web 开发时，两位专家从不同角度给出了各自的看法，但他们对适用场景的侧重有明显差异。
...

[expert start] 两条连着出现、done 的顺序还反了：两个 create_agent.invoke 在等网络时重叠。§8 那种 if python: invoke; if rust: invoke 做不到这一点。

合成提示里写死「下面只有 N 位专家的原文」是有用的。草稿阶段没写人数时，模型把两段笔记说成「五位专家」——原文没有的人数被编出来了。隔离文档解决不了合成步骤自己发挥。

add_conditional_edges("classify", fanout, ["expert", END]) 必须把 END 写进出口表：没有专家可派时 fanout 返回 END，漏写会 KeyError '__end__'（第 23 章同一条）。

10.3. jobs.py：子 Agent 后台任务 #

§5 的默认接法是同步：主 Agent 调用 ask_calendar 时会堵住，等子 Agent 跑完才继续。查会议这种秒级任务没问题。审一百页合同、跑一夜的数据任务，用户会觉得对话死了。

官方把这种「后台」叫 async，不是 Python 的 async/await。意思是：主 Agent 把活丢进作业系统，立刻拿回 job_id，回头继续和用户说话。标准模式是三个工具：

工具	职责
`start_review`	提交任务，返回 `job_id`
`check_status`	读 `pending / running / completed / failed`
`get_result`	完成后取结论；没完成不要编

下面用进程内字典 + 线程池模拟作业系统。真实项目把 JOBS 换成 Redis 即可，工具签名不用改。任务本身仍是一个普通 create_agent。

# 导入 time 模块，用于等待任务完成时的休眠与超时计时
import time

# 从 concurrent.futures 导入线程池执行器，用于在后台并行跑审阅任务
from concurrent.futures import ThreadPoolExecutor

# 从 pathlib 导入 Path，用于定位项目中的 .env 文件路径
from pathlib import Path

# 从 uuid 导入 uuid4，用于生成唯一的后台作业 ID
from uuid import uuid4

# 从 dotenv 导入查找与加载环境变量的工具函数
from dotenv import find_dotenv, load_dotenv

# 从 langchain.agents 导入创建_agent，用于创建主 Agent 与审阅子 Agent
from langchain.agents import create_agent

# 从 langchain.tools 导入 tool 装饰器，把普通函数注册成 Agent 可调用的工具
from langchain.tools import tool

# 从 langgraph 导入内存型检查点保存器，用于跨多轮对话保留会话状态
from langgraph.checkpoint.memory import InMemorySaver

# 加载环境变量：优先找当前工作目录下的 .env，找不到则用上一级目录的 .env，并允许覆盖已有变量
load_dotenv(
    # 先在当前工作目录查找名为 .env 的文件
    find_dotenv(filename=".env", usecwd=True)
    # 若找不到，则回退到本文件上一级目录中的 .env
    or Path(__file__).resolve().parents[1] / ".env",
    # 强制用 .env 中的值覆盖进程里已有的同名环境变量
    override=True,
)

# 进程内的作业表。真实项目会换成 Redis / 任务队列，结构一样：id → 状态和结果
# 用字典保存所有后台作业：键是 job_id，值是包含状态、结果、任务内容的字典
JOBS: dict[str, dict] = {}
# 后台线程池：主 Agent 调用 start_review 之后立刻返回，审阅在这里跑
# 创建最多 2 个工作线程的线程池，专门执行后台审阅任务
EXECUTOR = ThreadPoolExecutor(max_workers=2)


# 定义辅助函数：从 Agent 调用结果中取出最后一条消息的文本内容
def last_text(result: dict) -> str:
    # 取出结果里 messages 列表的最后一条消息的 content 字段
    content = result["messages"][-1].content
    # 若 content 已是字符串则直接返回，否则转成字符串再返回
    return content if isinstance(content, str) else str(content)


# 真正干活的子 Agent：故意写短，演示用；长文档审阅才值得丢到后台
# 创建专门负责合同审阅的子 Agent，不挂工具，只按系统提示输出简短风险点
reviewer = create_agent(
    # 指定使用的大模型为 deepseek-v4-flash
    model="deepseek:deepseek-v4-flash",
    # 审阅子 Agent 不需要额外工具，传空列表
    tools=[],
    # 系统提示：要求用三句中文指出主要风险，不要展开
    system_prompt="你是合同审阅助手。用三句中文指出主要风险，不要展开。",
)


# 定义在后台线程中真正执行审阅的内部函数
def _run_review(job_id: str, task: str) -> None:
    # 把该作业状态更新为 running，表示已开始执行
    JOBS[job_id]["status"] = "running"
    # 用 try/except 捕获审阅过程中的异常，避免线程静默失败
    try:
        # 调用审阅子 Agent，把任务文本作为用户消息传入
        result = reviewer.invoke({"messages": [{"role": "user", "content": task}]})
        # 从调用结果中提取最终文本，写入作业表的 result 字段
        JOBS[job_id]["result"] = last_text(result)
        # 审阅成功，把状态标记为 completed
        JOBS[job_id]["status"] = "completed"
    # 捕获任意异常，把失败信息写回作业表
    except Exception as exc:
        # 把作业状态标记为 failed
        JOBS[job_id]["status"] = "failed"
        # 把异常信息转成字符串存入 result，便于后续排查
        JOBS[job_id]["result"] = str(exc)


# 用 @tool 装饰器把 start_review 注册成主 Agent 可调用的工具
@tool
# 定义启动后台审阅的工具：接收任务描述，立刻返回 job_id
def start_review(task: str) -> str:
    # 工具说明：把合同审阅丢到后台，立刻返回 job_id，不要等它跑完
    """把一份合同审阅任务丢到后台，立刻返回 job_id。不要等它跑完。"""
    # 生成形如 job_xxxxxxxx 的短唯一作业 ID
    job_id = f"job_{uuid4().hex[:8]}"
    # 在作业表中登记该任务，初始状态为 pending，结果为空
    JOBS[job_id] = {"status": "pending", "result": None, "task": task}
    # 把真正的审阅函数提交到线程池异步执行
    EXECUTOR.submit(_run_review, job_id, task)
    # 立刻返回提示文案，告知已启动以及后续如何查进度
    return f"已启动后台审阅，job_id={job_id}。用户问起进度时用 check_status。"


# 用 @tool 装饰器把 check_status 注册成主 Agent 可调用的工具
@tool
# 定义查询后台任务状态的工具
def check_status(job_id: str) -> str:
    # 工具说明：查询后台任务状态，返回 pending / running / completed / failed
    """查询后台任务状态，返回 pending / running / completed / failed。"""
    # 根据 job_id 从作业表中取出对应记录，找不到则为 None
    job = JOBS.get(job_id)
    # 若作业不存在，返回找不到的提示
    if job is None:
        # 返回找不到该 job_id 的中文提示
        return f"找不到 {job_id}"
    # 返回该作业当前状态的中文描述
    return f"{job_id} 当前状态：{job['status']}"


# 用 @tool 装饰器把 get_result 注册成主 Agent 可调用的工具
@tool
# 定义取出已完成任务审阅结论的工具
def get_result(job_id: str) -> str:
    # 工具说明：取出已完成任务的审阅结论；未完成时不要编造
    """取出已完成任务的审阅结论。未完成时不要编造。"""
    # 根据 job_id 从作业表中取出对应记录
    job = JOBS.get(job_id)
    # 若作业不存在，返回找不到的提示
    if job is None:
        # 返回找不到该 job_id 的中文提示
        return f"找不到 {job_id}"
    # 若状态还不是 completed，则拒绝返回结论并提示稍后再查
    if job["status"] != "completed":
        # 返回尚未完成及当前状态的提示
        return f"{job_id} 尚未完成，状态是 {job['status']}，请稍后再查。"
    # 返回审阅结论；若结果为空则给出占位文案
    return job["result"] or "（空结果）"


# 定义打印 Agent 调用轨迹的调试辅助函数
def print_trace(result: dict, title: str) -> None:
    # 打印本轮轨迹的标题分隔线
    print(f"=== {title} ===")
    # 遍历结果中的每一条消息，带上序号
    for i, message in enumerate(result["messages"]):
        # 取出消息对象的类型名，便于区分 HumanMessage / AIMessage / ToolMessage 等
        kind = type(message).__name__
        # 先把工具调用信息初始化为空字符串
        calls = ""
        # 若该消息带有 tool_calls 属性且非空，则格式化工具名与参数
        if getattr(message, "tool_calls", None):
            # 把每次工具调用整理成 (名称, 参数) 列表的字符串
            calls = " " + str([(c["name"], c["args"]) for c in message.tool_calls])
        # 取出消息正文内容
        content = message.content
        # 若正文是过长字符串，则截断到 220 字符并加省略号，避免刷屏
        if isinstance(content, str) and len(content) > 220:
            # 截断过长内容并追加省略号
            content = content[:220] + "..."
        # 按统一格式打印序号、消息类型、工具调用与正文
        print(f"[{i}] {kind}{calls}: {content!r}")


# 定义阻塞等待所有后台作业结束（或超时）的辅助函数
def wait_jobs(timeout: float = 25.0) -> None:
    # 根据当前时间加上超时秒数，算出截止时间戳
    deadline = time.time() + timeout
    # 在截止时间之前循环轮询作业状态
    while time.time() < deadline:
        # 若作业表非空，且所有作业都已进入 completed 或 failed，则提前返回
        if JOBS and all(
            # 判断单个作业状态是否属于终态集合
            job["status"] in {"completed", "failed"}
            for job in JOBS.values()
        ):
            # 所有作业已结束，退出等待
            return
        # 短暂休眠 0.4 秒，避免空转占用 CPU
        time.sleep(0.4)


# 定义演示主流程：启动主 Agent，先发起审阅，再查询结果
def main() -> None:
    # 创建内存型检查点保存器，使同一 thread_id 下多轮对话能共享状态
    checkpointer = InMemorySaver()
    # 创建主 Agent：挂上启动审阅、查状态、取结果三个工具
    agent = create_agent(
        # 指定主 Agent 使用的大模型
        model="deepseek:deepseek-v4-flash",
        # 注册三个后台作业相关工具
        tools=[start_review, check_status, get_result],
        # 挂上检查点，实现多轮会话记忆
        checkpointer=checkpointer,
        # 系统提示：要求审合同时必须走后台工具，不要自己审
        system_prompt=(
            # 提示第一句：身份与必须调用 start_review 的规则
            "你是简洁的中文助理。用户要审合同或长文档时，必须调用 start_review，"
            # 提示第二句：拿到 job_id 后立刻告知用户，不要自己审
            "拿到 job_id 后立刻告诉用户任务已在后台运行，不要自己审。"
            # 提示第三句：问进度用 check_status，完成后用 get_result 转告结论
            "用户问进度时用 check_status；状态是 completed 再用 get_result 转告结论。"
        ),
    )
    # 构造会话配置，固定 thread_id 以便两轮调用共享检查点
    config = {"configurable": {"thread_id": "job-demo"}}

    # 第一轮调用：让用户提出合同审阅请求，主 Agent 应只启动后台任务
    r1 = agent.invoke(
        # 传入本轮消息字典
        {
            # messages 列表承载对话内容
            "messages": [
                {
                    # 角色为用户
                    "role": "user",
                    # 用户请求审阅合同要点：付款账期与违约金上限缺失
                    "content": "帮我审这份合同要点：货到后 30 天付款，违约金未写上限。",
                }
            ]
        },
        # 传入会话配置，绑定同一 thread_id
        config,
    )
    # 打印第一轮的完整消息轨迹，标题标明“只启动，不等待”
    print_trace(r1, "第 1 轮（只启动，不等待）")
    # 打印当前作业表中各 job 的状态快照
    print(f"作业表: { {k: v['status'] for k, v in JOBS.items()} }")

    # 阻塞等待后台审阅任务跑完（或超时）
    wait_jobs()
    # 等待结束后再次打印作业表状态，确认是否已完成
    print(f"等待后作业表: { {k: v['status'] for k, v in JOBS.items()} }")

    # 第二轮调用：用户追问是否审完及结论，主 Agent 应查状态并取结果
    r2 = agent.invoke(
        # 传入追问消息
        {"messages": [{"role": "user", "content": "审完了吗？结论是什么？"}]},
        # 继续使用同一会话配置，复用检查点中的上下文
        config,
    )
    # 打印第二轮的完整消息轨迹，标题标明“查状态并取结果”
    print_trace(r2, "第 2 轮（查状态并取结果）")


# 判断是否作为脚本直接运行（而非被 import）
if __name__ == "__main__":
    # 直接运行时执行演示主流程
    main()

输出：

=== 第 1 轮（只启动，不等待） ===
[0] HumanMessage: '帮我审这份合同要点：货到后 30 天付款，违约金未写上限。'
[1] AIMessage [('start_review', {'task': '审阅合同要点：货到后 30 天付款，违约金未写上限。'})]: ''
[2] ToolMessage: '已启动后台审阅，job_id=job_5b656942。用户问起进度时用 check_status。'
[3] AIMessage: '已把合同审阅任务放到后台了（任务号 job_5b656942），结果出来后我会第一时间告诉你。\n\n稍等片刻，随时可以问我进度。'
作业表: {'job_5b656942': 'running'}
等待后作业表: {'job_5b656942': 'completed'}
=== 第 2 轮（查状态并取结果） ===
...
[4] HumanMessage: '审完了吗？结论是什么？'
[5] AIMessage [('check_status', {'job_id': 'job_5b656942'})]: ''
[6] ToolMessage: 'job_5b656942 当前状态：completed'
[7] AIMessage [('get_result', {'job_id': 'job_5b656942'})]: ''
[8] ToolMessage: '付款账期长达30天，占用买方资金且影响现金流。违约金未设上限，违约时可能面临无法预估的巨额索赔风险。建议明确违约金计算方式及上限比例，并缩短付款周期或增加折扣条款。'
[9] AIMessage: '审阅结论如下：\n\n**主要风险**\n1. **付款账期偏长**：货到后 30 天付款，会占用买方资金、影响现金流。\n2. **违约金无上限**：这是最大隐患，违约时可能面临无法预估的巨额索赔。\n\n**建议**\n- 明确违约金的计算方式和**上限比例**（例如不超过合同总额的某个百分比）。\n- 尽量缩短付款周期，或争取提前付款折扣条款。\n\n如需进一步细化条款措辞，可以告诉我。'

第一轮结束时作业表仍是 running：主 Agent 没有等审阅。wait_jobs() 是演示脚本在等，产品里应换成通知（推送、轮询接口、点一下「查看 job_xxx」再发一条 HumanMessage）。checkpointer 让第二轮还记得那个 job_id，否则用户只说「审完了吗」模型会对不上号。

短任务继续用 §5 的同步调用。只有「用户不该干等」时才上这套三件套。

10.4. Deep Agents：现成架子，不是另一种 API #

官方 Multi-agent 首页的第一条提示是：若想要开箱即用的多 Agent，用 Deep Agents。它是 LangChain 之上的一个架子包（harness），不是 langchain 里的新函数。

把它拆开，里面几乎全是本章已经手写过的模式：

Deep Agents 自带	本章对应	差别
把子 Agent 当 `task` 调	§5.3 统一调度	它还带规划、隔离工作区
扫描 `SKILL.md` 做渐进披露	§7 / §10.1	它按目录约定扫盘，本章用字典
待办 / 规划	第 9 章中间件或一张小图	现成 `write_todos` 一类工具
虚拟文件系统	超出本章	子 Agent 在沙箱里读文件，避免把原对话撑爆
上下文压缩	第 11 章裁剪	长对话自动摘要

所以 Deep Agents 解决的是「胶水不想自己维护」，不是「又多了一种新模式」。把手写的四种跑懂再换架子，出了问题才知道该查监督者、技能还是后台任务。