Claude Agent SDK 架构实践：从 Skill 到 MCP，构建多 Agent 协同系统

做了 1 年 LLM 应用，最大的认知升级是：**”调 LLM API” 和 “构建 Agent 系统” 是两个完全不同的事**。

这篇文章复盘我们怎么用 Claude Agent SDK + Skills + MCP + LangGraph 搭出可观测、可降级、可扩展的多 Agent 协同架构。

---

一、为什么需要 Agent SDK

2024 年中，我们开始做政策智能体。第一版直接调 Anthropic API：

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    tools=[fetch_url_tool, search_policy_tool, calculate_tool],
    messages=[{"role": "user", "content": "营改增对小微企业有什么影响？"}]
)

跑了一段时间后发现问题：

1. 每次调用都要写 tool 描述——重复劳动
2. 无法跨调用共享状态——多轮对话要自己维护 history
3. 错误处理到处散落——每个调用都写 try-except
4. 多步决策的逻辑糊在一起——一个 system prompt 里塞太多规则
5. 没有可观测性——不知道 Agent 在想什么、哪步出错了

于是开始用 Claude Agent SDK——Anthropic 官方提供的 Agent 编排框架。

---

二、Claude Agent SDK 的核心概念

2.1 Skill（技能）

Skill = 可复用的能力单元。一个 Skill 包含：

• System Prompt：告诉 Agent 这个 Skill 是干什么的、什么时候用
• Tools：Skill 能调用的工具
• Examples：few-shot 例子
• Instructions：执行细节

# skills/policy-retrieval/SKILL.md
---
name: policy-retrieval
description: 从政策知识库中检索与用户问题最相关的政策原文
---

# Policy Retrieval Skill

## 用途
当用户提出政策相关问题时，使用此 Skill 检索相关政策原文。

## 工具
- `pgvector_query(query, top_k=5)`: 在政策向量库中检索
- `rerank(query, docs)`: 对检索结果精排

## 执行步骤
1. 理解用户问题，提取关键实体（如税种、年份、行业）
2. 用 pgvector_query 检索 top-20 文档
3. 用 rerank 精排，取 top-5
4. 输出文档 ID 列表（不直接生成答案）

## 注意事项
- 不要在此 Skill 中生成答案，交给上层
- 如果检索结果相关性都 < 0.7，返回空让上层决定

2.2 Tool（工具）

Tool = Agent 能调用的具体函数。

# tools/pgvector_query.py
from claude_agent_sdk import tool

@tool(
    name="pgvector_query",
    description="在 PostgreSQL pgvector 知识库中检索相关文档",
    input_schema={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "查询文本"},
            "top_k": {"type": "integer", "default": 5},
        },
        "required": ["query"],
    },
)
async def pgvector_query(query: str, top_k: int = 5) -> dict:
    """实际的检索逻辑"""
    embedding = await embed_text(query)
    results = await db.fetch("""
        SELECT id, title, content, 1 - (embedding <=> $1) AS score
        FROM policies
        ORDER BY embedding <=> $1
        LIMIT $2
    """, embedding, top_k)

    return {"results": [dict(r) for r in results]}

2.3 MCP（Model Context Protocol）

MCP = 标准化的 Tool 提供协议。一个 MCP server 可以提供多个 Tool。

# mcp_servers/policy_server.py
from mcp.server import Server
from mcp.types import Tool

app = Server("policy-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(name="pgvector_query", description="...", input_schema={...}),
        Tool(name="policy_search_by_date", description="...", input_schema={...}),
        Tool(name="policy_compare", description="...", input_schema={...}),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "pgvector_query":
        return await pgvector_query(**arguments)
    # ...

好处：MCP 是 Anthropic 推动的开放协议，不同 SDK 可以共用。

2.4 Agent Session（会话）

Agent Session = 一次完整的对话，包含：

• 多轮消息历史
• Skill 选择
• Tool 调用记录
• 中间状态

from claude_agent_sdk import Agent

agent = Agent(
    name="policy-assistant",
    skills=[
        "policy-retrieval",
        "policy-compare",
        "policy-rag",
        "site-analyze",
        "tag-selection",
    ],
    mcp_servers=[
        "policy-server",    # 政策检索 / 对比
        "crawler-server",   # 爬虫
        "asset-server",     # 资源处理
        "ingest-server",    # 数据入库
        "persist-server",   # 持久化
    ],
    max_turns=25,
    timeout_s=1800,  # 30 分钟
)

# 单次执行
result = await agent.run(
    user_query="营改增对小微企业有什么影响？",
    session_id="user_123_session_456",
)

---

三、我们的 6 大 Skill 设计

我们设计了 6 大 Skill，每个 Skill 职责单一：

┌─────────────────────────────────────────────────────────┐
│                  6 大 Skill 架构                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  policy-retrieval  ──→  单条政策检索                     │
│       ↓                                                │
│  policy-rag        ──→  RAG 摘要 + 引用                  │
│       ↓                                                │
│  policy-compare    ──→  两条政策对比（LangGraph）       │
│       ↓                                                │
│  site-analyze      ──→  爬虫站点分析                     │
│       ↓                                                │
│  tag-selection     ──→  政策自动打标签                   │
│       ↓                                                │
│  crawler-skill     ──→  跨站点爬取编排                   │
│                                                         │
└─────────────────────────────────────────────────────────┘

3.1 Skill 设计原则

我们总结出 4 条 Skill 设计原则：

原则 1：单一职责

每个 Skill 只做一件事。比如 policy-retrieval 只负责检索，不负责生成答案——交给上层 Skill。

原则 2：可独立测试

每个 Skill 都可以单独跑测试，不依赖其他 Skill：

# tests/test_policy_retrieval.py
async def test_retrieval_basic():
    agent = Agent(skills=["policy-retrieval"])
    result = await agent.run("增值税小微企业")
    assert len(result["results"]) > 0
    assert all(r["score"] > 0.7 for r in result["results"])

原则 3：显式输入输出

每个 Skill 必须有清晰的 schema：

# Input
{
  "query": str,
  "top_k": int  # default 5
}

# Output
{
  "documents": [
    {"id": str, "title": str, "content": str, "score": float}
  ],
  "total_retrieved": int
}

原则 4：失败显式化

不抛异常，而是返回 success=False + error_message：

{
  "success": False,
  "error_message": "向量库连接超时",
  "fallback_suggestion": "切换到 BM25 检索"
}

3.2 Skill 与 LangGraph 混用

复杂任务（如政策对比）需要多分支状态机，单 Skill 不够：

from langgraph.graph import StateGraph

# 政策对比 DAG
compare_graph = StateGraph()

compare_graph.add_node("fetch_a", fetch_policy_a)
compare_graph.add_node("fetch_b", fetch_policy_b)
compare_graph.add_node("extract_diff", extract_diff_sections)
compare_graph.add_node("render", render_comparison_table)

compare_graph.add_edge("fetch_a", "extract_diff")
compare_graph.add_edge("fetch_b", "extract_diff")
compare_graph.add_edge("extract_diff", "render")
compare_graph.set_finish_point("render")

# 编译后作为 LangGraph 节点嵌入 Claude Agent
compiled = compare_graph.compile()

agent = Agent(
    skills=["policy-retrieval", "policy-rag"],
    custom_nodes={
        "policy_compare": compiled,  # ← LangGraph 嵌入 Agent
    },
)

好处：简单的 Skill 用 Claude Agent SDK，复杂状态机用 LangGraph，各取所长。

---

四、7 个 MCP 工具 + 5 个 Server 设计

4.1 MCP 架构

┌─────────────────────────────────────────────────────────┐
│                    Claude Agent                           │
│                                                         │
│  skills:  [policy-retrieval, policy-rag, ...]            │
└────────────────────┬────────────────────────────────────┘
                     │ MCP Protocol
        ┌────────────┼────────────┬─────────────┐
        ▼            ▼            ▼             ▼
   ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐
   │policy- │  │crawler-│  │ asset- │  │ingest- │
   │ server │  │ server │  │ server │  │ server │
   │ (3 工具)│  │ (2 工具)│  │ (1 工具)│  │ (1 工具)│
   └────────┘  └────────┘  └────────┘  └────────┘

4.2 MCP Server 实现

# mcp_servers/crawler_server.py
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("crawler-server")


@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="fetch_url",
            description="通过爬虫抓取 URL 的 HTML 内容",
            input_schema={
                "type": "object",
                "properties": {
                    "url": {"type": "string"},
                    "render_js": {"type": "boolean", "default": True},
                },
                "required": ["url"],
            },
        ),
        Tool(
            name="parse_html",
            description="从 HTML 中提取结构化数据（CSS selector）",
            input_schema={
                "type": "object",
                "properties": {
                    "html": {"type": "string"},
                    "selectors": {"type": "object"},
                },
                "required": ["html", "selectors"},
            },
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "fetch_url":
        url = arguments["url"]
        render_js = arguments.get("render_js", True)

        # 调用实际的爬虫
        async with crawler_semaphore:
            html = await crawler.fetch(url, render_js=render_js)
        return [TextContent(type="text", text=html)]

    elif name == "parse_html":
        # 解析逻辑
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(arguments["html"], "html.parser")
        result = {}
        for key, selector in arguments["selectors"].items():
            elements = soup.select(selector)
            result[key] = [el.get_text(strip=True) for el in elements]
        return [TextContent(type="text", text=json.dumps(result))]

4.3 MCP Server 权限隔离

不同 Skill 应该只能访问必要的 MCP Server：

# config/skill_permissions.yaml
policy-retrieval:
  mcp_servers: [policy-server]
  tools: [pgvector_query, policy_search_by_date]

policy-compare:
  mcp_servers: [policy-server]
  tools: [pgvector_query, policy_compare]

site-analyze:
  mcp_servers: [crawler-server, policy-server]
  tools: [fetch_url, parse_html, pgvector_query]

crawler-skill:
  mcp_servers: [crawler-server, asset-server, ingest-server, persist-server]
  tools: [fetch_url, parse_html, download_asset, submit_policy, write_run_log]

好处：最小权限原则，避免 Skill 越权调用。

---

五、可观测性：让 Agent 行为可解释

5.1 全链路 Trace

接入 LangSmith，自动记录：

from langsmith import traceable

@traceable(name="agent.session")
async def run_agent_session(session_id, query):
    # 自动 trace：每次 Skill 选择、Tool 调用、Token 消耗、延迟
    return await agent.run(query, session_id=session_id)

打开 LangSmith 看一次完整 trace：

[Session: user_123_session_456]
  ↳ Query: "营改增对小微企业有什么影响？"

  [Turn 1] Skill Selection
    ↳ Selected: policy-retrieval (confidence 0.92)
    ↳ Latency: 50ms

  [Turn 2] Tool Call: pgvector_query
    ↳ Input: {"query": "营改增 小微企业", "top_k": 5}
    ↳ Output: 5 documents, score 0.78-0.92
    ↳ Latency: 120ms, tokens: 1500

  [Turn 3] Skill Selection
    ↳ Selected: policy-rag (confidence 0.88)

  [Turn 4] Tool Call: llm_generate
    ↳ Input prompt: 6500 tokens
    ↳ Output: 800 tokens
    ↳ Model: qwen-plus
    ↳ Latency: 2.3s, tokens: 7300

  [Final Answer]
    ↳ Citations: [doc_001, doc_003, doc_005]
    ↳ Faithfulness score: 0.94

没有 trace 时，Agent 失败 = 黑盒；有 trace，每一步都可解释、可优化。

5.2 失败模式分类

我们建立了 Agent 失败的 5 类模式：

失败模式	表现	解法
Skill 选错	选了不相关的 Skill	优化 Skill 描述 + Few-shot
Tool 调用错误	参数格式错	Tool schema 校验 + 严格类型
Token 超限	上下文窗口爆	摘要压缩 + 滑动窗口
无限循环	Agent 反复调用同一 Tool	max_turns 限制 + 状态检测
幻觉	输出与检索上下文不符	RAG + LLM-as-a-Judge 评估

5.3 评估体系

每月跑 Golden Dataset（200 条对话）评估：

test_cases = [
    {
        "query": "营改增对小微企业的影响",
        "expected_skills": ["policy-retrieval", "policy-rag"],
        "expected_doc_ids": ["doc_001", "doc_003"],
        "expected_citation_accuracy": 0.9,
    },
    # ... 199 more
]

async def evaluate_agent(agent, test_cases):
    results = []
    for case in test_cases:
        result = await agent.run(case["query"])

        # 1. Skill 选择准确率
        skill_accuracy = compute_skill_accuracy(
            result["skills_used"], case["expected_skills"]
        )

        # 2. 引用准确率
        citation_accuracy = compute_citation_accuracy(
            result["citations"], case["expected_doc_ids"]
        )

        # 3. Ragas 评估
        ragas_score = await ragas_evaluate(result, case)

        results.append({
            "skill_accuracy": skill_accuracy,
            "citation_accuracy": citation_accuracy,
            "ragas_faithfulness": ragas_score["faithfulness"],
        })

    return aggregate_metrics(results)

平均指标：

指标	数值
Skill 选择准确率	92%
引用准确率	88%
Faithfulness	0.94
Context Recall	0.86
平均任务延迟	3.2 秒

---

六、降级策略：Agent 失败时的兜底

Agent 失败是常态，必须有降级方案。

6.1 三层降级

async def execute_with_fallback(query, task_type):
    # 第 1 层：Agent（智能但慢 + 贵）
    try:
        return await agent.run(query)
    except AgentError:
        pass

    # 第 2 层：RAG 检索 + LLM 直接生成
    try:
        docs = await vector_search(query, top_k=5)
        context = "\n".join([d.content for d in docs])
        return await llm.complete(
            prompt=f"基于以下上下文：\n{context}\n\n回答：{query}"
        )
    except LLMError:
        pass

    # 第 3 层：纯检索结果（无 LLM 加工）
    docs = await vector_search(query, top_k=3)
    return {
        "answer": "未找到精确答案，以下是相关政策原文：",
        "documents": docs,
    }

6.2 失败告警

# Prometheus 告警
- alert: AgentFallbackSpike
  expr: rate(agent_fallback_total[10m]) > 0.5
  for: 5m
  annotations:
    summary: "Agent fallback 频率异常"

- alert: AgentExcessiveTurns
  expr: agent_turns_total{turns=">20"} > 10
  for: 5m
  annotations:
    summary: "Agent 单次会话超过 20 轮，可能是死循环"

---

七、踩过的坑

坑 1：Skill 描述太模糊

最初 Skill 描述写得太抽象：

description: 处理政策相关的任务

Agent 经常选错 Skill。改成：

description: 当用户提出**具体政策问题**（如某税种、某规定）时，调用此 Skill 检索相关政策原文。**不要用于一般性问答或闲聊**。

准确率从 60% 提升到 92%。

坑 2：Tool 太多导致选错

最初给 Agent 提供 20+ Tool。结果 Agent 经常选错。

原则：同时可用的 Tool ≤ 10 个。多了就拆成多个 Skill 隔离。

坑 3：循环调用 Tool

某次 Agent 反复调用 pgvector_query，每次用相同 query，陷入死循环。

解法：

• 加 max_turns=25 硬限制
• 检测”近 3 步相同 query”则强制停止

坑 4：Token 消耗失控

某次 Session 跑了 30 分钟，Token 消耗 ¥50。

解法：

• 加 max_tokens_per_session=10000 限制
• 超出后切换到精简模式

坑 5：Skill 之间的状态共享

不同 Skill 之间需要共享状态（如”已检索的文档”），最初用全局变量，并发时串了。

解法：用 Session-scoped state，由 Agent SDK 管理。

---

八、给类似场景的建议

如果你也要构建 Agent 系统：

1. Skill 设计先于实现——先列 5-10 个职责单一的 Skill，再写实现
2. Tool 数量控制在 10 个以内——多了就拆 Skill
3. 复杂任务用 LangGraph——单 Skill 表达不了的状态机
4. MCP 是协议不是锁——可以同时用 Anthropic / OpenAI 等不同 provider
5. 可观测性是基础设施——LangSmith / LangFuse 一开始就接入
6. 降级是必须——Agent 失败时必须有兜底
7. 评估驱动优化——没指标的 Agent 优化是盲改

---

九、总结

Agent SDK 不是”银弹”，是让多步决策可管理、可观测、可扩展的工具集。

核心原则：

1. Skill 单一职责——别让一个 Skill 既检索又生成
2. Tool 数量克制——同时 ≤ 10 个
3. MCP 是协议——可以混用不同 provider 的 Tool
4. 复杂任务用 LangGraph——状态机比 Prompt 链清晰
5. 可观测性第一——没有 trace 的 Agent 不可调试
6. 降级是必备——三层降级（Agent → RAG → 检索结果）

最后一句话：Agent 系统是软件工程 + AI 的结合，既要懂 LLM，也要懂架构。

---

十、参考资料

• Claude Agent SDK 文档
• Model Context Protocol
• LangGraph 状态机
• LangSmith Agent Tracing

---

作者：魏远标，贝斯平 AI 架构师。技术博客：javai.tech

你在用哪个 Agent SDK？踩过什么坑？留言聊聊～