164 lines
6.1 KiB
Markdown
164 lines
6.1 KiB
Markdown
---
|
||
title: "Any 类型与 except Exception 系统性治理策略"
|
||
date: 2026-07-01
|
||
category: conventions/
|
||
module: 全局（跨模块）
|
||
problem_type: convention
|
||
component: tooling
|
||
severity: low
|
||
applies_when:
|
||
- 消除 Any 类型残留
|
||
- 窄化 except Exception
|
||
- Pydantic v2 模型字段类型注解
|
||
- 异常捕获分类
|
||
tags: [type-safety, any-governance, except-classification, pydantic-v2, tech-debt]
|
||
---
|
||
|
||
# Any 类型与 except Exception 系统性治理策略
|
||
|
||
## Context
|
||
|
||
项目禁止使用 `any` 类型（AGENTS.md），但历史代码中积累了大量 `Any` 类型残留（~1000 处）和宽泛的 `except Exception`（~440 处）。这些残留导致：
|
||
- 类型安全缺失，IDE 无法提供有效提示
|
||
- 异常被静默吞没，线上问题难以诊断
|
||
- 代码审查时类型信息不足
|
||
|
||
本约定记录了在 PR #8-#11 中治理 1214+ 处残留时建立的策略，覆盖 `Any` 替代和 `except Exception` 分类两个维度。
|
||
|
||
## Guidance
|
||
|
||
### 1. Any 类型替代优先级（按优先级降序）
|
||
|
||
```python
|
||
# 1. TypeAlias — 用于有明确语义的 dict（非 Pydantic 模型字段）
|
||
from typing import TypeAlias
|
||
MetadataValue: TypeAlias = str | int | float | bool | list["MetadataValue"] | dict[str, "MetadataValue"] | None
|
||
MetadataDict: TypeAlias = dict[str, MetadataValue]
|
||
|
||
# 2. object — 最严格任意类型，适用于 dict/list/kwargs/返回值
|
||
def process(data: dict[str, object], **kwargs: object) -> object: ...
|
||
|
||
# 3. TYPE_CHECKING Protocol — 用于外部依赖对象（Redis/LLM gateway 等）
|
||
from typing import TYPE_CHECKING, Protocol
|
||
if TYPE_CHECKING:
|
||
class _RedisLike(Protocol):
|
||
async def get(self, key: str) -> bytes | None: ...
|
||
async def set(self, key: str, value: str, ex: int | None = None) -> None: ...
|
||
|
||
# 4. TYPE_CHECKING import + 字符串注解 — 避免循环引用
|
||
from typing import TYPE_CHECKING
|
||
if TYPE_CHECKING:
|
||
from agentkit.core.base import BaseAgent
|
||
|
||
# 5. Coroutine[Any, Any, Any] → Coroutine[object, object, object]
|
||
# 6. Callable[..., Awaitable[Any]] → Callable[..., Awaitable[object]]
|
||
```
|
||
|
||
### 2. except Exception 分类策略
|
||
|
||
按调用场景分类异常元组，而非一刀切：
|
||
|
||
```python
|
||
# WebSocket/网络 ops
|
||
except (ConnectionError, RuntimeError, asyncio.TimeoutError) as e:
|
||
|
||
# DB/Store/Redis
|
||
except (ConnectionError, OSError, asyncio.TimeoutError, ValueError, KeyError, RuntimeError) as e:
|
||
|
||
# SQLAlchemy DB 操作 — 用 DBAPIError（不继承 ConnectionError）
|
||
from sqlalchemy.exc import DBAPIError
|
||
except (DBAPIError, ValueError, KeyError, RuntimeError) as e:
|
||
|
||
# HTTP 请求 — httpx.HTTPError 涵盖 HTTPStatusError + RequestError
|
||
except (httpx.HTTPError, ValueError, KeyError, TypeError) as e:
|
||
|
||
# JSON 解析
|
||
except (json.JSONDecodeError, ValueError, TypeError) as e:
|
||
|
||
# SQLite
|
||
except (aiosqlite.Error, ValueError, KeyError, TypeError, RuntimeError) as e:
|
||
```
|
||
|
||
**框架边界保留**：FastAPI 请求处理器、WebSocket 主循环、CLI 主入口仍需 `except Exception` 兜底，但**必须加 CancelledError 守卫**：
|
||
|
||
```python
|
||
try:
|
||
await handle_websocket(ws)
|
||
except asyncio.CancelledError:
|
||
raise # 必须传播取消信号
|
||
except Exception: # noqa: BLE001 — 框架边界兜底
|
||
logger.exception("WebSocket handler failed")
|
||
```
|
||
|
||
### 3. 设计意图保留
|
||
|
||
某些方法的设计意图就是"任何异常都回退到默认值"，不应窄化：
|
||
|
||
- `health_check()` — 任何异常返回 False
|
||
- `list_sources()` — 任何异常回退到默认信息源
|
||
- 文档解析器（PDF/Docx/Xlsx）— 任何解析失败回退到文本
|
||
|
||
这些方法保留 `except Exception` + `asyncio.CancelledError: raise` 守卫 + `# noqa: BLE001` 注释。
|
||
|
||
## Why This Matters
|
||
|
||
- **类型安全**：`object` 强制调用方显式类型断言，避免 `Any` 导致的静默类型逃逸
|
||
- **可观测性**：窄化 except 后，未预期的异常会向上传播而非被静默吞没，便于线上问题诊断
|
||
- **Pydantic v2 兼容**：递归 TypeAlias 在 Pydantic v2 中会导致 RecursionError（见下文陷阱），需用 `dict[str, object]` 替代
|
||
- **CancelledError 安全**：`except Exception` 会捕获 `asyncio.CancelledError`，导致 async 任务无法正确取消；加守卫确保取消信号传播
|
||
|
||
## When to Apply
|
||
|
||
- 新代码禁止使用 `Any`，用 `object` 或更具体的类型
|
||
- 新代码的 `except Exception` 必须评估是否可窄化；框架边界必须加 CancelledError 守卫
|
||
- 重构现有代码时，按本策略治理 `Any` 和 `except Exception`
|
||
- `health_check`/`list_sources`/解析器方法保留 `except Exception`（设计意图）
|
||
|
||
## Examples
|
||
|
||
### 陷阱：Pydantic v2 递归 TypeAlias
|
||
|
||
```python
|
||
# ❌ 错误 — Pydantic v2 无法为递归命名别名构建 schema
|
||
FieldValue: TypeAlias = str | int | list["FieldValue"] | dict[str, "FieldValue"]
|
||
|
||
class Record(BaseModel):
|
||
values: dict[str, FieldValue] # RecursionError: maximum recursion depth exceeded
|
||
|
||
# ✅ 正确 — 用 object，Pydantic v2 运行时正确序列化 dict/list/primitive
|
||
class Record(BaseModel):
|
||
values: dict[str, object] = PydanticField(default_factory=dict)
|
||
# ponytail: JSONB columns hold arbitrary JSON; object is the most permissive
|
||
# type and Pydantic v2 serializes dict/list/primitive values fine at runtime.
|
||
```
|
||
|
||
### 陷阱：SQLAlchemy 异常不继承 ConnectionError
|
||
|
||
```python
|
||
# ❌ 错误 — SQLAlchemy 的 DBAPIError 不继承自 ConnectionError
|
||
except (ConnectionError, OSError) as e:
|
||
# 不会捕获 SQLAlchemy 的数据库错误
|
||
|
||
# ✅ 正确 — 显式导入 DBAPIError
|
||
from sqlalchemy.exc import DBAPIError
|
||
except (DBAPIError, ValueError, KeyError, RuntimeError) as e:
|
||
```
|
||
|
||
### 陷阱：测试用 side_effect=Exception(...) 模拟异常
|
||
|
||
```python
|
||
# 测试代码中常见：
|
||
mock.side_effect = Exception("Connection refused")
|
||
|
||
# 如果生产代码窄化为 except (httpx.HTTPError, ...):
|
||
# 测试会失败，因为裸 Exception 不在元组中
|
||
# 解决方案：health_check 等方法保留 except Exception（设计意图）
|
||
# 或修改测试用 httpx.ConnectError("...") 替代
|
||
```
|
||
|
||
## Related
|
||
|
||
- AGENTS.md — 项目规则（禁止 any 类型、API Key 恒定时间比较等）
|
||
- .trae/rules/ponytail.md — Lazy senior dev 模式，最小修改原则
|
||
- PR #8 (U1-U5)、#9、#10、#11 — 治理实施的 4 个 PR
|