Claude API Agent 框架实战
构建多步骤自动化工作流

一、多智能体系统架构概览

在传统的单次 LLM 调用模式中，一个请求只能返回一个回答。而 多智能体架构（Multi-Agent Architecture）打破了这一限制：通过一个「主 Agent（Lead Agent）」统筹规划，将复杂任务拆解为多个子任务，分发给各专精的「子 Agent（Subagent）」并行执行，最终汇总结果。

这套模式特别适合以下场景：

• 需要同时搜索多个信息源的研究任务
• 代码审查 + 测试 + 文档同步进行的 CI/CD 流水线
• 涉及多步骤决策、中间结果会影响后续步骤的开放式自动化

Anthropic 提供了两个层次的工具来实现这套架构：

二、环境准备

我们需要安装两个包，并正确配置 API Key。

mkdir claude-agent-demo && cd claude-agent-demo
npm init -y
npm install @anthropic-ai/sdk @anthropic-ai/claude-agent-sdk zod
npm install -D typescript @types/node tsx
npx tsc --init --target es2022 --module nodenext --moduleResolution nodenext

# .env 文件
ANTHROPIC_API_KEY=sk-ant-api03-xxxxxxxxxxxxxxxxxx

// 在代码里这样加载
import 'dotenv/config'; // 或直接在 shell 里 export ANTHROPIC_API_KEY=...
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();
// SDK 自动读取 process.env.ANTHROPIC_API_KEY

// hello.ts
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();

const response = await client.messages.create({
  model: 'claude-opus-4-6',
  max_tokens: 256,
  messages: [{ role: 'user', content: '用一句话介绍你自己' }],
});

console.log(response.content[0].text);

npx tsx hello.ts
# 输出类似：我是 Claude，Anthropic 开发的 AI 助手...

三、手写 Agent 循环：理解 Tool Use 核心机制

在使用高层 SDK 之前，先理解底层的 Agentic Loop（智能体循环）至关重要。这个循环的核心逻辑是：

1. 向 Claude 发送消息 + 可用工具列表
2. Claude 返回 tool_use 块，要求调用某个工具
3. 你的代码执行该工具，获取结果
4. 将工具结果作为 tool_result 消息发回给 Claude
5. 重复步骤 1-4，直到 Claude 返回 stop_reason: "end_turn"

下面是一个完整的例子：给 Claude 配备「天气查询」和「城市搜索」两个工具，让它自主完成一个多步骤研究任务：

TypeScript
// agent-loop.ts
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();

// ① 定义工具
const tools: Anthropic.Tool[] = [
  {
    name: 'get_weather',
    description: '获取指定城市的当前天气。返回温度（摄氏度）和天气状况。',
    input_schema: {
      type: 'object',
      properties: {
        city: { type: 'string', description: '城市名称，如"北京"或"上海"' },
      },
      required: ['city'],
    },
  },
  {
    name: 'search_attractions',
    description: '搜索指定城市的热门景点列表。',
    input_schema: {
      type: 'object',
      properties: {
        city: { type: 'string', description: '城市名称' },
        limit: { type: 'number', description: '返回结果数量，默认3' },
      },
      required: ['city'],
    },
  },
];

// ② 模拟工具执行函数
async function executeTool(
  name: string,
  input: Record
): Promise {
  if (name === 'get_weather') {
    const city = input.city as string;
    // 实际项目中调用真实天气 API
    return JSON.stringify({ city, temp: 18, condition: '晴天', humidity: '45%' });
  }
  if (name === 'search_attractions') {
    const city = input.city as string;
    const limit = (input.limit as number) || 3;
    const data: Record = {
      '北京': ['故宫', '长城', '颐和园', '天坛'],
      '上海': ['外滩', '豫园', '东方明珠', '南京路'],
    };
    return JSON.stringify((data[city] ?? ['景点A', '景点B']).slice(0, limit));
  }
  return '工具不存在';
}

// ③ 主 Agent 循环
async function runAgent(userQuery: string): Promise {
  let messages: Anthropic.MessageParam[] = [
    { role: 'user', content: userQuery },
  ];

  while (true) {
    const response = await client.messages.create({
      model: 'claude-opus-4-6',
      max_tokens: 4096,
      thinking: { type: 'adaptive' },
      tools,
      messages,
    });

    console.log(`[stop_reason: ${response.stop_reason}]`);

    // 任务完成，提取最终文本回答
    if (response.stop_reason === 'end_turn') {
      const textBlock = response.content.find((b) => b.type === 'text');
      return (textBlock as Anthropic.TextBlock)?.text ?? '';
    }

    // 需要调用工具
    if (response.stop_reason === 'tool_use') {
      // 将 Claude 的回复追加到对话历史
      messages.push({ role: 'assistant', content: response.content });

      // 收集所有工具调用结果
      const toolResults: Anthropic.ToolResultBlockParam[] = [];

      for (const block of response.content) {
        if (block.type === 'tool_use') {
          console.log(`  → 调用工具: ${block.name}`, block.input);
          const result = await executeTool(
            block.name,
            block.input as Record
          );
          console.log(`  ← 工具结果: ${result}`);

          toolResults.push({
            type: 'tool_result',
            tool_use_id: block.id,
            content: result,
          });
        }
      }

      // 将工具结果发回给 Claude
      messages.push({ role: 'user', content: toolResults });
    }
  }
}

// ④ 运行测试
const answer = await runAgent(
  '帮我对比北京和上海的今日天气，并各推荐3个必去景点'
);
console.log('\n=== 最终回答 ===\n', answer);

四、用 Tool Runner 简化工具循环

手写循环虽然灵活，但代码量较多。Anthropic SDK 提供了 Tool Runner（Beta） 来自动处理整个循环，只需定义工具和 run 函数即可：

TypeScript
// tool-runner.ts
import Anthropic from '@anthropic-ai/sdk';
import { betaZodTool } from '@anthropic-ai/sdk/helpers/beta/zod';
import { z } from 'zod';

const client = new Anthropic();

// 用 Zod 定义工具——类型安全，不需要手写 JSON Schema
const getWeather = betaZodTool({
  name: 'get_weather',
  description: '获取城市天气，返回温度和天气状况',
  inputSchema: z.object({
    city: z.string().describe('城市名称'),
    unit: z.enum(['celsius', 'fahrenheit']).optional().default('celsius'),
  }),
  run: async ({ city, unit }) => {
    // 这里接入真实天气 API
    return `${city}：18°C，晴天（${unit}）`;
  },
});

const searchAttractions = betaZodTool({
  name: 'search_attractions',
  description: '搜索城市热门景点',
  inputSchema: z.object({
    city: z.string(),
    limit: z.number().min(1).max(10).default(3),
  }),
  run: async ({ city, limit }) => {
    const db: Record = {
      北京: ['故宫', '长城', '颐和园', '天坛', '798艺术区'],
      上海: ['外滩', '豫园', '东方明珠', '南京路', '新天地'],
    };
    const spots = (db[city] ?? ['景点A', '景点B']).slice(0, limit);
    return JSON.stringify(spots);
  },
});

// Tool Runner 自动处理循环，返回最终 Message
const finalMessage = await client.beta.messages.toolRunner({
  model: 'claude-opus-4-6',
  max_tokens: 4096,
  tools: [getWeather, searchAttractions],
  messages: [
    {
      role: 'user',
      content: '对比北京和上海的今日天气，各推荐3个必去景点，最后给出旅行建议',
    },
  ],
});

// 提取文本回答
const text = finalMessage.content
  .filter((b) => b.type === 'text')
  .map((b) => (b as Anthropic.TextBlock).text)
  .join('');

console.log(text);

Tool Runner 的优势对比：

五、多 Agent 编排：主 Agent + 子 Agent 并行

真正的多步骤自动化工作流往往需要多个专精 Agent 协同工作。Claude Agent SDK 的 query() + agents 选项，让你能够轻松定义子 Agent 并让主 Agent 在需要时召唤它们。

场景：自动竞品分析报告

我们构建一个系统：主 Agent 负责规划和整合，派出「研究员 Agent」和「写作 Agent」分工完成任务：

TypeScript
// multi-agent.ts
import { query } from '@anthropic-ai/claude-agent-sdk';

const result: string[] = [];

for await (const message of query({
  prompt: `
    请完成以下竞品分析任务：
    1. 使用 researcher 子Agent，分别调研 Notion 和 Obsidian 的核心功能
    2. 使用 writer 子Agent，将调研结果整理成一份结构化的对比报告
    3. 给出最终的选型建议
  `,
  options: {
    model: 'claude-opus-4-6',
    maxTurns: 20,
    allowedTools: ['Read', 'Glob', 'WebSearch', 'Agent'],
    systemPrompt: `
      你是一位高级产品经理。在处理复杂任务时，
      善用专精子Agent来分工提升效率。
    `,

    // 定义可调用的子 Agent
    agents: {
      researcher: {
        description:
          '专业信息研究员，擅长收集和整理产品信息、技术文档和用户评价。',
        prompt: `
          你是一位细致的产品研究员。
          为每个产品收集：核心功能列表、定价策略、目标用户、优缺点。
          以结构化 JSON 格式返回结果。
        `,
        tools: ['WebSearch', 'Read'],
      },

      writer: {
        description:
          '技术写作专家，能将研究数据转化为清晰易读的分析报告。',
        prompt: `
          你是一位经验丰富的技术写作者。
          接收 JSON 格式的研究数据，输出完整的 Markdown 格式对比报告。
          报告需包含：执行摘要、功能对比表、适用场景、选型建议。
        `,
        tools: ['Read'],
      },
    },
  },
})) {
  if ('result' in message) {
    result.push(message.result);
  } else if (message.type === 'assistant') {
    // 实时打印 Agent 思考过程（可选）
    const text = message.message.content
      .filter((b) => b.type === 'text')
      .map((b) => b.text)
      .join('');
    if (text) process.stdout.write(text);
  }
}

console.log('\n=== 最终报告 ===\n', result.join('\n'));

从外部并行启动多个独立 Agent

如果任务之间完全独立，可以用 Promise.all 并行启动多个 query 调用，大幅缩短总耗时：

TypeScript
// parallel-agents.ts
import { query } from '@anthropic-ai/claude-agent-sdk';

async function runResearchAgent(topic: string): Promise {
  const results: string[] = [];
  for await (const msg of query({
    prompt: `深度调研：${topic}。输出结构化摘要（500字以内）`,
    options: {
      model: 'claude-opus-4-6',
      allowedTools: ['WebSearch'],
      maxTurns: 8,
    },
  })) {
    if ('result' in msg) results.push(msg.result);
  }
  return results.join('');
}

// 并行调研三个主题
const topics = [
  'Claude API 的最新功能和定价',
  'OpenAI API 的最新功能和定价',
  'Google Gemini API 的最新功能和定价',
];

console.log('并行启动 3 个研究 Agent...');
const [claudeReport, openaiReport, geminiReport] = await Promise.all(
  topics.map(runResearchAgent)
);

console.log('全部完成，开始整合...');
// 再调用一个整合 Agent 汇总结果
for await (const msg of query({
  prompt: `
    以下是三个 AI API 的调研报告，请整合成一份对比分析：

    === Claude API ===
    ${claudeReport}

    === OpenAI API ===
    ${openaiReport}

    === Google Gemini API ===
    ${geminiReport}

    请输出：功能对比表、价格对比、适用场景建议。
  `,
  options: { model: 'claude-opus-4-6', maxTurns: 5 },
})) {
  if ('result' in msg) console.log(msg.result);
}

六、会话保持与长对话上下文管理

多步骤工作流往往需要跨多轮保持状态。Claude Agent SDK 提供了 Session Resumption（会话恢复），让你能在 Agent 暂停后以完整上下文继续工作：

TypeScript
// session-resume.ts
import { query } from '@anthropic-ai/claude-agent-sdk';

let sessionId: string | undefined;

// 第一轮：让 Agent 读取并分析代码库
console.log('=== 第一轮：分析阶段 ===');
for await (const message of query({
  prompt: '扫描当前目录下所有 TypeScript 文件，找出潜在的类型错误和不规范写法',
  options: {
    cwd: process.cwd(),
    allowedTools: ['Read', 'Glob', 'Grep'],
    model: 'claude-opus-4-6',
  },
})) {
  if (message.type === 'system' && message.subtype === 'init') {
    sessionId = message.session_id; // 记录 session ID
    console.log('Session ID:', sessionId);
  }
  if ('result' in message) {
    console.log('分析结果：', message.result);
  }
}

// 第二轮：基于第一轮的发现，继续修复（携带完整上下文）
console.log('\n=== 第二轮：修复阶段 ===');
for await (const message of query({
  prompt: '现在逐个修复你找到的问题，每处修改前先说明原因',
  options: {
    resume: sessionId, // 恢复之前的 session，保留所有上下文
    allowedTools: ['Read', 'Edit', 'Write'],
    permissionMode: 'acceptEdits',
  },
})) {
  if ('result' in message) {
    console.log('修复完成：', message.result);
  }
}

长对话的 Compaction（压缩）

当对话历史接近 200K token 上限时，可以开启 Compaction 自动压缩（仅 Opus 4.6，Beta 功能）：

TypeScript
// long-conversation.ts
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();
const messages: Anthropic.Beta.BetaMessageParam[] = [];

async function chat(userMessage: string): Promise {
  messages.push({ role: 'user', content: userMessage });

  const response = await client.beta.messages.create({
    betas: ['compact-2026-01-12'],
    model: 'claude-opus-4-6',
    max_tokens: 4096,
    messages,
    context_management: {
      edits: [{ type: 'compact_20260112' }],
    },
  });

  // 关键：必须把完整 content（包括压缩块）追加回 messages
  messages.push({ role: 'assistant', content: response.content });

  return response.content
    .filter((b) => b.type === 'text')
    .map((b) => (b as Anthropic.TextBlock).text)
    .join('');
}

// 可以持续对话而不担心超过上下文窗口
await chat('分析这个 50KB 的代码文件...');
await chat('针对刚才发现的 Bug，设计修复方案');
await chat('把修复方案转成具体的 PR 描述');

七、错误处理与生产最佳实践

将 Agent 系统从 Demo 推向生产环境，需要做好几个关键点：

1. 类型化异常处理

始终使用 SDK 提供的类型化异常类，不要用字符串匹配：

TypeScript
// error-handling.ts
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();

async function safeAgentCall(prompt: string): Promise {
  try {
    const response = await client.messages.create({
      model: 'claude-opus-4-6',
      max_tokens: 2048,
      messages: [{ role: 'user', content: prompt }],
    });
    return (response.content[0] as Anthropic.TextBlock).text;
  } catch (error) {
    if (error instanceof Anthropic.RateLimitError) {
      // 429：等待后重试（SDK 会自动重试，但这里演示手动处理）
      const retryAfter = (error as any).headers?.['retry-after'] ?? 30;
      console.warn(`触发速率限制，${retryAfter}秒后重试`);
      await new Promise((r) => setTimeout(r, Number(retryAfter) * 1000));
      return safeAgentCall(prompt); // 递归重试
    } else if (error instanceof Anthropic.AuthenticationError) {
      console.error('API Key 无效，请检查 ANTHROPIC_API_KEY 环境变量');
      process.exit(1);
    } else if (error instanceof Anthropic.BadRequestError) {
      console.error('请求格式错误：', error.message);
      return null;
    } else if (error instanceof Anthropic.APIError) {
      console.error(`API 错误 ${error.status}：`, error.message);
      return null;
    }
    throw error; // 未知错误，向上抛出
  }
}

2. 设置 maxTurns 防止无限循环

TypeScript
for await (const msg of query({
  prompt: '...',
  options: {
    maxTurns: 15,        // 防止 Agent 陷入死循环
    maxBudgetUsd: 0.5,   // 最多消耗 $0.5（生产环境必备）
    model: 'claude-opus-4-6',
  },
})) { ... }

3. 使用 Hooks 做审计日志

TypeScript
import { query, HookCallback } from '@anthropic-ai/claude-agent-sdk';
import { appendFileSync } from 'fs';

// 每次工具调用后记录审计日志
const auditHook: HookCallback = async (input) => {
  const record = {
    ts: new Date().toISOString(),
    tool: (input as any).tool_name,
    file: (input as any).tool_input?.file_path,
  };
  appendFileSync('./audit.log', JSON.stringify(record) + '\n');
  return {};
};

for await (const msg of query({
  prompt: '重构 src/ 目录下的所有文件',
  options: {
    cwd: '/your/project',
    allowedTools: ['Read', 'Edit', 'Write'],
    permissionMode: 'acceptEdits',
    hooks: {
      PostToolUse: [{ matcher: 'Edit|Write', hooks: [auditHook] }],
    },
  },
})) { ... }

4. 模型选择策略

不同任务用不同模型，节省成本：

TypeScript
function selectModel(task: 'simple' | 'medium' | 'complex'): string {
  const map = {
    simple: 'claude-haiku-4-5',    // 分类、提取：速度快、成本低
    medium: 'claude-sonnet-4-6',   // 写作、代码：均衡
    complex: 'claude-opus-4-6',    // 推理、Agent编排：最强
  };
  return map[task];
}

// 主 Agent 用 Opus，子 Agent 可以用 Sonnet
const agents = {
  researcher: {
    // 研究子任务用 Sonnet 省成本
    prompt: '...',
    model: 'claude-sonnet-4-6',
    tools: ['WebSearch'],
  },
};

常见问题解答