Claude Agent SDK 教程合集

课程专栏

第 9 课 / 共 9 课

返回课程目录

平台能跑了，但要用在实际项目里，还需要解决三个问题：花了多少钱？出错了怎么办？日志怎么更有用？

第9课：进阶优化 —— 日志、成本控制、错误处理

本课目标

平台能跑了，但要用在实际项目里，还需要解决三个问题：花了多少钱？出错了怎么办？日志怎么更有用？

一、成本控制

问题：一次研究花多少钱？

每次调 Claude API 都在花钱。研究平台一次完整运行可能涉及： - 组长 1 次调用（sonnet） - 3-4 个研究员各搜索 5-10 次（haiku × 4） - 1 个分析师（haiku） - 1 个报告员（haiku）

大概 $0.5 - $2 一次。

方案1：设置预算上限

code
options = ClaudeAgentOptions(
    max_budget_usd=1.00,  # 最多花 1 美元
    # ... 其他配置
)

超过预算后 Claude 会自动停止。简单粗暴但有效。

方案2：子Agent用便宜模型

code
agents = {
    "researcher": AgentDefinition(
        model="haiku",       # 搜索任务用最便宜的
        # ...
    ),
    "data-analyst": AgentDefinition(
        model="haiku",       # 数据处理也用便宜的
        # ...
    ),
    "report-writer": AgentDefinition(
        model="haiku",       # 写报告用便宜的
        # ...
    ),
}

options = ClaudeAgentOptions(
    model="sonnet",          # 只有组长用好模型（需要理解和决策）
    # ...
)

模型费用差异很大，haiku 比 sonnet 便宜很多。研究员主要是搜索和整理，不需要最强的推理能力。

方案3：限制搜索次数

在 researcher 的 prompt 里写死上限：

code
搜索次数：最少 3 次，最多 8 次。不要超过 8 次。

或者用 Hook 强制限制：

code
class SearchLimiter:
    def __init__(self, max_searches=8):
        self.max_searches = max_searches
        self._count = 0

    async def pre_tool_use_hook(self, hook_input, tool_use_id, context):
        if hook_input["tool_name"] == "WebSearch":
            self._count += 1
            if self._count > self.max_searches:
                print(f"⚠️ 搜索次数超限 ({self._count}/{self.max_searches})，拦截")
                return {
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": "搜索次数超过上限",
                    }
                }
        return {}

方案4：从 ResultMessage 中获取费用

code
from claude_agent_sdk import ResultMessage

async for msg in client.receive_response():
    if isinstance(msg, ResultMessage):
        print(f"会话ID: {msg.session_id}")
        # ResultMessage 中可能包含 usage/cost 信息
        # 具体字段取决于 SDK 版本

二、错误处理

问题：搜索失败了怎么办？网络断了怎么办？

AI 系统最常见的错误：

错误类型 原因 应对
CLINotFoundError Claude CLI 没装 检查安装
CLIConnectionError 网络问题 重试
ProcessError 进程异常退出 查看 exit_code
CLIJSONDecodeError 返回数据格式错误 重试或跳过
API 限流 请求太频繁 等一等再试

基础错误处理

code
from claude_agent_sdk import (
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
)


async def safe_research(client, topic, max_retries=3):
    """带重试的研究函数"""
    for attempt in range(max_retries):
        try:
            await client.query(topic)
            async for msg in client.receive_response():
                process_assistant_message(msg, tracker, print)
            return  # 成功了，直接返回

        except CLIConnectionError as e:
            print(f"⚠️ 网络错误 (尝试 {attempt+1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                print("  等待 5 秒后重试...")
                await anyio.sleep(5)
            else:
                print("❌ 多次重试失败，放弃")
                raise

        except ProcessError as e:
            print(f"❌ 进程异常: exit_code={e.exit_code}")
            raise

        except CLIJSONDecodeError as e:
            print(f"⚠️ 数据解析错误: {e}")
            if attempt < max_retries - 1:
                print("  重试中...")
            else:
                raise

用 Hook 捕获工具失败

code
async def handle_tool_failure(hook_input, tool_use_id, context):
    """PostToolUseFailure: 工具调用失败时触发"""
    tool_name = hook_input["tool_name"]
    error = hook_input.get("error", "未知错误")
    print(f"❌ [{tool_name}] 执行失败: {error}")

    # 你可以在这里：
    # 1. 记录到日志文件
    # 2. 发送告警通知
    # 3. 返回修改后的结果让 Claude 继续
    return {}


hooks = {
    "PostToolUseFailure": [
        HookMatcher(matcher=None, hooks=[handle_tool_failure]),
    ],
    # ... 其他 hooks
}

三、日志系统增强

双通道日志

研究平台的日志设计：控制台显示简要信息，文件记录完整细节。

code
"""
增强版日志系统
"""
import sys
from pathlib import Path
from datetime import datetime


class ResearchLogger:
    def __init__(self, log_dir: Path):
        self.log_dir = log_dir
        self._transcript = open(log_dir / "transcript.txt", "a", encoding="utf-8")
        self._jsonl = open(log_dir / "tool_calls.jsonl", "a", encoding="utf-8")

    def console(self, text):
        """只打印到控制台（给用户看的简要信息）"""
        print(text)

    def file_only(self, text):
        """只写入文件（详细信息，不打扰用户）"""
        self._transcript.write(text + "\n")
        self._transcript.flush()

    def both(self, text):
        """同时输出到控制台和文件"""
        print(text)
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._transcript.write(f"[{timestamp}] {text}\n")
        self._transcript.flush()

    def log_tool_call(self, agent_id, tool_name, tool_input):
        """结构化日志（JSONL格式，方便后续分析）"""
        import json
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": agent_id,
            "tool": tool_name,
            "input_preview": str(tool_input)[:200],
        }
        self._jsonl.write(json.dumps(entry, ensure_ascii=False) + "\n")
        self._jsonl.flush()

    def close(self):
        self._transcript.close()
        self._jsonl.close()

JSONL 日志的好处

tool_calls.jsonl 每行一个 JSON 对象，方便后续用 Python 分析：

code
"""
分析日志：统计每个Agent调用了多少次工具
"""
import json
from collections import Counter

with open("logs/session_xxx/tool_calls.jsonl") as f:
    calls = [json.loads(line) for line in f]

# 按 Agent 统计
by_agent = Counter(c["agent_id"] for c in calls)
print("各Agent工具调用次数:")
for agent, count in by_agent.most_common():
    print(f"  {agent}: {count} 次")

# 按工具统计
by_tool = Counter(c["tool"] for c in calls)
print("\n各工具使用次数:")
for tool, count in by_tool.most_common():
    print(f"  {tool}: {count} 次")

输出类似：

bash
各Agent工具调用次数:
  RESEARCHER-1: 6 次
  RESEARCHER-2: 5 次
  RESEARCHER-3: 4 次
  DATA-ANALYST-1: 5 次
  REPORT-WRITER-1: 4 次

各工具使用次数:
  WebSearch: 12 次
  Write: 5 次
  Read: 4 次
  Glob: 3 次
  Bash: 3 次

四、其他优化技巧

1. 思考深度控制

code
options = ClaudeAgentOptions(
    effort="low",     # 思考浅一点，速度快（适合简单搜索）
    # effort="medium",  # 中等
    # effort="high",    # 深度思考（适合复杂分析）
)

研究员用 "low"，组长和报告员用 "medium" 或 "high"。

2. 结构化输出

让 Claude 返回固定格式的 JSON，方便程序处理：

code
options = ClaudeAgentOptions(
    output_format={
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "subtopics": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "拆解出的子课题列表"
                },
                "estimated_time": {
                    "type": "string",
                    "description": "预计耗时"
                }
            }
        }
    }
)

3. 加载项目级配置

code
options = ClaudeAgentOptions(
    setting_sources=["project"],  # 加载 .claude/ 目录下的配置
    # 这样 Skills 和 Commands 就能被子Agent使用
)

研究平台的 demo 把 PDF 生成技巧放在 .claude/skills/pdf/SKILL.md 里，report-writer 通过 Skill 工具来读取这些最佳实践。

五、完整的优化版配置

把所有优化整合在一起：

code
options = ClaudeAgentOptions(
    # 角色
    system_prompt=lead_prompt,

    # 工具和子Agent
    allowed_tools=["Task"],
    agents=agents,

    # 安全和权限
    permission_mode="bypassPermissions",
    hooks=hooks,

    # 模型和性能
    model="sonnet",           # 组长用好模型
    max_turns=20,             # 留足空间
    effort="medium",          # 中等思考深度

    # 成本控制
    max_budget_usd=2.00,      # 单次上限 2 美元

    # 项目配置
    setting_sources=["project"],  # 加载 .claude/ 配置
    cwd="/path/to/project",       # 工作目录
)

本课小结

  1. 成本控制：max_budget_usd + 便宜模型 + 限制搜索次数
  2. 错误处理：try/except 捕获 SDK 异常 + 重试机制 + PostToolUseFailure Hook
  3. 日志系统：双通道（控制台 + 文件）+ JSONL 结构化日志
  4. 性能优化：effort 参数 + 结构化输出 + 项目配置

教程总结

恭喜你完成了全部 9 课！回顾一下你学到了什么：

code
第1课  理解了系统的整体架构
第2课  搭好了开发环境
第3课  学会了 query() 基础问答
第4课  学会了 ClaudeSDKClient 多轮对话
第5课  学会了给 AI 装工具（内置 + 自定义）
第6课  学会了定义和调度多个子Agent
第7课  学会了用 Hooks 监控和控制 AI 行为
第8课  从零搭建了完整的智能研究平台
第9课  优化了成本、日志、错误处理

这套知识不仅能搭研究平台，还能用来做任何多Agent协作的系统。原理都是一样的：定义角色 → 分配工具 → 设定流程 → 监控执行。

去做点有意思的东西吧！

继续阅读

沿着当前专题继续，或返回课程目录重新整理阅读顺序。

返回课程目录