Deep Agents 技术剖析：LangChain 的开源 Agent Harness

┌─────────────────────────────────────────┐
│  Deep Agents (Harness)                  │
│  planning · filesystem · subagents      │
│  context compaction · skills · memory   │
├─────────────────────────────────────────┤
│  LangChain (Framework)                  │
│  models · messages · tools · agents     │
├─────────────────────────────────────────┤
│  LangGraph (Runtime)                    │
│  state · checkpointing · streaming      │
│  durable execution · human-in-the-loop  │
└─────────────────────────────────────────┘

def create_deep_agent(
    model: str | BaseChatModel | None = None,          # 默认 claude-sonnet-4-6
    tools: Sequence[BaseTool | Callable | dict] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware] = (),          # 用户自定义中间件
    subagents: Sequence[SubAgent | CompiledSubAgent | AsyncSubAgent] | None = None,
    skills: list[str] | None = None,                    # SKILL.md 目录路径
    memory: list[str] | None = None,                    # AGENTS.md 路径
    response_format: ResponseFormat | type | dict | None = None,
    context_schema: type | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    backend: BackendProtocol | BackendFactory | None = None,
    interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
    debug: bool = False,
    name: str | None = None,
    cache: BaseCache | None = None,
) -> CompiledStateGraph:

1.  TodoListMiddleware          ← planning (write_todos 工具)
2.  SkillsMiddleware            ← 渐进式技能加载 (如果提供了 skills)
3.  FilesystemMiddleware        ← 文件工具 (ls/read/write/edit/glob/grep)
4.  SubAgentMiddleware          ← task 工具 (同步 subagent)
5.  SummarizationMiddleware     ← 自动上下文压缩
6.  PatchToolCallsMiddleware    ← 修复悬挂的 tool calls
7.  AsyncSubAgentMiddleware     ← 远程/后台 subagent (如果提供了)
8.  >>> 用户自定义中间件插入点 <<<
9.  AnthropicPromptCachingMiddleware  ← prompt 缓存优化 (非 Anthropic 模型静默跳过)
10. MemoryMiddleware            ← AGENTS.md 注入到 system prompt
11. HumanInTheLoopMiddleware    ← 审批门控 (如果配置了 interrupt_on)

{
    "content": "Query Prometheus for CPU metrics across all nodes",
    "status": "in_progress"  # pending | in_progress | completed
}

class TaskToolSchema(BaseModel):
    description: str    # 任务描述
    subagent_type: str  # subagent 类型名

# 1. 声明式同步 — 最常用
SubAgent = TypedDict("SubAgent", {
    "name": str,                   # 必需
    "description": str,            # 必需
    "system_prompt": str,          # 必需
    "tools": Sequence[...],        # 可选，默认继承父 agent
    "model": str | BaseChatModel,  # 可选，默认继承父 agent
    "middleware": list[...],       # 可选，额外中间件
    "skills": list[str],           # 可选，不继承父 agent（需显式指定）
})

# 2. 预编译 — 自定义 runnable
CompiledSubAgent = TypedDict("CompiledSubAgent", {
    "name": str,
    "description": str,
    "runnable": Runnable,  # state schema 须包含 'messages' key
})

# 3. 异步远程 — 后台执行
AsyncSubAgent = TypedDict("AsyncSubAgent", {
    "name": str,
    "description": str,
    "graph_id": str,      # 远程 LangGraph 服务器上的 graph ID
    "url": str,           # Agent Protocol 服务器 URL
    "headers": dict,      # 认证头
})

class TruncateArgsSettings(TypedDict, total=False):
    trigger: ContextSize | None    # 何时激活
    keep: ContextSize              # 保留最近多少消息不动
    max_length: int                # 每个参数值的字符上限（默认 2000）
    truncation_text: str           # 截断后缀（默认 "...(argument truncated)"）

SummarizationMiddleware(
    model=model,
    backend=backend,
    trigger=("fraction", 0.85),  # 达到模型 max_input_tokens 的 85% 时触发
    keep=("fraction", 0.10),     # 保留最后 10% 的消息
)

from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend(routes={
    "/memories/": StoreBackend(config),   # 持久化记忆
    "/": StateBackend(config),            # 临时工作文件
})

class FileData(TypedDict):
    content: str          # UTF-8 文本或 base64 编码二进制
    encoding: str         # "utf-8" 或 "base64"
    created_at: str       # ISO 8601（可选）
    modified_at: str      # ISO 8601（可选）

skills/
  alert-investigation/
    SKILL.md          # YAML frontmatter + markdown 指令
    helper.py         # 可选辅助文件
  blog-post/
    SKILL.md

---
name: alert-investigation
description: >
  Investigates production alerts by correlating metrics, logs,
  and infrastructure state.
---

# Alert Investigation Workflow
## Phase 1: Triage
...

agent = create_deep_agent(
    memory=["~/.deepagents/AGENTS.md", "./.deepagents/AGENTS.md"],
    backend=FilesystemBackend(root_dir="/"),
)

from deepagents import create_deep_agent, SubAgent
from deepagents.backends import FilesystemBackend
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.checkpoint.memory import MemorySaver

metric_tools = await load_mcp_tools(connection=grafana_connection)
infra_tools = await load_mcp_tools(connection=infra_connection)

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-5-20250929",
    tools=[*metric_tools, *infra_tools],
    system_prompt="""\
You are a Cloud Operations Agent. Investigate production incidents.

## Decision Flow
1. Match request to a skill -> read SKILL.md -> execute
2. Pasted alert data -> use alert-investigation skill
3. No match -> plan with write_todos -> execute

## Rules
- Complete every investigation in <=12 tool calls
- Batch parallel tool calls where possible
""",
    memory=["./AGENTS.md"],
    skills=["./skills/"],
    subagents=[{
        "name": "metrics-collector",
        "description": "Query Prometheus/Grafana metrics in isolation",
        "system_prompt": "You are a metrics specialist.",
        "model": "anthropic:claude-haiku-4-5-20251001",  # 便宜模型做子任务
    }],
    backend=FilesystemBackend(root_dir="./workspace"),
    checkpointer=MemorySaver(),
    interrupt_on={"execute": True},  # shell 执行需要人工审批
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "CPU >95% on web-us-east-1-app3"}]},
    config={"configurable": {"thread_id": "incident-42"}},
)

# 安装
curl -LsSf https://raw.githubusercontent.com/langchain-ai/deepagents/main/libs/cli/scripts/install.sh | bash

# 带额外模型提供商
DEEPAGENTS_EXTRAS="nvidia,ollama" curl -LsSf ... | bash

# 或直接
uv tool install 'deepagents-cli[nvidia,ollama]'

from deepagents_acp.server import AgentServerACP

agent = create_deep_agent(tools=[...], checkpointer=MemorySaver())
server = AgentServerACP(agent, models=[
    {"value": "anthropic:claude-opus-4-6", "name": "Claude Opus 4"},
    {"value": "openai:gpt-4-turbo", "name": "GPT-4 Turbo"},
])
await run_agent(server)

{
  "agent": {
    "profiles": {
      "deepagents": {
        "name": "DeepAgents",
        "command": "npx",
        "args": ["deepagents-acp"]
      }
    }
  }
}

LangGraph Dev Server (127.0.0.1:2024)
    │
    ▼
Deep Agent (Nemotron Super 3)
    │
    ├── OpenShellBackend (gRPC → 沙箱执行)
    └── FilesystemBackend (持久化 memory/skills)
    │
    ▼
OpenShell Gateway (k3s in Docker)
    │
    ▼
Sandbox Container (策略强制执行)

uv run openshell policy get deepagent-sandbox --full > policy.yaml
# 编辑 policy.yaml（文件系统路径、网络访问、进程权限）
uv run openshell policy set deepagent-sandbox --policy policy.yaml --wait