Skip to content

Chapter 12 · Capstone: files + skills + delegation + approval ​

Combine file tools, skills, subtask delegation, and human approval into one loop: the agent reads workspace files, drafts a report per the skill, pauses before the publish tool, and finishes only after approval.

Mechanics and design choices ​

This chapter is a minimal end-to-end example, deliberately without a vector store or cloud deployment. Each part is independently replaceable so it can be compared against the previous 11 chapters.

Example ​

python
import os
from langchain_openai import ChatOpenAI

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"),
)

import pathlib, shutil, tempfile
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

root = tempfile.mkdtemp(prefix="deepagents-cap-")
try:
    ws = pathlib.Path(root) / "workspace"
    ws.mkdir()
    (ws / "notes.md").write_text("# Notes\n- Ship v1 this week\n")
    skill = pathlib.Path(root) / "skills" / "report"
    skill.mkdir(parents=True)
    (skill / "SKILL.md").write_text(
        "---\nname: report\ndescription: Produce concise reports\n---\nUse bullets, keep under 10 lines.\n"
    )

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

    agent = create_deep_agent(
        model=model,
        tools=[publish_report],
        backend=FilesystemBackend(root_dir=str(ws), virtual_mode=True),
        skills=[str(skill.parent)],
        interrupt_on={"publish_report": True},
        checkpointer=MemorySaver(),
        system_prompt=(
            "Read /workspace/notes.md with read_file, draft a short report following the report skill, "
            "then call publish_report."
        ),
    )
    cfg = {"configurable": {"thread_id": "cap-1"}}
    paused = agent.invoke(
        {"messages": [{"role": "user", "content": "Prepare the release report."}]},
        config=cfg,
    )
    print("Paused before publish:", bool(paused["__interrupt__"]))

    from langgraph.types import Command
    final = agent.invoke(Command(resume=True), config=cfg, version="v2")
    print(final.value["messages"][-1].content)
finally:
    shutil.rmtree(root, ignore_errors=True)

Run and verify ​

API key required. Acceptance: the first invoke pauses before publish_report (Paused is True); after resuming with True, the tool runs and a final message is printed.

Pitfalls ​

Approval is not idempotency: node reruns can still duplicate external side effects, so the publish side must enforce its own idempotency (unique request_id plus a state table).

🧪 随堂测验

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

1. Which capabilities are combined?

2. What to guard against on publish reruns?

3. Do skills carry security?

4. Where does the first invoke stop?

Hands-on exercises ​

  1. Reject once and observe how the rejection is recorded in history.
  2. Replace publish_report with a real webhook test endpoint.
  3. Swap the skill for your project standards and verify acceptance.

Official references ​

Verified on 2026-09-17 with Deep Agents 0.7.15.