ECC python-patterns Skill 深度解析：从 Protocol、Dataclass 到 Async/Await 的 Pythonic 模式库

ECC（Agent Harness Performance Optimization System）在其技能体系中内置了一份面向 Python 代码库的规范技能 python-patterns。该技能定义了一套“Python 专属设计模式”清单——从结构化子类型的 Protocol、用作 DTO 的 dataclass，到资源管理的上下文管理器、惰性求值的生成器、装饰器、async/await 并发、进阶类型注解、依赖注入、包组织、自定义异常与函数式组合。当你在 ECC 生态（Claude Code、Codex、Opencode、Cursor 等 harness）中编写、重构或评审 Python 代码时，本技能会自动挂载，用于约束 Agent 写出符合 Pythonic 惯例的实现。读完本文，你将完整掌握该技能的每一条模式，并在 ECC 仓库自带的 src/llm 抽象层源码中找到这些模式在生产代码中的真实印证。

技能定位：它是一份“可被 Agent 调用”的 Python 编码契约

python-patterns 技能本体存放在 .kiro/skills/python-patterns/SKILL.md，其 frontmatter 决定了它的启用方式：

---
name: python-patterns
description: >
  Python-specific design patterns and best practices including protocols,
  dataclasses, context managers, decorators, async/await, type hints, and
  package organization. Use when working with Python code to apply Pythonic
  patterns.
metadata:
  origin: ECC
  globs: ["**/*.py", "**/*.pyi"]
---

关键信息有三点：

• globs 声明该技能只对 **/*.py 与 **/*.pyi 文件生效，即只要 Agent 正在操作 Python 源码或类型桩文件，就会考虑套用本文模式；
• description 直接列出了技能覆盖的主题（protocols、dataclasses、context managers、decorators、async/await、type hints、package organization），并明确了适用时机——“working with Python code”；
• origin: ECC 标明这是 ECC 原生孵化的技能。

在 ECC 仓库内，这一技能与规则库 rules/python/patterns.md 形成配套：规则文件负责给出最精简的强制要点，并在末尾显式指向技能本体：

See skill: python-patterns for comprehensive patterns including decorators, concurrency, and package organization.

也就是说，rules/python/patterns.md 是“骨架规则”，python-patterns 技能是“完整参考实现手册”，两者共同约束仓库中的 Python 风格（其相邻规则还包括 coding-style.md、fastapi.md、hooks.md、security.md、testing.md）。仓库同样保留了技能的多语言镜像，例如 docs/es/skills/python-patterns/SKILL.md。

Protocol：带类型提示的鸭子类型（结构化子类型）

技能给出的第一条模式是使用 typing.Protocol 实现“结构化子类型”——即鸭子类型与静态类型检查的折中方案。只要一个类拥有协议所声明的全部方法签名，就自动满足该协议，无需显式继承：

from typing import Protocol

class Repository(Protocol):
    def find_by_id(self, id: str) -> dict | None: ...
    def save(self, entity: dict) -> dict: ...

# Any class with these methods satisfies the protocol
class UserRepository:
    def find_by_id(self, id: str) -> dict | None:
        # implementation
        pass

    def save(self, entity: dict) -> dict:
        # implementation
        pass

def process_entity(repo: Repository, id: str) -> None:
    entity = repo.find_by_id(id)
    # ... process

技能总结的收益是：

• Type safety without inheritance：获得类型安全性而不必强行走继承体系；
• Flexible, loosely coupled code：模块间只依赖“形状”（方法签名）而非具体类，耦合更松散；
• Easy testing and mocking：测试中构造一个满足协议的替身对象即可，天然便于 mock。

在 ECC 的实际 Python 工程中，这一思想更多以“抽象基类 + 接口契约”的形式落地：src/llm/core/interface.py 定义了统一的 LLMProvider 抽象（声明 generate、list_models、validate_config 等抽象方法），而 src/llm/providers/resolver.py 通过一张 _PROVIDER_MAP: dict[ProviderType, type[LLMProvider]] 注册表将 claude、openai、ollama、astraflow、atlas 等多个 Provider 实现统一接入。从该设计可以推断：任何新增 Provider 只要满足接口契约并调用 register_provider 注册（resolver.py），即可无缝接入调用方——这正是“面向协议而非面向具体实现”的解耦价值。

Dataclass 作为 DTO：数据载体的一等公民

技能建议用 @dataclass 表达 DTO（数据传输对象）与值对象（value object），因为它自动生成样板代码：

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CreateUserRequest:
    name: str
    email: str
    age: Optional[int] = None
    tags: list[str] = field(default_factory=list)

@dataclass(frozen=True)
class User:
    """Immutable user entity"""
    id: str
    name: str
    email: str

技能总结的能力要点：

• 自动生成 __init__、__repr__、__eq__；
• frozen=True 提供不可变性（适合不可变实体/值对象）；
• field() 支撑可变默认值（如 default_factory=list，避免共享可变默认值的经典陷阱）；
• 字段类型注解天然构成轻量校验基础。

这套模式在 ECC 的 src/llm 中得到了教科书级的应用。文件 src/llm/core/types.py 通篇使用 @dataclass(frozen=True) 定义核心数据模型：Message（L26-L45）、ToolDefinition（L48-L79）、ToolCall（L82-L86）、ToolResult（L89-L93）、LLMInput（L96-L118）、LLMOutput（L121-L147）以及 ModelInfo（L150-L166）。例如 LLMInput 同时示范了两个关键用法：

1. field(default_factory=dict) 为可变字段提供安全默认值；
2. frozen=True + to_dict() 让不可变 DTO 通过显式方法完成对外序列化，并且用 result | self.metadata 合并额外元数据：

@dataclass(frozen=True)
class LLMInput:
    messages: list[Message]
    model: str | None = None
    temperature: float = 1.0
    max_tokens: int | None = None
    tools: list[ToolDefinition] | None = None
    stream: bool = False
    metadata: dict[str, Any] = field(default_factory=dict)

值得注意，Message、ToolDefinition 等类型内部还在定义阶段引用了尚未声明的 ToolCall（L32），这依赖文件顶部的 from __future__ import annotations 延迟求值注解——这也是 Python 3.11+ 项目中组织互相引用的 DTO 时的高频技巧（可对照 types.py）。而 ToolDefinition 提供了三个 to_* 方法（to_dict/to_openai_tool/to_anthropic_tool），把同一份 DTO 转换成不同厂商的 tool schema，是“dataclass 承载格式适配逻辑”的绝佳范例（types.py）。

Context Manager：用 with 收敛资源生命周期

技能强调用上下文管理器（with 语句）管理资源。首选基于 contextlib.contextmanager 的生成器式写法，最典型的场景是数据库事务的提交/回滚：

from contextlib import contextmanager
from typing import Generator

@contextmanager
def database_transaction(db) -> Generator[None, None, None]:
    """Context manager for database transactions"""
    try:
        yield
        db.commit()
    except Exception:
        db.rollback()
        raise

# Usage
with database_transaction(db):
    db.execute("INSERT INTO users ...")

该实现的关键语义：yield 之前的代码在进入 with 块时执行，yield 之后的 db.commit() 在块正常退出后执行；一旦块内抛出异常则跳过 commit 直接 rollback() 并 raise 原异常。配合返回值注解 Generator[None, None, None]，静态检查工具也能理解“生成器型上下文管理器”的边界。

对于需要持有显式状态（如文件句柄）的场景，技能给出了类式上下文管理器——实现 __enter__/__exit__ 双协议方法：

class FileProcessor:
    def __init__(self, filename: str):
        self.filename = filename
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, 'r')
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        return False  # Don't suppress exceptions

注意 __exit__ 的返回语义：返回 False 表示不吞掉异常，异常会继续向外传播；只有确实想“消化”异常（如记录日志后让程序继续）时才返回 True。这一点技能注释特别标注为“Don't suppress exceptions”。

Generators：惰性求值与内存友好的迭代

技能推荐用生成器处理大数据集的惰性求值，经典用例是逐行读取大文件——任意时刻内存中只保留一行：

def read_large_file(filename: str):
    """Generator for reading large files line by line"""
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

# Memory-efficient processing
for line in read_large_file('huge.txt'):
    process(line)

更进一步，技能对比了生成器表达式与列表推导的内存差异，并展示了“生成器管道（pipeline）”的衔接方式：

# Instead of list comprehension
squares = (x**2 for x in range(1000000))  # Lazy evaluation

# Pipeline pattern
numbers = (x for x in range(100))
evens = (x for x in numbers if x % 2 == 0)
squares = (x**2 for x in evens)

(x**2 for x in range(1000000)) 不立即构造百万级列表，而是按需产出；多层生成器之间通过迭代协议串联，形成“生产-过滤-映射”的流水线，整体复杂度是 O(1) 内存。该模式与下文函数式编程小节的 pipe 组合互为补充。

Decorators：行为织入的两种形态

函数装饰器：测量/日志等横切逻辑

技能示范的 timing 装饰器强调一个关键细节——必须用 functools.wraps 保留被装饰函数的元信息（__name__、__doc__ 等），否则调试与文档工具都会失真：

from functools import wraps
import time

def timing(func):
    """Decorator to measure execution time"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.2f}s")
        return result
    return wrapper

@timing
def slow_function():
    time.sleep(1)

类装饰器：单例等实例化控制

类装饰器同样可用 @wraps 包装类。技能给出的 singleton 用闭包字典缓存类实例，确保同一进程内只存在一份配置类对象：

def singleton(cls):
    """Decorator to make a class a singleton"""
    instances = {}

    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance

@singleton
class Config:
    pass

需要说明的是：singleton 本质上是共享全局状态，使用时应权衡并发与测试隔离成本；技能将其定位为“class-level 行为织入”的教学示例。若需要保留类型提示与更可控的语义，可进一步结合下文 Generic/ParamSpec 实现“类型安全装饰器”。

Async/Await：I/O 密集型并发的标准范式

并发收集：asyncio.gather

技能给出 I/O 密集场景的异步函数范式——把并发子任务收集为 task 列表后统一 await asyncio.gather(...)，将 N 次串行网络等待压成一次并发窗口：

import asyncio
from typing import List

async def fetch_user(user_id: str) -> dict:
    """Async function for I/O-bound operations"""
    await asyncio.sleep(0.1)  # Simulate network call
    return {"id": user_id, "name": "Alice"}

async def fetch_all_users(user_ids: List[str]) -> List[dict]:
    """Concurrent execution with asyncio.gather"""
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

# Run async code
asyncio.run(fetch_all_users(["1", "2", "3"]))

顶层调用统一使用 asyncio.run(...) 作为入口。ECC 的测试基座同样为此做好了铺垫：根目录 pyproject.toml 的 pytest 配置启用了 asyncio_mode = "auto"，测试目录中的 test_*.py（如 tests/test_claude_provider.py、tests/test_atlas_provider.py 等）可直接编写 async 测试函数而不必手工包装事件循环。

异步上下文管理器：__aenter__ / __aexit__

数据库连接这类“异步建立、异步释放”的资源，应实现异步上下文管理器协议：

class AsyncDatabase:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()

async with AsyncDatabase() as db:
    await db.query("SELECT * FROM users")

它与同步 __enter__/__exit__ 一一对应，只是全部替换为 async def，并配合 async with 使用。

Type Hints：进阶泛型与类型安全装饰器

技能在类型注解上给出了两条进阶路线。

TypeVar / Generic / ParamSpec：泛型仓储与签名保持

from typing import TypeVar, Generic, Callable, ParamSpec, Concatenate

T = TypeVar('T')
P = ParamSpec('P')

class Repository(Generic[T]):
    """Generic repository pattern"""
    def __init__(self, entity_type: type[T]):
        self.entity_type = entity_type

    def find_by_id(self, id: str) -> T | None:
        # implementation
        pass

# Type-safe decorator
def log_call(func: Callable[P, T]) -> Callable[P, T]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        print(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

要点拆解：

• TypeVar('T') 绑定类型变量，让 Repository[T] 能按实体类型参数化，find_by_id 返回 T | None；
• ParamSpec('P') 捕获被装饰函数的参数规格，配合 P.args/P.kwargs 让装饰器包装层也能被静态类型检查精确追踪——这是把上文 timing 装饰器“类型化”的官方推荐方式；
• Callable[P, T] -> Callable[P, T] 表示“接受原签名、返回同签名”。

ECC 的真实 Python 工程同样全面采用了这类现代类型语法。根目录 pyproject.toml 声明 requires-python = ">=3.11"，ruff 的 target-version = "py311"（L67-L69），因此在 src/llm 中可以自由使用 X | None、list[str]、dict[str, Any] 等内建泛型写法（如 resolver.py 的 dict[str, str] 解析逻辑），并且仓库还以 mypy 配置强制校验（pyproject.toml 中 warn_return_any = true、warn_unused_ignores = true）。

Union Types（Python 3.10+）与模式匹配

def process(value: str | int | None) -> str:
    match value:
        case str():
            return value.upper()
        case int():
            return str(value)
        case None:
            return "empty"

str | int | None 取代了 Optional[str]/Union[...] 的冗长写法，而 match 语句（structural pattern matching）可对联合类型做穷尽式的分支处理。由于 ECC 要求 Python ≥ 3.11（见 pyproject.toml），这两套语法都可以放心使用。

Dependency Injection：构造器注入

技能推荐的依赖注入形态是构造器注入（Constructor Injection）——把依赖作为 __init__ 参数显式传入，并允许可选依赖（如缓存）缺省：

class UserService:
    def __init__(
        self,
        repository: Repository,
        logger: Logger,
        cache: Cache | None = None
    ):
        self.repository = repository
        self.logger = logger
        self.cache = cache

    def get_user(self, user_id: str) -> User | None:
        if self.cache:
            cached = self.cache.get(user_id)
            if cached:
                return cached

        user = self.repository.find_by_id(user_id)
        if user and self.cache:
            self.cache.set(user_id, user)

        return user

该模式的价值在测试中最明显：无需真实数据库即可注入内存 fake 仓储。ECC 的 Provider 工厂正是“注入思想”的工程化样例——resolver.py 的 get_provider() 根据 ProviderType 查表实例化对应 Provider 类并透传 **kwargs 配置，调用方只依赖 LLMProvider 接口而非具体厂商 SDK。

Package Organization：src 布局与模块导出

技能给出了一套企业级 Python 包布局（src layout）：

project/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── domain/          # Business logic
│       │   ├── __init__.py
│       │   └── models.py
│       ├── services/        # Application services
│       │   ├── __init__.py
│       │   └── user_service.py
│       └── infrastructure/  # External dependencies
│           ├── __init__.py
│           └── database.py
├── tests/
│   ├── unit/
│   └── integration/
├── pyproject.toml
└── README.md

• src/ 包裹真实包体，避免源码树与仓库根目录混排引发的导入错位；
• 按 domain（领域逻辑）/ services（应用服务）/ infrastructure（外部依赖适配）分层，依赖方向由外向内收敛。

模块导出的规范做法是通过 __init__.py 汇聚公共 API，并用 __all__ 显式声明“对外契约”：

# __init__.py
from .models import User, Product
from .services import UserService

__all__ = ['User', 'Product', 'UserService']

ECC 的 src/llm 就是该布局的直接落地：代码全部置于 src/ 下（pyproject.toml 中 hatchling 打包目标正是 packages = ["src/llm"]），按 core（类型与接口）、providers（各厂商实现）、prompt（提示词构建）、tools、cli 等子包切分职责；各子目录均以 __init__.py 组织导出。

Error Handling：分层异常与 except*

技能建议为领域层建立自定义异常体系，先定义一个基类再逐级派生，让调用方可以用一个 except DomainError 捕获整棵异常树：

class DomainError(Exception):
    """Base exception for domain errors"""
    pass

class UserNotFoundError(DomainError):
    """Raised when user is not found"""
    def __init__(self, user_id: str):
        self.user_id = user_id
        super().__init__(f"User {user_id} not found")

class ValidationError(DomainError):
    """Raised when validation fails"""
    def __init__(self, field: str, message: str):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")

派生异常通过 super().__init__(...) 构造带上下文的错误消息，并保留结构化字段（如 user_id、field）供上层程序化处理。

ECC 的 src/llm 正是这样建模的：文件 src/llm/core/interface.py 定义根异常 LLMError（携带 provider、code、details 三个结构化字段），再派生出 AuthenticationError、RateLimitError、ContextLengthError、ModelNotFoundError、ToolExecutionError。从该结构可以看出，调用方既能一次性兜底 except LLMError，也能对“限流”“超上下文”等场景精确降级重试。

对于多个并发任务各自失败的情况，技能给出了 Python 3.11+ 的 Exception Groups 与 except* 语法——按异常类型分别批量处理，且不会因一个分支消费掉全部异常：

try:
    # Multiple operations
    pass
except* ValueError as eg:
    # Handle all ValueError instances
    for exc in eg.exceptions:
        print(f"ValueError: {exc}")
except* TypeError as eg:
    # Handle all TypeError instances
    for exc in eg.exceptions:
        print(f"TypeError: {exc}")

except* ValueError 只消费组内的 ValueError 成员，组中残留的其他类型异常会继续被后续 except* 或外层处理，这让 asyncio.gather 返回的多异常结果处理变得优雅。

Property Decorators：受控属性访问

技能用 @property 与 setter 实现对属性的读写控制——对外暴露只读属性，或在校验通过后才允许写入：

class User:
    def __init__(self, name: str):
        self._name = name
        self._email = None

    @property
    def name(self) -> str:
        """Read-only property"""
        return self._name

    @property
    def email(self) -> str | None:
        return self._email

    @email.setter
    def email(self, value: str) -> None:
        if '@' not in value:
            raise ValueError("Invalid email")
        self._email = value

name 只暴露 getter（无 setter 即只读），email 则在 setter 中执行格式校验并抛出 ValueError——把“不变式维护”从散落的赋值点收敛到单一入口，是与上文自定义异常体系协同使用的经典手段。

Functional Programming：高阶函数与管道组合

技能示范了高阶函数与函数组合。借助 functools.reduce 实现从左到右的 pipe 组合器：

from functools import reduce
from typing import Callable, TypeVar

T = TypeVar('T')
U = TypeVar('U')

def pipe(*functions: Callable) -> Callable:
    """Compose functions left to right"""
    def inner(arg):
        return reduce(lambda x, f: f(x), functions, arg)
    return inner

# Usage
process = pipe(
    str.strip,
    str.lower,
    lambda s: s.replace(' ', '_')
)
result = process("  Hello World  ")  # "hello_world"

pipe(str.strip, str.lower, lambda s: ...) 定义了一条纯函数数据流，前一个函数的输出自动成为下一个函数的输入，得到 "hello_world"。这种声明式组合适合构建可读性高的字符串/数据变换链，也与上文“生成器管道”形成同步/惰性两种互补范式。

何时启用该技能（技能给出的决策清单）

技能在文末给出了适用场景清单，可用于判断何时应显式应用这些模式：

• Designing Python APIs and packages —— 设计 Python API 与包结构时；
• Implementing async/concurrent systems —— 实现异步/并发系统时；
• Structuring Python projects —— 组织 Python 工程目录时；
• Writing Pythonic code —— 日常编写需要“地道 Python”风格的代码时；
• Refactoring Python codebases —— 对既有 Python 代码库做重构时；
• Type-safe Python development —— 追求类型安全的开发时。

在 ECC 仓库中的“证据链”速查

为了让读者能顺着真实代码进一步学习，下表汇总了本文每一模式在 ECC 仓库内可对照的落点：

技能模式	仓库内对照实现	相对路径
Dataclass 作为 DTO / frozen=True / field(default_factory=...)	Message、LLMInput、LLMOutput、ModelInfo 等不可变模型	src/llm/core/types.py
接口契约（Protocol/ABC 思想）	LLMProvider 抽象基类与多厂商实现	src/llm/core/interface.py
工厂与注入	_PROVIDER_MAP 注册表 + get_provider()	src/llm/providers/resolver.py
分层自定义异常	LLMError 及各派生异常	src/llm/core/interface.py
现代类型语法（≥3.11）	`X	None、list[...]/dict[...]、asyncio_mode` 测试
技能配套规则	精简版 Python 模式要点	rules/python/patterns.md

结语

ECC 的 python-patterns 技能本质上是一份“可直接被 Agent 执行”的 Python 编码契约：它不空谈抽象原则，而是把从 Protocol 鸭子类型、dataclass DTO、contextmanager 资源管理、生成器惰性管道，到 async/await、TypeVar/ParamSpec 类型安全装饰器、构造器注入、src 布局、分层异常与函数式 pipe 的每一类模式，都给出了可直接复制的最小可运行示例与适用边界。而 ECC 仓库自身 src/llm 中大量使用 frozen dataclass、抽象接口、结构化异常与 ≥3.11 现代类型语法的事实，恰好验证了这套模式在真实多 Provider 抽象层中的工程价值——在你下一次编写或评审 Python 代码时，这份清单即是值得逐条对照的 Pythonic 检查表。