Coverage for agentos/errors/handler.py: 44%
101 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 12:20 +0800
1"""v0.80 — 用户友好错误处理：分类 + 格式化 + 建议。"""
3from __future__ import annotations
5import sys
6import traceback
7from dataclasses import dataclass, field
8from enum import Enum, auto
11class ErrorCategory(Enum):
12 """错误分类枚举。"""
14 NETWORK = auto() # 网络/API 调用失败
15 AUTH = auto() # 认证/API Key 问题
16 CONFIG = auto() # 配置错误
17 RATE_LIMIT = auto() # 限流/配额
18 VALIDATION = auto() # 输入验证
19 TIMEOUT = auto() # 超时
20 RESOURCE = auto() # 资源不足（内存/磁盘）
21 MODEL = auto() # 模型相关
22 PLUGIN = auto() # 插件错误
23 INTERNAL = auto() # 内部错误
24 UNKNOWN = auto() # 未分类
27CATEGORY_HINTS = {
28 ErrorCategory.NETWORK: "请检查网络连接或 API 端点地址。",
29 ErrorCategory.AUTH: "请确认 API Key 是否正确设置（环境变量或配置文件）。",
30 ErrorCategory.CONFIG: "请检查 agentos.yaml 配置文件，确保字段拼写正确。",
31 ErrorCategory.RATE_LIMIT: "请求频率过高，请稍后重试。可调整 RateLimitCfg.max_rps。",
32 ErrorCategory.VALIDATION: "输入参数不符合预期格式，请参考文档修正。",
33 ErrorCategory.TIMEOUT: "操作超时。可增大 LoopCfg.step_timeout 或 ModelConfig.timeout。",
34 ErrorCategory.RESOURCE: "系统资源不足，请检查内存/磁盘或降低并发。",
35 ErrorCategory.MODEL: "模型返回异常或调用失败，可尝试切换备用 Provider。",
36 ErrorCategory.PLUGIN: "插件加载失败，请检查插件路径和依赖。",
37 ErrorCategory.INTERNAL: "内部错误，请联系开发者并提供 trace_id。",
38 ErrorCategory.UNKNOWN: "未知错误，请查看详细日志。",
39}
42@dataclass
43class ErrorContext:
44 """错误上下文信息。"""
46 trace_id: str = ""
47 category: ErrorCategory = ErrorCategory.UNKNOWN
48 message: str = ""
49 suggestion: str = ""
50 detail: str = ""
51 recovery_actions: list[str] = field(default_factory=list)
54class HumanError(Exception):
55 """包装原始异常，附带用户友好的上下文。"""
57 def __init__(self, original: Exception, context: ErrorContext):
58 super().__init__(str(original))
59 self.original = original
60 self.context = context
62 def __str__(self) -> str:
63 return self.context.message or super().__str__()
66class ErrorFormatter:
67 """将 Python 异常转换为用户友好的格式化输出。"""
69 @staticmethod
70 def categorize(exc: Exception) -> ErrorCategory:
71 """根据异常类型和消息自动分类。"""
72 msg = str(exc).lower()
73 type_name = type(exc).__name__.lower()
75 if any(kw in msg for kw in ["timeout", "timed out", "connect timeout"]):
76 return ErrorCategory.TIMEOUT
77 if any(kw in msg for kw in ["rate limit", "too many requests", "429"]):
78 return ErrorCategory.RATE_LIMIT
79 if any(
80 kw in msg
81 for kw in ["unauthorized", "forbidden", "401", "403", "api key", "invalid key"]
82 ):
83 return ErrorCategory.AUTH
84 if any(kw in msg for kw in ["connection", "network", "dns", "refused", "unreachable"]):
85 return ErrorCategory.NETWORK
86 if any(
87 kw in msg for kw in ["validation", "invalid", "expected", "type error", "value error"]
88 ):
89 return ErrorCategory.VALIDATION
90 if any(kw in msg for kw in ["memory", "disk", "quota", "out of"]):
91 return ErrorCategory.RESOURCE
92 if any(kw in type_name for kw in ["plugin", "load"]):
93 return ErrorCategory.PLUGIN
94 if any(kw in msg for kw in ["config", "cfg", "yaml"]):
95 return ErrorCategory.CONFIG
96 if any(kw in type_name for kw in ["model", "llm", "provider"]):
97 return ErrorCategory.MODEL
98 return ErrorCategory.UNKNOWN
100 @staticmethod
101 def extract_recovery(original: Exception, category: ErrorCategory) -> list[str]:
102 """根据异常给出可操作的恢复建议。"""
103 actions = [CATEGORY_HINTS.get(category, "")]
104 msg = str(original)
106 if ErrorFormatter._has_retry(category):
107 actions.append("框架已自动重试，若持续失败请检查上游服务状态。")
108 if "api key" in msg.lower() or "key" in msg.lower():
109 actions.append("运行 `agentos config set api_key <your-key>` 或设置环境变量。")
110 if "model" in msg.lower() and "not found" in msg.lower():
111 actions.append("请确认 ModelConfig.model_name 拼写正确，或使用 RECOMMENDED_CONFIG。")
112 return [a for a in actions if a]
114 @staticmethod
115 def _has_retry(category: ErrorCategory) -> bool:
116 return category in (ErrorCategory.NETWORK, ErrorCategory.TIMEOUT, ErrorCategory.RATE_LIMIT)
118 @classmethod
119 def format(cls, exc: Exception, trace_id: str = "") -> ErrorContext:
120 """将异常格式化为 ErrorContext。"""
121 category = cls.categorize(exc)
122 return ErrorContext(
123 trace_id=trace_id,
124 category=category,
125 message=cls._friendly_message(exc, category),
126 suggestion=CATEGORY_HINTS.get(category, ""),
127 detail=cls._extract_key_detail(exc),
128 recovery_actions=cls.extract_recovery(exc, category),
129 )
131 @staticmethod
132 def _friendly_message(exc: Exception, category: ErrorCategory) -> str:
133 type_msg = str(exc)
134 prefix = {
135 ErrorCategory.NETWORK: "网络连接失败",
136 ErrorCategory.AUTH: "认证失败",
137 ErrorCategory.CONFIG: "配置错误",
138 ErrorCategory.RATE_LIMIT: "请求被限流",
139 ErrorCategory.VALIDATION: "输入校验失败",
140 ErrorCategory.TIMEOUT: "操作超时",
141 ErrorCategory.RESOURCE: "资源不足",
142 ErrorCategory.MODEL: "模型调用异常",
143 ErrorCategory.PLUGIN: "插件错误",
144 ErrorCategory.INTERNAL: "内部错误",
145 ErrorCategory.UNKNOWN: "发生错误",
146 }.get(category, "错误")
147 return f"{prefix}: {type_msg[:120]}"
149 @staticmethod
150 def _extract_key_detail(exc: Exception) -> str:
151 lines = traceback.format_exception_only(type(exc), exc)
152 return "".join(lines[-2:]).strip()
155def format_error(exc: Exception, trace_id: str = "") -> str:
156 """一行调用：输出用户友好的错误信息。"""
157 ctx = ErrorFormatter.format(exc, trace_id)
158 parts = [f"[{ctx.category.name}] {ctx.message}"]
159 if ctx.suggestion:
160 parts.append(f" 建议: {ctx.suggestion}")
161 for action in ctx.recovery_actions:
162 parts.append(f" -> {action}")
163 return "\n".join(parts)
166def friendly_error(func):
167 """装饰器：自动捕获异常并输出友好信息。"""
169 def wrapper(*args, **kwargs):
170 try:
171 return func(*args, **kwargs)
172 except Exception as e:
173 friendly_msg = format_error(e)
174 print(friendly_msg, file=sys.stderr)
175 raise
177 return wrapper