Atomic Agents 错误处理指南：验证错误、API 故障与自定义重试策略完整清单

Atomic Agents 是一款轻量级、模块化的 Python AI 智能体框架。这篇文章是一份面向新手的错误处理完整清单，覆盖 验证错误（Pydantic Schema 校验）、API 故障（限流、超时、5xx）与 自定义重试策略（指数退避、降级链、Hook 监控），帮助你在生产环境中稳定运行 AI Agent。

一、为什么 Agent 必须做错误处理？

一次 agent.run() 调用背后藏着多个可能失败的环节：

层级	错误类型	典型场景
1️⃣ Schema 验证	验证错误（ValidationError）	用户输入为空、LLM 输出不符合结构
2️⃣ API 调用	API 故障（APIError / RateLimitError）	限流、网络超时、服务端 5xx
3️⃣ Hook 系统	监控与告警	记录错误指标、触发重试
4️⃣ 自定义异常	业务级容错	降级、熔断、兜底回复

Atomic Agents 基于 Pydantic 做运行时校验，输入数据在到达大模型之前就会被检查；输出则必须符合你定义的 BaseIOSchema，不符合会触发 Instructor 的解析重试。理解这条数据流，你就理解了错误从哪来、往哪挡：

二、验证错误：在数据到达 LLM 前拦截

1. 用 Pydantic 规则约束输入

所有 Schema 都继承自 base_io_schema.py 中的 BaseIOSchema，你可以直接用 Pydantic 的 Field 和 field_validator 写校验规则：

class ValidatedInputSchema(BaseIOSchema):
    """带校验规则的输入 Schema"""
    query: str = Field(..., min_length=1, max_length=1000)
    max_results: int = Field(default=10, ge=1, le=100)

    @field_validator('query')
    @classmethod
    def query_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("查询不能为空或仅含空白")
        return v.strip()

调用时传入空字符串，agent.run(ValidatedInputSchema(query="")) 会立刻抛出 ValueError，根本不会浪费一次 API 调用。多字段之间的关联校验（比如日期范围）可以用 model_validator(mode='after') 完成。

2. 用 parse:error Hook 捕获校验失败

当 LLM 的输出不符合 Schema 时，会触发 parse:error 事件。AtomicAgent 提供了完整的 Hook 注册接口（见 atomic_agent.py）：

def on_parse_error(error):
    print(f"校验失败: {error}")

agent.register_hook("parse:error", on_parse_error)

在 Handler 里可以遍历 error.errors() 拿到具体出错字段（err['loc']）和原因（err['msg']），方便做精细化的重试或日志记录。

三、API 故障：重试与指数退避

1. 分清「值得重试」和「不值得重试」的错误

OpenAI SDK 抛出的异常大致分三类：

• 🐢 RateLimitError（限流）→ 值得重试，等待时间按指数增长
• 🌐 APIConnectionError（网络断开）→ 值得重试
• ⚠️ APIError 且状态码 ≥ 500 → 重试；4xx 客户端错误 → 直接抛出不重试

经典的指数退避写法：等待时间 = base_delay × 2^attempt，例如 1s、2s、4s，最多重试 3 次后放弃。

2. 优雅降级：备用 Agent 链

主模型挂了怎么办？定义一条 FallbackAgentChain：按顺序尝试「主力 Agent（如 gpt-4o）→ 备用 Agent（更便宜的模型）」，任何一个成功就返回结果。这样即使主力模型持续故障，用户也能拿到响应，只是质量略降——这叫 Graceful Degradation（优雅降级）。

3. 别忘了设置超时

在创建客户端时传入 openai.OpenAI(timeout=30.0)，避免请求无限挂起；同时在 model_api_parameters 里限制 max_tokens，控制响应长度。

四、用 Hook 系统构建自定义重试策略

Hook 是 Atomic Agents 处理错误的「中枢神经」。它和 Instructor 的事件系统深度集成，支持四个关键事件：

事件	触发时机	用途
parse:error	Pydantic 校验失败	记录字段级错误、智能重试
completion:kwargs	发出 API 请求前	统计请求量、记录起点时间
completion:response	收到 LLM 响应后	计算耗时、重置重试计数
completion:error	API 或网络错误	判断错误类型、决定是否重试

自定义重试的核心思路是：给 completion:error 注册一个 Handler，在 Handler 里识别错误字符串（如 "rate limit"、"timeout"、"503"），符合条件才执行 time.sleep(base_delay * 2**attempt) 后重试，成功时（completion:response）把计数器归零。

几个实用要点：

• 🪶 保持 Hook 轻量：Hook 是同步执行的，别在里面做写数据库这类重操作
• 🧯 隔离 Hook 异常：框架的 _dispatch_hook 已对 Handler 异常做了隔离，Handler 崩溃不会中断主流程
• ⚡ 零开销：未注册 Hook 时没有任何性能损耗；需要临时关闭可用 agent.disable_hooks()

完整的可运行演示请看 hooks-example/main.py，它展示了请求量、成功率、平均响应时间等指标的实时统计面板。

五、生产环境最佳实践清单 ✅

策略	解决的问题	关键做法
Schema 验证	非法输入/输出	Field 约束 + field_validator
指数退避重试	限流、瞬时网络故障	2 倍递增等待，4xx 不重试
超时设置	请求挂起	客户端 timeout=30.0
Fallback 链	主力模型不可用	多 Agent 依次降级
熔断器	防止级联故障	连续失败 N 次后短路
全量日志	事后排查	Hook + logging 记录 JSON 日志

📌 核心原则：多种策略组合使用。 验证挡在最前面，重试兜住瞬时故障，降级保证可用性，Hook 让一切可观测。

六、相关资料索引

• 📖 官方错误处理指南：docs/guides/error-handling.md
• 📖 Hook 系统详解：docs/guides/hooks.md
• 🧪 Hook 演示项目说明：atomic-examples/hooks-example/README.md
• 🏗️ Agent 核心实现（含 Hook 管理方法）：atomic_agent.py
• 💬 对话历史管理：chat_history.py

掌握以上清单，你的 Atomic Agents 应用就能从容应对验证错误、API 故障与各种边界情况，稳定地跑在生产环境里。