第 14 章 · 综合实战三：规划、委派与审批的研究流水线

把第 5、8、9、10 章合成一条研究流水线：write_todos 规划 → researcher 子智能体收集 → 按 skill 起草报告 → publish_report 前暂停等人工批准。检索走本地 fixture，审批是唯一的真实副作用点。

机制与设计选择

子智能体是 stateless 的一次性收集器，只把结论带回主 agent，搜索过程不进主上下文；TodoListMiddleware 提供可检查的规划状态；interrupt_on 让发布可被拒绝。结构自检只验证 task 工具在位，不需要调用模型。

示例

python

import os, pathlib, tempfile, shutil
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents.middleware import TodoListMiddleware
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from deepagents import create_deep_agent

FIXTURE_RESULTS = {
    "langgraph": "LangGraph: durable execution, streaming, human-in-the-loop.",
    "deepagents": "Deep Agents: harness with file tools, subagents, and context management.",
}

@tool
def search_topics(query: str) -> str:
    """Search the local course knowledge base (fixture)."""
    key = query.strip().lower().split()[0] if query else ""
    return FIXTURE_RESULTS.get(key, "No results.")

@tool
def publish_report(report: str) -> str:
    """Publish the final research report (mock side effect)."""
    return "published"

root = tempfile.mkdtemp(prefix="deepagents-pipeline-")
try:
    skill = pathlib.Path(root) / "skills" / "research-report"
    skill.mkdir(parents=True)
    (skill / "SKILL.md").write_text(
        "---\nname: research-report\ndescription: Produce a short structured research report\n---\n"
        "Structure: 1) Scope, 2) Findings (with source labels), 3) Open questions. Under 15 lines.\n"
    )
    model = ChatOpenAI(
        model=os.environ["MODEL_ID"],
        api_key=os.environ["DEEPAGENT_API_KEY"],
        base_url=os.environ.get("DEEPAGENT_BASE_URL", "https://api.openai.com/v1"),
    )
    researcher = {
        "name": "researcher",
        "description": "Collects findings from search_topics and returns concise bullets",
        "system_prompt": "Collect only what the task asks for. Return findings as short bullets with source labels.",
        "tools": [search_topics],
    }
    agent = create_deep_agent(
        model=model,
        tools=[search_topics, publish_report],
        subagents=[researcher],
        skills=[str(skill.parent)],
        middleware=[TodoListMiddleware()],
        interrupt_on={"publish_report": True},
        checkpointer=MemorySaver(),
        system_prompt=(
            "Plan with write_todos, delegate collection to the researcher subagent, "
            "draft the report per the research-report skill, then call publish_report."
        ),
    )
    # Structural check (no model call needed):
    available = sorted(agent.nodes["tools"].bound.tools_by_name)
    assert "task" in available, available
    print("task tool available; planning + subagent + approval wired.")

    if os.environ.get("DEEPAGENT_API_KEY") not in ("", "sk-fake"):
        cfg = {"configurable": {"thread_id": "ch14"}}
        paused = agent.invoke(
            {"messages": [{"role": "user", "content": "Compare langgraph and deepagents, then publish the report."}]},
            cfg,
        )
        assert paused["__interrupt__"]
        final = agent.invoke(Command(resume=True), config=cfg, version="v2")
        print(final.value["messages"][-1].content)
    else:
        print("model step skipped: DEEPAGENT_API_KEY not configured; structure check passed.")
finally:
    shutil.rmtree(root, ignore_errors=True)

运行与验收

无 key 运行：打印 task tool available 与 model step skipped。配置 key 运行：第一次 invoke 停在 publish_report 之前（interrupt 非空），resume=True 后输出报告并打印 published。

容易踩的坑

把 fixture 检索换成真实搜索后，子智能体的返回质量直接决定报告质量，委派描述要写清验收标准。审批不能替代幂等：resume 重放仍可能重复执行发布，生产要在 publish 侧加唯一请求 ID。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 子智能体回传什么？

2. task 工具缺了会怎样？

3. 发布副作用要防什么？

4. 规划工具的状态？

动手实践

把 researcher 换成并行两个子智能体并合并结论。
publish_report 加幂等键并测试重复 resume。
把规划状态推到前端进度条。

官方参考

核验日期：2026-09-17，示例已在本机离线运行验证。

第 14 章 · 综合实战三：规划、委派与审批的研究流水线 ​

机制与设计选择 ​

示例 ​

运行与验收 ​

容易踩的坑 ​