Chapter 14 · Capstone 3: planning, delegation, and approval research pipeline

Fuse chapters 5, 8, 9, and 10 into a research pipeline: write_todos plans, the researcher subagent collects, a skill drafts the report, and publish_report pauses for human approval. Retrieval is a local fixture; approval is the only real side effect.

Mechanics and design choices

The subagent is a one-shot, stateless collector: it brings conclusions back to the main agent, and the search process never enters the main context. TodoListMiddleware supplies inspectable planning state; interrupt_on makes publishing rejectable. The structural check only verifies the task tool is present, with no model call.

Example

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)

Run and verify

Run without a key: it prints task tool available and model step skipped. With a key: the first invoke pauses before publish_report (interrupt is non-empty), and resume=True prints the report and published.

Pitfalls

Once the fixture is replaced by real search, subagent output quality decides report quality; state acceptance criteria in the delegation description. Approval does not replace idempotency: resume replays the node, so production publishing needs a unique request ID.

🧪 随堂测验

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

1. What does a subagent return?

2. What if the task tool is missing?

3. What must publishing guard against?

4. What does the planning tool track?

Hands-on exercises

1. Use two parallel researchers and merge conclusions.
2. Add an idempotency key to publish_report and test repeated resume.
3. Push planning state to a frontend progress bar.

Official references

• subagents
• human-in-the-loop
• context-engineering

Verified on 2026-09-17; examples were run locally without network access.