openai-agents-python 沙箱 PTY 输出收集机制解析从collect_pty_output到PtyExecUpdate【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python导读本文围绕 docs/ref/sandbox/session/pty_output.md 所指向的agents.sandbox.session.pty_output模块展开深入剖析 openai-agents-python 沙箱体系中 PTY伪终端交互输出的采集、解码、截断与进程生命周期管理实现。读完本文你将掌握collect_pty_output的完整运行循环与参数语义、不完整 UTF-8 序列的跨批次携带策略、输出 token 截断的近似算法以及 Docker / Unix Local 等沙箱后端如何复用这套公共收集逻辑为 Agent 提供稳定的交互式命令行输出。说明docs/ref/下的参考页为面向 API 的自动生成文档pty_output.md仅以 mkdocstrings 的::: agents.sandbox.session.pty_output指令渲染模块签名。其全部技术实质位于 src/agents/sandbox/session/pty_output.py 与配套的 src/agents/sandbox/session/pty_types.py本文即以这两份源码为主体展开。一、PTY 输出收集在沙箱体系中的定位openai-agents-python 的沙箱会话SandboxSession为 Agent 提供两类命令执行能力一次性执行的exec与可交互的 PTY 会话pty_exec_start/pty_write_stdin/pty_terminate_all。交互式 PTY 场景下进程可能长期运行、持续向终端写入输出而 Agent 的每次调用只希望在有限时间窗口内取回一段增量输出同时保证输出字节流可被安全解码为 UTF-8 文本、且不超过上下文窗口的 token 预算。pty_output模块正是这一需求的核心实现。从 src/agents/sandbox/session/sandbox_session.py 可以看到对外接口形态async def pty_exec_start( self, *command: str | Path, timeout: float | None None, shell: bool | list[str] True, user: str | User | None None, tty: bool False, yield_time_s: float | None None, max_output_tokens: int | None None, ) - PtyExecUpdate: ... async def pty_write_stdin( self, *, session_id: int, chars: str, yield_time_s: float | None None, max_output_tokens: int | None None, ) - PtyExecUpdate: ...两者统一返回PtyExecUpdate数据类定义于 src/agents/sandbox/session/pty_types.pydataclass(frozenTrue) class PtyExecUpdate: process_id: int | None output: bytes exit_code: int | None original_token_count: int | Noneprocess_id本次 PTY 会话的分配 ID注意这是沙箱内部逻辑 ID非宿主系统 PIDoutput本次采集到的原始输出字节可能已被截断并附带截断标记exit_code进程退出码进程仍在运行时为Noneoriginal_token_count当输出触发截断时返回截断前的近似 token 总数供上层判断信息丢失量。pty_exec_start的yield_time_s与max_output_tokens参数正是向下传递给pty_output.collect_pty_output的核心控制项。二、collect_pty_output核心采集器的运行循环collect_pty_output是模块的入口函数src/agents/sandbox/session/pty_output.py签名如下async def collect_pty_output( *, output_chunks: deque[bytes], output_lock: asyncio.Lock, output_notify: asyncio.Event, is_done: Callable[[], bool], yield_time_ms: int, max_output_tokens: int | None, poll_output: Callable[[float], Awaitable[None]] | None None, settle_output: Callable[[], Awaitable[None]] | None None, wait_for_output: Callable[[float], Awaitable[None]] | None None, ) - tuple[bytes, int | None, bool]:2.1 参数语义参数类型含义output_chunksdeque[bytes]生产端写入的原始输出字节块队列线程/协程与采集器之间的缓冲区output_lockasyncio.Lock保护队列所有权的互斥锁防止并发读写竞态output_notifyasyncio.Event生产端有新数据时置位的事件驱动采集器及时醒来is_doneCallable[[], bool]判断 PTY 会话是否已结束进程退出yield_time_msint本次采集的时间窗口毫秒到期即返回当前已收集的输出max_output_tokensint \| None输出 token 预算None表示不截断poll_outputCallable[[float], Awaitable[None]] \| None可选的轮询型适配器将“拉取式”后端如 Docker 原始 socket的数据灌入output_chunkssettle_outputCallable[[], Awaitable[None]] \| None可选的收尾钩子在 deadline 前后再执行一次拉取确保边界时刻的字节不丢失wait_for_outputCallable[[float], Awaitable[None]] \| None可选的等待原语替代默认的output_notify.wait()超时等待返回值三元组为(output: bytes, original_token_count: int | None, output_closed: bool)其中output_closed标识本批次结束时进程是否已确定退出决定后续是否对不完整 UTF-8 后缀做“携带”而非“丢弃”。2.2 主循环逻辑函数内部使用time.monotonic()计算deadline now yield_time_ms / 1000然后循环执行到达 deadline 立即退出先查 deadline 再 poll避免超时后多余拉取见 tests/sandbox/test_pty_output.py 中test_collect_pty_output_checks_deadline_before_next_poll对poll_count的断言若提供了poll_output先调用它把最新字节灌入队列再_drain_output_chunks将队列清空到output字节缓冲若is_done()为真标记output_closed True调用settle_output或再次poll_output并做最后一次排空后退出——保证“done 判定瞬间前后”入队的字节都被带走对应测试 test_collect_pty_output_drains_chunks_added_when_done否则以剩余时间为超时等待wait_for_output(remaining_s)默认实现为asyncio.wait_for(output_notify.wait(), timeoutremaining_s)超时则退出循环对应测试 test_collect_pty_output_drains_chunks_queued_when_wait_times_out。退出循环后还有一段“收尾结算”逻辑即使主循环因 deadline 退出也会再调用一次settle_output/poll_output并重新检查is_done()确保进程恰在采集窗口结束时退出时最后一批输出与退出状态都能被正确捕获test_collect_pty_output_waits_for_notification 验证了 notify 驱动路径。2.3 取消安全采集过程中若外层任务被取消例如 Agent 提前中断代码捕获asyncio.CancelledError将已排空的output字节原样appendleft回output_chunks队列头部再向上抛出取消异常。由于队列操作不包含任何await所有权恢复是同步完成的后续采集者仍能看到完整数据test_collect_pty_output_restores_carry_when_next_collection_is_cancelled。三、时序节流yield_time_ms的钳制与空输入策略pty_types.py顶部定义了四组与调度相关的常量src/agents/sandbox/session/pty_types.pyPTY_YIELD_TIME_MS_MIN 250 PTY_EMPTY_YIELD_TIME_MS_MIN 5_000 PTY_YIELD_TIME_MS_MAX 30_000 PTY_PROCESSES_MAX 64 PTY_PROCESSES_WARNING 60 PTY_PROCESSES_PROTECTED_RECENT 8 PTY_PROCESS_ID_MIN 1_000 PTY_PROCESS_ID_MAX_EXCLUSIVE 100_000配套两个工具函数def clamp_pty_yield_time_ms(yield_time_ms: int) - int: return max(PTY_YIELD_TIME_MS_MIN, min(PTY_YIELD_TIME_MS_MAX, yield_time_ms)) def resolve_pty_write_yield_time_ms(*, yield_time_ms: int, input_empty: bool) - int: normalized clamp_pty_yield_time_ms(yield_time_ms) if input_empty: return max(normalized, PTY_EMPTY_YIELD_TIME_MS_MIN) return normalized设计意图最小 250ms / 最大 30s无论调用方传入多小或多大的yield_time_ms都被钳制在合理区间避免高频空轮询拖垮事件循环也避免单次采集阻塞过久空输入加长到 5spty_write_stdin写入空字符串相当于一次“纯读取”操作时若进程仍在运行且无新输出等待窗口自动放大到至少 5 秒减少无谓唤醒tests/sandbox/test_pty_types.py 对input_emptyTrue/False两条路径均有断言Docker 后端的默认窗口在 src/agents/sandbox/sandboxes/docker.py 中pty_exec_start的默认yield_time_sNone会被转换为10_000ms再经clamp_pty_yield_time_ms归一后传给collect_pty_output。四、UTF-8 边界处理不完整多字节序列的“携带”PTY 输出是字节流一个多字节 UTF-8 字符如中文、emoji完全可能被网络包或 socket 读操作切成两段落在两次不同的采集批次里。若每次采集都独立decode就会在字符中间产生替换符。pty_output模块用两个函数解决此问题。4.1_incomplete_utf8_suffix_length该函数pty_output.py从缓冲区尾部向前扫描判断末尾是否存在“只要再来若干字节就能拼成一个合法 UTF-8 标量”的不完整后缀返回应携带的字节数。其判定规则包括检查末尾的 1~3 个连续续字节0x80..0xBF数量根据首字节区间判断期望总长0xC2..0xDF为 2 字节、0xE0..0xEF为 3 字节、0xF0..0xF4为 4 字节对受限前缀做排除0xE0后必须 0xA0拒绝过长的编码、0xED后必须 0x9F拒绝代理区、0xF0后必须 0x90、0xF4后必须 0x8F拒绝超出 U10FFFF纯续字节或 ASCII 结尾返回 0无需携带。参数化测试 tests/sandbox/test_pty_output.py 覆盖了空输入、纯 ASCII、2/3/4 字节前缀以及全部受限组合e0-overlong、ed-surrogate、f0-overlong、f4-out-of-range均返回 0。4.2_drain_and_carry_incomplete_suffix与采集结束语义async def _drain_and_carry_incomplete_suffix( output_chunks: deque[bytes], output_lock: asyncio.Lock, output: bytearray, ) - None: async with output_lock: while output_chunks: output.extend(output_chunks.popleft()) carry _incomplete_utf8_suffix_length(output) if carry: tail bytes(output[-carry:]) del output[-carry:] output_chunks.appendleft(tail)在锁内完成“排空 → 计算携带长度 → 截出尾部 → 放回队列头部”期间不 yield杜绝所有权切换窗口。collect_pty_output在结尾处据此分流pty_output.pyoutput_closed True进程已退出不携带直接把全部字节解码返回——进程结束意味着不会再有后续字节保留后缀只会无限等待output_closed False进程仍在运行携带不完整后缀本次仅返回前面的完整部分。测试 test_collect_pty_output_preserves_valid_utf8_at_every_split 对é2 字节、€3 字节、4 字节的每一种切分点验证第一次采集返回完整前缀且output_closedFalse第二次采集补上剩余字节两次拼接结果与原始字符串的 UTF-8 编码完全一致。而 test_collect_pty_output_replaces_restricted_utf8_prefixes_without_carry 则验证对于无法补全的受限前缀如\xe0\x80会以替换符解码输出而不是无限期携带。五、输出截断与 token 统计当max_output_tokens非None时pty_output.truncate_text_by_tokens委托给 src/agents/sandbox/util/token_truncation.py 的formatted_truncate_text_with_token_countdef formatted_truncate_text_with_token_count( content: str, max_output_tokens: int | None ) - tuple[str, int | None]: if max_output_tokens is None: return content, None policy TruncationPolicy.tokens(max_output_tokens) if _byte_len(content) policy.byte_budget(): return content, None total_lines len(content.splitlines()) prefix fTotal output lines: {total_lines}\n\n truncated _truncate_token_output(content, policy, prefixprefix) return truncated, approx_token_count(content)关键实现事实近似换算系数APPROX_BYTES_PER_TOKEN 4即按每 token 约 4 字节做估算approx_token_count、approx_bytes_for_tokens均基于该系数仅在超预算时截断输出未超预算时original_token_count返回None一旦超预算返回(截断文本, 截断前近似 token 数)collect_pty_output将其原样透出为PtyExecUpdate.original_token_count对称截断 标记_truncate_token_output将预算在头尾间对半分配split_budget保留首尾内容、删除中间并插入形如…N tokens truncated…的标记format_truncation_marker同时前缀附上Total output lines: N提示行让 Agent 明确感知信息缺失量UTF-8 安全切分split_string按字符边界切分字节预算保证截断点不会落在多字节字符中间。这一机制保证了即使交互命令如top、日志 tail持续产出海量输出Agent 拿到的每次增量也始终在 token 预算内且可安全解码。六、PTY 会话与进程生命周期管理pty_types.py还负责 PTY 会话注册表的资源治理def allocate_pty_process_id(used_process_ids: set[int]) - int: while True: process_id random.randrange(PTY_PROCESS_ID_MIN, PTY_PROCESS_ID_MAX_EXCLUSIVE) if process_id not in used_process_ids: return process_idID 分配在[1000, 100000)区间随机分配且避开已占用 IDtest_allocate_pty_process_id_avoids_used_ids上限与告警同时活跃 PTY 会话上限PTY_PROCESSES_MAX 64达到 60 时打印告警日志Docker 后端在 docker.py 中触发淘汰策略process_id_to_prune_from_meta依据(process_id, last_used, exited)元数据做 LRU 风格淘汰——先保护最近使用过的 8 个PTY_PROCESSES_PROTECTED_RECENT优先淘汰“已退出且最久未用”的会话其次淘汰最久未用的存活会话test_process_id_to_prune_from_meta_prefers_exited_unprotected_sessions取消与清理的语义保真_settle_pty_cleanup在清理协程执行期间屏蔽重复取消保证最初的取消原因含 3.11 的取消参数在清理完成后原样向上抛出即使清理本身失败也不会吞掉取消信号test_settle_pty_cleanup_preserves_cancel_reason_when_cleanup_fails。Docker 后端在pty_exec_start中演示了完整生命周期docker.py通过sh -lc包装命令把 shell PID 写入容器内 pid 文件 → 创建exec并打开原始 socket → 启动后台读取线程_pump_pty_socket持续把 socket 字节灌入output_chunks→ 注册进程 ID必要时先淘汰超限会话→ 调用collect_pty_output收集首个输出窗口 →_finalize_pty_update组装PtyExecUpdate。七、后端适配与复用一个收集器、多种沙箱collect_pty_output的适配点设计使其对所有后端透明。poll_output/settle_output/wait_for_output三个可选钩子统一了两种输出模型事件驱动型后端如 Unix Local读取协程拿到数据后output_chunks.append(...)并output_notify.set()采集器通过 notify 等待被唤醒拉取型后端如 Docker 原始 socket 流通过poll_output(deadline)在每次循环迭代主动灌入数据wait_for_output可传自定义等待。除内置的 src/agents/sandbox/sandboxes/docker.py 与unix_local.py外扩展生态中的 E2B、Daytona、Blaxel、Cloudflare、Modal 等沙箱扩展src/agents/extensions/sandbox/下同样引用该模块印证了“队列排空、超时结算、UTF-8 携带、解码”作为公共语义被各后端共享源码注释亦明确写道Queue draining, timeout settlement, UTF-8 carry, and decoding remain shared for every backend。实际使用入口可参考示例 examples/sandbox/unix_local_pty.pydocs/sandbox/guide.md中亦建议“当 PTY 状态重要时用该示例让 Agent 保持在单个交互进程中”以及 docs/sandbox/guide.md 对Shell能力exec_command PTY 支持时的write_stdin的说明。八、质量保障测试矩阵PTY 输出逻辑由两套单元测试全量覆盖tests/sandbox/test_pty_output.py覆盖采集主循环的 notify 唤醒、done 后补排空、超时入队、UTF-8 各切分点携带、受限前缀替换、deadline 前置检查、终态携带只结算一次、取消后队列所有权恢复共 8 组场景tests/sandbox/test_pty_types.py覆盖 yield 时间钳制、空输入加长策略、进程 ID 分配、LRU 淘汰优先级、清理协程的取消语义保持共 5 组场景。这些测试既是行为契约也是阅读源码的最佳索引每个边界条件如\xf4\x90非法前缀、被切成 3 段的极端情形都能在测试参数表中找到对应输入与期望输出。九、小结agents.sandbox.session.pty_output是 openai-agents-python 沙箱交互能力的“输出中枢”以collect_pty_output为核心通过 deadline、notify、settle 三层时序控制实现“取一段、不阻塞、不丢字节”的增量读取通过_incomplete_utf8_suffix_length与“携带”机制保证多字节字符跨批次不产生乱码通过truncate_text_by_tokens与 4 字节/token 的近似换算把输出约束在 token 预算内通过clamp_pty_yield_time_ms、进程 ID 分配与 LRU 淘汰实现资源使用的自保护。理解这套机制有助于你在使用 PTY 交互式沙箱时正确设置yield_time_s与max_output_tokens也为你阅读或扩展自定义沙箱后端提供了清晰的实现范式。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考