Coverage for agentos/errors/handler.py: 44%
102 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +0800
1"""v0.80 — 用户友好错误处理：分类 + 格式化 + 建议。"""
3from __future__ import annotations
5import traceback
6import sys
7from dataclasses import dataclass, field
8from enum import Enum, auto
9from typing import Any, Optional
12class ErrorCategory(Enum):
13 """错误分类枚举。"""
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 """错误上下文信息。"""
45 trace_id: str = ""
46 category: ErrorCategory = ErrorCategory.UNKNOWN
47 message: str = ""
48 suggestion: str = ""
49 detail: str = ""
50 recovery_actions: list[str] = field(default_factory=list)
53class HumanError(Exception):
54 """包装原始异常，附带用户友好的上下文。"""
56 def __init__(self, original: Exception, context: ErrorContext):
57 super().__init__(str(original))
58 self.original = original
59 self.context = context
61 def __str__(self) -> str:
62 return self.context.message or super().__str__()
65class ErrorFormatter:
66 """将 Python 异常转换为用户友好的格式化输出。"""
68 @staticmethod
69 def categorize(exc: Exception) -> ErrorCategory:
70 """根据异常类型和消息自动分类。"""
71 msg = str(exc).lower()
72 type_name = type(exc).__name__.lower()
74 if any(kw in msg for kw in ["timeout", "timed out", "connect timeout"]):
75 return ErrorCategory.TIMEOUT
76 if any(kw in msg for kw in ["rate limit", "too many requests", "429"]):
77 return ErrorCategory.RATE_LIMIT
78 if any(kw in msg for kw in ["unauthorized", "forbidden", "401", "403", "api key", "invalid key"]):
79 return ErrorCategory.AUTH
80 if any(kw in msg for kw in ["connection", "network", "dns", "refused", "unreachable"]):
81 return ErrorCategory.NETWORK
82 if any(kw in msg for kw in ["validation", "invalid", "expected", "type error", "value error"]):
83 return ErrorCategory.VALIDATION
84 if any(kw in msg for kw in ["memory", "disk", "quota", "out of"]):
85 return ErrorCategory.RESOURCE
86 if any(kw in type_name for kw in ["plugin", "load"]):
87 return ErrorCategory.PLUGIN
88 if any(kw in msg for kw in ["config", "cfg", "yaml"]):
89 return ErrorCategory.CONFIG
90 if any(kw in type_name for kw in ["model", "llm", "provider"]):
91 return ErrorCategory.MODEL
92 return ErrorCategory.UNKNOWN
94 @staticmethod
95 def extract_recovery(original: Exception, category: ErrorCategory) -> list[str]:
96 """根据异常给出可操作的恢复建议。"""
97 actions = [CATEGORY_HINTS.get(category, "")]
98 msg = str(original)
100 if ErrorFormatter._has_retry(category):
101 actions.append("框架已自动重试，若持续失败请检查上游服务状态。")
102 if "api key" in msg.lower() or "key" in msg.lower():
103 actions.append("运行 `agentos config set api_key <your-key>` 或设置环境变量。")
104 if "model" in msg.lower() and "not found" in msg.lower():
105 actions.append("请确认 ModelConfig.model_name 拼写正确，或使用 RECOMMENDED_CONFIG。")
106 return [a for a in actions if a]
108 @staticmethod
109 def _has_retry(category: ErrorCategory) -> bool:
110 return category in (ErrorCategory.NETWORK, ErrorCategory.TIMEOUT, ErrorCategory.RATE_LIMIT)
112 @classmethod
113 def format(cls, exc: Exception, trace_id: str = "") -> ErrorContext:
114 """将异常格式化为 ErrorContext。"""
115 category = cls.categorize(exc)
116 return ErrorContext(
117 trace_id=trace_id,
118 category=category,
119 message=cls._friendly_message(exc, category),
120 suggestion=CATEGORY_HINTS.get(category, ""),
121 detail=cls._extract_key_detail(exc),
122 recovery_actions=cls.extract_recovery(exc, category),
123 )
125 @staticmethod
126 def _friendly_message(exc: Exception, category: ErrorCategory) -> str:
127 type_msg = str(exc)
128 prefix = {
129 ErrorCategory.NETWORK: "网络连接失败",
130 ErrorCategory.AUTH: "认证失败",
131 ErrorCategory.CONFIG: "配置错误",
132 ErrorCategory.RATE_LIMIT: "请求被限流",
133 ErrorCategory.VALIDATION: "输入校验失败",
134 ErrorCategory.TIMEOUT: "操作超时",
135 ErrorCategory.RESOURCE: "资源不足",
136 ErrorCategory.MODEL: "模型调用异常",
137 ErrorCategory.PLUGIN: "插件错误",
138 ErrorCategory.INTERNAL: "内部错误",
139 ErrorCategory.UNKNOWN: "发生错误",
140 }.get(category, "错误")
141 return f"{prefix}: {type_msg[:120]}"
143 @staticmethod
144 def _extract_key_detail(exc: Exception) -> str:
145 lines = traceback.format_exception_only(type(exc), exc)
146 return "".join(lines[-2:]).strip()
149def format_error(exc: Exception, trace_id: str = "") -> str:
150 """一行调用：输出用户友好的错误信息。"""
151 ctx = ErrorFormatter.format(exc, trace_id)
152 parts = [f"[{ctx.category.name}] {ctx.message}"]
153 if ctx.suggestion:
154 parts.append(f" 建议: {ctx.suggestion}")
155 for action in ctx.recovery_actions:
156 parts.append(f" -> {action}")
157 return "\n".join(parts)
160def friendly_error(func):
161 """装饰器：自动捕获异常并输出友好信息。"""
162 def wrapper(*args, **kwargs):
163 try:
164 return func(*args, **kwargs)
165 except Exception as e:
166 friendly_msg = format_error(e)
167 print(friendly_msg, file=sys.stderr)
168 raise
169 return wrapper