AI AgentAgent 框架MCP 服务后端【免费下载链接】atomic-agentsBuilding AI agents, atomically项目地址https://gitcode.com/gh_mirrors/at/atomic-agents点击查看免费下载导读本文基于 Atomic Agents 官方 错误处理指南系统讲解如何在基于 Pydantic Instructor 构建的 AI Agent 应用中建立分层、可观测、可恢复的错误处理体系。你将掌握四类核心手段——Schema 运行时校验、LLM Provider API 失败重试、Hook 事件监控、以及兜底降级与熔断模式并学会如何结合仓库源码中的AtomicAgent、BaseIOSchema与 Hook 事件实现构建可用于生产环境的健壮 Agent 应用。概览Atomic Agents 的四层错误处理从框架设计上Atomic Agents 的错误处理被划分为四个相互独立、可组合叠加的层次各层实现均可从源码逐一印证Schema 校验Schema ValidationBaseIOSchema继承自 PydanticBaseModel在输入到达 LLM 之前、以及 LLM 输出被解析回结构化对象时进行运行时校验拦截非法数据API 错误处理API Error Handling针对 LLM Provider 的限流RateLimitError、连接异常APIConnectionError、服务端错误APIError等瞬时故障设计重试策略Hook 系统Hook System通过 Instructor 事件系统监控与响应错误支持日志、指标采集与智能重试自定义异常处理Custom Exception Handling以装饰器、统一错误处理器、Fallback 链与熔断器等模式构建完整的故障恢复机制。这四层并非互斥官方文档明确建议生产应用应始终组合多种策略Always combine multiple strategies for robust production applications。从源码看核心载体是 atomic_agent.py 中的AtomicAgent[InputSchema, OutputSchema]泛型类它在初始化时持有 Instructor 客户端、模型名、历史记录与SystemPromptGenerator并在类文档中完整声明了所支持的 Hook 事件parse:error、completion:kwargs、completion:response、completion:error、completion:last_attempt。base_io_schema.py 则定义了所有输入/输出 Schema 的公共基类BaseIOSchema。Schema 校验错误把非法数据挡在 LLM 之外Pydantic Schema 是错误处理的第一道防线它在数据进入 LLM 请求、以及 LLM 返回被反序列化时即完成校验避免把格式错误的数据浪费在昂贵的模型调用上。BaseIOSchema还强制要求每个 Schema 具备非空 docstring 作为其描述详见 base_io_schema.py 的_validate_description这保证了模型能准确理解输入/输出结构的语义。基础校验通过Field约束 field_validator可以定义字段级校验规则。以下示例定义了一个带查询长度、结果数量约束的输入 Schema以及带置信度范围与来源列表约束的输出 Schemaimport os from typing import List from pydantic import Field, field_validator import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema from atomic_agents.context import ChatHistory class ValidatedInputSchema(BaseIOSchema): Input schema with validation rules. query: str Field(..., descriptionUser query, min_length1, max_length1000) max_results: int Field(default10, ge1, le100, descriptionMaximum results to return) field_validator(query) classmethod def query_not_empty(cls, v: str) - str: if not v.strip(): raise ValueError(Query cannot be empty or whitespace only) return v.strip() class ValidatedOutputSchema(BaseIOSchema): Output schema with validation. answer: str Field(..., descriptionThe response) confidence: float Field(..., ge0.0, le1.0, descriptionConfidence score 0-1) sources: List[str] Field(default_factorylist, descriptionSource references) # Initialize client and agent client instructor.from_openai(openai.OpenAI()) agent AtomicAgentValidatedInputSchema, ValidatedOutputSchema ) ) # Handle validation errors try: response agent.run(ValidatedInputSchema(query, max_results5)) except ValueError as e: print(fValidation error: {e})值得注意的字段约束语义min_length/max_length作用于字符串长度ge/le作用于数值上下界而field_validator可以执行更复杂的逻辑校验如去除空白后重新判定。当agent.run(...)收到非法输入时会在进入模型调用前抛出ValueError/ValidationError因此 try/except 应包住agent.run调用本身。自定义校验器对于跨字段依赖如日期范围或枚举白名单类校验Pydantic 提供model_validator与field_validator的组合能力from pydantic import Field, field_validator, model_validator from typing import Optional from atomic_agents import BaseIOSchema class SearchInputSchema(BaseIOSchema): Search input with complex validation. query: str Field(..., descriptionSearch query) category: Optional[str] Field(None, descriptionCategory filter) date_from: Optional[str] Field(None, descriptionStart date YYYY-MM-DD) date_to: Optional[str] Field(None, descriptionEnd date YYYY-MM-DD) field_validator(category) classmethod def validate_category(cls, v: Optional[str]) - Optional[str]: valid_categories [technology, science, business, health] if v is not None and v.lower() not in valid_categories: raise ValueError(fCategory must be one of: {valid_categories}) return v.lower() if v else None model_validator(modeafter) def validate_dates(self): if self.date_from and self.date_to: if self.date_from self.date_to: raise ValueError(date_from must be before date_to) return selffield_validator聚焦单个字段此处对category做白名单校验并统一小写model_validator(modeafter)则在整个模型组装完成后校验字段间关系日期区间前后一致性。这套机制同样作用于 LLM 返回结果的解析当模型输出不符合输出 Schema 时Pydantic 校验失败会触发parse:error事件后文详述。API 错误处理为瞬时故障设计指数退避重试LLM Provider 的失败通常分两类瞬时故障限流、网络抖动、5xx 服务端错误适合重试永久性错误4xx 客户端错误、鉴权失败重试无意义应直接抛出。OpenAI 客户端提供RateLimitError、APIConnectionError、APIError等异常类型可据此差异化处理。基础重试模式import os import time from typing import Optional import instructor import openai from openai import APIError, RateLimitError, APIConnectionError from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory def create_agent_with_retry( max_retries: int 3, retry_delay: float 1.0 ) - AtomicAgent: Create an agent with retry configuration. client instructor.from_openai(openai.OpenAI()) return AtomicAgentBasicChatInputSchema, BasicChatOutputSchema, model_api_parameters{ max_tokens: 1000, temperature: 0.7 } ) ) def run_with_retry( agent: AtomicAgent, input_data: BasicChatInputSchema, max_retries: int 3, retry_delay: float 1.0 ) - Optional[BasicChatOutputSchema]: Run agent with automatic retry on transient failures. last_error None for attempt in range(max_retries): try: return agent.run(input_data) except RateLimitError as e: last_error e wait_time retry_delay * (2 ** attempt) # Exponential backoff print(fRate limited. Waiting {wait_time}s before retry {attempt 1}/{max_retries}) time.sleep(wait_time) except APIConnectionError as e: last_error e print(fConnection error. Retry {attempt 1}/{max_retries}) time.sleep(retry_delay) except APIError as e: last_error e if e.status_code and e.status_code 500: print(fServer error. Retry {attempt 1}/{max_retries}) time.sleep(retry_delay) else: raise # Dont retry client errors (4xx) print(fAll retries failed. Last error: {last_error}) return None # Usage agent create_agent_with_retry() user_input BasicChatInputSchema(chat_messageExplain quantum computing) response run_with_retry(agent, user_input) if response: print(fResponse: {response.chat_message}) else: print(Failed to get response after retries)关键设计决策限流使用指数退避retry_delay * (2 ** attempt)因为限流本身就是请求过快的信号连接错误使用固定延迟避免在恢复阶段叠加拥塞仅对 5xx 服务端错误重试4xx 客户端错误直接raise交由上层处理。model_api_parameters传入的max_tokens、temperature会被AtomicAgent透传给 Instructor 的 completion 调用见 atomic_agent.py 的_get_completion_kwargs。利用 Hook 系统进行错误监控Hook 系统是 Atomic Agents 错误处理的核心观测层。框架整合了 Instructor 的事件系统AtomicAgent提供了register_hook、unregister_hook、clear_hooks、enable_hooks、disable_hooks与hooks_enabled属性等完整管理 API见 atomic_agent.py。支持的 Hook 事件包括事件说明触发时机parse:errorPydantic 校验失败LLM 输出不符合输出 Schema 时completion:kwargs请求参数就绪请求发送给 LLM 之前completion:response完成响应LLM 返回响应后completion:errorAPI / 网络错误连接失败、超时等completion:last_attempt最后一次重试重试耗尽前的最终尝试源码类文档声明此外get_context_token_count()在统计完成后还会分发token:counted事件用于用量监控见 atomic_agent.py。更完整的 Hook 使用说明可参考 docs/guides/hooks.md。错误日志 Hookimport os import logging from datetime import datetime from typing import Any, Optional import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory # Configure logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) def on_error_hook(error: Exception, context: dict) - None: Hook called when an error occurs during agent execution. logger.error(fAgent error: {type(error).__name__}: {error}) logger.error(fContext: {context}) def on_completion_hook(response: Any, duration_ms: float) - None: Hook called on successful completion. logger.info(fAgent completed in {duration_ms:.2f}ms) # Create agent with hooks using Instructors hook system client instructor.from_openai(openai.OpenAI()) # Register hooks with the instructor client client.on(completion, lambda *args: on_completion_hook(*args)) agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) )这里演示了两种 Hook 接入方式的其中一种直接调用 Instructor 客户端的client.on(...)。AtomicAgent.register_hook(event, handler)的底层实现本质上也是委托给客户端——只要客户端实现了on/off/clear方法注册与注销就会同步到 Instructor见 atomic_agent.py测试用例 test_atomic_agent.py 验证了该委托行为。统一错误处理器对于需要集中管理错误分类与处理的场景官方文档提供了一个装饰器 回调注入的AgentErrorHandlerimport os from typing import Callable, Optional, TypeVar from functools import wraps import instructor import openai from pydantic import ValidationError from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema T TypeVar(T, boundBaseIOSchema) class AgentErrorHandler: Centralized error handler for Atomic Agents. def __init__( self, on_validation_error: Optional[Callable[[ValidationError], None]] None, on_api_error: Optional[Callable[[Exception], None]] None, on_unknown_error: Optional[Callable[[Exception], None]] None ): self.on_validation_error on_validation_error or self._default_validation_handler self.on_api_error on_api_error or self._default_api_handler self.on_unknown_error on_unknown_error or self._default_unknown_handler def _default_validation_handler(self, error: ValidationError) - None: print(fValidation failed: {error.error_count()} errors) for err in error.errors(): print(f - {err[loc]}: {err[msg]}) def _default_api_handler(self, error: Exception) - None: print(fAPI error: {type(error).__name__}: {error}) def _default_unknown_handler(self, error: Exception) - None: print(fUnknown error: {type(error).__name__}: {error}) def wrap(self, func: Callable) - Callable: Decorator to wrap agent calls with error handling. wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except ValidationError as e: self.on_validation_error(e) return None except (openai.APIError, openai.APIConnectionError) as e: self.on_api_error(e) return None except Exception as e: self.on_unknown_error(e) return None return wrapper # Usage error_handler AgentErrorHandler() error_handler.wrap def ask_agent(agent: AtomicAgent, question: str): from atomic_agents import BasicChatInputSchema return agent.run(BasicChatInputSchema(chat_messagequestion)) # Create and use agent client instructor.from_openai(openai.OpenAI()) agent AtomicAgent( configAgentConfig( clientclient, modelgpt-5-mini ) ) response ask_agent(agent, What is machine learning?)设计要点三类回调可注入校验错误ValidationError、API 错误openai.APIError及其子类、未知异常分别对应不同处理入口默认可打印error.error_count()与error.errors()明细返回值统一为None表示失败被包装函数的调用方只需判空即可无需层层 try/except这种集中式分类处理与 Hook 的事件订阅是互补方案前者适合同步业务流后者适合横切监控。优雅降级Fallback Agent 链当主 Agent 持续失败时一个成熟的生产方案是准备多个不同模型/提供商的 Agent按优先级依次尝试直到成功。官方文档给出的FallbackAgentChain是该模式的最小实现import os from typing import Optional, List import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory class FallbackAgentChain: Chain of agents with automatic fallback on failure. def __init__(self, agents: List[AtomicAgent]): self.agents agents def run(self, input_data: BasicChatInputSchema) - Optional[BasicChatOutputSchema]: Try each agent in order until one succeeds. last_error None for i, agent in enumerate(self.agents): try: print(fTrying agent {i 1}/{len(self.agents)}) return agent.run(input_data) except Exception as e: last_error e print(fAgent {i 1} failed: {e}) continue print(fAll agents failed. Last error: {last_error}) return None # Create primary and fallback agents with different models/providers def create_fallback_chain() - FallbackAgentChain: # Primary: GPT-4 primary_client instructor.from_openai(openai.OpenAI()) primary_agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) ) # Fallback: GPT-4o-mini (cheaper, faster) fallback_client instructor.from_openai(openai.OpenAI()) fallback_agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) ) return FallbackAgentChain([primary_agent, fallback_agent]) # Usage chain create_fallback_chain() response chain.run(BasicChatInputSchema(chat_messageExplain quantum computing)) if response: print(response.chat_message)该模式的实践要点主备模型差异化主 Agent 用高质量模型如gpt-4o备选 Agent 用更快更便宜的小模型如gpt-5-mini以成本换取可用性异常粒度可细化生产环境可按前文的重试策略瞬时故障 vs 永久错误决定是立即降级还是先重试避免因一次抖动就切换模型由于每个 Agent 都是独立实例且共享同一输入 SchemaBasicChatInputSchema链式切换对调用方完全透明。最佳实践四条官方文档总结了四条生产级最佳实践其中三条与防御 观测直接相关。1. 始终校验输入在输入 Schema 中做防御性校验甚至可以过滤提示注入特征from pydantic import Field, field_validator from atomic_agents import BaseIOSchema class SafeInputSchema(BaseIOSchema): Input schema with comprehensive validation. message: str Field(..., min_length1, max_length10000) field_validator(message) classmethod def sanitize_message(cls, v: str) - str: # Remove potential prompt injection attempts dangerous_patterns [ignore previous, disregard instructions] for pattern in dangerous_patterns: if pattern.lower() in v.lower(): raise ValueError(Invalid input detected) return v.strip()2. 记录所有错误用装饰器统一记录 Agent 操作的异常栈logger.exception会自动附带当前异常的 tracebackimport logging from functools import wraps logger logging.getLogger(__name__) def log_errors(func): Decorator to log all errors from agent operations. wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.exception(fError in {func.__name__}: {e}) raise return wrapper需要说明的是AtomicAgent内部对 Hook 处理器本身也做了错误隔离当某个 Hook handler 抛异常时_dispatch_hook会记录 warning 并继续执行后续 handler不中断 Agent 主流程见 atomic_agent.py对应测试 test_atomic_agent.py。因此你的日志/监控 Hook 自身即使出问题也不会拖垮业务调用。3. 设置超时在客户端层面配置超时避免无界等待import os import instructor import openai from atomic_agents import AtomicAgent, AgentConfig from atomic_agents.context import ChatHistory # Configure timeout at client level client instructor.from_openai( openai.OpenAI(timeout30.0) # 30 second timeout ) agent AtomicAgent( configAgentConfig( clientclient, modelgpt-5-mini, historyChatHistory(), model_api_parameters{ max_tokens: 500 # Limit response length } ) )超时timeout30.0与max_tokens限制输出长度双管齐下是防止慢请求与超长输出拖垮系统的基础手段。结合前文重试模式超时抛出的异常可按APIConnectionError路径处理。4. 实现熔断器熔断器用于防止级联故障当失败次数达到阈值后开路拒绝调用经过冷却时间后转入半开状态试探性放行一次调用import time from typing import Optional, Callable from dataclasses import dataclass dataclass class CircuitBreaker: Simple circuit breaker for agent calls. failure_threshold: int 5 reset_timeout: float 60.0 _failure_count: int 0 _last_failure_time: float 0 _state: str closed # closed, open, half-open def call(self, func: Callable, *args, **kwargs): Execute function with circuit breaker protection. if self._state open: if time.time() - self._last_failure_time self.reset_timeout: self._state half-open else: raise Exception(Circuit breaker is open) try: result func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): self._failure_count 0 self._state closed def _on_failure(self): self._failure_count 1 self._last_failure_time time.time() if self._failure_count self.failure_threshold: self._state open # Usage circuit_breaker CircuitBreaker(failure_threshold3, reset_timeout30.0) def safe_agent_call(agent, input_data): return circuit_breaker.call(agent.run, input_data)三态状态机语义closed正常放行、open直接拒绝并快速失败、half-open冷却期满后放行一次探测。当熔断器处于 open 状态时调用立即抛异常避免对已失效的 Provider 持续发起昂贵的请求。与框架上下文管理的联动值得补充的一点错误处理并非只发生在调用失败时刻。AtomicAgent在每次run/run_stream/run_async调用前会执行_trim_context()对历史进行裁剪见 atomic_agent.py当配置了max_context_tokens且上下文超限时框架会按 turn 从最旧开始删除若单个 turn 本身就超出限制则会抛出带明确修复建议的ValueError——这是非模型错误的一类典型处理场景。其 token 统计基于 token_counter.py 的count_context通过 LiteLLM 实现与模型无关的计数并可返回utilization上下文占用比例。总结策略选型速查表策略适用场景实现方式Schema 校验阻止非法输入进入模型Pydantic validatorsField约束、field_validator、model_validator重试逻辑瞬时故障限流、连接抖动、5xx指数退避 固定延迟4xx 直接抛出Hook 系统监控、日志、指标采集Instructor 事件parse:error、completion:*register_hookFallback 链高可用与降级多 Agent不同模型/提供商按序尝试熔断器防止级联故障三态状态机closed / open / half-open落地建议将 Schema 校验作为第一道闸门前置拦截非法输入与注入特征以 Hook 系统承载横切观测日志、指标、审计对外部 API 调用统一走重试 → Fallback → 熔断的递进恢复路径并将统一错误处理器作为最终兜底出口。这样每一层各司其职才能构建出既健壮又可观测的 Atomic Agents 生产应用。赞分享AI AgentAgent 框架MCP 服务后端【免费下载链接】atomic-agentsBuilding AI agents, atomically项目地址https://gitcode.com/gh_mirrors/at/atomic-agents点击查看免费下载相关推荐从Hystrix到Resilience4jJava熔断降级方案的完整技术演进与实践指南从Hystrix到Resilience4jJava熔断降级方案的完整技术演进与实践指南 在当今微服务架构盛行的时代 Java熔断降级方案 已成为保障分布式系文档教程后端X-Plane Connect完全指南如何用代码实时控制飞行模拟器X Plane Connect完全指南如何用代码实时控制飞行模拟器 X Plane ConnectXPC是一款开源研究工具能让开发者通过C、C、J开发工具一个页面能拦掉多少广告请求uBlock Origin 免费开源广告拦截扩展快速上手一个页面能拦掉多少广告请求uBlock Origin 免费开源广告拦截扩展快速上手 默认配置自带 5 份社区过滤列表已知的广告与跟踪请求在离开本机前就被取消网络安全应用安全上一篇Sliver框架终极扩展开发指南5步创建自定义渗透测试模块下一篇Peritext性能优化指南高效处理大规模文本协作的10个技巧创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/10/10 5:29:01 · 阅读 1.2k Atomic Agents 错误处理指南：从 Schema 校验到熔断降级的完整实战方案
AI AgentAgent 框架MCP 服务后端【免费下载链接】atomic-agentsBuilding AI agents, atomically项目地址https://gitcode.com/gh_mirrors/at/atomic-agents点击查看免费下载导读本文基于 Atomic Agents 官方 错误处理指南系统讲解如何在基于 Pydantic Instructor 构建的 AI Agent 应用中建立分层、可观测、可恢复的错误处理体系。你将掌握四类核心手段——Schema 运行时校验、LLM Provider API 失败重试、Hook 事件监控、以及兜底降级与熔断模式并学会如何结合仓库源码中的AtomicAgent、BaseIOSchema与 Hook 事件实现构建可用于生产环境的健壮 Agent 应用。概览Atomic Agents 的四层错误处理从框架设计上Atomic Agents 的错误处理被划分为四个相互独立、可组合叠加的层次各层实现均可从源码逐一印证Schema 校验Schema ValidationBaseIOSchema继承自 PydanticBaseModel在输入到达 LLM 之前、以及 LLM 输出被解析回结构化对象时进行运行时校验拦截非法数据API 错误处理API Error Handling针对 LLM Provider 的限流RateLimitError、连接异常APIConnectionError、服务端错误APIError等瞬时故障设计重试策略Hook 系统Hook System通过 Instructor 事件系统监控与响应错误支持日志、指标采集与智能重试自定义异常处理Custom Exception Handling以装饰器、统一错误处理器、Fallback 链与熔断器等模式构建完整的故障恢复机制。这四层并非互斥官方文档明确建议生产应用应始终组合多种策略Always combine multiple strategies for robust production applications。从源码看核心载体是 atomic_agent.py 中的AtomicAgent[InputSchema, OutputSchema]泛型类它在初始化时持有 Instructor 客户端、模型名、历史记录与SystemPromptGenerator并在类文档中完整声明了所支持的 Hook 事件parse:error、completion:kwargs、completion:response、completion:error、completion:last_attempt。base_io_schema.py 则定义了所有输入/输出 Schema 的公共基类BaseIOSchema。Schema 校验错误把非法数据挡在 LLM 之外Pydantic Schema 是错误处理的第一道防线它在数据进入 LLM 请求、以及 LLM 返回被反序列化时即完成校验避免把格式错误的数据浪费在昂贵的模型调用上。BaseIOSchema还强制要求每个 Schema 具备非空 docstring 作为其描述详见 base_io_schema.py 的_validate_description这保证了模型能准确理解输入/输出结构的语义。基础校验通过Field约束 field_validator可以定义字段级校验规则。以下示例定义了一个带查询长度、结果数量约束的输入 Schema以及带置信度范围与来源列表约束的输出 Schemaimport os from typing import List from pydantic import Field, field_validator import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema from atomic_agents.context import ChatHistory class ValidatedInputSchema(BaseIOSchema): Input schema with validation rules. query: str Field(..., descriptionUser query, min_length1, max_length1000) max_results: int Field(default10, ge1, le100, descriptionMaximum results to return) field_validator(query) classmethod def query_not_empty(cls, v: str) - str: if not v.strip(): raise ValueError(Query cannot be empty or whitespace only) return v.strip() class ValidatedOutputSchema(BaseIOSchema): Output schema with validation. answer: str Field(..., descriptionThe response) confidence: float Field(..., ge0.0, le1.0, descriptionConfidence score 0-1) sources: List[str] Field(default_factorylist, descriptionSource references) # Initialize client and agent client instructor.from_openai(openai.OpenAI()) agent AtomicAgentValidatedInputSchema, ValidatedOutputSchema ) ) # Handle validation errors try: response agent.run(ValidatedInputSchema(query, max_results5)) except ValueError as e: print(fValidation error: {e})值得注意的字段约束语义min_length/max_length作用于字符串长度ge/le作用于数值上下界而field_validator可以执行更复杂的逻辑校验如去除空白后重新判定。当agent.run(...)收到非法输入时会在进入模型调用前抛出ValueError/ValidationError因此 try/except 应包住agent.run调用本身。自定义校验器对于跨字段依赖如日期范围或枚举白名单类校验Pydantic 提供model_validator与field_validator的组合能力from pydantic import Field, field_validator, model_validator from typing import Optional from atomic_agents import BaseIOSchema class SearchInputSchema(BaseIOSchema): Search input with complex validation. query: str Field(..., descriptionSearch query) category: Optional[str] Field(None, descriptionCategory filter) date_from: Optional[str] Field(None, descriptionStart date YYYY-MM-DD) date_to: Optional[str] Field(None, descriptionEnd date YYYY-MM-DD) field_validator(category) classmethod def validate_category(cls, v: Optional[str]) - Optional[str]: valid_categories [technology, science, business, health] if v is not None and v.lower() not in valid_categories: raise ValueError(fCategory must be one of: {valid_categories}) return v.lower() if v else None model_validator(modeafter) def validate_dates(self): if self.date_from and self.date_to: if self.date_from self.date_to: raise ValueError(date_from must be before date_to) return selffield_validator聚焦单个字段此处对category做白名单校验并统一小写model_validator(modeafter)则在整个模型组装完成后校验字段间关系日期区间前后一致性。这套机制同样作用于 LLM 返回结果的解析当模型输出不符合输出 Schema 时Pydantic 校验失败会触发parse:error事件后文详述。API 错误处理为瞬时故障设计指数退避重试LLM Provider 的失败通常分两类瞬时故障限流、网络抖动、5xx 服务端错误适合重试永久性错误4xx 客户端错误、鉴权失败重试无意义应直接抛出。OpenAI 客户端提供RateLimitError、APIConnectionError、APIError等异常类型可据此差异化处理。基础重试模式import os import time from typing import Optional import instructor import openai from openai import APIError, RateLimitError, APIConnectionError from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory def create_agent_with_retry( max_retries: int 3, retry_delay: float 1.0 ) - AtomicAgent: Create an agent with retry configuration. client instructor.from_openai(openai.OpenAI()) return AtomicAgentBasicChatInputSchema, BasicChatOutputSchema, model_api_parameters{ max_tokens: 1000, temperature: 0.7 } ) ) def run_with_retry( agent: AtomicAgent, input_data: BasicChatInputSchema, max_retries: int 3, retry_delay: float 1.0 ) - Optional[BasicChatOutputSchema]: Run agent with automatic retry on transient failures. last_error None for attempt in range(max_retries): try: return agent.run(input_data) except RateLimitError as e: last_error e wait_time retry_delay * (2 ** attempt) # Exponential backoff print(fRate limited. Waiting {wait_time}s before retry {attempt 1}/{max_retries}) time.sleep(wait_time) except APIConnectionError as e: last_error e print(fConnection error. Retry {attempt 1}/{max_retries}) time.sleep(retry_delay) except APIError as e: last_error e if e.status_code and e.status_code 500: print(fServer error. Retry {attempt 1}/{max_retries}) time.sleep(retry_delay) else: raise # Dont retry client errors (4xx) print(fAll retries failed. Last error: {last_error}) return None # Usage agent create_agent_with_retry() user_input BasicChatInputSchema(chat_messageExplain quantum computing) response run_with_retry(agent, user_input) if response: print(fResponse: {response.chat_message}) else: print(Failed to get response after retries)关键设计决策限流使用指数退避retry_delay * (2 ** attempt)因为限流本身就是请求过快的信号连接错误使用固定延迟避免在恢复阶段叠加拥塞仅对 5xx 服务端错误重试4xx 客户端错误直接raise交由上层处理。model_api_parameters传入的max_tokens、temperature会被AtomicAgent透传给 Instructor 的 completion 调用见 atomic_agent.py 的_get_completion_kwargs。利用 Hook 系统进行错误监控Hook 系统是 Atomic Agents 错误处理的核心观测层。框架整合了 Instructor 的事件系统AtomicAgent提供了register_hook、unregister_hook、clear_hooks、enable_hooks、disable_hooks与hooks_enabled属性等完整管理 API见 atomic_agent.py。支持的 Hook 事件包括事件说明触发时机parse:errorPydantic 校验失败LLM 输出不符合输出 Schema 时completion:kwargs请求参数就绪请求发送给 LLM 之前completion:response完成响应LLM 返回响应后completion:errorAPI / 网络错误连接失败、超时等completion:last_attempt最后一次重试重试耗尽前的最终尝试源码类文档声明此外get_context_token_count()在统计完成后还会分发token:counted事件用于用量监控见 atomic_agent.py。更完整的 Hook 使用说明可参考 docs/guides/hooks.md。错误日志 Hookimport os import logging from datetime import datetime from typing import Any, Optional import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory # Configure logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) def on_error_hook(error: Exception, context: dict) - None: Hook called when an error occurs during agent execution. logger.error(fAgent error: {type(error).__name__}: {error}) logger.error(fContext: {context}) def on_completion_hook(response: Any, duration_ms: float) - None: Hook called on successful completion. logger.info(fAgent completed in {duration_ms:.2f}ms) # Create agent with hooks using Instructors hook system client instructor.from_openai(openai.OpenAI()) # Register hooks with the instructor client client.on(completion, lambda *args: on_completion_hook(*args)) agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) )这里演示了两种 Hook 接入方式的其中一种直接调用 Instructor 客户端的client.on(...)。AtomicAgent.register_hook(event, handler)的底层实现本质上也是委托给客户端——只要客户端实现了on/off/clear方法注册与注销就会同步到 Instructor见 atomic_agent.py测试用例 test_atomic_agent.py 验证了该委托行为。统一错误处理器对于需要集中管理错误分类与处理的场景官方文档提供了一个装饰器 回调注入的AgentErrorHandlerimport os from typing import Callable, Optional, TypeVar from functools import wraps import instructor import openai from pydantic import ValidationError from atomic_agents import AtomicAgent, AgentConfig, BaseIOSchema T TypeVar(T, boundBaseIOSchema) class AgentErrorHandler: Centralized error handler for Atomic Agents. def __init__( self, on_validation_error: Optional[Callable[[ValidationError], None]] None, on_api_error: Optional[Callable[[Exception], None]] None, on_unknown_error: Optional[Callable[[Exception], None]] None ): self.on_validation_error on_validation_error or self._default_validation_handler self.on_api_error on_api_error or self._default_api_handler self.on_unknown_error on_unknown_error or self._default_unknown_handler def _default_validation_handler(self, error: ValidationError) - None: print(fValidation failed: {error.error_count()} errors) for err in error.errors(): print(f - {err[loc]}: {err[msg]}) def _default_api_handler(self, error: Exception) - None: print(fAPI error: {type(error).__name__}: {error}) def _default_unknown_handler(self, error: Exception) - None: print(fUnknown error: {type(error).__name__}: {error}) def wrap(self, func: Callable) - Callable: Decorator to wrap agent calls with error handling. wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except ValidationError as e: self.on_validation_error(e) return None except (openai.APIError, openai.APIConnectionError) as e: self.on_api_error(e) return None except Exception as e: self.on_unknown_error(e) return None return wrapper # Usage error_handler AgentErrorHandler() error_handler.wrap def ask_agent(agent: AtomicAgent, question: str): from atomic_agents import BasicChatInputSchema return agent.run(BasicChatInputSchema(chat_messagequestion)) # Create and use agent client instructor.from_openai(openai.OpenAI()) agent AtomicAgent( configAgentConfig( clientclient, modelgpt-5-mini ) ) response ask_agent(agent, What is machine learning?)设计要点三类回调可注入校验错误ValidationError、API 错误openai.APIError及其子类、未知异常分别对应不同处理入口默认可打印error.error_count()与error.errors()明细返回值统一为None表示失败被包装函数的调用方只需判空即可无需层层 try/except这种集中式分类处理与 Hook 的事件订阅是互补方案前者适合同步业务流后者适合横切监控。优雅降级Fallback Agent 链当主 Agent 持续失败时一个成熟的生产方案是准备多个不同模型/提供商的 Agent按优先级依次尝试直到成功。官方文档给出的FallbackAgentChain是该模式的最小实现import os from typing import Optional, List import instructor import openai from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema from atomic_agents.context import ChatHistory class FallbackAgentChain: Chain of agents with automatic fallback on failure. def __init__(self, agents: List[AtomicAgent]): self.agents agents def run(self, input_data: BasicChatInputSchema) - Optional[BasicChatOutputSchema]: Try each agent in order until one succeeds. last_error None for i, agent in enumerate(self.agents): try: print(fTrying agent {i 1}/{len(self.agents)}) return agent.run(input_data) except Exception as e: last_error e print(fAgent {i 1} failed: {e}) continue print(fAll agents failed. Last error: {last_error}) return None # Create primary and fallback agents with different models/providers def create_fallback_chain() - FallbackAgentChain: # Primary: GPT-4 primary_client instructor.from_openai(openai.OpenAI()) primary_agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) ) # Fallback: GPT-4o-mini (cheaper, faster) fallback_client instructor.from_openai(openai.OpenAI()) fallback_agent AtomicAgentBasicChatInputSchema, BasicChatOutputSchema ) ) return FallbackAgentChain([primary_agent, fallback_agent]) # Usage chain create_fallback_chain() response chain.run(BasicChatInputSchema(chat_messageExplain quantum computing)) if response: print(response.chat_message)该模式的实践要点主备模型差异化主 Agent 用高质量模型如gpt-4o备选 Agent 用更快更便宜的小模型如gpt-5-mini以成本换取可用性异常粒度可细化生产环境可按前文的重试策略瞬时故障 vs 永久错误决定是立即降级还是先重试避免因一次抖动就切换模型由于每个 Agent 都是独立实例且共享同一输入 SchemaBasicChatInputSchema链式切换对调用方完全透明。最佳实践四条官方文档总结了四条生产级最佳实践其中三条与防御 观测直接相关。1. 始终校验输入在输入 Schema 中做防御性校验甚至可以过滤提示注入特征from pydantic import Field, field_validator from atomic_agents import BaseIOSchema class SafeInputSchema(BaseIOSchema): Input schema with comprehensive validation. message: str Field(..., min_length1, max_length10000) field_validator(message) classmethod def sanitize_message(cls, v: str) - str: # Remove potential prompt injection attempts dangerous_patterns [ignore previous, disregard instructions] for pattern in dangerous_patterns: if pattern.lower() in v.lower(): raise ValueError(Invalid input detected) return v.strip()2. 记录所有错误用装饰器统一记录 Agent 操作的异常栈logger.exception会自动附带当前异常的 tracebackimport logging from functools import wraps logger logging.getLogger(__name__) def log_errors(func): Decorator to log all errors from agent operations. wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logger.exception(fError in {func.__name__}: {e}) raise return wrapper需要说明的是AtomicAgent内部对 Hook 处理器本身也做了错误隔离当某个 Hook handler 抛异常时_dispatch_hook会记录 warning 并继续执行后续 handler不中断 Agent 主流程见 atomic_agent.py对应测试 test_atomic_agent.py。因此你的日志/监控 Hook 自身即使出问题也不会拖垮业务调用。3. 设置超时在客户端层面配置超时避免无界等待import os import instructor import openai from atomic_agents import AtomicAgent, AgentConfig from atomic_agents.context import ChatHistory # Configure timeout at client level client instructor.from_openai( openai.OpenAI(timeout30.0) # 30 second timeout ) agent AtomicAgent( configAgentConfig( clientclient, modelgpt-5-mini, historyChatHistory(), model_api_parameters{ max_tokens: 500 # Limit response length } ) )超时timeout30.0与max_tokens限制输出长度双管齐下是防止慢请求与超长输出拖垮系统的基础手段。结合前文重试模式超时抛出的异常可按APIConnectionError路径处理。4. 实现熔断器熔断器用于防止级联故障当失败次数达到阈值后开路拒绝调用经过冷却时间后转入半开状态试探性放行一次调用import time from typing import Optional, Callable from dataclasses import dataclass dataclass class CircuitBreaker: Simple circuit breaker for agent calls. failure_threshold: int 5 reset_timeout: float 60.0 _failure_count: int 0 _last_failure_time: float 0 _state: str closed # closed, open, half-open def call(self, func: Callable, *args, **kwargs): Execute function with circuit breaker protection. if self._state open: if time.time() - self._last_failure_time self.reset_timeout: self._state half-open else: raise Exception(Circuit breaker is open) try: result func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): self._failure_count 0 self._state closed def _on_failure(self): self._failure_count 1 self._last_failure_time time.time() if self._failure_count self.failure_threshold: self._state open # Usage circuit_breaker CircuitBreaker(failure_threshold3, reset_timeout30.0) def safe_agent_call(agent, input_data): return circuit_breaker.call(agent.run, input_data)三态状态机语义closed正常放行、open直接拒绝并快速失败、half-open冷却期满后放行一次探测。当熔断器处于 open 状态时调用立即抛异常避免对已失效的 Provider 持续发起昂贵的请求。与框架上下文管理的联动值得补充的一点错误处理并非只发生在调用失败时刻。AtomicAgent在每次run/run_stream/run_async调用前会执行_trim_context()对历史进行裁剪见 atomic_agent.py当配置了max_context_tokens且上下文超限时框架会按 turn 从最旧开始删除若单个 turn 本身就超出限制则会抛出带明确修复建议的ValueError——这是非模型错误的一类典型处理场景。其 token 统计基于 token_counter.py 的count_context通过 LiteLLM 实现与模型无关的计数并可返回utilization上下文占用比例。总结策略选型速查表策略适用场景实现方式Schema 校验阻止非法输入进入模型Pydantic validatorsField约束、field_validator、model_validator重试逻辑瞬时故障限流、连接抖动、5xx指数退避 固定延迟4xx 直接抛出Hook 系统监控、日志、指标采集Instructor 事件parse:error、completion:*register_hookFallback 链高可用与降级多 Agent不同模型/提供商按序尝试熔断器防止级联故障三态状态机closed / open / half-open落地建议将 Schema 校验作为第一道闸门前置拦截非法输入与注入特征以 Hook 系统承载横切观测日志、指标、审计对外部 API 调用统一走重试 → Fallback → 熔断的递进恢复路径并将统一错误处理器作为最终兜底出口。这样每一层各司其职才能构建出既健壮又可观测的 Atomic Agents 生产应用。赞分享AI AgentAgent 框架MCP 服务后端【免费下载链接】atomic-agentsBuilding AI agents, atomically项目地址https://gitcode.com/gh_mirrors/at/atomic-agents点击查看免费下载相关推荐从Hystrix到Resilience4jJava熔断降级方案的完整技术演进与实践指南从Hystrix到Resilience4jJava熔断降级方案的完整技术演进与实践指南 在当今微服务架构盛行的时代 Java熔断降级方案 已成为保障分布式系文档教程后端X-Plane Connect完全指南如何用代码实时控制飞行模拟器X Plane Connect完全指南如何用代码实时控制飞行模拟器 X Plane ConnectXPC是一款开源研究工具能让开发者通过C、C、J开发工具一个页面能拦掉多少广告请求uBlock Origin 免费开源广告拦截扩展快速上手一个页面能拦掉多少广告请求uBlock Origin 免费开源广告拦截扩展快速上手 默认配置自带 5 份社区过滤列表已知的广告与跟踪请求在离开本机前就被取消网络安全应用安全上一篇Sliver框架终极扩展开发指南5步创建自定义渗透测试模块下一篇Peritext性能优化指南高效处理大规模文本协作的10个技巧创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考