Python 资源管理实战指南基于 agents24 仓库 python-resource-management Skill 的上下文管理器、清理与流式处理全解析【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents本篇技术指南以本仓库plugins/python-development/skills/python-resource-management/SKILL.md为核心骨架系统讲解如何利用上下文管理器context manager对数据库连接、文件句柄、网络套接字等资源进行确定性管理。文中覆盖类式协议实现、async 协议、contextmanager装饰器、异常抑制、流式累积、指标追踪与ExitStack多资源编排等 9 大实战模式并引入仓库内references/details.md的进阶示例帮助读者写出即使异常发生也能可靠释放资源的健壮 Python 代码。该 Skill 在仓库中的定位与使用场景python-resource-management是本仓库python-development插件下的一个技能Skill与同目录下的python-resilience重试、超时、熔断、python-error-handling、python-observability等技能互为补充共同构成生产级 Python 开发的技能矩阵。该插件的python-proAgent 将 Context managers and thewithstatement for resource management 列为现代 Python 能力的核心要点之一可见资源管理在整个插件知识体系中的基础地位。从 SKILL.md 的元数据看该技能在以下场景应被主动调用管理数据库连接与连接池database connections and connection pools处理文件句柄与 I/Ofile handles and I/O实现自定义上下文管理器custom context managers构建带状态的流式响应streaming responses with state处理嵌套资源清理nested resource cleanup创建异步上下文管理器async context managers核心思想只有一句话资源数据库连接、文件句柄、网络套接字等应当被确定性地释放即使发生异常也不例外。核心概念上下文管理器协议1.with语句保证自动释放with语句确保资源在代码块退出时自动释放无论块内是正常结束还是抛出了异常。这是 Python 中最基础也最重要的资源管理手段。2. 协议方法同步与异步上下文管理器协议由两组方法构成同步协议__enter__/__exit__异步协议__aenter__/__aexit__与async with配合使用3. 无条件清理__exit__总是执行无论块内是否发生异常。这是确定性释放的根本保证——即使with块内部抛错清理逻辑也一定会运行。4. 异常处理语义__exit__的返回值决定异常的去向返回True抑制suppress异常with块外的代码不会看到该异常返回False/None异常继续向上传播返回值语义是整个异常抑制机制的核心后面会在选择性异常抑制模式中展开。快速开始装饰器一行实现对于简单场景contextmanager装饰器是最快上手的方式from contextlib import contextmanager contextmanager def managed_resource(): resource acquire_resource() try: yield resource finally: resource.cleanup() with managed_resource() as r: r.do_work()关键在于try/finally结构yield之前是进入逻辑yield之后即finally块是退出逻辑。无论with块内是否抛异常finally中的清理代码都会执行。基础模式从类式协议到装饰器Pattern 1类式上下文管理器复杂资源对于需要维护内部状态、生命周期复杂的资源建议实现完整的类式协议。以下是一个数据库连接的完整示例class DatabaseConnection: Database connection with automatic cleanup. def __init__(self, dsn: str) - None: self._dsn dsn self._conn: Connection | None None def connect(self) - None: Establish database connection. self._conn psycopg.connect(self._dsn) def close(self) - None: Close connection if open. if self._conn is not None: self._conn.close() self._conn None def __enter__(self) - DatabaseConnection: Enter context: connect and return self. self.connect() return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Exit context: always close connection. self.close() # Usage with context manager (preferred) with DatabaseConnection(dsn) as db: result db.execute(query) # Manual management when needed db DatabaseConnection(dsn) db.connect() try: result db.execute(query) finally: db.close()值得注意的细节__exit__接收三个参数异常类型、异常实例、traceback。块内无异常时三者均为None。close()中使用if self._conn is not None判空保证幂等——重复调用也不会出错。示例同时展示了两种用法优先使用with语句需要手动管理时也必须用try/finally包裹绝不能裸写db.connect()后忘记close()。由于返回类型为None隐式返回None所有异常都会被正常传播——这正是默认且推荐的行为。Pattern 2异步上下文管理器对于asyncpg、aiohttp等异步生态的资源需要实现异步协议__aenter__/__aexit__class AsyncDatabasePool: Async database connection pool. def __init__(self, dsn: str, min_size: int 1, max_size: int 10) - None: self._dsn dsn self._min_size min_size self._max_size max_size self._pool: asyncpg.Pool | None None async def __aenter__(self) - AsyncDatabasePool: Create connection pool. self._pool await asyncpg.create_pool( self._dsn, min_sizeself._min_size, max_sizeself._max_size, ) return self async def __aexit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Close all connections in pool. if self._pool is not None: await self._pool.close() async def execute(self, query: str, *args) - list[dict]: Execute query using pooled connection. async with self._pool.acquire() as conn: return await conn.fetch(query, *args) # Usage async with AsyncDatabasePool(dsn) as pool: users await pool.execute(SELECT * FROM users WHERE active $1, True)示例中的min_size/max_size参数控制连接池的规模下限与上限是连接池性能调优的关键参数async with self._pool.acquire() as conn则展示了嵌套的async with——从池中借出的每个连接同样经由上下文管理器归还。Pattern 3使用contextmanager装饰器简化对于不需要维护实例状态的场景装饰器可以大幅减少样板代码。该模式也演示了如何使用structlog结构化日志记录资源生命周期信息from contextlib import contextmanager, asynccontextmanager import time import structlog logger structlog.get_logger() contextmanager def timed_block(name: str): Time a block of code. start time.perf_counter() try: yield finally: elapsed time.perf_counter() - start logger.info(f{name} completed, duration_secondsround(elapsed, 3)) # Usage with timed_block(data_processing): process_large_dataset() asynccontextmanager async def database_transaction(conn: AsyncConnection): Manage database transaction. await conn.execute(BEGIN) try: yield conn await conn.execute(COMMIT) except Exception: await conn.execute(ROLLBACK) raise # Usage async with database_transaction(conn) as tx: await tx.execute(INSERT INTO users ...) await tx.execute(INSERT INTO audit_log ...)这个事务管理示例是资源管理的经典应用正常路径提交COMMIT任何异常路径回滚ROLLBACK并raise重新抛出从而保证事务的原子性。注意这里与异常抑制正好相反——事务场景必须重新抛出异常让调用方感知失败。Pattern 4无条件资源释放清理逻辑必须在__exit__中无条件执行。以下FileProcessor展示了主文件与临时文件的双重清理class FileProcessor: Process file with guaranteed cleanup. def __init__(self, path: str) - None: self._path path self._file: IO | None None self._temp_files: list[Path] [] def __enter__(self) - FileProcessor: self._file open(self._path, r) return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Clean up all resources unconditionally. # Close main file if self._file is not None: self._file.close() # Clean up any temporary files for temp_file in self._temp_files: try: temp_file.unlink() except OSError: pass # Best effort cleanup # Return None/False to propagate any exception两个细节值得学习幂等判空关闭前检查self._file is not None避免重复关闭报错。尽力而为清理临时文件删除使用try/except OSError包裹并pass——清理操作本身不应掩盖原始异常。如果删除失败保留临时文件也比丢失原始错误信息更好。末尾的注释明确提示返回None/False意味着传播所有异常这是默认推荐行为。进阶模式异常抑制、流式处理与多资源编排references/details.md收录了 SKILL.md 中Detailed worked examples所指的进阶内容以## Advanced Patterns开头包含 Pattern 5 至 Pattern 9。Pattern 5选择性异常抑制异常抑制必须精准打击只抑制经过确认、属于预期行为的异常绝不能无差别吞掉错误class StreamWriter: Writer that handles broken pipe gracefully. def __init__(self, stream) - None: self._stream stream def __enter__(self) - StreamWriter: return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - bool: Clean up, suppressing BrokenPipeError on shutdown. self._stream.close() # Suppress BrokenPipeError (client disconnected) # This is expected behavior, not an error if exc_type is BrokenPipeError: return True # Exception suppressed return False # Propagate all other exceptions这是__exit__返回True抑制异常的唯一合理用例之一客户端主动断开导致的BrokenPipeError属于预期行为不应让服务端进程带着错误退出。但除此之外的所有异常都被原样传播。Pattern 6带累积状态的流式处理构建流式响应时常常需要同时维护增量块和累积状态。以下模式用一个dataclass封装累积逻辑from collections.abc import Generator from dataclasses import dataclass, field dataclass class StreamingResult: Accumulated streaming result. chunks: list[str] field(default_factorylist) _finalized: bool False property def content(self) - str: Get accumulated content. return .join(self.chunks) def add_chunk(self, chunk: str) - None: Add chunk to accumulator. if self._finalized: raise RuntimeError(Cannot add to finalized result) self.chunks.append(chunk) def finalize(self) - str: Mark stream complete and return content. self._finalized True return self.content def stream_with_accumulation( response: StreamingResponse, ) - Generator[tuple[str, str], None, str]: Stream response while accumulating content. Yields: Tuple of (accumulated_content, new_chunk) for each chunk. Returns: Final accumulated content. result StreamingResult() for chunk in response.iter_content(): result.add_chunk(chunk) yield result.content, chunk return result.finalize()该生成器是协程式返回值的经典用法每轮yield (累积内容, 新块)供调用方消费流结束后通过生成器return返回最终累积内容由StopIteration.value携带。_finalized标志防止流结束后继续写入。Pattern 7高效的字符串累积流式累积时必须避免字符串拼接造成的 O(n²) 复杂度def accumulate_stream(stream) - str: Efficiently accumulate stream content. # BAD: O(n²) due to string immutability # content # for chunk in stream: # content chunk # Creates new string each time # GOOD: O(n) with list and join chunks: list[str] [] for chunk in stream: chunks.append(chunk) return .join(chunks) # Single allocation原因在于 Python 字符串不可变content chunk每次都会创建全新的字符串对象并拷贝已有内容累积 N 个块的总代价是 O(n²)。改用list.append 单次.join()后复杂度降为 O(n)且join只需一次内存分配。Pattern 8追踪流式指标对生产环境的流式服务时间到首字节time-to-first-byte是衡量用户感知延迟的关键指标import time from collections.abc import Generator def stream_with_metrics( response: StreamingResponse, ) - Generator[str, None, dict]: Stream response while collecting metrics. Yields: Content chunks. Returns: Metrics dictionary. start time.perf_counter() first_chunk_time: float | None None chunk_count 0 total_bytes 0 for chunk in response.iter_content(): if first_chunk_time is None: first_chunk_time time.perf_counter() - start chunk_count 1 total_bytes len(chunk.encode()) yield chunk total_time time.perf_counter() - start return { time_to_first_byte_ms: round((first_chunk_time or 0) * 1000, 2), total_time_ms: round(total_time * 1000, 2), chunk_count: chunk_count, total_bytes: total_bytes, }该生成器在流式消费的同时采集四项指标首字节延迟、总耗时、块数与字节数并通过生成器return以字典形式返回给调用方。这些指标可直接接入 python-observability 技能所述的观测体系。Pattern 9用 ExitStack / AsyncExitStack 管理动态数量的资源当资源数量在运行期才确定时不能写死多层with。ExitStack允许在运行时动态压入任意数量的上下文管理器并在块退出时按后进先出LIFO顺序统一清理from contextlib import ExitStack, AsyncExitStack from pathlib import Path def process_files(paths: list[Path]) - list[str]: Process multiple files with automatic cleanup. results [] with ExitStack() as stack: # Open all files - theyll all be closed when block exits files [stack.enter_context(open(p)) for p in paths] for f in files: results.append(f.read()) return results async def process_connections(hosts: list[str]) - list[dict]: Process multiple async connections. results [] async with AsyncExitStack() as stack: connections [ await stack.enter_async_context(connect_to_host(host)) for host in hosts ] for conn in connections: results.append(await conn.fetch_data()) return results同步侧用stack.enter_context(open(p))压入文件句柄即使中途某个open抛错此前已压入的资源也会全部关闭。异步侧用await stack.enter_async_context(...)压入异步上下文管理器配合async with AsyncExitStack()使用。最佳实践速查10 条将 SKILL.md 与进阶示例结合沉淀出以下十条可直接落地的工程准则始终使用上下文管理器—— 任何需要清理的资源都应走with语句无条件清理——__exit__在异常时也必然执行把清理逻辑放在那里不要意外抑制异常—— 默认返回False只有确认该异常是预期行为时才返回True如 Pattern 5 的BrokenPipeError优先使用contextmanager—— 简单资源用装饰器减少样板代码两种协议都实现—— 让资源同时支持with与手动管理try/finally使用ExitStack—— 应对运行期才能确定的资源数量高效累积—— 用listjoin不要用字符串避免 O(n²)追踪指标—— 流式场景务必关注 time-to-first-bytePattern 8文档化行为—— 尤其是异常抑制逻辑必须在 docstring 中说明测试清理路径—— 编写测试验证资源在出错时依然被正确释放。在仓库中继续探索技能主体plugins/python-development/skills/python-resource-management/SKILL.md进阶示例plugins/python-development/skills/python-resource-management/references/details.md配套技能plugins/python-development/skills/python-resilience/SKILL.md重试、超时与容错与plugins/python-development/skills/python-error-handling异常处理对应 Agentplugins/python-development/agents/python-pro.md将上下文管理器列为现代 Python 核心能力同插件其他技能异步模式、性能优化、观测性等见plugins/python-development/skills目录将这些模式与仓库的 agent/skill 体系结合使用即可在 Claude Code、Codex、Cursor、OpenCode、GitHub Copilot 与 Google Antigravity 等不同 harness 中复用同一套经过验证的资源管理实践。【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Python 资源管理实战指南基于 agents24 仓库 python-resource-management Skill 的上下文管理器、清理与流式处理全解析【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents本篇技术指南以本仓库plugins/python-development/skills/python-resource-management/SKILL.md为核心骨架系统讲解如何利用上下文管理器context manager对数据库连接、文件句柄、网络套接字等资源进行确定性管理。文中覆盖类式协议实现、async 协议、contextmanager装饰器、异常抑制、流式累积、指标追踪与ExitStack多资源编排等 9 大实战模式并引入仓库内references/details.md的进阶示例帮助读者写出即使异常发生也能可靠释放资源的健壮 Python 代码。该 Skill 在仓库中的定位与使用场景python-resource-management是本仓库python-development插件下的一个技能Skill与同目录下的python-resilience重试、超时、熔断、python-error-handling、python-observability等技能互为补充共同构成生产级 Python 开发的技能矩阵。该插件的python-proAgent 将 Context managers and thewithstatement for resource management 列为现代 Python 能力的核心要点之一可见资源管理在整个插件知识体系中的基础地位。从 SKILL.md 的元数据看该技能在以下场景应被主动调用管理数据库连接与连接池database connections and connection pools处理文件句柄与 I/Ofile handles and I/O实现自定义上下文管理器custom context managers构建带状态的流式响应streaming responses with state处理嵌套资源清理nested resource cleanup创建异步上下文管理器async context managers核心思想只有一句话资源数据库连接、文件句柄、网络套接字等应当被确定性地释放即使发生异常也不例外。核心概念上下文管理器协议1.with语句保证自动释放with语句确保资源在代码块退出时自动释放无论块内是正常结束还是抛出了异常。这是 Python 中最基础也最重要的资源管理手段。2. 协议方法同步与异步上下文管理器协议由两组方法构成同步协议__enter__/__exit__异步协议__aenter__/__aexit__与async with配合使用3. 无条件清理__exit__总是执行无论块内是否发生异常。这是确定性释放的根本保证——即使with块内部抛错清理逻辑也一定会运行。4. 异常处理语义__exit__的返回值决定异常的去向返回True抑制suppress异常with块外的代码不会看到该异常返回False/None异常继续向上传播返回值语义是整个异常抑制机制的核心后面会在选择性异常抑制模式中展开。快速开始装饰器一行实现对于简单场景contextmanager装饰器是最快上手的方式from contextlib import contextmanager contextmanager def managed_resource(): resource acquire_resource() try: yield resource finally: resource.cleanup() with managed_resource() as r: r.do_work()关键在于try/finally结构yield之前是进入逻辑yield之后即finally块是退出逻辑。无论with块内是否抛异常finally中的清理代码都会执行。基础模式从类式协议到装饰器Pattern 1类式上下文管理器复杂资源对于需要维护内部状态、生命周期复杂的资源建议实现完整的类式协议。以下是一个数据库连接的完整示例class DatabaseConnection: Database connection with automatic cleanup. def __init__(self, dsn: str) - None: self._dsn dsn self._conn: Connection | None None def connect(self) - None: Establish database connection. self._conn psycopg.connect(self._dsn) def close(self) - None: Close connection if open. if self._conn is not None: self._conn.close() self._conn None def __enter__(self) - DatabaseConnection: Enter context: connect and return self. self.connect() return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Exit context: always close connection. self.close() # Usage with context manager (preferred) with DatabaseConnection(dsn) as db: result db.execute(query) # Manual management when needed db DatabaseConnection(dsn) db.connect() try: result db.execute(query) finally: db.close()值得注意的细节__exit__接收三个参数异常类型、异常实例、traceback。块内无异常时三者均为None。close()中使用if self._conn is not None判空保证幂等——重复调用也不会出错。示例同时展示了两种用法优先使用with语句需要手动管理时也必须用try/finally包裹绝不能裸写db.connect()后忘记close()。由于返回类型为None隐式返回None所有异常都会被正常传播——这正是默认且推荐的行为。Pattern 2异步上下文管理器对于asyncpg、aiohttp等异步生态的资源需要实现异步协议__aenter__/__aexit__class AsyncDatabasePool: Async database connection pool. def __init__(self, dsn: str, min_size: int 1, max_size: int 10) - None: self._dsn dsn self._min_size min_size self._max_size max_size self._pool: asyncpg.Pool | None None async def __aenter__(self) - AsyncDatabasePool: Create connection pool. self._pool await asyncpg.create_pool( self._dsn, min_sizeself._min_size, max_sizeself._max_size, ) return self async def __aexit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Close all connections in pool. if self._pool is not None: await self._pool.close() async def execute(self, query: str, *args) - list[dict]: Execute query using pooled connection. async with self._pool.acquire() as conn: return await conn.fetch(query, *args) # Usage async with AsyncDatabasePool(dsn) as pool: users await pool.execute(SELECT * FROM users WHERE active $1, True)示例中的min_size/max_size参数控制连接池的规模下限与上限是连接池性能调优的关键参数async with self._pool.acquire() as conn则展示了嵌套的async with——从池中借出的每个连接同样经由上下文管理器归还。Pattern 3使用contextmanager装饰器简化对于不需要维护实例状态的场景装饰器可以大幅减少样板代码。该模式也演示了如何使用structlog结构化日志记录资源生命周期信息from contextlib import contextmanager, asynccontextmanager import time import structlog logger structlog.get_logger() contextmanager def timed_block(name: str): Time a block of code. start time.perf_counter() try: yield finally: elapsed time.perf_counter() - start logger.info(f{name} completed, duration_secondsround(elapsed, 3)) # Usage with timed_block(data_processing): process_large_dataset() asynccontextmanager async def database_transaction(conn: AsyncConnection): Manage database transaction. await conn.execute(BEGIN) try: yield conn await conn.execute(COMMIT) except Exception: await conn.execute(ROLLBACK) raise # Usage async with database_transaction(conn) as tx: await tx.execute(INSERT INTO users ...) await tx.execute(INSERT INTO audit_log ...)这个事务管理示例是资源管理的经典应用正常路径提交COMMIT任何异常路径回滚ROLLBACK并raise重新抛出从而保证事务的原子性。注意这里与异常抑制正好相反——事务场景必须重新抛出异常让调用方感知失败。Pattern 4无条件资源释放清理逻辑必须在__exit__中无条件执行。以下FileProcessor展示了主文件与临时文件的双重清理class FileProcessor: Process file with guaranteed cleanup. def __init__(self, path: str) - None: self._path path self._file: IO | None None self._temp_files: list[Path] [] def __enter__(self) - FileProcessor: self._file open(self._path, r) return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - None: Clean up all resources unconditionally. # Close main file if self._file is not None: self._file.close() # Clean up any temporary files for temp_file in self._temp_files: try: temp_file.unlink() except OSError: pass # Best effort cleanup # Return None/False to propagate any exception两个细节值得学习幂等判空关闭前检查self._file is not None避免重复关闭报错。尽力而为清理临时文件删除使用try/except OSError包裹并pass——清理操作本身不应掩盖原始异常。如果删除失败保留临时文件也比丢失原始错误信息更好。末尾的注释明确提示返回None/False意味着传播所有异常这是默认推荐行为。进阶模式异常抑制、流式处理与多资源编排references/details.md收录了 SKILL.md 中Detailed worked examples所指的进阶内容以## Advanced Patterns开头包含 Pattern 5 至 Pattern 9。Pattern 5选择性异常抑制异常抑制必须精准打击只抑制经过确认、属于预期行为的异常绝不能无差别吞掉错误class StreamWriter: Writer that handles broken pipe gracefully. def __init__(self, stream) - None: self._stream stream def __enter__(self) - StreamWriter: return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) - bool: Clean up, suppressing BrokenPipeError on shutdown. self._stream.close() # Suppress BrokenPipeError (client disconnected) # This is expected behavior, not an error if exc_type is BrokenPipeError: return True # Exception suppressed return False # Propagate all other exceptions这是__exit__返回True抑制异常的唯一合理用例之一客户端主动断开导致的BrokenPipeError属于预期行为不应让服务端进程带着错误退出。但除此之外的所有异常都被原样传播。Pattern 6带累积状态的流式处理构建流式响应时常常需要同时维护增量块和累积状态。以下模式用一个dataclass封装累积逻辑from collections.abc import Generator from dataclasses import dataclass, field dataclass class StreamingResult: Accumulated streaming result. chunks: list[str] field(default_factorylist) _finalized: bool False property def content(self) - str: Get accumulated content. return .join(self.chunks) def add_chunk(self, chunk: str) - None: Add chunk to accumulator. if self._finalized: raise RuntimeError(Cannot add to finalized result) self.chunks.append(chunk) def finalize(self) - str: Mark stream complete and return content. self._finalized True return self.content def stream_with_accumulation( response: StreamingResponse, ) - Generator[tuple[str, str], None, str]: Stream response while accumulating content. Yields: Tuple of (accumulated_content, new_chunk) for each chunk. Returns: Final accumulated content. result StreamingResult() for chunk in response.iter_content(): result.add_chunk(chunk) yield result.content, chunk return result.finalize()该生成器是协程式返回值的经典用法每轮yield (累积内容, 新块)供调用方消费流结束后通过生成器return返回最终累积内容由StopIteration.value携带。_finalized标志防止流结束后继续写入。Pattern 7高效的字符串累积流式累积时必须避免字符串拼接造成的 O(n²) 复杂度def accumulate_stream(stream) - str: Efficiently accumulate stream content. # BAD: O(n²) due to string immutability # content # for chunk in stream: # content chunk # Creates new string each time # GOOD: O(n) with list and join chunks: list[str] [] for chunk in stream: chunks.append(chunk) return .join(chunks) # Single allocation原因在于 Python 字符串不可变content chunk每次都会创建全新的字符串对象并拷贝已有内容累积 N 个块的总代价是 O(n²)。改用list.append 单次.join()后复杂度降为 O(n)且join只需一次内存分配。Pattern 8追踪流式指标对生产环境的流式服务时间到首字节time-to-first-byte是衡量用户感知延迟的关键指标import time from collections.abc import Generator def stream_with_metrics( response: StreamingResponse, ) - Generator[str, None, dict]: Stream response while collecting metrics. Yields: Content chunks. Returns: Metrics dictionary. start time.perf_counter() first_chunk_time: float | None None chunk_count 0 total_bytes 0 for chunk in response.iter_content(): if first_chunk_time is None: first_chunk_time time.perf_counter() - start chunk_count 1 total_bytes len(chunk.encode()) yield chunk total_time time.perf_counter() - start return { time_to_first_byte_ms: round((first_chunk_time or 0) * 1000, 2), total_time_ms: round(total_time * 1000, 2), chunk_count: chunk_count, total_bytes: total_bytes, }该生成器在流式消费的同时采集四项指标首字节延迟、总耗时、块数与字节数并通过生成器return以字典形式返回给调用方。这些指标可直接接入 python-observability 技能所述的观测体系。Pattern 9用 ExitStack / AsyncExitStack 管理动态数量的资源当资源数量在运行期才确定时不能写死多层with。ExitStack允许在运行时动态压入任意数量的上下文管理器并在块退出时按后进先出LIFO顺序统一清理from contextlib import ExitStack, AsyncExitStack from pathlib import Path def process_files(paths: list[Path]) - list[str]: Process multiple files with automatic cleanup. results [] with ExitStack() as stack: # Open all files - theyll all be closed when block exits files [stack.enter_context(open(p)) for p in paths] for f in files: results.append(f.read()) return results async def process_connections(hosts: list[str]) - list[dict]: Process multiple async connections. results [] async with AsyncExitStack() as stack: connections [ await stack.enter_async_context(connect_to_host(host)) for host in hosts ] for conn in connections: results.append(await conn.fetch_data()) return results同步侧用stack.enter_context(open(p))压入文件句柄即使中途某个open抛错此前已压入的资源也会全部关闭。异步侧用await stack.enter_async_context(...)压入异步上下文管理器配合async with AsyncExitStack()使用。最佳实践速查10 条将 SKILL.md 与进阶示例结合沉淀出以下十条可直接落地的工程准则始终使用上下文管理器—— 任何需要清理的资源都应走with语句无条件清理——__exit__在异常时也必然执行把清理逻辑放在那里不要意外抑制异常—— 默认返回False只有确认该异常是预期行为时才返回True如 Pattern 5 的BrokenPipeError优先使用contextmanager—— 简单资源用装饰器减少样板代码两种协议都实现—— 让资源同时支持with与手动管理try/finally使用ExitStack—— 应对运行期才能确定的资源数量高效累积—— 用listjoin不要用字符串避免 O(n²)追踪指标—— 流式场景务必关注 time-to-first-bytePattern 8文档化行为—— 尤其是异常抑制逻辑必须在 docstring 中说明测试清理路径—— 编写测试验证资源在出错时依然被正确释放。在仓库中继续探索技能主体plugins/python-development/skills/python-resource-management/SKILL.md进阶示例plugins/python-development/skills/python-resource-management/references/details.md配套技能plugins/python-development/skills/python-resilience/SKILL.md重试、超时与容错与plugins/python-development/skills/python-error-handling异常处理对应 Agentplugins/python-development/agents/python-pro.md将上下文管理器列为现代 Python 核心能力同插件其他技能异步模式、性能优化、观测性等见plugins/python-development/skills目录将这些模式与仓库的 agent/skill 体系结合使用即可在 Claude Code、Codex、Cursor、OpenCode、GitHub Copilot 与 Google Antigravity 等不同 harness 中复用同一套经过验证的资源管理实践。【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
