1. 为什么“失败”才是工具运行时的核心资产1.1 从一次线上事故说起去年冬天我负责的一个智能体项目在凌晨两点崩了。日志里只有一行冷冰冰的agent execution terminated due to error.没有堆栈没有上下文连是哪个工具调用挂掉的都不知道。团队花了整整四个小时才定位到问题一个天气查询工具在返回结果时因为上游接口超时返回了一个空对象而下游的解析逻辑直接对空对象取属性触发了未捕获异常整个 agent 链路瞬间断裂。那次事故之后我彻底改变了对“工具运行时”的认知。以前我总觉得工具调用嘛成功返回结果就完事了失败就是异常异常就该被 catch 掉然后重试或者跳过。但真正做过 agent 开发的人都知道失败不是噪音失败是信号失败是数据。一个成熟的工具运行时不是要把失败消灭掉而是要把失败变成可观测、可分析、可复用的结构化信息。这个项目标题“P04 工具运行时失败是数据”说的就是这件事。它不是一个具体的代码库而是一种设计理念在 agent 框架里工具执行的结果不应该只有“成功”和“失败”两种状态而应该是一份完整的执行记录包含输入参数、输出内容、错误类型、耗时、重试次数、上下文快照等等。这份记录本身就是数据可以用来做调试、做监控、做评测、做模型微调甚至用来训练 agent 自己学会规避错误。如果你正在搭建 agent或者正在设计 agent 框架与编排层又或者你只是好奇“agent 是什么”“agent 开发学习路线”该怎么走那这篇文章应该能给你一些从实战里摔出来的经验。我会从设计思路、核心细节、实操过程、常见问题四个维度把“失败是数据”这件事拆开揉碎讲清楚。1.2 工具运行时的定位agent 的手和脚在聊失败之前先明确工具运行时在 agent 架构里的位置。一个典型的 agent 系统大致可以分成四层规划层决定做什么、记忆层记住什么、工具层执行什么、编排层协调什么。工具运行时属于工具层的核心组件它负责接收规划层下发的工具调用请求执行具体的函数或 API然后把结果返回给编排层。你可以把 agent 想象成一个项目经理规划层是大脑记忆层是笔记本工具运行时就是项目经理的手和脚。手和脚干活的时候不可能每次都顺顺利利。有时候是工具本身有问题比如 API 挂了有时候是参数传错了比如日期格式不对有时候是环境问题比如网络超时有时候是权限问题比如 token 过期。这些失败如果只是被简单记录成“调用失败”那项目经理就永远不知道到底是手的问题还是脚的问题下次还会踩同样的坑。所以工具运行时的设计目标不是追求 100% 成功率而是让每一次执行——无论成功还是失败——都留下足够的信息让系统能够理解发生了什么并据此做出更好的决策。这就是“失败是数据”的第一层含义失败不是终点而是下一次决策的输入。1.3 为什么传统异常处理在 agent 场景下不够用传统后端开发里异常处理有一套成熟的模式try-catch-finally日志记录错误码返回。这套模式在单体应用里很好用但在 agent 场景下会暴露三个致命问题。第一异常粒度太粗。一个ToolExecutionError可能包含几十种不同的失败原因参数校验失败、网络超时、限流、认证失败、返回格式不符合 JSON Schema、业务逻辑错误等等。如果全部归为一类排查的时候就像大海捞针。第二异常丢失上下文。传统异常往往只记录错误信息和堆栈但 agent 工具调用需要知道当时传了什么参数是第几次重试上游 agent 的意图是什么这些信息在传统异常里通常没有。第三异常不可复用。一次失败如果只是被记录到日志文件里那它就只是历史。但如果把失败结构化存储它就可以变成评测集、变成 few-shot 示例、变成模型微调的数据。传统异常处理没有为这种复用设计接口。我试过在一个早期项目里直接用 Python 的try/except包裹工具调用结果就是日志里全是KeyError和TimeoutError根本看不出哪个工具在什么场景下容易挂。后来改成结构化记录才真正把失败变成了可分析的数据。2. 核心细节解析失败数据的结构化设计2.1 一次工具调用的完整生命周期要理解失败数据该怎么设计先得看清楚一次工具调用从生到死经历了什么。我把它拆成六个阶段请求构造阶段编排层根据 agent 的决策构造工具调用请求包括工具名、参数、调用 ID、超时设置等。参数校验阶段工具运行时根据工具的 JSON Schema 校验参数是否合法。这一步能拦掉大量低级错误。执行阶段真正调用工具函数或 API可能涉及网络请求、文件读写、数据库查询等。结果解析阶段把工具返回的原始结果解析成 agent 能理解的格式通常要求符合预定义的输出 Schema。状态记录阶段把执行结果成功或失败写入执行记录供后续查询和分析。反馈阶段把结果返回给编排层编排层决定下一步是重试、换工具、还是终止。失败可能发生在任何一个阶段。参数校验失败和执行超时虽然都是“失败”但处理策略完全不同。所以失败数据的设计必须能区分失败发生在哪个阶段。2.2 失败数据的字段设计一份可落地的 Schema下面是我在实际项目里用过的一套失败数据结构用 JSON Schema 描述。这套结构经过多次迭代基本能覆盖 agent 工具运行时的常见场景。{ $schema: http://json-schema.org/draft-07/schema#, title: ToolExecutionRecord, type: object, required: [execution_id, tool_name, status, timestamp], properties: { execution_id: { type: string, description: 本次执行的唯一标识用于追踪和关联 }, tool_name: { type: string, description: 被调用的工具名称 }, status: { type: string, enum: [success, failure, timeout, rejected], description: 执行状态 }, failure_stage: { type: string, enum: [request_build, param_validation, execution, result_parse, feedback], description: 失败发生的阶段 }, failure_type: { type: string, description: 失败类型如 network_error, schema_mismatch, auth_error 等 }, input_params: { type: object, description: 实际传入的参数快照 }, raw_output: { type: string, description: 工具返回的原始输出截断到合理长度 }, error_message: { type: string, description: 错误信息 }, error_stack: { type: string, description: 错误堆栈仅内部调试用 }, duration_ms: { type: integer, description: 执行耗时毫秒 }, retry_count: { type: integer, description: 当前是第几次重试从 0 开始 }, context_snapshot: { type: object, description: 执行时的上下文快照如 agent 意图、会话 ID 等 } } }这套 Schema 有几个设计要点值得展开说。execution_id 是追踪的锚点。每次工具调用生成一个唯一 ID所有日志、指标、追踪都围绕这个 ID 展开。排查问题时只要拿到这个 ID就能还原整个调用链路。failure_stage 和 failure_type 分开。阶段回答“在哪挂的”类型回答“为什么挂的”。这两个维度交叉分析能快速定位系统性问题。比如如果大量失败集中在param_validation阶段说明 agent 的规划层需要优化如果集中在execution阶段且类型是network_error说明需要加强重试和熔断。input_params 必须记录实际值。很多框架只记录参数 Schema不记录实际传入的值。但排查问题时你往往需要知道“当时到底传了什么”。比如一个日期参数Schema 说是 string但实际传的是2024-13-45这种错误只有看到实际值才能发现。raw_output 要截断。工具返回的原始输出可能非常大比如一个网页抓取工具返回几十 KB 的 HTML。全部存下来会撑爆存储所以需要截断到合理长度比如 2000 字符同时记录原始长度。context_snapshot 是 agent 特有的。传统 API 调用不需要知道“调用者的意图”但 agent 工具调用需要。因为同一个工具在不同意图下失败的处理策略可能不同。比如一个搜索工具在“查天气”意图下失败可以跳过在“查合同条款”意图下失败就必须重试。2.3 失败分类体系从混沌到有序有了字段还需要一套分类体系。否则failure_type会变成每个人随手写的字符串最后没法聚合分析。我参考了 HTTP 状态码和 gRPC 错误码的设计思路把工具失败分成六大类失败类别典型场景是否可重试处理策略参数错误Schema 不匹配、必填缺失、类型错误否反馈给规划层要求修正参数认证错误Token 过期、权限不足、签名错误视情况刷新凭证后重试一次网络错误超时、连接拒绝、DNS 失败是指数退避重试最多 3 次限流错误429、配额耗尽是等待后重试或降级到备用工具业务错误查询无结果、状态冲突否返回给 agent由 agent 决定下一步系统错误工具内部异常、依赖服务宕机视情况熔断切换到备用方案这套分类的价值在于它把“失败”从一个笼统的概念变成了可操作的决策依据。编排层拿到失败记录后可以根据failure_type直接查表决定下一步动作而不需要每次都用 if-else 硬编码。注意分类体系不要设计得太细。我见过一个项目把失败分成 47 种类型结果维护成本极高而且很多类型一年都遇不到一次。六大类基本够用如果确实需要更细的粒度可以在error_message里补充细节。2.4 JSON Schema 在参数校验中的实战用法前面提到参数校验阶段能拦掉大量低级错误这里展开说说 JSON Schema 的具体用法。在 agent 工具运行时里每个工具都应该有一份输入 Schema 和一份输出 Schema。输入 Schema 用于校验 agent 传来的参数输出 Schema 用于校验工具返回的结果。输入 Schema 的编写有几个容易踩的坑。第一不要用additionalProperties: false除非你确定。agent 在生成参数时有时会多传一些字段如果严格禁止额外属性会导致大量本可避免的失败。第二枚举值要留扩展空间。比如一个排序参数如果你只允许asc和desc那 agent 传ascending就会失败。可以在校验前做一层归一化。第三默认值要显式声明。agent 可能不传某些可选参数工具运行时应该根据 Schema 里的default自动填充。输出 Schema 的校验同样重要。我遇到过好几次工具返回了“看起来正常”的结果但字段类型不对导致下游解析崩溃。比如一个返回数字的工具因为上游接口变更返回了字符串123如果不在输出阶段校验问题会一直传到 agent 的决策层才暴露。from jsonschema import validate, ValidationError def validate_tool_input(tool_schema, params): try: validate(instanceparams, schematool_schema) return True, None except ValidationError as e: return False, { failure_stage: param_validation, failure_type: schema_mismatch, error_message: str(e), input_params: params }这段代码看起来简单但它是工具运行时的第一道防线。实测下来加上参数校验之后执行阶段的失败率能降低 40% 以上因为很多问题在进入真正执行之前就被拦住了。3. 实操过程从零搭建一个带失败数据记录的工具运行时3.1 整体架构与模块划分下面是我在一个实际项目里落地的工具运行时架构。整个运行时分成五个模块注册中心管理所有工具的元信息包括名称、描述、输入 Schema、输出 Schema、超时设置、重试策略。校验器负责输入参数校验和输出结果校验。执行器真正调用工具函数处理超时、重试、熔断。记录器把每次执行的结果结构化写入存储。反馈器把结果返回给编排层附带失败分类和建议动作。这五个模块的职责边界要清晰。我见过一些项目把校验、执行、记录混在一个函数里结果代码越来越臃肿最后没人敢改。分开之后每个模块可以独立测试、独立替换。3.2 工具注册与 Schema 定义先看工具注册。每个工具在注册时必须提供完整的元信息。下面是一个天气查询工具的注册示例TOOL_REGISTRY {} def register_tool(name, description, input_schema, output_schema, timeout_ms5000, max_retries2, retry_backoff1.5): def decorator(func): TOOL_REGISTRY[name] { name: name, description: description, input_schema: input_schema, output_schema: output_schema, timeout_ms: timeout_ms, max_retries: max_retries, retry_backoff: retry_backoff, func: func } return func return decorator register_tool( nameget_weather, description查询指定城市的当前天气, input_schema{ type: object, properties: { city: {type: string, minLength: 1}, unit: {type: string, enum: [celsius, fahrenheit], default: celsius} }, required: [city] }, output_schema{ type: object, properties: { city: {type: string}, temperature: {type: number}, condition: {type: string} }, required: [city, temperature, condition] }, timeout_ms3000, max_retries2 ) def get_weather(city, unitcelsius): # 实际调用天气 API ...这里有几个参数需要解释。timeout_ms是单次执行的超时时间超过就判定为timeout失败。max_retries是最大重试次数注意这是“额外重试次数”所以总执行次数是max_retries 1。retry_backoff是退避系数第 n 次重试的等待时间是base_delay * (retry_backoff ** n)。为什么超时时间要按工具单独设置因为不同工具的正常耗时差异很大。一个本地计算工具可能 10ms 就返回一个外部 API 可能要 2s。如果统一设 5s本地工具的超时保护就形同虚设如果统一设 1s外部 API 又会频繁超时。所以超时时间必须按工具配置甚至可以根据历史执行数据动态调整。3.3 执行器的核心逻辑重试、超时与熔断执行器是整个运行时最复杂的部分。下面是一个简化但可运行的实现import time import signal from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout class ToolExecutor: def __init__(self, registry, recorder): self.registry registry self.recorder recorder self.executor ThreadPoolExecutor(max_workers10) self.circuit_breaker {} # 熔断状态 def execute(self, tool_name, params, contextNone): tool self.registry.get(tool_name) if not tool: return self._record_failure( tool_name, params, request_build, tool_not_found, f工具 {tool_name} 未注册, context ) # 熔断检查 if self._is_circuit_open(tool_name): return self._record_failure( tool_name, params, execution, circuit_open, f工具 {tool_name} 处于熔断状态, context ) # 参数校验 valid, error validate_tool_input(tool[input_schema], params) if not valid: return self._record_failure( tool_name, params, param_validation, schema_mismatch, error[error_message], context ) # 执行与重试 last_error None for attempt in range(tool[max_retries] 1): start time.time() try: future self.executor.submit(tool[func], **params) raw_output future.result(timeouttool[timeout_ms] / 1000) duration int((time.time() - start) * 1000) # 输出校验 valid, error validate_tool_output(tool[output_schema], raw_output) if not valid: last_error (result_parse, output_schema_mismatch, error) continue self._reset_circuit(tool_name) return self._record_success( tool_name, params, raw_output, duration, attempt, context ) except FutureTimeout: duration int((time.time() - start) * 1000) last_error (execution, timeout, f执行超时 {duration}ms) except Exception as e: duration int((time.time() - start) * 1000) last_error (execution, runtime_error, str(e)) # 退避等待 if attempt tool[max_retries]: wait 0.5 * (tool[retry_backoff] ** attempt) time.sleep(wait) # 全部重试失败 self._record_circuit_failure(tool_name) return self._record_failure( tool_name, params, last_error[0], last_error[1], last_error[2], context, retry_counttool[max_retries] )这段代码里有几个关键设计。熔断机制如果某个工具连续失败超过阈值比如 5 次就暂时把它标记为不可用避免 agent 反复调用一个已经挂掉的服务。熔断状态有冷却时间冷却后自动恢复。退避重试重试不是立刻重试而是等待一段时间且等待时间逐次增加。这能避免在服务端已经过载时继续施压。输出校验即使执行成功也要校验输出是否符合 Schema不符合就当作失败处理。实操心得重试策略要区分失败类型。网络错误和限流错误适合重试参数错误和业务错误重试没有意义。我早期版本对所有失败都重试结果参数错误重试了 3 次还是失败白白浪费了时间和配额。后来改成按failure_type决定是否重试效率提升明显。3.4 记录器的存储设计记录器负责把执行记录写入存储。存储选型要看规模小规模用 SQLite 或 JSON 文件就够中大规模用 PostgreSQL 或 ClickHouse超大规模用对象存储加索引。我推荐用 PostgreSQL因为它的 JSONB 类型非常适合存储半结构化的执行记录而且支持 GIN 索引可以快速查询input_params里的任意字段。CREATE TABLE tool_executions ( execution_id UUID PRIMARY KEY, tool_name VARCHAR(128) NOT NULL, status VARCHAR(32) NOT NULL, failure_stage VARCHAR(32), failure_type VARCHAR(64), input_params JSONB, raw_output TEXT, error_message TEXT, duration_ms INTEGER, retry_count INTEGER DEFAULT 0, context_snapshot JSONB, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_tool_status ON tool_executions(tool_name, status); CREATE INDEX idx_failure_type ON tool_executions(failure_type) WHERE status ! success; CREATE INDEX idx_created_at ON tool_executions(created_at DESC);有了这张表你可以做很多分析。比如查“过去 24 小时失败率最高的工具”SELECT tool_name, COUNT(*) FILTER (WHERE status ! success) * 100.0 / COUNT(*) AS failure_rate FROM tool_executions WHERE created_at NOW() - INTERVAL 24 hours GROUP BY tool_name ORDER BY failure_rate DESC LIMIT 10;再比如查“某个工具最常见的失败类型”SELECT failure_type, COUNT(*) AS cnt FROM tool_executions WHERE tool_name get_weather AND status ! success GROUP BY failure_type ORDER BY cnt DESC;这些查询看起来简单但它们是“失败是数据”理念的直接体现。没有结构化存储这些分析根本做不了。3.5 反馈器把失败变成 agent 的决策输入记录器解决的是“事后分析”反馈器解决的是“事中决策”。当工具执行失败时反馈器不只是返回一个错误而是返回一份包含建议动作的结构化反馈。def build_feedback(execution_record): failure_type execution_record.get(failure_type) feedback_map { schema_mismatch: { action: retry_with_fixed_params, hint: 参数不符合 Schema请检查字段类型和必填项 }, timeout: { action: retry, hint: 执行超时建议重试或换用备用工具 }, auth_error: { action: refresh_credential, hint: 认证失败请刷新凭证后重试 }, rate_limit: { action: wait_and_retry, hint: 触发限流建议等待后重试 }, business_error: { action: try_alternative, hint: 业务逻辑错误建议换用其他工具或调整策略 }, circuit_open: { action: skip, hint: 工具已熔断建议跳过或使用备用方案 } } return feedback_map.get(failure_type, { action: abort, hint: 未知错误建议终止当前任务 })这份反馈会随执行记录一起返回给编排层。编排层根据action字段决定下一步是重试、换工具、还是终止。这样就把“失败处理”从硬编码的 if-else变成了数据驱动的决策。4. 常见问题与排查技巧实录4.1 失败数据记录本身的坑坑一记录太多导致性能下降。我一开始把每次执行的完整输入输出都存下来结果一个高频工具每天产生几十万条记录数据库写入成了瓶颈。后来改成成功记录只存摘要工具名、耗时、参数哈希失败记录存完整信息。因为成功记录通常不需要逐条分析而失败记录才是重点。坑二敏感信息泄露。工具参数里可能包含 API Key、用户密码、个人身份信息。如果直接存进数据库就是安全隐患。我的做法是在记录前做一层脱敏对password、token、secret等字段名的值替换成***对疑似身份证号、手机号的字符串做掩码处理。坑三错误堆栈太长。Python 的异常堆栈动辄几十行全部存下来浪费空间。我通常只保留最后 5 层堆栈加上异常类型和消息足够定位问题。坑四时间戳不一致。分布式系统里不同机器的时钟可能有偏差。如果记录器用本地时间分析时会出现“失败发生在成功之后”的诡异现象。统一用 UTC 时间并且在记录里同时存created_at和duration_ms这样即使时钟有偏差也能通过耗时推算真实顺序。4.2 排查技巧速查表现象可能原因排查方法解决方向某工具失败率突然飙升上游服务变更或宕机查该工具近 1 小时失败类型分布熔断该工具切换备用参数校验失败集中出现agent 规划层输出格式漂移抽样查看input_params优化规划层 prompt 或加 few-shot超时失败集中在特定时段服务端负载高峰按小时聚合超时次数调整超时阈值或错峰调用重试后仍然失败失败类型不可重试检查failure_type是否属于可重试类修正重试策略输出 Schema 校验失败上游返回格式变更对比raw_output和output_schema更新 Schema 或加兼容层熔断频繁触发阈值设置过严查看熔断触发时的连续失败次数调整阈值或冷却时间这张表是我从多次线上排查里总结出来的基本覆盖了 80% 的常见问题。遇到新问题时先查表如果表里没有再深入分析。4.3 从失败数据到 agent 自我改进“失败是数据”的终极价值是让 agent 能够从失败中学习。具体怎么做我试过两条路径。第一条路径是失败案例回灌到 prompt。把高频失败案例整理成 few-shot 示例加到规划层的 prompt 里让 agent 知道“这种参数会失败应该这样传”。实测下来参数校验失败率能降低 60% 以上。第二条路径是失败数据用于微调。把执行记录里的input_params和failure_type配对构造训练数据微调一个小的分类模型用来预测“给定参数这次调用会不会失败”。如果预测会失败就提前拦截避免无效调用。这条路径成本较高适合调用量大的场景。注意失败数据用于改进时要注意数据偏差。如果某个工具因为长期熔断而很少被调用它的失败数据就很少模型可能会低估它的风险。所以熔断期间也要定期做探针调用保持数据新鲜度。4.4 工具运行时的监控指标最后分享一套我常用的监控指标这些指标都从执行记录里聚合而来工具成功率按工具、按小时聚合低于 95% 告警。P95 执行耗时按工具聚合超过超时阈值 80% 告警。失败类型分布按天聚合某类型占比突然上升告警。重试率重试次数大于 0 的调用占比高于 20% 告警。熔断触发次数按天聚合任何熔断触发都值得关注。参数校验失败率按工具聚合高于 10% 说明规划层需要优化。这些指标不需要复杂的监控系统用 SQL 定时查询加简单告警就能实现。关键是坚持记录坚持分析。我见过太多项目工具运行时只记录“成功/失败”出了问题全靠猜。有了这套数据排查从“猜”变成了“查”效率完全不是一个量级。我个人在实际操作中的体会是工具运行时的失败数据就像飞机的黑匣子。平时你可能不会天天去看它但一旦出事它就是唯一能还原真相的东西。而且看得多了你甚至能提前发现隐患——比如某个工具的 P95 耗时在缓慢上升虽然还没超时但趋势已经不对了。这种提前量就是“失败是数据”带来的最大价值。