1. 多代理投研平台为什么绕不开 SubAgent 编排一份完整的投研报告拆开看其实是三条互不干扰的信息流财报数据要读 PDF、算指标、做同比行业新闻要抓时效、去重、判断信源质量风险预警要盯公告、盯异动、盯舆情。这三件事的输入源、工具链、判断逻辑完全不同。如果塞给一个 Agent 串行做它会频繁在「财务分析师」「新闻编辑」「风控专员」三个角色之间切换上下文互相污染最后产出一份四平八稳但没什么洞察的综述。Claude Agent SDK 给出的解法是 SubAgent-as-tool主代理Orchestrator把每个子代理当成一把工具什么时候调、调哪个由主代理自己判断。子代理运行时上下文完全隔离它读的文件、搜的结果、中间推理过程都留在自己的窗口里主代理只能看到最终结论。这个隔离特别关键——Anthropic 在官方工程博客里披露过内部多代理研究系统的数据多代理架构相比单代理在内部研究基准上提升了 90.2%代价是 token 消耗大约是单代理的 15 倍。听起来成本很高但三个子代理并行跑每个都在自己的领域深耕比一个代理在三个领域之间来回切换角色质量上的提升远超 token 的代价。而且他们自己也说这种架构只适合真正需要多方向探索的复杂任务简单问答用它反而亏。我一开始想过自己手搓一个 Function-Calling 的编排逻辑写asyncio.gather()做并行调度。但后来发现 SDK 已经把这些都封装好了你只需要在 prompt 里告诉主代理「这些任务可以并行调用」它自己就会在一条消息里同时发多个 Agent 工具调用SDK 底层自动并行执行。这个设计让我省了至少几百行编排代码。# backend/agents/registry.py # 三个子代理的注册表key 就是 Agent 工具调用时使用的名字 from backend.agents.financial import financial_analyzer_agent from backend.agents.industry_news import industry_news_collector_agent from backend.agents.risk_alert import risk_alert_agent def build_agents_config() - dict[str, AgentDefinition]: Return the SubAgent config dict registered on the orchestrator. return { financial-analyzer: financial_analyzer_agent(), industry_news_collector: industry_news_collector_agent(), a-share-risk-alert: risk_alert_agent(), } # 单例所有 research run 复用同一份配置 agents_config build_agents_config()不过这里有个坑部分国产模型自主规划能力有限它会倾向于直接调用 Skill而不是走 SubAgent。所以我在前端提示词里硬编码了一句「必须使用以下子 agent 完成任务不能自行调用 skills」强制主代理走 Agent 工具路由。这句话在backend/api/research.py的_build_prompt()里def _build_prompt(req: ResearchRequest) - str: Augment the user prompt with file path and explicit SubAgent routing. if req.mode ! fresh: # Resume / fork 复用已有上下文不需要再注入路由指令 return req.prompt parts: list[str] [] if req.agents: # 显式指定要调度的子代理防止主代理绕过 Agent 工具直接调用 Skill names 、.join(req.agents) parts.append(f必须使用以下子agent完成任务不能自行调用skills{names}。) parts.append(req.prompt) if req.file_id: # 把上传的 PDF 路径拼进提示词financial-analyzer 会读取它 path f{settings.upload_dir}/{req.file_id}.pdf parts.append( f财报PDF文件路径为{path}请使用 financial-analyzer agent 读取并分析。 ) return \n.join(parts)这段路由指令看起来简单但它决定了整个编排是否按预期走。我试过不加这句主代理有大概三成概率会跳过 SubAgent直接自己读 PDF 然后写一段泛泛的分析子代理面板全程空白用户以为系统卡住了。加上之后调度路径稳定了前端三个 SubAgent 面板也能正常亮起来。还有一个容易被忽略的点子代理的description字段是主代理决定何时调用的唯一依据必须写得具体。我见过有人写「分析财务数据」结果主代理经常不调它因为描述太模糊主代理判断不出这个工具和当前任务的关联。改成「财务报告分析专家擅长解析 PDF 财报并生成结构化 Markdown 报告」之后调用命中率明显上升。prompt字段是子代理的系统提示词决定它的专业领域和行为边界tools限制它能调用的工具避免越权modelinherit表示继承主代理的模型配置无需单独指定。2. TaoToken 统一 Key 接入把模型调用通道收敛到一处多代理系统跑起来之后模型调用会变得很分散主代理要调一次三个子代理各调若干次加上工具调用后的二次推理一次投研任务可能产生几十次 API 请求。如果每个 Agent 各自配置一套 Key 和 Base URL管理成本会迅速失控——换模型要改四处限流要分别处理出问题排查时根本不知道是哪条通道挂了。我的做法是通过 TaoToken 统一 Key/API 通道完成模型调用接入把所有 Agent 的模型请求收敛到一个入口。TaoToken 是一个模型 API 聚合通道官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。它的价值在于你只需要维护一个 Key、一个 Base URL主代理和所有子代理共用同一套配置换模型时改一个环境变量就全局生效。具体接入方式是在ClaudeAgentOptions的env里注入 Anthropic 兼容的环境变量。Claude Agent SDK 底层走的是 Anthropic 协议所以只要把ANTHROPIC_BASE_URL指向 TaoToken 的 API 端点把ANTHROPIC_API_KEY设成你的 TaoToken KeySDK 就会自动走这条通道。# backend/agents/options.py import os from claude_agent_sdk import ClaudeAgentOptions def build_model_env() - dict[str, str]: Inject unified model channel env vars for all agents. return { # 统一走 TaoToken 的 Anthropic 兼容端点 ANTHROPIC_BASE_URL: os.environ.get( TAOTOKEN_BASE_URL, https://taotoken.net/api ), ANTHROPIC_API_KEY: os.environ[TAOTOKEN_API_KEY], # 主代理和子代理共用同一个模型 ID换模型只改这一处 ANTHROPIC_MODEL: os.environ.get(TAOTOKEN_MODEL, claude-sonnet-4-5), }这里有个细节值得展开。ANTHROPIC_MODEL设成环境变量之后子代理配置里的modelinherit就会继承这个值不需要在每个AgentDefinition里重复写模型 ID。如果你想让某个子代理用更便宜的模型比如新闻采集这种偏检索的任务可以在它的AgentDefinition里显式指定modelclaude-haiku-4-5覆盖继承值。这样主代理用强模型做汇总子代理按任务复杂度分配模型成本能压下来不少。Key 的获取在 TaoToken 控制台的 API Keys 页面地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。创建之后复制出来写进.env文件# .env TAOTOKEN_API_KEYsk-xxxxxxxxxxxxxxxxxxxxxxxx TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELclaude-sonnet-4-5注意.env文件必须加进.gitignore这个在后面的治理章节还会提到——我们的 PreToolUse Hook 会拦截对.env的写操作但读操作是放行的所以千万别把 Key 硬编码进代码。统一通道还有一个隐性好处可观测性。所有模型请求都经过同一个端点OpenTelemetry 导出的 span 里service.name和enduser.id能统一归因不会出现「这个子代理走了 A 通道、那个走了 B 通道」导致成本统计对不上的情况。我在 Grafana 里按tenant.id聚合 token 用量时数据来源是干净的。如果你还没创建 Key可以先到模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 试一下通道是否通畅确认能正常返回再接入 SDK。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的 Base URL 和参数说明。3. 可复制配置Agent 定义、会话模式与 OTel 环境变量这一节把前面提到的配置片段整合成可以直接复制的完整版本。先看子代理定义这是整个编排的基础单元。# backend/agents/financial.py from claude_agent_sdk import AgentDefinition def financial_analyzer_agent() - AgentDefinition: 财务报告分析子代理负责读取 PDF 财报并生成结构化报告。 return AgentDefinition( # description 是主代理决定何时调用的唯一依据必须写得具体 description( 财务报告分析专家擅长解析 PDF 财报并生成结构化 Markdown 报告 包括营收、利润、现金流、同比环比等核心指标。 ), # prompt 是子代理的系统提示词决定它的专业领域和行为边界 prompt( 你是一位资深财务分析师。读取指定的财报 PDF提取三大报表核心数据 计算同比环比识别异常科目最终输出结构化 Markdown 报告。 所有数字必须标注来源页码不确定的数据明确标注「待核实」。 ), # tools 限制子代理能调用的工具避免越权 tools[Read, Grep, Glob, Bash, Write, Edit], # skills 挂载黑盒复用的 Skill这里直接复用参考实现 skills[financial-report-analyzer], # modelinherit 表示继承主代理的模型配置无需单独指定 modelinherit, )会话管理是 SDK 设计里最优雅的部分。传统 API 调用是无状态的发一个请求拿一个响应就结束。但投研不是这样分析师可能会先跑一份完整分析然后追问「补充最近一周的行业利空」或者分岔出一条「用不同折现率做 DCF 估值」的探索分支。SDK 原生支持三种会话模式# fresh全新会话生成新 session_id # 显式传入 UUID 是关键否则 SDK 可能使用临时会话后续无法 resume/fork options ClaudeAgentOptions( session_idstr(uuid.uuid4()), agentsagents_config, # 注册三个子代理 mcp_servers{websearch: websearch_server}, # 挂载 Bocha 搜索 hooksbuild_hooks(), # 注入治理 Hook envbuild_otel_env(), # 注入 OpenTelemetry 环境变量 ) # resume继续追问同一个 session_id # 注意参考文档说 resume 只需传 session_id但实际测试发现 SDK 不总是可靠恢复 SubAgent 定义 options ClaudeAgentOptions( resumesession_id, agentsagents_config, # 必须显式传入 mcp_servers{websearch: websearch_server}, # 必须显式传入 hooksbuild_hooks(), envbuild_otel_env(), ) # fork分岔探索克隆出新 session_id原会话不受影响 # max_turns 限制分支深度避免无限探索 options ClaudeAgentOptions( resumesession_id, fork_sessionTrue, max_turns5, agentsagents_config, mcp_servers{websearch: websearch_server}, hooksbuild_hooks(), envbuild_otel_env(), )这里有个坑我踩了很久。最初按照参考文档的说法resume 和 fork 只传resumesession_id就行SDK 会从存储的 session 文件里恢复 agents 和 MCP 配置。但实际上SDK 并不总是可靠地恢复 SubAgent 定义。我在测试 fork 的时候发现fork 出来的会话启动了一个general-purpose类型的默认代理而不是我配置的三个专用子代理。后端日志显示 SubagentStart hook 触发了但agent_type是general-purpose完全不对。修复方式很简单但文档没说resume 和 fork 的时候也要显式传入agentsagents_config和mcp_servers不能依赖 SDK 的 session 恢复机制。另一个值得注意的点是session_id的获取。SDK 在运行结束后会返回一个ResultMessage里面的session_id字段就是这次会话的 ID。对于 fresh 和 fork这是一个新 ID对于 resume跟之前一样。我们把这个 ID 附在 SSE 的final_result事件里返回给前端前端就能基于它发起 resume 或 fork。# backend/agents/orchestrator.py async def _translate(msg: Any, ctx: _RunContext) - AsyncIterator[dict[str, Any]]: Convert a single SDK message into typed SSE events. if isinstance(msg, ResultMessage): report msg.result or # 从 SDK 返回的 ResultMessage 中提取 session_id # fresh/resume 与输入一致fork 会返回全新的分支 session_id session_id getattr(msg, session_id, None) yield { type: final_result, agent: orchestrator, data: { report: report, disclaimer: DISCLAIMER, session_id: session_id or , }, } returnOpenTelemetry 的环境变量配置是另一个必须可复制的片段。Claude Agent SDK 在 CLI 子进程内置了 OTel 插桩你不需要改一行业务代码只需要通过ClaudeAgentOptions.env传入正确的环境变量。# backend/agents/telemetry.py from urllib.parse import quote def build_otel_env( service_name: str iiras, enduser_id: str , tenant_id: str , ) - dict[str, str]: Build the OpenTelemetry env dict for ClaudeAgentOptions.env. Returns an empty dict when OTEL_EXPORTER_OTLP_ENDPOINT is unset, so telemetry is opt-in and never blocks a run. endpoint settings.otel_exporter_otlp_endpoint if not endpoint: # 端点没配就关闭遥测开发环境无感知 return {} # Resource attributes 会附在每条 span / metric / log event 上 attrs [ service.version1.0.0, deployment.environmentdevelopment, ] if enduser_id: # 对特殊字符做 percent-encode防止破坏 OTEL_RESOURCE_ATTRIBUTES 解析 attrs.append(fenduser.id{quote(enduser_id)}) if tenant_id: attrs.append(ftenant.id{quote(tenant_id)}) return { # 总开关没有它 CLI 不会导出任何信号 CLAUDE_CODE_ENABLE_TELEMETRY: 1, # Traces 的 beta 开关调用链目前还是 beta 特性 CLAUDE_CODE_ENHANCED_TELEMETRY_BETA: 1, # 三种信号各选 otlp exporter千万不要用 console OTEL_TRACES_EXPORTER: otlp, OTEL_METRICS_EXPORTER: otlp, OTEL_LOGS_EXPORTER: otlp, # OTLP 传输协议用 HTTP/protobuf比 gRPC 少一个依赖 OTEL_EXPORTER_OTLP_PROTOCOL: http/protobuf, OTEL_EXPORTER_OTLP_ENDPOINT: endpoint, # 服务名与资源属性用于在 Jaeger/Grafana 中筛选 OTEL_SERVICE_NAME: service_name, OTEL_RESOURCE_ATTRIBUTES: ,.join(attrs), # 导出间隔降到 1 秒 # CLI 默认 metrics 60 秒、traces/logs 5 秒导出一批 # 投研任务可能几分钟就跑完不调低的话进程结束时数据还没发出去 OTEL_METRIC_EXPORT_INTERVAL: 1000, OTEL_LOGS_EXPORT_INTERVAL: 1000, OTEL_TRACES_EXPORT_INTERVAL: 1000, }这个函数有几个设计决策值得展开。opt-in 设计端点没配就返回空字典遥测完全关闭。这样开发的时候不用管 Collector上线的时候在.env里加一个OTEL_EXPORTER_OTLP_ENDPOINThttp://your-collector:4318就开了。用户和租户归因enduser.id和tenant.id会附在每条 span、每个 metric、每个 log event 上在 Grafana 里你可以按分析师筛选成本按团队聚合 token 用量。1 秒导出间隔这个很重要。CLI 默认 metrics 每 60 秒导出一批traces 和 logs 每 5 秒。但投研任务一次可能就跑三五分钟如果用默认间隔进程结束时可能还有一大批数据没发出去。全部降到 1 秒基本能保证数据不丢。还有一个雷区永远不要把 exporter 设为console。因为 console 输出会占用 SDK 和 CLI 子进程之间的 stdio 通信通道导致运行异常。本地调试用 Jaeger all-in-one 这种本地 Collector 就行。4. 验证请求一轮端到端跑通与成功结果判读配置写完接下来要验证整条链路是否真的通了。我设计了一轮最小化的端到端验证动作覆盖模型调用、SubAgent 调度、SSE 回传、OTel 导出四个环节。第一步确认 TaoToken 通道本身可用。用一个最简单的 curl 请求打一下 API 端点curl -X POST https://taotoken.net/api/v1/messages \ -H x-api-key: $TAOTOKEN_API_KEY \ -H anthropic-version: 2023-06-01 \ -H content-type: application/json \ -d { model: claude-sonnet-4-5, max_tokens: 64, messages: [{role: user, content: 回复 OK 两个字母即可}] }如果返回的 JSON 里有content字段且文本是「OK」说明 Key 和 Base URL 都对。如果返回 401说明 Key 有问题如果返回 404说明 Base URL 路径不对注意 TaoToken 的端点是https://taotoken.net/apiSDK 会自动拼接/v1/messages。第二步跑一个单子代理的最小任务验证 SubAgent 调度。构造一个只触发financial-analyzer的请求# scripts/verify_single_agent.py import asyncio from claude_agent_sdk import query, ClaudeAgentOptions from backend.agents.registry import build_agents_config from backend.agents.options import build_model_env async def main(): options ClaudeAgentOptions( agentsbuild_agents_config(), envbuild_model_env(), permission_modebypassPermissions, ) async for msg in query( prompt请使用 financial-analyzer agent 分析这份财报/tmp/test-report.pdf, optionsoptions, ): print(type(msg).__name__, getattr(msg, result, )[:120]) asyncio.run(main())预期输出里应该能看到AssistantMessage、ToolUseMessage工具名是Agent参数里subagent_type是financial-analyzer、ResultMessage。如果只看到AssistantMessage和ResultMessage中间没有Agent工具调用说明主代理没走 SubAgent 路由回去检查_build_prompt()里的强制路由指令是否生效。第三步验证 SSE 流式回传。启动后端用 curl 打 SSE 端点curl -N -X POST http://localhost:8000/api/research/stream \ -H Content-Type: application/json \ -d { prompt: 分析这份财报并给出风险提示, agents: [financial-analyzer, a-share-risk-alert], mode: fresh }-N参数关闭 curl 的缓冲让你能实时看到事件流。预期会依次看到subagent_dispatch、tool_call、partial_message、subagent_result、final_result、done这几类事件。每个事件的格式是event: type\ndata: json\n\n。如果只收到final_result和done中间过程事件全丢了说明 SSE 的yield被缓冲了检查 FastAPI 的StreamingResponse是否设置了media_typetext/event-stream以及有没有中间件在缓冲响应。第四步验证 OTel 数据导出。本地起一个 Jaeger all-in-onedocker run -d --name jaeger \ -p 16686:16686 \ -p 4318:4318 \ jaegertracing/all-in-one:latest然后在.env里设置OTEL_EXPORTER_OTLP_ENDPOINThttp://localhost:4318重跑一次任务打开http://localhost:16686搜索服务名iiras。你应该能看到一条完整的 trace从 Orchestrator 的根 span 开始下面挂着三个子代理的 span每个子代理 span 下面又挂着工具调用和 LLM 请求的 span。如果只看到根 span 没有子 span检查CLAUDE_CODE_ENHANCED_TELEMETRY_BETA1是否设置traces 目前还是 beta 特性不开这个开关只有 metrics 和 logs。成功结果的判读标准我总结成三条模型调用返回 200 且内容非空SubAgent 面板三个都亮起且各自有工具调用记录Jaeger 里能看到 Agent → SubAgent → Tool → LLM 的完整调用链。三条都满足说明整条链路通了。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节把我在搭建过程中真实遇到的报错和排查路径整理出来对照着看能省不少时间。401 Unauthorized。这是最常见的报错通常有三个原因。第一ANTHROPIC_API_KEY没设或者设成了空字符串检查.env是否被正确加载Python 里用os.environ[TAOTOKEN_API_KEY]会在缺失时直接抛 KeyError比os.environ.get()更早暴露问题。第二Key 复制时带了首尾空格TaoToken 控制台复制出来的 Key 有时会带换行写进.env时要注意。第三Base URL 写成了https://taotoken.net而不是https://taotoken.net/apiSDK 会往https://taotoken.net/v1/messages发请求路径不对自然 401。排查命令# 确认环境变量已加载 python -c import os; print(os.environ.get(ANTHROPIC_BASE_URL)); print(os.environ.get(ANTHROPIC_API_KEY, )[:8])local proxy failed。这个报错通常出现在 SDK 启动 CLI 子进程的时候提示本地代理连接失败。原因是 SDK 会尝试通过一个本地代理端口和 CLI 通信如果这个端口被占用或者防火墙拦截就会报这个错。排查方向检查是否有其他进程占用了 SDK 默认的代理端口用lsof -i :port查一下如果是容器环境确认容器网络策略没有拦截 localhost 通信。这个报错和模型通道无关是 SDK 内部通信问题不要往 TaoToken 配置上找原因。reading choices 相关报错。这个报错出现在解析模型响应的时候提示读取choices字段失败。根本原因是响应格式不匹配——SDK 期望的是 Anthropic 格式content数组但实际收到的是 OpenAI 格式choices数组。这通常发生在 Base URL 指向了一个 OpenAI 兼容端点而不是 Anthropic 兼容端点。TaoToken 的https://taotoken.net/api是 Anthropic 兼容的如果你误配成了其他路径就会出现这个错。确认方式用第 4 节的 curl 命令打一下看返回的 JSON 顶层是content还是choices。OAuth 相关报错。Claude Agent SDK 的 CLI 子进程默认会尝试 OAuth 登录流程在无头服务器上会卡住或报错。解决办法是显式设置 API Key 模式通过环境变量ANTHROPIC_API_KEY提供凭证SDK 检测到 API Key 存在就不会走 OAuth。如果还是报 OAuth 错检查是否有残留的 CLI 配置文件在干扰清理~/.claude/下的凭证缓存后重试。SubAgent 不触发。这个不算报错但很常见。表现是任务跑完了但子代理面板全程空白最终报告是主代理自己写的。原因有两个一是description写得太模糊主代理判断不出何时调用二是提示词里没有强制路由指令主代理倾向于自己干。修复方式在 §1 已经讲过把description写具体在_build_prompt()里加「必须使用以下子agent」的硬约束。SSE 事件丢失。表现是前端只收到final_result中间过程事件全没有。原因是响应被缓冲了。FastAPI 的StreamingResponse需要设置media_typetext/event-stream并且要确保没有中间件比如 GZip 中间件在缓冲响应体。另外 Nginx 反代的话要加proxy_buffering off;。OTel 数据不导出。表现是 Jaeger 里什么都搜不到。排查顺序先确认CLAUDE_CODE_ENABLE_TELEMETRY1设了再确认OTEL_EXPORTER_OTLP_ENDPOINT指向的 Collector 可达用curl http://localhost:4318/v1/traces测一下最后确认导出间隔调到了 1 秒默认 60 秒的话短任务数据可能还没发出去进程就结束了。CC Switch / Cline MCP / Codex auth.json 三件套。如果你在项目里用到了这些工具做模型切换或 MCP 配置记住任何一处配置都要写全三件套Base URL、Key、Model ID。缺任何一个都会导致调用失败。Base URL 统一用https://taotoken.net/apiKey 用 TaoToken 控制台创建的 KeyModel ID 用你实际要调的模型名。三件套不一致是排查时最容易忽略的坑比如 Base URL 改了但 Model ID 还是旧的就会报模型不存在。6. 治理与观测Hook 三防线与成本归因把多代理系统放到服务器上跑最大的问题是什么没人看着它。CLI 脚本在本地跑的时候工具调用的权限确认弹出来你点一下就行。但在后端服务里没有人去点「允许」。SDK 的解决方案是permission_modebypassPermissions直接跳过所有权限确认。但这等于完全不设防一个 hallucination 就可能让代理执行rm -rf。这就是 Hook 机制的价值。SDK 提供了多个生命周期 Hook在工具调用前后、子代理启停、会话结束等节点都会触发回调。我用其中几个搭建了三道防线。第一道防线PreToolUse 权限守卫。在每次工具调用前触发自动判断是放行还是拒绝。只读工具如 Read、Grep、Glob 直接放行高危操作如 Write、Edit 敏感文件、destructive Bash 命令直接拒绝。# backend/agents/hooks.py READONLY_TOOLS {Read, Glob, Grep} # 高危 Bash 模式只要命中就拦截 DANGEROUS_BASH_PATTERNS ( rrm\s-rf\s/, rmkfs\., r:\(\)\s*\{\s*:\|\:\s*\s*\}, # fork bomb r\s*/dev/(sda|sd[b-z]|nvme|disk), rdd\sif.*of/dev/, ) # 受保护路径禁止写操作 PROTECTED_PATH_PREFIXES (/etc, /sys, /proc, C:\\Windows) PROTECTED_FILE_NAMES {.env} async def pre_tool_guard( input_data: HookInput, tool_use_id: str | None, context: HookContext, ) - dict[str, Any]: Layered permission guard for headless agent runs. Returns: - allow: for read-only tools. - deny: for protected-file writes or destructive Bash commands. - {}: for anything else, letting the SDK fall back to its default. if input_data.get(hook_event_name) ! PreToolUse: return {} data PreToolUseHookInput(input_data) tool_name data.get(tool_name, ) tool_input data.get(tool_input) or {} # 快速路径只读工具自动放行无需人工确认 if tool_name in READONLY_TOOLS: return { hookSpecificOutput: { hookEventName: PreToolUse, permissionDecision: allow, permissionDecisionReason: read-only tool, } } # 拦截对 .env 或系统目录的写操作 if tool_name in {Write, Edit, MultiEdit}: file_path tool_input.get(file_path) or tool_input.get(path) or if _is_protected_path(str(file_path)): reason f{tool_name} on protected path is not allowed in headless mode return _deny(reason) # 拦截破坏性 Bash 命令 if tool_name Bash: command tool_input.get(command, ) if _is_dangerous_command(str(command)): return _deny(destructive Bash command blocked) # 其余情况不干预让 SDK 走默认流程 return {}这里有个巨坑。permissionDecision字段必须嵌套在hookSpecificOutput里面不能直接放在返回字典的顶层。我一开始按照直觉写了顶层结果 CLI 完全无视我的决策该放行的没放行该拦截的没拦截。查了 SDK 的SyncHookJSONOutput类型定义才发现嵌套结构的要求。第二道防线PostToolUse 审计日志。每次工具调用后触发把调用的完整信息写入 JSONL 文件。这样每一份投研报告都可以回溯到它调用了哪些工具、读了哪些文件、搜了什么关键词。async def audit_logger( input_data: HookInput, tool_use_id: str | None, context: HookContext, ) - dict[str, Any]: Append a JSONL record for every completed or failed tool call. event_name input_data.get(hook_event_name) if event_name not in {PostToolUse, PostToolUseFailure}: return {} try: record _build_audit_record(input_data, tool_use_id) _append_jsonl(record) except Exception as exc: # 审计失败不能影响主流程必须吞掉异常 logger.exception(audit_logger failed to record tool call: %s, exc) return {}第三道防线SubagentStart / SubagentStop 生命周期追踪。子代理启动和结束时触发记录agent_id和transcript_path。这样即使主代理只看到最终结果我们也能通过日志知道哪个子代理先启动、哪个卡住了、哪个失败了。三道防线加在一起让这个无人值守的系统有了基本的治理能力。不是完美的但至少不会因为一个 hallucination 就把服务器搞炸。观测层面OpenTelemetry 导出的 metrics 里会自动记录cache_read_input_tokens你在 Grafana 里可以做一个 cached_vs_uncached 的对比图来看缓存效果。成本的计算公式其实不复杂C (Ni × Pi No × Po Nc × Pc) / 1,000,000其中 C 是单次调用总成本单位美元Ni 是输入 token 数即发送给模型的 prompt、历史消息、工具结果的总 token 量多代理场景下每个子代理都有自己的输入所以 Ni 会累加Pi 是输入 token 单价单位美元/百万 token不同模型价格不同No 是输出 token 数模型生成的文本量输出通常比输入贵所以 No 对成本的影响往往比 Ni 更大Po 是输出 token 单价通常比输入价高 3-5 倍是成本的大头Nc 是缓存命中 token 数prompt caching 命中后模型只需读取已缓存的 token这部分有独立计价Pc 是缓存读取单价通常是标准输入价格的 10-20%也就是说比正常输入便宜 80-90%1,000,000 是因为 Pi、Po、Pc 都是「每百万 token」的报价所以总费用要除以一百万来得到实际美元数。多代理场景下一次投研任务的总成本是所有子代理和主代理各自调用的 C 之和。这就是为什么 token 消耗会是单代理的 15 倍——三个子代理各有自己的输入输出主代理还要汇总token 走的是加法。但缓存命中率高的话实际花费会低很多。内容级日志默认关闭。OTel 默认只导出结构信息耗时、模型名、工具名、token 数不会记录 prompt 内容和工具输出。但有几个开关可以按需开启OTEL_LOG_USER_PROMPTS1会记录 prompt 文本OTEL_LOG_TOOL_CONTENT1会记录完整的工具输入输出。在投研场景中财报 PDF 内容、搜索关键词、内部文件路径都可能涉及敏感信息所以我的建议是默认不开只有当你的可观测 pipeline 经过安全审批后按需打开。本地审计日志已经覆盖了内容级合规需求。如果你需要长期跑编码类或 Agent 类任务可以考虑 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它针对高频调用场景做了额度优化。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite API Keys 管理在 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。Claude Code 相关的接入说明在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite 里面有 Anthropic 协议的完整配置示例。最后说一个我踩过的坑SSE 事件里的done事件可能为空数据前端需要单独处理不能直接JSON.parse。这个细节在useSSEStream.ts里处理了done事件没有data字段时直接标记完成即可。另外 CLI 子进程返回的 Agent 工具结果里会附带大量协议元数据比如agentId: a78cf5ef4ead7f68e (internal ID - do not mention to user...)这些是 CLI 内部的代理间通信指令不应该展示给用户。我在后端写了_clean_subagent_report()函数用几个正则把这些噪声剥离前端也做了一层兜底清理。不做这一步的话用户会在报告里看到一堆莫名其妙的内部 ID 和指令文本体验很差。