1. 本章目标 #

前面十三章，每一章讲一个部件。这一章把它们全部接起来，做一个能跑的东西。

我们要做的是企业知识治理助手：员工问「P1 工单的升级时限是多少、现在有没有超时的」，它去查知识库和工单系统、核对规范、写一份带来源编号的报告，如果需要发布到知识库，先停下来等人批准。

选这个场景是因为它同时压到了前面所有部件：

需求 用到的部件 出处
查知识库、查工单、接内部系统 tools + MCP 第 41 章
报告草稿不塞进对话 虚拟文件系统 + FilesystemBackend 第 42 章
别让它读到 .env permissions 第 43 章
报告格式要统一 Skills 第 44 章
记住组织约定和用户偏好 Memory + StoreBackend 第 45 章
长任务不炸上下文 摘要 + 卸载 第 46 章
检索和审校分开做 Subagents 第 47 章
发布前必须人批 interrupt_on 第 48 章
前端要能看到进度 事件流 第 49 章
装配顺序要对 Middleware 栈 第 50 章
上线不能一崩全崩 容错 + rubric + 评测 第 51 章

学完你应有一份可以直接改成自己业务的项目骨架，以及四个只有真接起来才会遇到的坑——它们都是本章实测撞出来的，前面章节没有覆盖。

前置依赖： 第 39～51 章。

本章验证环境：deepagents 0.7.12、langchain 1.3.18。全部实验零 token：所有模型都是脚本化的假类，用来驱动一条确定的业务流。这套自检脚本本身就是产出的一部分，建议挂进 CI。

2. 项目结构 # 

kb-agent/
├── langgraph.json               # LangGraph Server 部署声明
├── pyproject.toml
├── .env.example
├── src/agent/
│   ├── config.py                # 唯一的配置来源
│   ├── build.py                 # 唯一调用 create_deep_agent 的地方
│   ├── graph.py                 # Server 入口（一行）
│   ├── permissions.py           # 文件系统权限规则
│   ├── subagents.py             # 子智能体分工
│   ├── middleware.py            # 容错栈
│   └── tools/
│       ├── __init__.py          # 工具汇总 + MCP 合流
│       ├── kb.py                # 知识库检索（只读）
│       └── biz.py               # 业务工具（含副作用的）
├── skills/                      # 技能库（父目录！）
│   ├── research-report/SKILL.md
│   └── kb-governance/
│       ├── SKILL.md
│       └── references/notify-template.md
├── memories/AGENTS.md           # 长期记忆
├── workspace/                   # 智能体的草稿纸（落盘）
├── kb/                          # 知识库快照（只读）
├── published/                   # 正式发布区（写需审批）
├── evals/run_eval.py            # 回归评测
└── verify.py                    # 装配自检（CI 跑这个）

两条组织原则，值得在自己项目里照搬：

  1. create_deep_agent 只在 build.py 里出现一次。 测试、评测、Server 入口都调 build_agent()，配置改一处就全生效。散着写 create_deep_agent 的项目，三个月后没人能说清线上跑的到底是哪套配置。
  2. 配置全部集中在 config.py。 尤其是那几个容错阈值——它们需要按环境调，散在代码里就只能改代码。

3. 配置层 # 

"""src/agent/config.py —— 集中管理配置。"""
from __future__ import annotations

import os
from dataclasses import dataclass, field
from pathlib import Path

# 项目根目录：从本文件往上两层（src/agent/config.py -> 项目根）
ROOT = Path(__file__).resolve().parents[2]


@dataclass(frozen=True)
class Settings:
    """一份不可变的配置快照。"""

    # --- 模型 --------------------------------------------------------------
    main_model: str = os.getenv("MAIN_MODEL", "anthropic:claude-sonnet-4-5")
    # 子智能体用便宜的模型：它们做的是检索和整理，不需要主模型的推理能力
    worker_model: str = os.getenv("WORKER_MODEL", "anthropic:claude-haiku-4-5")
    # 评分员单独一个模型，避免「自己评自己」的偏袒
    grader_model: str = os.getenv("GRADER_MODEL", "anthropic:claude-sonnet-4-5")

    # --- 模型视角的虚拟路径 -------------------------------------------------
    # 关键：这些是 backend 内的路径。`skills=` 传的也是这种虚拟路径，
    # 传磁盘绝对路径会被 FilesystemBackend 判成「超出 root 目录」而报错（见 §8.2）
    workspace: str = "/workspace"        # 草稿、中间产物
    kb_dir: str = "/kb"                  # 知识库快照（只读）
    memories_dir: str = "/memories"      # 跨会话记忆
    published_dir: str = "/published"    # 正式发布区（写操作需审批）
    skills_dir: str = "/skills"          # 技能库

    # --- 落盘位置（真实磁盘路径）-------------------------------------------
    # backend 的 root 设成项目根，上面那些虚拟路径就一一对应到根下的子目录
    disk_root: Path = field(default_factory=lambda: ROOT)
    memories_file: Path = field(default_factory=lambda: ROOT / "memories" / "AGENTS.md")

    # --- 容错阈值（第 51 章）------------------------------------------------
    model_run_limit: int = int(os.getenv("MODEL_RUN_LIMIT", "40"))
    model_thread_limit: int = int(os.getenv("MODEL_THREAD_LIMIT", "200"))
    search_run_limit: int = int(os.getenv("SEARCH_RUN_LIMIT", "15"))
    tool_run_limit: int = int(os.getenv("TOOL_RUN_LIMIT", "100"))
    recursion_limit: int = int(os.getenv("RECURSION_LIMIT", "150"))

    # --- 开关 --------------------------------------------------------------
    enable_hitl: bool = os.getenv("ENABLE_HITL", "1") == "1"


settings = Settings()

虚拟路径和磁盘路径的对应关系是整个项目最容易搞错的地方，所以在配置里就写清楚：disk_root 是项目根，/workspace 对应 <根>/workspace，/skills 对应 <根>/skills。这个约定会在 §8.2 救你一次。

4. 工具层 #

4.1. 只读工具与副作用工具要分开 #

这不是洁癖，是因为它们在容错栈里的待遇完全不同：只读工具可以自动重试（第 51 章），副作用工具必须审批 + 幂等。所以从定义的时候就分开列名单。

"""src/agent/tools/kb.py —— 知识库检索。复用第 12～15 章的 RAG 组件。"""
from __future__ import annotations

from langchain_core.tools import tool

# 演示用的假知识库。生产里换成第 14 章的向量库（Chroma / PGVector / Milvus）
_DOCS = [
    {"id": "POL-001", "title": "报销制度", "owner": "财务部", "updated": "2026-03-11",
     "text": "差旅报销须在行程结束后 15 个工作日内提交，超过 30 天不予受理。"},
    {"id": "POL-014", "title": "数据分级规范", "owner": "安全部", "updated": "2025-08-02",
     "text": "客户手机号、身份证号属于 L3 敏感数据，禁止导出到外部系统。"},
    {"id": "SOP-207", "title": "工单升级流程", "owner": "客服部", "updated": "2026-01-20",
     "text": "P1 工单 30 分钟未响应自动升级到值班经理；P2 为 4 小时。"},
]


@tool
def kb_search(query: str, top_k: int = 3) -> str:
    """在企业知识库中检索相关条目。

    Args:
        query: 检索关键词或自然语言问题。
        top_k: 返回条数，默认 3。

    Returns:
        命中条目的编号、标题、责任部门、更新日期与正文。
    """
    # 生产实现：retriever.invoke(query)，见第 15 章
    hits = [d for d in _DOCS if any(w in d["title"] + d["text"] for w in query)]
    hits = (hits or _DOCS)[:top_k]
    return "\n\n".join(
        f"[{d['id']}] {d['title']}（{d['owner']}，更新于 {d['updated']}）\n{d['text']}"
        for d in hits
    )


@tool
def kb_get(doc_id: str) -> str:
    """按编号取出知识库条目全文。

    Args:
        doc_id: 条目编号，例如 "POL-014"。
    """
    for d in _DOCS:
        if d["id"] == doc_id:
            return (f"[{d['id']}] {d['title']}\n责任部门：{d['owner']}\n"
                    f"更新日期：{d['updated']}\n\n{d['text']}")
    # 返回可读的错误而不是抛异常，模型能自己纠正
    return f"未找到编号 {doc_id}。可用编号：{', '.join(d['id'] for d in _DOCS)}"


@tool
def kb_stale_check(days: int = 365) -> str:
    """列出超过指定天数未更新的知识库条目，用于知识治理。

    Args:
        days: 天数阈值，默认 365。
    """
    from datetime import date

    today = date(2026, 9, 3)          # 演示用固定「今天」
    stale = []
    for d in _DOCS:
        y, m, dd = (int(x) for x in d["updated"].split("-"))
        age = (today - date(y, m, dd)).days
        if age > days:
            stale.append(f"[{d['id']}] {d['title']} —— 已 {age} 天未更新，"
                         f"责任部门 {d['owner']}")
    return "\n".join(stale) if stale else f"没有超过 {days} 天未更新的条目。"


KB_TOOLS = [kb_search, kb_get, kb_stale_check]

注意 kb_get 找不到条目时返回可读文本而不是抛异常。这是给智能体用的工具的通用写法：抛异常需要 ToolErrorMiddleware 才能救回来，而返回一句「可用编号是 A、B、C」模型立刻就能自己改。

4.2. 副作用工具自带幂等 # 

"""src/agent/tools/biz.py —— 业务工具。"""
from __future__ import annotations

from langchain_core.tools import tool

# 只读工具：幂等、可重试、失败无害
READONLY_TOOL_NAMES = ["kb_search", "kb_get", "kb_stale_check", "ticket_query"]

# 副作用工具：必须过审批，且自身要做幂等
SIDE_EFFECT_TOOL_NAMES = ["publish_kb_doc", "notify_owner"]

_TICKETS = [
    {"id": "T-9001", "level": "P1", "title": "支付回调超时", "status": "处理中",
     "owner": "支付组", "waited_min": 42},
    {"id": "T-9002", "level": "P2", "title": "导出报表乱码", "status": "待处理",
     "owner": "报表组", "waited_min": 310},
]

# 幂等台账：记录已执行过的操作。生产里这应该是数据库表，不是内存 set
_DONE: set[str] = set()


@tool
def ticket_query(level: str = "") -> str:
    """查询工单列表。

    Args:
        level: 按等级过滤，如 "P1"；留空返回全部。
    """
    rows = [t for t in _TICKETS if not level or t["level"] == level]
    if not rows:
        return f"没有等级为 {level} 的工单。可用等级：P1、P2。"
    return "\n".join(
        f"{t['id']} [{t['level']}] {t['title']} —— {t['status']}，"
        f"负责 {t['owner']}，已等待 {t['waited_min']} 分钟"
        for t in rows
    )


@tool
def publish_kb_doc(doc_id: str, title: str, body: str, idempotency_key: str) -> str:
    """把一份文档正式发布到企业知识库。这是不可逆的外部写操作。

    Args:
        doc_id: 目标条目编号。
        title: 标题。
        body: 正文全文。
        idempotency_key: 幂等键。同一个键重复调用只会真正执行一次。
    """
    # 幂等检查放在最前面：断点续跑或工具重试时，这一步防止重复发布
    if idempotency_key in _DONE:
        return f"该发布请求（key={idempotency_key}）此前已执行，本次跳过。"
    _DONE.add(idempotency_key)
    return f"已发布 [{doc_id}] {title}，正文 {len(body)} 字。"


@tool
def notify_owner(department: str, message: str, idempotency_key: str) -> str:
    """给责任部门发送通知。这是不可逆的外部写操作。

    Args:
        department: 部门名。
        message: 通知正文。
        idempotency_key: 幂等键。
    """
    if idempotency_key in _DONE:
        return f"该通知（key={idempotency_key}）此前已发送，本次跳过。"
    _DONE.add(idempotency_key)
    return f"已通知 {department}：{message[:40]}"


BIZ_TOOLS = [ticket_query, publish_kb_doc, notify_owner]

把 idempotency_key 做成工具的显式参数，而不是在内部生成 UUID。这样：

如果幂等键是工具内部生成的，第三条就不成立——重放会生成新键，重复执行。

4.3. MCP 工具合流 # 

"""src/agent/tools/__init__.py —— 工具汇总。"""
from __future__ import annotations

from .biz import BIZ_TOOLS, READONLY_TOOL_NAMES, SIDE_EFFECT_TOOL_NAMES
from .kb import KB_TOOLS

LOCAL_TOOLS = [*KB_TOOLS, *BIZ_TOOLS]


async def mcp_tools() -> list:
    """从 MCP 服务器拉工具（第 17、41 章）。没配就返回空列表。

    生产里典型的 MCP 来源：Jira、Confluence、内部数据平台。
    """
    import os

    url = os.getenv("MCP_SERVER_URL")
    if not url:
        return []
    from langchain_mcp_adapters.client import MultiServerMCPClient

    client = MultiServerMCPClient({
        "internal": {"url": url, "transport": "streamable_http"},
    })
    tools = await client.get_tools()
    # 关键：MCP 工具名可能和内置文件工具撞名（第 41 章的坑），撞了就重命名
    builtin = {"ls", "read_file", "write_file", "edit_file", "delete", "glob",
               "grep", "task", "execute"}
    for t in tools:
        if t.name in builtin:
            t.name = f"mcp_{t.name}"
    return tools

那个改名循环不是多余的。第 41 章验证过：同名工具会静默覆盖内置件。一个第三方 MCP 服务器提供了叫 grep 的工具，你的智能体就再也搜不了文件了，而且不报任何错。

5. 权限层 #

5.1. 规则 # 

"""src/agent/permissions.py —— 文件系统权限（第 43 章）。
规则按声明顺序匹配，首条命中即生效。"""
from __future__ import annotations

from deepagents import FilesystemPermission

from .config import settings

# 顺序非常重要：越具体的规则越靠前，兜底规则放最后
MAIN_PERMISSIONS = [
    # 1. 凭据类：彻底不可见。注意要同时写 `.*` 和 `**/.*`——
    #    `/**` 这个 glob 不匹配点文件，只写它会把 .env / .ssh 漏出去
    FilesystemPermission(operations=["read", "write"],
                         paths=["/**/.env", "/.env", "/**/.*", "/.*"],
                         mode="deny"),
    # 路径必须以 `/` 开头，写 `**/credentials*` 会直接抛 ValueError（见 §5.2）
    FilesystemPermission(operations=["read", "write"],
                         paths=["/**/credentials*", "/**/*secret*",
                                "/**/*token*", "/credentials*", "/*secret*"],
                         mode="deny"),

    # 2. 知识库快照：只读。写操作要走 publish_kb_doc 工具，不能直接改文件
    FilesystemPermission(operations=["write"],
                         paths=[f"{settings.kb_dir}{{,/**}}"],
                         mode="deny"),

    # 3. 正式发布区：写之前必须人工确认
    FilesystemPermission(operations=["write"],
                         paths=[f"{settings.published_dir}{{,/**}}"],
                         mode="interrupt"),

    # 4. 记忆目录：允许智能体自己更新（第 45 章）
    FilesystemPermission(operations=["read", "write"],
                         paths=[f"{settings.memories_dir}{{,/**}}"],
                         mode="allow"),

    # 5. 工作区：随便读写，这是它的草稿纸
    FilesystemPermission(operations=["read", "write"],
                         paths=[f"{settings.workspace}{{,/**}}"],
                         mode="allow"),

    # 6. 兜底：上面没覆盖到的一律拒绝。写死这一条，不要依赖「无命中则放行」
    FilesystemPermission(operations=["read", "write"], paths=["/**", "/*"],
                         mode="deny"),
]

# 研究员子智能体：只读知识库 + 只写自己的草稿目录。
# 第 47 章的关键结论——子智能体的 permissions 是「完全替换」而非叠加，
# 所以这里必须把凭据保护重新写一遍，不能指望继承父规则。
RESEARCHER_PERMISSIONS = [
    FilesystemPermission(operations=["read", "write"],
                         paths=["/**/.*", "/.*", "/**/*secret*", "/**/*token*"],
                         mode="deny"),
    FilesystemPermission(operations=["write"],
                         paths=[f"{settings.workspace}/research{{,/**}}"],
                         mode="allow"),
    FilesystemPermission(operations=["read"],
                         paths=[f"{settings.kb_dir}{{,/**}}",
                                f"{settings.workspace}{{,/**}}"],
                         mode="allow"),
    FilesystemPermission(operations=["read", "write"], paths=["/**", "/*"],
                         mode="deny"),
]

注意 {,/**} 这个写法（第 43 章）：/workspace{,/**} 同时匹配目录本身和它下面的一切，而 /workspace/** 只匹配下面的内容、漏掉目录本身。

5.2. 坑一：权限路径必须以 / 开头 #

写这份规则时第一次撞的坑。直觉上 glob 写 **/credentials* 是对的，但：

ValueError: Permission path must start with '/': '**/credentials*'

所有权限路径都必须以 / 开头。 好消息是这个报错发生在构造期、消息也说得很清楚，属于「立刻发现」的那类问题。麻烦在于这意味着相对 glob 完全不能用——想匹配任意深度的 credentials*，得同时写 /**/credentials*（子目录里的）和 /credentials*（根目录下的）两条。

漏写第二条的话，根目录下的 credentials.json 就落到兜底规则去了。我们的兜底是 deny，所以安全；但如果你的兜底是放行，这就是个洞。这也是为什么第 6 条兜底规则要显式写成 deny，而不是依赖「无命中则放行」的默认行为。

6. 子智能体层 #

6.1. 坑二：tools 只收对象，不收名字 #

按第 47 章的知识，我最初这么写：

# 这样写会报错
SubAgent(name="researcher", tools=["kb_search", "kb_get", "read_file"], ...)
AttributeError: 'function' object has no attribute 'name'

这个报错信息毫无指向性——堆栈深处在 ToolNode.__init__，完全看不出问题在 SubAgent.tools。实测四种写法：

字符串名（业务工具）        失败  AttributeError: 'function' object has no attribute 'name'
字符串名（内置文件工具）    失败  AttributeError: 'function' object has no attribute 'name'
字符串名混合               失败  AttributeError: 'function' object has no attribute 'name'
工具对象                   通过
工具对象 + 内置字符串       失败  AttributeError: 'function' object has no attribute 'name'

看类型定义就明白了：

tools: NotRequired[Sequence[BaseTool | Callable | dict[str, Any]]]

SubAgent.tools 只接受工具对象，一个字符串都不能有。 这跟同一个库里其它地方的约定不一致，很容易搞混：

参数 接受字符串名？
SubAgent(tools=...) 不接受，只要对象
FilesystemMiddleware(tools=...) 接受（白名单，第 41 章）
HarnessProfile(excluded_tools=...) 接受（黑名单，第 50 章）
ToolRetryMiddleware(tools=...) 接受

6.2. 坑三：tools 管不住内置文件工具 #

修好上一个坑，写成 tools=[kb_search, kb_get, ticket_query]，以为工具面就收窄了。实测子智能体真正看到的是：

A. 子智能体 tools=[alpha]（只一个业务工具）
  主智能体工具面: ['alpha', 'beta', 'delete', 'edit_file', 'glob', 'grep', 'ls',
                  'read_file', 'task', 'write_file']
  子智能体工具面: ['alpha', 'delete', 'edit_file', 'glob', 'grep', 'ls',
                  'read_file', 'write_file']

B. 子智能体不设 tools（继承父）
  子智能体工具面: ['alpha', 'beta', 'delete', 'edit_file', 'glob', 'grep', 'ls',
                  'read_file', 'write_file']

C. tools=[alpha] + middleware=[FilesystemMiddleware(tools=["read_file", "ls"])]
  子智能体工具面: ['alpha', 'ls', 'read_file']

三条结论：

  1. SubAgent.tools 只控制业务工具。 七个内置文件工具由 FilesystemMiddleware 单独注入，tools 管不着——包括 delete。你以为给检索员配了只读工具面，实际上它能删文件。
  2. task 不在子智能体的工具面里，无论哪种写法。这印证了第 47 章「委派只有一层」的结论。
  3. 要真正收窄文件工具，得给子智能体单独挂一个 FilesystemMiddleware 白名单（C 组）。这就是第 41 章讲的白名单机制，只是用在了子智能体上。

6.3. 最终写法 # 

"""src/agent/subagents.py —— 子智能体分工（第 47 章）。

三条设计原则来自第 47 章的实测结论：
1. 对话历史完全隔离 —— description 里必须写清全部上下文，子智能体看不到主对话；
2. 文件系统完全共享 —— 大产出走文件，返回值只回一句「写到哪了」；
3. permissions 完全替换 —— 每个子智能体的规则要自成体系。

外加两条本章实测出来的：
4. `SubAgent.tools` 只接受工具对象，传字符串名会抛 AttributeError；
5. `SubAgent.tools` 只管业务工具，内置文件工具照给不误（包括 delete），
   要收窄得靠 `middleware=[FilesystemMiddleware(tools=[...])]`。
"""
from __future__ import annotations

from deepagents import FilesystemMiddleware, SubAgent

from .config import settings
from .permissions import RESEARCHER_PERMISSIONS
from .tools.biz import ticket_query
from .tools.kb import kb_get, kb_search


def build_subagents(backend) -> list[SubAgent]:
    """子智能体要和主智能体共享同一个 backend，否则看不到彼此写的文件。"""
    researcher = SubAgent(
        name="researcher",
        # description 是主智能体挑人的唯一依据，写「什么时候用它」而不是「它是谁」
        description=(
            "检索型子智能体。当需要从企业知识库、工单系统查证事实时派给它。"
            "调用时必须在 description 里写明：要查什么、时间范围、以及结果写到哪个文件。"
            "它会把整理好的材料写进 /workspace/research/ 并只回一句文件路径。"
        ),
        system_prompt=(
            "你是企业知识检索员。工作方式：\n"
            "1. 用 kb_search / kb_get / ticket_query 收集材料，宁可多查一轮也不要猜；\n"
            "2. 每条结论后面必须跟来源编号，格式 `[POL-014]`；\n"
            "3. 把整理结果写进指定文件，正文里不要出现手机号、身份证号等 L3 数据；\n"
            "4. 最后只回一句话：写到了哪个路径、包含几条材料。不要复述正文。"
        ),
        # 只给这三个业务工具（必须传对象，不能传名字字符串）
        tools=[kb_search, kb_get, ticket_query],
        # 文件工具面单独收窄：只能读和写，拿不到 delete / edit_file
        middleware=[FilesystemMiddleware(
            backend=backend, tools=["read_file", "write_file", "ls", "glob"])],
        model=settings.worker_model,
        permissions=RESEARCHER_PERMISSIONS,
    )

    reviewer = SubAgent(
        name="reviewer",
        description=(
            "审校子智能体。当一份草稿写完、需要检查引用是否齐全和是否泄露敏感数据时派给它。"
            "调用时在 description 里给出草稿文件路径和适用的规范编号。"
            "它会把问题清单写进 /workspace/review/ 并回一句结论。"
        ),
        system_prompt=(
            "你是合规审校员。逐条检查：\n"
            "1. 每个事实性结论是否带来源编号；\n"
            "2. 是否出现手机号、身份证号、邮箱等 L3 敏感数据（见 POL-014）；\n"
            "3. 是否有超过一年未更新的知识条目被当成现行规范引用。\n"
            "把问题写成清单存文件，最后回一句：通过 / 不通过 + 问题数量。"
        ),
        tools=[kb_get],
        middleware=[FilesystemMiddleware(
            backend=backend, tools=["read_file", "write_file", "ls", "grep"])],
        model=settings.worker_model,
    )
    return [researcher, reviewer]

注意这里是工厂函数而不是模块级常量。因为子智能体的 FilesystemMiddleware 需要拿到和主智能体同一个 backend 实例——它们共享文件系统靠的就是这个。写成常量的话，backend 要么在导入时创建（顺序不好控），要么两边各建一个（就不共享了）。

researcher 的 description 里有一句「调用时必须在 description 里写明：要查什么、时间范围、以及结果写到哪个文件」。这是在给主智能体写调用说明——因为子智能体看不到主对话，主智能体如果只写「查一下工单」，研究员完全不知道查什么。把要求写进 description，主智能体读到了就会照做。

7. 容错层 # 

"""src/agent/middleware.py —— 容错与观测栈（第 51 章）。
所有默认值都在这里被重新设过一遍。"""
from __future__ import annotations

import logging

import langchain.agents.middleware as M

from .config import settings
from .tools import READONLY_TOOL_NAMES

logger = logging.getLogger(__name__)


def _retry_on(exc: Exception) -> bool:
    """只重试真正的瞬时故障。默认的 default_retry_on 什么都重试，太宽。"""
    if isinstance(exc, (TimeoutError, ConnectionError)):
        return True
    status = getattr(getattr(exc, "response", None), "status_code", None)
    return status is not None and (status == 429 or status >= 500)


def _on_tool_error(exc: Exception, request) -> str | None:
    """把预期内的工具异常翻译给模型；预期外的返回 None 让它崩出来报警。"""
    name = request.tool_call["name"]
    if isinstance(exc, (TimeoutError, ConnectionError)):
        return f"`{name}` 暂时不可用（{type(exc).__name__}）。可以换个来源，或先做别的。"
    if isinstance(exc, (ValueError, KeyError)):
        return f"`{name}` 入参有问题：{exc}。请检查参数后重试。"
    logger.exception("工具 %s 抛出未预期异常", name)
    return None       # 不吞，让监控能看到


def _on_model_failure(exc: Exception) -> str:
    """重试全部失败后的兜底回答。注意这里必须是函数，传字符串会静默失效。"""
    logger.error("模型调用最终失败: %s", exc)
    return "模型服务暂时不可用，本次请求未能完成。请稍后重试，或联系管理员。"


def build_middleware(*, with_rubric: bool = True, fallback_model=None,
                     grader_model=None) -> list:
    """按第 51 章的清单装配容错栈。

    Args:
        with_rubric: 是否挂评分中间件。
        fallback_model: 覆盖降级模型。注意 ModelFallbackMiddleware 在**构造时**
            就会 init_chat_model，所以测试里必须注入假模型对象，
            否则会因为模型名无效直接报 ValueError。
        grader_model: 覆盖评分模型，同理。
    """
    mws: list = [
        # --- 任务规划：v0.7 起 write_todos 需要显式开启（第 47 章）------------
        M.TodoListMiddleware(),

        # --- 模型层 ----------------------------------------------------------
        M.ModelRetryMiddleware(max_retries=3, retry_on=_retry_on,
                               on_failure=_on_model_failure),
        M.ModelFallbackMiddleware(fallback_model or settings.worker_model),

        # --- 工具层：先重试，重试无效再转成消息 --------------------------------
        M.ToolRetryMiddleware(max_retries=2, retry_on=_retry_on,
                              tools=READONLY_TOOL_NAMES),   # 只对幂等工具重试
        M.ToolErrorMiddleware(_on_tool_error),

        # --- 循环层 ------------------------------------------------------------
        M.ModelCallLimitMiddleware(run_limit=settings.model_run_limit,
                                   thread_limit=settings.model_thread_limit),
        # 单个昂贵工具限量，用默认的 continue 让它换别的方式继续
        M.ToolCallLimitMiddleware(tool_name="kb_search",
                                  run_limit=settings.search_run_limit),
        # 全局兜底必须显式 end——默认的 continue 拦得住工具却停不下图
        M.ToolCallLimitMiddleware(run_limit=settings.tool_run_limit,
                                  exit_behavior="end"),
    ]

    if with_rubric:
        from deepagents import RubricMiddleware

        def _log_eval(ev) -> None:
            logger.info("rubric 第 %s 轮: %s — %s",
                        ev["iteration"], ev["result"], ev["explanation"])

        mws.append(RubricMiddleware(model=grader_model or settings.grader_model,
                                    max_iterations=3, on_evaluation=_log_eval))
    return mws

那两个 fallback_model / grader_model 参数看着像是测试用的杂物，其实是被迫加的——ModelFallbackMiddleware 在构造函数里就调 init_chat_model：

ValueError: Unable to infer model provider for model='x'.

也就是说只要你 build_middleware()，它就会尝试解析模型名，哪怕这个智能体根本不会被 invoke。零 token 测试里想传假模型，就只能开一个注入口。这类「构造期就有副作用」的中间件在写测试时要特别留意。

8. 装配 #

8.1. build.py # 

"""src/agent/build.py —— 把所有部件装成一个 deep agent。
这是整个项目唯一调用 create_deep_agent 的地方。"""
from __future__ import annotations

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, FilesystemBackend, StoreBackend

from .config import settings
from .middleware import build_middleware
from .permissions import MAIN_PERMISSIONS
from .subagents import build_subagents
from .tools import LOCAL_TOOLS

SYSTEM_PROMPT = f"""\
你是企业知识治理助手，服务于内部员工的政策查询、工单分析与知识库维护。

## 工作方式
1. 复杂请求先用 `write_todos` 列计划，再逐项推进。
2. 查证事实派给 `researcher` 子智能体；草稿完成后派给 `reviewer` 审校。
   派任务时要把全部上下文写进 description——子智能体看不到我们的对话。
3. 大段材料写文件（`{settings.workspace}/`），不要堆在对话里。

## 硬性规范
- 每个事实性结论必须带来源编号，格式 `[POL-014]`。
- 不确定的地方明确写「未在知识库中找到依据」，不要推测。
- 绝不在输出中出现客户手机号、身份证号等 L3 敏感数据（依据 [POL-014]）。
- 引用超过一年未更新的条目时，必须提示「该条目可能已过时」。

## 目录约定
- `{settings.workspace}/` 草稿与中间产物，可自由读写
- `{settings.kb_dir}/` 知识库快照，只读
- `{settings.published_dir}/` 正式发布区，写入需人工批准
- `{settings.memories_dir}/` 你的长期记忆
"""


def _interrupt_config() -> dict:
    """哪些操作必须过人工（第 48 章）。只拦不可逆的外部写操作。"""
    if not settings.enable_hitl:
        return {}

    def need_approval(req) -> bool:
        """条件拦截：`when` 是单参回调（第 48 章的坑）。"""
        args = req["args"] if isinstance(req, dict) else req.tool_call["args"]
        return args.get("department") != "自己组"

    return {
        # 发布到知识库：无条件拦
        "publish_kb_doc": True,
        # 对外通知：按条件拦
        "notify_owner": {"when": need_approval,
                         "allowed_decisions": ["approve", "edit", "reject"]},
    }


def build_backend():
    """文件落盘 + 记忆跨会话（第 42、45 章）。"""
    for sub in ("workspace", "kb", "published", "skills"):
        (settings.disk_root / sub).mkdir(parents=True, exist_ok=True)
    return CompositeBackend(
        default=FilesystemBackend(root_dir=str(settings.disk_root)),
        routes={
            # 记忆走 StoreBackend 才能跨线程。注意 store 的 key 是「去掉路由前缀后」
            # 的路径（第 45 章的坑）：/memories/AGENTS.md 在 store 里的 key 是 /AGENTS.md
            f"{settings.memories_dir}/": StoreBackend(
                namespace=lambda rt: ("memories", "org")),
        },
    )


def build_agent(*, model=None, checkpointer=None, store=None,
                with_rubric: bool = True, with_permissions: bool = True,
                fallback_model=None, grader_model=None):
    """构造智能体。

    Args:
        model: 覆盖主模型。测试时传假模型可以做到零 token 验证装配。
        checkpointer: 生产传 PostgresSaver；LangGraph Server 会自己注入。
        store: 跨会话记忆的存储。
        with_rubric: 是否挂评分中间件。
        with_permissions: 关掉可用于本地调试，生产必须开。
    """
    # 子智能体必须拿到同一个 backend 实例，否则看不到主智能体写的文件
    backend = build_backend()
    kwargs = dict(
        model=model or settings.main_model,
        tools=LOCAL_TOOLS,
        system_prompt=SYSTEM_PROMPT,
        subagents=build_subagents(backend),
        middleware=build_middleware(with_rubric=with_rubric,
                                    fallback_model=fallback_model,
                                    grader_model=grader_model),
        backend=backend,
        # 指技能库的父目录（虚拟路径），不是某个技能目录，也不是磁盘绝对路径
        skills=[settings.skills_dir],
        memory=[f"{settings.memories_dir}/AGENTS.md"],
        interrupt_on=_interrupt_config(),
    )
    if with_permissions:
        kwargs["permissions"] = MAIN_PERMISSIONS
    if checkpointer is not None:
        kwargs["checkpointer"] = checkpointer
    if store is not None:
        kwargs["store"] = store
    return create_deep_agent(**kwargs)

8.2. 坑四：skills= 走的是 backend，不是磁盘 #

第 44 章说「skills= 要指父目录」。我照做了，传了磁盘绝对路径：

skills=[str(ROOT / "skills")]     # 看起来很合理

结果运行时炸了：

ValueError: Path: C:\...\ch62\skills outside root directory: C:\...\ch62\workspace
  During task with name 'SkillsMiddleware.before_agent'

看堆栈就懂了：SkillsMiddleware.before_agent → backend.ls(source_path)。skills= 的路径是交给 backend 解析的虚拟路径，不是 Python 直接读的磁盘路径。 我的 FilesystemBackend root 是 workspace/，而技能库在它外面，于是判定越界。

两种修法：

这个坑的普适版本是：Deep Agents 里凡是传给它的路径，默认都是「模型视角的虚拟路径」——skills=、memory=、permissions 的 paths、工具调用里的 file_path，全都是。只有 FilesystemBackend(root_dir=...) 这一处是真实磁盘路径。混淆这两者是新手最常见的错误来源。

9. 技能与记忆 #

9.1. 两个技能 #

技能负责「怎么做」这类按需加载的规范（第 44 章）。这里放两个：写报告的格式规范、知识库治理的作业流程。

---
name: research-report
description: 撰写企业内部研究报告。当用户要求「出一份报告」「做个分析」「总结一下现状」时使用，规定了报告的固定结构、引用格式与措辞要求。
---

# 企业研究报告规范

## 固定结构

报告必须按以下五段组织，段落标题原样使用：

1. **结论先行** —— 不超过 3 句话，直接给答案。
2. **依据** —— 每条依据一行，行尾必须带来源编号 `[POL-014]`。
3. **风险与缺口** —— 明确写出「哪些问题当前无法回答」。没有就写「无」。
4. **建议动作** —— 每条包含责任部门和时限。
5. **附录** —— 引用条目清单，含各自的更新日期。

## 引用格式

- 知识库条目：`[POL-014]`
- 工单：`[T-9001]`
- 一句话里有多个来源时并列：`[POL-014][SOP-207]`
- **禁止**出现无来源的事实性陈述。判断性表述要写明「据此推断」。

## 时效提示

引用更新日期距今超过 365 天的条目时，该行末尾追加：`（该条目可能已过时）`。
用 `kb_stale_check` 可以批量确认。

## 数据红线

正文与附录中不得出现客户手机号、身份证号、完整邮箱地址（依据 [POL-014]，L3 数据）。
如果检索结果里带了这类数据，用 `***` 遮蔽后再引用。

## 落盘位置

草稿写到 `/workspace/reports/<主题>-draft.md`，审校通过后才考虑发布。

description 那一行是唯一会常驻系统提示的部分（第 44 章的渐进式披露）。所以它必须写「什么时候用」而不是「这是什么」——模型靠它做检索判断。「当用户要求『出一份报告』『做个分析』时使用」这种触发词枚举，比「这是报告写作规范」有用得多。

第二个技能演示带资源目录的形态：

---
name: kb-governance
description: 知识库治理作业。当用户要求「清理知识库」「检查哪些文档过期了」「更新某条规范」时使用，规定了过期判定、责任人通知与发布流程。
---

# 知识库治理作业

## 治理三步

1. **盘点** —— `kb_stale_check(days=365)` 拿到过期清单。
2. **定责** —— 每条过期条目找到 `owner` 字段里的责任部门。
3. **推动** —— 用 `notify_owner` 逐个通知，通知正文见 `references/notify-template.md`。

## 过期判定标准

| 类别 | 阈值 | 说明 |
|---|---|---|
| 安全规范（POL-0xx，安全部） | 180 天 | 合规要求，从严 |
| 业务制度（POL-其他） | 365 天 | 常规 |
| 操作手册（SOP-xxx） | 365 天 | 常规 |

阈值不一致，所以不能只跑一次 `kb_stale_check(365)` 就完事——安全部的条目要再用
`kb_stale_check(180)` 单独过一遍。

## 发布纪律

- `publish_kb_doc` 是不可逆操作，**每次调用必须带 `idempotency_key`**，
  取值用 `<doc_id>-<日期>` 这种可复现的形式，不要用随机数。
  这样即使流程中断重跑，也不会重复发布。
- 发布前必须已经过 `reviewer` 子智能体审校并返回「通过」。
- 发布后在 `/workspace/publish-log.md` 追加一行记录。

## 不要做的事

- 不要直接写 `/kb/` 下的文件——那是只读快照，改了不会生效。
- 不要为了让报告好看而省略「未找到依据」的条目。

「幂等键用 <doc_id>-<日期> 而不是随机数」这条规矩写在技能里，而不是系统提示里。因为它只在做发布时才用得上，塞进系统提示是每轮都付费的浪费——这正是技能存在的意义。

9.2. 记忆 #

记忆负责「每次都要知道」的长期事实（第 45 章）：

# 组织约定

这份文件每次启动都会被完整加载进系统提示，所以只放**长期稳定、每次都用得上**的东西。
一次性的任务上下文不要写进来。

## 组织信息

- 知识库编号规则：`POL-` 政策制度，`SOP-` 操作手册，`T-` 工单。
- 部门简称：安全部 = 信息安全部，客服部 = 客户成功部。
- 财年划分：4 月 1 日起为新财年，Q1 = 4～6 月。

## 已知的坑

- `kb_search` 是关键词匹配，同义词命中率差。查「报销」要顺带试「差旅」「费用」。
- 工单系统的 `waited_min` 字段在跨天工单上会偏大，超过 1440 的值需要人工复核。

## 用户偏好

（这一节由助手自己维护，用 `edit_file` 更新）

- 暂无

「已知的坑」那一节特别值得抄。把工具的缺陷写进记忆，等于给模型打补丁。 kb_search 同义词命中率差是个客观事实，与其寄望模型每次都想到，不如直接告诉它「查报销要顺带试差旅」。

10. 装配自检 #

这是本章最实用的产出：一个零 token 的脚本，验证所有部件真的装上了。建议挂进 CI，改任何配置都跑一遍。

"""verify.py —— 装配自检：零 token 验证整个项目能装起来、各部件都在位。"""
from __future__ import annotations

import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src"))

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

RUNTIME = {}


class Fake(BaseChatModel):
    """零 token 假模型：只用来触发装配，不产生真实调用。

    关键：Skills 和 Memory 是在 before_agent 阶段注入系统提示的，
    构造期抓不到。必须在这里、模型真正被调用时抓 messages[0]。
    """

    @property
    def _llm_type(self) -> str:
        return "fake"

    def bind_tools(self, tools, **kw):
        RUNTIME["tools"] = sorted(getattr(t, "name", str(t)) for t in tools)
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        RUNTIME.setdefault("system_prompt", str(messages[0].content))
        return ChatResult(generations=[ChatGeneration(
            message=AIMessage(content="ok"))])


# 拦截 create_agent，把实际传进去的 middleware 列表抓出来（第 50 章的手法）
import deepagents.graph as G

CAP = {}
_orig = G.create_agent


def spy(model, **kw):
    CAP["middleware"] = [m.name for m in kw.get("middleware", [])]
    return _orig(model, **kw)


G.create_agent = spy

from agent.build import build_agent
from agent.config import settings

store = InMemoryStore()
# 注意 key 是去掉路由前缀后的路径（第 45 章）
store.put(("memories", "org"), "/AGENTS.md",
          {"content": settings.memories_file.read_text(encoding="utf-8")})

agent = build_agent(model=Fake(), checkpointer=InMemorySaver(), store=store,
                    with_rubric=True, fallback_model=Fake(), grader_model=Fake())

print("1. middleware 装配顺序")
for i, n in enumerate(CAP["middleware"], 1):
    print(f"  {i:2d}. {n}")

print("\n2. 权限规则自检：兜底规则必须能挡住点文件")
from agent.permissions import MAIN_PERMISSIONS

for i, p in enumerate(MAIN_PERMISSIONS, 1):
    print(f"  {i}. [{p.mode:9s}] {'/'.join(p.operations):11s} {p.paths}")
dot_covered = any(p.mode == "deny" and any(".*" in x for x in p.paths)
                  for p in MAIN_PERMISSIONS)
print(f"\n  点文件（.env / .ssh）被显式 deny: "
      f"{'是' if dot_covered else '否 —— 有风险！'}")

print("\n3. 端到端冒烟：跑一轮（假模型，零 token）")
cfg = {"configurable": {"thread_id": "smoke"},
       "recursion_limit": settings.recursion_limit}
out = agent.invoke({"messages": [{"role": "user", "content": "自检"}]}, cfg)
print(f"  跑通，返回 {len(out['messages'])} 条消息")
print(f"  checkpoint 已写入，next = {agent.get_state(cfg).next}")

print("\n4. 运行时工具面（模型真正看到的）")
names = RUNTIME["tools"]
for i in range(0, len(names), 4):
    print("  " + "  ".join(f"{n:22s}" for n in names[i:i + 4]))
print(f"  合计 {len(names)} 个")

print("\n5. 运行时系统提示各段是否到位")
sp = RUNTIME["system_prompt"]
for label, needle in [
    ("业务提示（硬性规范）", "每个事实性结论必须带来源编号"),
    ("Skills 段（research-report）", "research-report"),
    ("Skills 段（kb-governance）", "kb-governance"),
    ("Memory 段（AGENTS.md 内容）", "知识库编号规则"),
    ("子智能体 researcher", "researcher"),
    ("子智能体 reviewer", "reviewer"),
    ("任务规划 write_todos", "write_todos"),
]:
    print(f"  {label:34s} {'在' if needle.lower() in sp.lower() else '缺失'}")
print(f"\n  系统提示总长 {len(sp)} 字符（约 {len(sp) // 4} token）")

实测输出：

==============================================================================
1. middleware 装配顺序
==============================================================================
   1. SkillsMiddleware
   2. FilesystemMiddleware
   3. SubAgentMiddleware
   4. SummarizationMiddleware
   5. PatchToolCallsMiddleware
   6. TodoListMiddleware
   7. ModelRetryMiddleware
   8. ModelFallbackMiddleware
   9. ToolRetryMiddleware
  10. ToolErrorMiddleware
  11. ModelCallLimitMiddleware
  12. ToolCallLimitMiddleware[kb_search]
  13. ToolCallLimitMiddleware
  14. RubricMiddleware
  15. AnthropicPromptCachingMiddleware
  16. MemoryMiddleware
  17. HumanInTheLoopMiddleware

==============================================================================
2. 权限规则自检：兜底规则必须能挡住点文件
==============================================================================
  1. [deny     ] read/write  ['/**/.env', '/.env', '/**/.*', '/.*']
  2. [deny     ] read/write  ['/**/credentials*', '/**/*secret*', '/**/*token*',
                              '/credentials*', '/*secret*']
  3. [deny     ] write       ['/kb{,/**}']
  4. [interrupt] write       ['/published{,/**}']
  5. [allow    ] read/write  ['/memories{,/**}']
  6. [allow    ] read/write  ['/workspace{,/**}']
  7. [deny     ] read/write  ['/**', '/*']

  点文件（.env / .ssh）被显式 deny: 是

==============================================================================
3. 端到端冒烟：跑一轮（假模型，零 token）
==============================================================================
  跑通，返回 2 条消息
  checkpoint 已写入，next = ()

==============================================================================
4. 运行时工具面（模型真正看到的）
==============================================================================
  delete        edit_file       glob            grep
  kb_get        kb_search       kb_stale_check  ls
  notify_owner  publish_kb_doc  read_file       task
  ticket_query  write_file      write_todos
  合计 15 个

==============================================================================
5. 运行时系统提示各段是否到位
==============================================================================
  业务提示（硬性规范）               在
  Skills 段（research-report）      在
  Skills 段（kb-governance）        在
  Memory 段（AGENTS.md 内容）        在
  子智能体 researcher               在
  子智能体 reviewer                 在
  任务规划 write_todos              在

  系统提示总长 9775 字符（约 2443 token）

三点值得说：

装配顺序完全符合第 50 章的规律。 我们传的 8 个 middleware 整齐地插在 PatchToolCallsMiddleware（第 5 位）和 AnthropicPromptCachingMiddleware（第 15 位）之间。SkillsMiddleware 在最前、MemoryMiddleware 在提示缓存之后、HumanInTheLoopMiddleware 在最后，都是框架排的。

同类 middleware 靠 .name 区分。 第 12 位显示为 ToolCallLimitMiddleware[kb_search]——带 tool_name 的实例会把工具名写进 .name，所以两个 ToolCallLimitMiddleware 不会互相替换（第 50 章的替换规则）。如果两个都不带 tool_name，后一个会顶掉前一个。

Skills 和 Memory 必须在运行时抓。 我一开始在 spy 里抓 create_agent(system_prompt=...)，结果这两段全部显示「缺失」——因为它们是 before_agent 钩子注入的，构造期还不存在。改到假模型的 _generate 里抓 messages[0] 才看得到。要验证系统提示，必须真的跑一轮。

系统提示 9775 字符 ≈ 2443 token。对照第 46 章的基线（工具 schema 约 10894 字符、记忆固定开销 5245 字符、技能固定开销约 2011 字符），这个量级是合理的：业务提示本身不到 600 字，剩下的都是框架的固定成本。

11. 端到端演示 #

自检证明「装上了」，还要证明「连起来了」。用脚本化假模型驱动一条完整业务流：

"""demo.py —— 规划 → 派子智能体 → 审批拦截 → 幂等发布。零 token。"""


def ai(content="", calls=None):
    """构造一条 AIMessage，可带工具调用。"""
    tcs = [{"name": n, "args": a, "id": f"c{i}", "type": "tool_call"}
           for i, (n, a) in enumerate(calls or [], 1)]
    return ChatResult(generations=[ChatGeneration(
        message=AIMessage(content=content, tool_calls=tcs))])


STEP = {"main": 0, "sub": 0}


class ScriptedMain(BaseChatModel):
    """主智能体剧本：列计划 → 派研究员 → 写草稿 → 请求发布 → 收尾。"""

    @property
    def _llm_type(self) -> str:
        return "scripted"

    def bind_tools(self, tools, **kw):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        STEP["main"] += 1
        n = STEP["main"]
        if n == 1:
            return ai(calls=[("write_todos", {"todos": [
                {"content": "查 P1 工单现状", "status": "in_progress"},
                {"content": "写治理报告", "status": "pending"},
                {"content": "发布到知识库", "status": "pending"},
            ]})])
        if n == 2:
            # 注意 description 里带齐全部上下文——子智能体看不到我们的对话
            return ai(calls=[("task", {
                "description": ("查 P1 等级工单的当前状态和等待时长，"
                                "对照 SOP-207 的升级时限，"
                                "把结果写到 /workspace/research/p1.md"),
                "subagent_type": "researcher"})])
        if n == 3:
            return ai(calls=[("write_file", {
                "file_path": "/workspace/reports/p1-draft.md",
                "content": ("## 结论先行\nP1 工单 T-9001 已等待 42 分钟，"
                            "超过 SOP-207 规定的 30 分钟升级时限 "
                            "[T-9001][SOP-207]。\n## 依据\n..."), })])
        if n == 4:
            # 这一步会被 HITL 拦下来
            return ai(calls=[("publish_kb_doc", {
                "doc_id": "RPT-001", "title": "P1 工单升级合规检查",
                "body": "见 /workspace/reports/p1-draft.md",
                "idempotency_key": "RPT-001-2026-09-03"})])
        return ai(content="已完成：报告已发布为 RPT-001，依据 [T-9001][SOP-207]。")


class ScriptedSub(BaseChatModel):
    """子智能体剧本：查工单 → 写文件 → 回一句路径。"""

    @property
    def _llm_type(self) -> str:
        return "sub"

    def bind_tools(self, tools, **kw):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kw) -> ChatResult:
        STEP["sub"] += 1
        if STEP["sub"] == 1:
            return ai(calls=[("ticket_query", {"level": "P1"})])
        if STEP["sub"] == 2:
            return ai(calls=[("write_file", {
                "file_path": "/workspace/research/p1.md",
                "content": "T-9001 P1 支付回调超时，已等待 42 分钟 [T-9001]"})])
        return ai(content="已写入 /workspace/research/p1.md，共 1 条材料。")


app = build_agent(model=ScriptedMain(), checkpointer=InMemorySaver(),
                  store=store, with_rubric=False, fallback_model=ScriptedMain())

cfg = {"configurable": {"thread_id": "demo-001"},
       "recursion_limit": settings.recursion_limit}

out = app.invoke({"messages": [{"role": "user", "content":
                                "检查 P1 工单是否违反升级时限，出报告并发布"}]}, cfg)

输出：

==============================================================================
第一段：规划 → 委派 → 写草稿，直到撞上审批
==============================================================================
  HumanMessage '检查 P1 工单是否违反升级时限，出报告并发布'
  AIMessage    -> 调用 ['write_todos']
  ToolMessage  "Updated todo list to [{'content': '查 P1 工单现状', ...}]"
  AIMessage    -> 调用 ['task']
  ToolMessage  '已写入 /workspace/research/p1.md，共 1 条材料。'
  AIMessage    -> 调用 ['write_file']
  ToolMessage  'Updated file /workspace/reports/p1-draft.md'
  AIMessage    -> 调用 ['publish_kb_doc']

  被中断: True
  action_requests[0] 的字段: ['args', 'description', 'name']
  待批准操作: publish_kb_doc
  参数: doc_id='RPT-001', idempotency_key='RPT-001-2026-09-03'
  可选处置: ['approve', 'edit', 'reject', 'respond']

==============================================================================
第二段：人工批准后继续
==============================================================================
  AIMessage    ''
  ToolMessage  '已发布 [RPT-001] P1 工单升级合规检查，正文 32 字。'
  AIMessage    '已完成：报告已发布为 RPT-001，依据 [T-9001][SOP-207]。'

==============================================================================
第三段：文件系统共享验证 —— 子智能体写的文件主智能体能看到
==============================================================================
  /workspace/research/p1.md  （49 字节，写盘成功）
      首行: T-9001 P1 支付回调超时，已等待 42 分钟 [T-9001]
  /workspace/reports/p1-draft.md  （112 字节，写盘成功）
      首行: ## 结论先行

==============================================================================
第四段：幂等验证 —— 同一个 key 再发一次不会重复执行
==============================================================================
  第二次调用: 该发布请求（key=RPT-001-2026-09-03）此前已执行，本次跳过。
  换个 key:  已发布 [RPT-001] P1 工单升级合规检查，正文 1 字。

主智能体共 5 轮，子智能体共 3 轮。演示完成。

对照前面各章的结论逐条验收：

第一段一共 5 轮模型调用（包括子智能体的 3 轮就是 8 轮），远低于 model_run_limit=40。真实模型会更啰嗦一些，但 40 这个上限对这类任务是合适的量级。

12. 部署与评测 #

12.1. Server 入口 # 

{
  "dependencies": ["."],
  "graphs": {
    "kb-agent": "./src/agent/graph.py:agent"
  },
  "env": ".env"
}
"""src/agent/graph.py —— LangGraph Server 的入口。

注意：这里不要传 checkpointer 和 store —— Server 会注入托管的 Postgres 实现。
自己传会被覆盖，或者更糟：本地开发时用了 InMemorySaver 而不自知。
"""
from .build import build_agent

agent = build_agent()

整个入口就一行，因为所有装配逻辑都在 build.py。这就是 §2 那条组织原则的回报。

12.2. 回归评测 # 

"""evals/run_eval.py —— 回归评测（第 51 章 §8）。
改提示词、换模型之前后各跑一次，对比分数。

    export LANGSMITH_API_KEY=ls__...
    EXP=v2-haiku python evals/run_eval.py
"""
from __future__ import annotations

import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from langsmith import Client, evaluate

from agent.build import build_agent
from agent.config import settings

DATASET = "kb-agent-regression"

CASES = [
    {"inputs": {"q": "P1 工单的升级时限是多少？现在有没有超时的？"},
     "outputs": {"must_contain": ["SOP-207", "30"], "must_not": ["身份证"]}},
    {"inputs": {"q": "客户手机号可以导出到外部报表系统吗？"},
     "outputs": {"must_contain": ["POL-014", "L3"], "must_not": []}},
    {"inputs": {"q": "知识库里哪些文档超过一年没更新了？"},
     "outputs": {"must_contain": ["POL-014"], "must_not": []}},
]


def ensure_dataset(client: Client) -> None:
    """幂等地建数据集。重复跑不会重复插入。"""
    if client.has_dataset(dataset_name=DATASET):
        return
    ds = client.create_dataset(DATASET, description="知识治理助手回归集")
    client.create_examples(dataset_id=ds.id, examples=CASES)


def target(inputs: dict) -> dict:
    """被测目标：一条输入跑成一份输出。"""
    agent = build_agent()
    out = agent.invoke(
        {"messages": [{"role": "user", "content": inputs["q"]}],
         # 顺带用 rubric 做运行时自评，评测里能一起看到它的效果
         "rubric": "每个事实性结论必须带来源编号；不确定处要写明未找到依据。"},
        # 每个用例独立 thread，否则用例之间会串上下文
        {"configurable": {"thread_id": f"eval-{abs(hash(inputs['q']))}"},
         "recursion_limit": settings.recursion_limit},
    )
    return {"answer": out["messages"][-1].content}


def citation_check(outputs: dict, reference_outputs: dict) -> dict:
    """规则型评估器：必须出现的编号在不在、禁止出现的词有没有漏出来。"""
    ans = outputs.get("answer", "")
    missing = [k for k in reference_outputs["must_contain"] if k not in ans]
    leaked = [k for k in reference_outputs["must_not"] if k in ans]
    score = 1.0 if not missing and not leaked else 0.0
    return {"key": "citation_and_safety", "score": score,
            "comment": f"缺失={missing} 泄露={leaked}" if score == 0 else "通过"}


def rubric_judge(inputs: dict, outputs: dict) -> dict:
    """LLM-as-judge：用和线上 rubric 一致的标准打分。"""
    from langchain.chat_models import init_chat_model

    judge = init_chat_model(settings.grader_model)
    verdict = judge.invoke(
        "按标准给回答打分，只回 1 或 0。\n"
        "标准：结论有来源编号支撑，不确定处明确写了「未找到依据」。\n\n"
        f"问题：{inputs['q']}\n回答：{outputs.get('answer', '')}"
    ).content.strip()
    return {"key": "rubric_quality", "score": 1.0 if verdict.startswith("1") else 0.0}


if __name__ == "__main__":
    os.environ.setdefault("LANGSMITH_TRACING", "true")
    client = Client()
    ensure_dataset(client)
    results = evaluate(
        target,
        data=DATASET,
        evaluators=[citation_check, rubric_judge],
        experiment_prefix=os.getenv("EXP", "baseline"),
        max_concurrency=2,        # deep agent 单例开销大，别开太高
    )
    print(results)

两个评估器一规则一模型，是有意的搭配：规则型评估器便宜、稳定、能查具体缺了哪个编号，适合做守门；LLM-as-judge 能评规则表达不出来的东西（比如「不确定处有没有诚实标注」），但自己也会波动。只用后者，你会分不清分数下降是模型退步还是评委抽风。

13. 从 49 到 62：这门课的地图 #

十四章下来，Deep Agents 的四组能力在这个项目里各自落到了哪：

能力组 部件 在本项目里的体现
Harness 内置工具面、middleware 栈、profile verify.py 输出的 17 层装配、15 个工具
上下文管理 文件系统、Skills、Memory、摘要、卸载 草稿走 /workspace、两个技能、AGENTS.md、9775 字符系统提示
委派 write_todos、task、子智能体 研究员 + 审校员，主智能体只见摘要
操控 interrupt_on、permissions 发布必批、.env 不可见、发布区写入需确认

更值得回味的是为什么这门课把 Deep Agents 放在最后。这个项目里的每个部件，前面都手写过一遍：

先手写一遍再用内置件，你才知道它替你省掉了什么、以及默认值不合适时该改哪一层。 本章那四个坑就是例子——如果不清楚「skills 路径要过 backend」「子智能体权限是替换不是叠加」这些底层机制，撞上去只会一头雾水。

14. 常见坑 #

按撞到的先后顺序：

坑 1：权限路径不以 / 开头。 ValueError: Permission path must start with '/'。相对 glob 完全不能用，要匹配任意位置得写两条：/**/x 和 /x。

坑 2：SubAgent.tools 传字符串名。 报错是 AttributeError: 'function' object has no attribute 'name'，堆栈在 ToolNode.__init__，完全看不出真正原因。只能传工具对象。注意这和 FilesystemMiddleware(tools=)、excluded_tools、ToolRetryMiddleware(tools=) 的约定都不一样，后三者都收字符串。

坑 3：以为 SubAgent.tools 收窄了工具面。 它只管业务工具，七个内置文件工具照给不误，包括 delete。要收窄得单独挂 FilesystemMiddleware(tools=[白名单])。

坑 4：skills= 传磁盘绝对路径。 它是交给 backend 解析的虚拟路径。ValueError: Path ... outside root directory。推广一下：Deep Agents 里除了 FilesystemBackend(root_dir=)，其它所有路径参数都是虚拟路径。

坑 5：在构造期检查系统提示。 Skills 和 Memory 是 before_agent 钩子注入的，构造期抓到的 system_prompt 里没有它们。要验证必须真跑一轮，在模型的 _generate 里抓 messages[0]。

坑 6：ModelFallbackMiddleware 构造期就解析模型名。 零 token 测试里传假模型名会直接 ValueError，得留注入口。

坑 7：action_requests 里是 name 不是 action。 取待批准的工具名时容易写错，报 KeyError。

坑 8：子智能体和主智能体用了不同的 backend 实例。 表现是「研究员说文件写好了，主智能体读不到」。把 build_subagents 写成接收 backend 的工厂函数就能避免。

坑 9：MCP 工具和内置工具撞名。 静默覆盖，不报错。批量改名是唯一可靠的防法。

坑 10：幂等键由工具内部生成。 断点续跑重放时会生成新键，重复执行。做成显式参数，让它跟着工具调用一起被恢复。

15. 练习 #

  1. 换成真模型跑通。 配好 ANTHROPIC_API_KEY，把 demo.py 的脚本化模型换掉，跑一次「检查 P1 工单是否违反升级时限」。对照 research-report 技能，看它有没有按五段结构写、有没有带来源编号。没按规范的地方，是技能文档写得不够明确，回去改技能而不是改系统提示。

  2. 接一个真的 MCP 服务器。 用第 17 章的方式起一个本地 MCP server 暴露两个工具，通过 mcp_tools() 接进来。故意让其中一个叫 grep，验证改名逻辑生效。

  3. 补第三个子智能体。 加一个 summarizer，专门把长报告压成三句话的摘要。给它配 response_format（第 47 章）让它返回结构化的 {summary, word_count}，而不是自由文本。

  4. 权限渗透测试。 写一个脚本，让智能体尝试读 /.env、/workspace/../secrets.txt、/kb/POL-001.md（写）。确认前两个被 deny、第三个被拒绝写入。把这个脚本加进 verify.py。

  5. 断点续跑演练。 把 publish_kb_doc 改成第一次调用抛 ConnectionError。跑一遍，确认崩在发布这一步；然后修好、用同一 thread_id 调 invoke(None, cfg)，确认前面的检索和写草稿都没重跑，且幂等键保证不会重复发布。

  6. 上前端。 用第 49 章的 stream_events(version="v3") 做一个执行面板，把 stream.subagents 的每个句柄渲染成一张卡片（研究员在查什么、审校员进行到哪），write_todos 的结果渲染成任务清单。

  7. 过一遍第 51 章的上线清单。 逐条对照本项目，把不满足的补上。特别注意：_DONE 那个内存 set 在多实例部署下完全失效，改成数据库表。

16. 小结 #

这一章没有新的 API，全是前面十三章的组装。真正的收获是四个只有真接起来才会遇到的坑，以及它们背后的共同原因：

Deep Agents 里有两套路径体系。 磁盘路径只出现在 FilesystemBackend(root_dir=) 一处，其它全是模型视角的虚拟路径。skills= 的坑（§8.2）就是把两者混了。

「限制」这个词在不同参数里含义不同。 SubAgent.tools 限制的是业务工具，不含文件工具；FilesystemMiddleware(tools=) 限制的是文件工具；permissions 限制的是路径。三者正交，要三管齐下才是真的收窄。这也是为什么 §6.2 那个坑那么容易踩——名字都叫 tools，管的东西不一样。

继承规则要一个个记。 permissions 是替换、tools 指定则覆盖、skills 不继承、middleware 按名替换。没有统一规律，第 47 章那张表值得贴在墙上。

验证要在运行时做。 构造期看到的和模型实际看到的是两回事（§10 的第三点）。verify.py 这类「跑一轮假模型然后检查」的自检脚本，比读文档和读代码都可靠。

最后回到这门课开头的问题：什么时候该用 Deep Agents，什么时候该自己搭？

这个项目给出的答案是——当你需要的东西超过三样内置能力时。 我们用到了文件系统、技能、记忆、子智能体、审批、容错六样。自己搭这六样，光是让它们正确地互相配合（比如子智能体共享文件系统、审批能中断在工具执行前、摘要不破坏提示缓存）就是几个月的工作。而如果你只需要「一个会调三个工具的问答机器人」，create_agent 加两个 middleware 就够了，上 harness 反而要为一堆用不到的能力付 token。

工具的价值不在于它能做什么，而在于它替你省掉了什么。这门课从第 1 章手写 create_agent 到这里，就是为了让你能算清这笔账。

← 上一节 51.生产化容错评测与部署 下一节 all-MiniLM-L6-v2 →