Claude Agent SDK 四层架构拆解：官方最佳实践到底长什么样

三月底的时候 Anthropic 悄悄发了一版 Agent SDK 1.4，配套文档里第一次把”四层架构”写进去了。之前大家做 Agent 基本都是野路子——谁有经验谁按自己理解搞，抽象到底该切几刀、每刀切在哪儿，全凭个人品味。

我过去一年接过三四个 Agent 项目，踩过的架构坑可太多了。有个项目一开始把工具调用、Prompt 组装、状态管理全塞在一个类里，改到第三个功能的时候那个类已经 2000 多行，谁都不敢动。后来我照着 Agent SDK 的思路重构，砍到四个模块，才算活过来。

这篇文章把 Agent SDK 官方的四层架构拆开讲，每一层我尽量给出接口是什么样、扩展点在哪里、生产上怎么用。如果你正在从零搭一个 Agent，这套骨架可以直接抄。

架构总览：四层是哪四层

简单说就是：

1. 应用层（Application）：用户意图、任务定义、UI 对接
2. Agent 层（Agent）：规划、工具选择、对话管理
3. 运行时（Runtime）：执行循环、重试、超时、熔断
4. 服务层（Service）：模型 API、工具实现、持久化

每一层对下层依赖、对上层提供接口，之间用清晰的数据结构交互。下面挨个拆。

一、应用层：把”用户要什么”翻译成”Agent 能处理什么”

这一层最容易被忽视，因为它看起来”没什么技术含量”。但我踩坑最多的就是这里。

应用层的核心职责是任务定义——把用户的模糊需求翻译成 Agent 能吃的结构化输入。比如用户说”帮我把这个项目重构一下”，这在 Agent 眼里是完全不可执行的。应用层要做的是：

1. 识别意图（重构）
2. 约束范围（哪些文件？哪些不能动？）
3. 定义成功标准（测试全过？性能不降？代码行数减少？）
4. 配置资源上限（最多跑多久？最多多少次 LLM 调用？）

实际代码里一般长这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from anthropic_agent_sdk import TaskSpec, ResourceBudget

task = TaskSpec(
    intent="refactor_module",
    scope={
        "target_files": ["src/payment/*.py"],
        "excluded_files": ["src/payment/legacy_bridge.py"],
    },
    success_criteria=[
        "pytest tests/payment passes",
        "line_count_delta <= 0",
    ],
    budget=ResourceBudget(
        max_llm_calls=50,
        max_duration_seconds=1800,
        max_tokens=1_000_000,
    ),
)

应用层的关键扩展点是意图分类器和成功判定器。意图分类器可以是一个小模型（Haiku 非常合适）做路由，不同意图走不同的 Agent 配置。成功判定器可以是纯代码（比如跑一下测试）、也可以是 LLM 自己判（LLM-as-judge）。

我的建议是：能用代码判的绝对不要用 LLM 判。LLM-as-judge 看起来灵活，但不稳定、还费钱。

二、Agent 层：规划、选工具、记状态

这是大家最熟的一层，但也是最容易写烂的一层。Agent 层对外只有一个核心接口：

1
2
3

class Agent:
    def run(self, task: TaskSpec) -> TaskResult:
        ...

它内部要做三件事：

1. Planning（规划）

拿到 task 之后，Agent 要先产出一个执行计划。Agent SDK 官方推荐的是分层规划——先出一个 high-level plan（3-5 步），每一步再在运行时动态展开成具体的子任务。

这比”一步到位画详细计划”好太多了。我之前试过让 Claude 一次性产出 50 步的详细计划，结果前 5 步执行完之后，环境已经和它最初假设的完全不一样了，后面 45 步全白搭。分层规划让 Agent 每次只需要对下一步做细节规划，灵活性高很多。

2. 工具选择

SDK 1.4 里引入了 ToolPicker 抽象。它的作用是：当你注册了 50 个工具的时候，不要把这 50 个工具的描述全塞进 prompt——那样既浪费 token 又会让模型 confused。ToolPicker 会根据当前任务阶段只暴露相关工具。

简化版长这样：

1
2
3
4
5
6
7
8

class ContextAwareToolPicker:
    def pick(self, task_context):
        if task_context.phase == "exploration":
            return [self.tools["read_file"], self.tools["grep"]]
        if task_context.phase == "modification":
            return [self.tools["edit"], self.tools["write"]]
        if task_context.phase == "verification":
            return [self.tools["run_tests"], self.tools["read_file"]]

3. 状态管理

Agent 的状态不只是对话历史。完整的状态应该包括：

• 已读取的文件及其内容快照（避免重复读）
• 已做的修改列表（方便回滚）
• 已失败的尝试（避免重复犯错）
• 资源消耗计数（token、时间、API 调用数）

这一层也是 Prompt Caching 落地的主战场。你可以设计成前面是不变的 system prompt + 任务定义 + 工具定义，后面是动态变化的对话历史，在”系统+任务+工具”的末尾打一个 cache_control，就能把最大的静态部分缓存起来。我在之前讲 成本优化 的时候详细写过这招。

三、运行时：核心循环 + 生产韧性

运行时是 Agent SDK 最”工程化”的一层。它不懂业务，只做三件事：跑循环、处理错误、管资源。

核心循环是这样的伪代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

while not task.is_done() and budget.has_remaining():
    # 1. 收集上下文
    context = agent.gather_context(task, state)

    # 2. 执行（模型调用或工具调用）
    try:
        step_result = agent.execute(context)
    except TransientError as e:
        retry(e)
        continue
    except FatalError as e:
        return TaskResult.failed(e)

    # 3. 验证
    verification = verify(step_result, task.success_criteria)
    state.update(step_result, verification)

    if verification.passed:
        return TaskResult.success(state)

三个阶段——收集上下文 → 执行 → 验证——这是 Agent SDK 官方文档反复强调的核心循环。我实操下来最大的心得是：**”验证”这一步绝对不能省**。

很多人写 Agent 就是”让模型说它做完了就算完了”。这是灾难。模型经常”幻觉性完成”——它告诉你测试过了，实际上它根本没跑测试。必须在 runtime 里显式验证。

熔断和指数退避的落地

生产环境最容易出问题的是两种情况：

• API 间歇性故障：模型服务偶发 5xx、429
• Agent 陷入死循环：同一个错误反复试

熔断器（Circuit Breaker）解决第一种。简化实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "CLOSED"
        self.opened_at = None

    def call(self, fn, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.opened_at > self.reset_timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitOpenError()

        try:
            result = fn(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.state = "OPEN"
                self.opened_at = time.time()
            raise

指数退避配合 jitter 解决重试风暴：

1
2
3
4

def backoff_with_jitter(attempt, base=1, max_delay=60):
    delay = min(base * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.3)
    return delay + jitter

我现在的生产配置一般是：熔断阈值 5 次、重置 60 秒；重试最多 3 次、base 1 秒、max 30 秒。这个参数在我跑的几个项目上稳定了半年没出过事。

死循环检测

Agent 陷死循环的特征是连续 N 步的 action 和 state 没有实质变化。简单做法是对每步的 action hash，如果相邻 3 步 hash 一样（或者语义相似度 > 0.95），就强制打断让 Agent 换思路。更彻底的做法是 runtime 里带一个 “review loop”——每 10 步停下来反思”我们在朝目标推进吗”，如果判断没进展就 abort。

四、服务层：模型、工具、持久化

服务层负责对接外部——LLM API、数据库、第三方工具。这一层的设计原则是可替换。

以模型为例，服务层暴露一个接口：

1
2

class ModelProvider(Protocol):
    def complete(self, messages, tools, **kwargs) -> ModelResponse: ...

然后可以有 AnthropicProvider、RouterProvider（同时挂多个模型、按场景路由）、MockProvider（测试用）。上层完全不知道下面是哪家。

我的 RouterProvider 实现里一般会挂两个策略：

• 按任务难度路由：简单任务 Haiku、中等 Sonnet、复杂 Opus
• 按失败降级：Opus 失败自动降到 Sonnet，Sonnet 再失败降到 Haiku

这种混搭可以把成本控制在一个相对稳定的区间。生产上我跑的 Agent 平均成本比”纯 Opus”低 60%，比”纯 Sonnet”还能再低 15-20%。

工具这一层也要注意——工具的实现和声明要分离。工具声明（名字、参数、描述）给 LLM 看；工具实现（实际代码）是服务层的职责。这样你可以在不同环境下用不同实现（开发环境用 mock、生产用真实 API），声明完全不变。

持久化我不展开了，Agent SDK 给了一个 StateStore 抽象接口，你可以接 Redis、Postgres、甚至本地 SQLite。核心是能随时把 Agent 的完整状态 dump 出来，出问题能复现。

写在最后

四层架构听起来工程味很重，但实际跑起来你会发现它省下的时间远大于额外的抽象成本。一个没分层的 Agent 遇到 bug，你得在两千行代码里大海捞针；一个分好层的 Agent，bug 的位置基本一眼能看出在哪层。

Agent SDK 现在还在快速演进，但四层这个骨架我觉得短期不会变。如果你在做 Agent 相关项目，openclaw.cocoloop.cn 上有一些开源实现可以参考，源码读下来很能启发架构思考。

最后给一个建议：**不要上来就追求”完美架构”**。第一版能跑就行，跑起来之后按痛点重构——哪一层经常要改，就说明那一层的抽象没做好，用痛点来驱动架构演进，比纸上谈兵画图靠谱得多。