企业官网建设流程全解析

• 可观测性
• AI 评测
• LLMOps
• AI 应用
• 人工智能

本文是 Phoenix（AI Observability & Evaluation 平台）内部对OpenTelemetry 上下文（contextvars）跨异步边界失效问题的完整技术复盘，聚焦 Playground 的聊天、流式输出与评估（evaluators）场景。读者读完将掌握：async for+ 异常退出时token.reset()抛出"Token was created in a different Context"的 CPython/asyncio 根因、contextlib.aclosing与生成器侧修复两种方案的取舍，以及 Phoenix 在 playground_clients.py 等处的最终设计。

1. 问题陈述：contextvars 跨 async 挂起/恢复会"漂移"

在 Phoenix Playground 的流式链路中，聊天补全、评估器、订阅（subscriptions）都依赖 OpenTelemetry 的start_as_current_span。该 API（以及任何依赖contextvars的 API）会把"当前上下文"存在一个 contextvar 中。一旦发生async 挂起/恢复（例如async for chunk in stream()），这个上下文就可能悄悄改变：

• 生成器侧：异步生成器先cv.set(...)拿到 token，然后yield；当消费者抛异常或提前退出后，生成器的finally可能在一个不同的上下文里执行。此时在finally中调用token.reset()会抛出"Token was created in a different Context"（OpenTelemetry 侧则表现为"Failed to detach context"）。
• 消费者侧：async for遍历生成器的消费者也可能通过 contextvars 挂 span，跨 async 边界时同样存在清理时上下文错位的风险。

映射到 Phoenix 代码：

核心目标有三：用纯 Python 实证这个失败模式；从 CPython 源码与官方文档层面搞清"为什么"；设计出绝不依赖跨边界 contextvars 的架构。

2. 根因：CPython 与 asyncio 的行为

2.1 观察到的现象

异步生成器设置 contextvar 并拿到 token、yield之后，如果消费者在async for期间抛异常，生成器的finally会在另一个执行上下文中运行，此时token.reset()抛出：

ValueError: <Token ...> was created in a different Context

该校验逻辑位于 CPython 的Python/context.c：

// PyContextVar_Reset(): PyContext *ctx = context_get(); // 当前上下文（取自线程状态） if (ctx != tok->tok_ctx) { // token 创建时的上下文（set() 调用时） PyErr_Format(PyExc_ValueError, "%R was created in a different Context", tok); return -1; }

也就是说，生成器的finally执行时，其"当前上下文"与 token 创建时的上下文不是同一个。

2.2 为什么：async for在异常时不会调用aclose()

相关源码：Python/compile.c（compiler_async_for）与Python/ceval.c（END_ASYNC_FOR）。

• async for的结构为：GET_AITER→ 循环体（SETUP_FINALLY、GET_ANEXT、YIELD_FROM即 await、循环体）→except 块（END_ASYNC_FOR）。整个流程中没有任何对iterator.aclose()的调用。
• END_ASYNC_FOR的行为：如果异常是StopAsyncIteration，弹出并继续；对于任何其他异常（包括循环体中的RuntimeError），它只会重新抛出，不会调用iterator.aclose()。于是异常发生后异步生成器保持打开状态，直到被finalize（例如被 GC 回收）才真正关闭。

2.3 生成器实际是如何关闭的：finalizer

Lib/asyncio/base_events.py中注册了异步生成器终结钩子：

def _asyncgen_finalizer_hook(self, agen): self._asyncgens.discard(agen) if not self.is_closed(): self.call_soon_threadsafe(self.create_task, agen.aclose())

事件循环在run_forever()中注册 async-gen 钩子后，异步生成器被销毁时finalizer会通过新建 Task来调度agen.aclose()：

• self.create_task(agen.aclose())创建了一个新Task；
• 每个Task在创建时捕获当时的上下文：self._context = contextvars.copy_context()（Lib/asyncio/tasks.py）；
• 因此执行aclose()的 Task 拿到的是 finalizer 运行时"当前"的上下文（例如某个回调或事件循环默认上下文），并非原消费者 Task 的上下文。

该 Task 运行时通过context.run(callback)执行回调（Lib/asyncio/events.py的Handle._run()），于是生成器代码（包括其finally）跑在了这个新 Task 的上下文里。于是：

• Token 创建于Context A（原消费者 Task 中生成器执行cv.set()时）；
• 生成器的finally运行于Context B（finalizer 为执行aclose()新建的 Task）；
• context_get()返回 B，而tok->tok_ctx是 A → 抛出"created in a different Context"。

一句话总结：消费者异常 →async for不调用aclose()→ 生成器稍后被 finalize → finalizer 执行create_task(agen.aclose())→ 生成器的finally在一个上下文不同的新 Task 中运行。

CPython 参考路径：Python/context.c（PyContextVar_Reset、context_get()）、Python/compile.c（compiler_async_for）、Python/ceval.c（END_ASYNC_FOR）、Lib/asyncio/base_events.py（_asyncgen_finalizer_hook）、Lib/asyncio/tasks.py（Task 的copy_context()）、Lib/asyncio/events.py（Handle._run()）。

3. 官方 Python 文档给出的两条出路

• 异步生成器函数（语言参考）：若异步生成器提前退出（break、调用方取消或其他异常），其异步清理会在意外的上下文中执行——例如在它依赖的 Task 生命周期结束之后，或在事件循环关闭、触发异步生成器 GC 钩子时。调用方必须显式调用aclose()来终结生成器并把它从事件循环中摘除。
• contextlib.aclosing（标准库）：使用async with aclosing(agen):可以保证生成器的异步退出代码在与迭代相同的上下文中执行（异常与 contextvar 行为符合预期，退出代码不会在某个它所依赖的 Task 生命周期结束后才运行）。

由此语言层面给出两种规避方式：

1. 调用方修复：显式关闭生成器，例如async with aclosing(stream): ...；
2. 生成器侧修复：不在其清理代码中依赖 contextvars——在finally里用start_span+span.end()，而不是start_as_current_span。

Phoenix 为流式客户端选择了方案 (2)，这样就不必依赖每个消费者都记得使用aclosing()。

4. 实证测试（纯 Python，不引入 OTel）

位置：internal_docs/vignettes/otel-contextvars-async/contextvars_async_gen_demo.py运行方式（仓库根目录）：

uv run python internal_docs/vignettes/otel-contextvars-async/contextvars_async_gen_demo.py

脚本刻意不引入 pytest，只用 asyncio + assert，以避免测试框架引入干扰变量。脚本先运行若干 print 演示（main()、main_normal()、main_consumer_raises()、main_consumer_with_sleep()），最后执行run_empirical_tests()中的六组断言。

4.1 四个核心场景

关键结论：该失败可复现且与 finalizer 强相关；aclosing在直接消费者使用它时能实证修复；消费者侧的 token 在这些场景中保持有效；问题属于异常/拆除路径专属，并非每次挂起都会发生。

为什么消费者（evaluate）可以继续用start_as_current_span：消费者是 async函数而非 async 生成器。当流抛异常或循环 break 时，异常通过正常的栈展开传播，消费者的with块在同一个 Task里退出，因此它的 contextvar token 在上下文管理器执行__exit__时依然有效。场景 3、4 已实证确认消费者的reset(token)成功。所以 evaluators.py 中的evaluate可以放心使用start_as_current_span，其清理在同一 Task 中运行。

4.2aclosing能为我们做什么

• 调用方侧修复：如果消费者用async with aclosing(stream):包裹再async for chunk in stream: ...，那么 break/raise/cancel 时上下文管理器会 awaitstream.aclose()，生成器的finally在**同一 Task（同一上下文）**中运行，生成器内的 contextvar token 依然有效。
• 收益：① 在同一 Task 中显式、及时清理；② 若日后生成器再次引入上下文相关清理，可保证同上下文；③ 是官方文档记载的标准库模式。
• 为何不依赖它：Phoenix 有多个调用点（evaluators、subscriptions、chat_mutations、playground_clients），要求每个调用方都用aclosing()极易遗漏，且对历史代码或第三方消费者无效。修复生成器本身（清理时不使用 contextvars：finally中用start_span+span.end()）能让流在任何调用方迭代方式下都安全。在自己的调用点再加aclosing作为可选加固。

4.3 两种均有效的修复（澄清）

只要在每个对流执行async for的调用点使用aclosing，并不需要在生成器中放弃start_as_current_span。因为aclosing会让生成器由消费它的同一 Task 关闭，其finally在同一上下文运行，contextvar token 保持有效。两种修复任一即可：

Phoenix 选择生成器侧修复，使正确性不依赖于每个调用方（包括未来代码与第三方代码）都记得使用aclosing。

4.4 为什么只在调用点加aclosing仍不够：生成器链

实践中 Phoenix 曾给所有调用chat_completion_create的位置（subscriptions、chat_mutations、evaluators、playground_clients）都加上aclosing(stream)，但"Token was created in a different Context"依然出现。

原因：流的"消费者"往往本身就是一个async 生成器——例如订阅处理器执行async for chunk in stream: yield chunk把块转发给客户端。于是形成链条：外层生成器 A（订阅）使用async with aclosing(stream B): async for chunk in B: yield chunk。当客户端断开或请求被放弃时，驱动 A 的 Task 可能被取消，A 本身也可能被丢弃。A 只会被finalizer关闭——finalizer 在新 Task中执行create_task(A.aclose())。该 Task 运行 A 的aclose()，其中会退出 A 的async with aclosing(B)并调用B.aclose()。于是 B 是从 finalizer 的 Task 里被关闭的，而非原请求 Task。B 的finally（以及 OTel 的 detach）跑在了错误的上下文中。

aclosing(B)确实执行了，但它是在A 的aclose()内部执行的，而 A 的aclose()运行在 finalizer 的 Task 里。文档所述"生成器 finally 与迭代同 Task"只对直接迭代 B 的代码成立——那段代码是 A 的函数体；当 A 自身被 finalize 时，该函数体已不在原 Task 中运行。

结论：当直接调用点本身是一个可能被 finalizer 关闭的异步生成器时，仅在直接调用点加aclosing不够。此时必须采用生成器侧修复（B 的清理不依赖 contextvars）。

4.5 为什么框架栈（Strawberry、Starlette、ASGI）与此相关

aclosing并非因为 Strawberry、Starlette 或 ASGI 层"破坏"而失效。Phoenix 确实正确地在流 (B) 周围使用了aclosing：对流执行async for chunk in stream的代码是订阅 resolver，它运行了async with aclosing(stream): ...。微妙之处在上一层：

流的"消费者"是订阅 resolver——一个异步生成器 (A)，把块 yield 给客户端。真正迭代该 resolver 的是框架（Strawberry 的订阅传输，构建于 Starlette/ASGI 之上）。客户端断开或请求结束时，该层通常不会对订阅 resolver 调用aclosing，resolver 被丢弃后稍后被 finalize。于是：

• 我们用 aclosing 关闭流 (B)；迭代 B 的代码是 resolver (A)；
• 框架不会用 aclosing 关闭 resolver (A)，而是把它留给 GC 与 finalizer；
• finalizer 在新 Task 中执行A.aclose()，这会运行我们写在aclosing(B).__aexit__里的逻辑，于是 B 在错误的上下文中被关闭。

所以底层栈之所以相关，不是因为它破坏 aclosing，而是因为它定义了谁在迭代订阅，并且不保证该迭代器（resolver）在同一 Task 中被关闭。resolver 常常留给 finalizer，导致流最终从 finalizer 的 Task 里被关闭。因此无论框架如何关闭（或不关闭）订阅，生成器侧修复（B 的清理不使用 contextvars）都是必需的。

5. Phoenix 源码中的落地实现

5.1 生成器侧：chat_completion_create用start_span+finally里的span.end()

在 playground_clients.py 中，chat_completion_create的实现（约第 191–240 行）正是生成器侧修复的直接体现：

# 使用 start_span（而非 start_as_current_span），并在 finally 中 span.end()， # 使生成器绝不挂接 contextvars。可避免生成器在另一 Task 中被关闭时的 # "Failed to detach context" / "Token was created in a different Context"。 span = tracer_.start_span( "ChatCompletion", context=otel_context, attributes=attributes, set_status_on_exception=False, # 状态手动设置 ) try: async for chunk in self._chat_completion_create( messages=messages, tools=tools, response_format=response_format, invocation_parameters=invocation_parameters, span=span, stream_model_output=stream_model_output, ): yield chunk span.set_status(Status(StatusCode.OK)) except Exception as e: span.set_status(Status(StatusCode.ERROR, str(e))) span.record_exception(e) raise finally: span.end()

要点：span 由start_span创建并不设为当前 span，异常时手动设置ERROR状态并record_exception，最终在finally里无条件span.end()。因为span.end()不触碰任何 contextvar，即使生成器的finally在 finalizer 的另一个 Task 上下文中执行也不会出错。源码注释明确记录了这条设计原则及其针对的两个报错。

5.2 消费者侧：evaluators.py保持start_as_current_span

在 evaluators.py 中，评估器大量使用tracer_.start_as_current_span(...)（例如约第 314、327、358、418 行等）。这与 §4.1 的结论一致：evaluate是 async函数而非异步生成器，异常沿栈展开时其with块在同一 Task 中退出，token 始终有效，因此这里保留 contextvars 是安全的。

5.3 订阅链路：subscriptions.py的流式转发

在 subscriptions.py 中，订阅 resolver 约第 105–115 行对llm_client.chat_completion_create(...)执行async for chunk in ...后yield chunk（并在异常时 yield 一个ChatCompletionSubscriptionError）。这正对应 §4.4 的"链式生成器"：resolver 本身是 async 生成器，可能被框架留给 finalizer。此时chat_completion_create内部的生成器侧修复保证流的清理安全，而框架是否对 resolver 调用aclosing已无关紧要。

5.4 应用层测试验证

test_playground_clients.py 中覆盖了客户端行为与 span 属性的断言（如test_text_response_records_expected_attributes、test_authentication_error_records_error_status_on_span等），用于验证生成器侧修复后产生的 trace 符合预期，且不再出现"Failed to detach context"或 token 相关错误。

6. 测试策略

1. 先用纯 Python：只用contextvars+ 异步生成器（不引入 OTel）证明 token 在"挂起 + 异常"后可能失效，产出可泛化的 Python 语言层知识；
2. 映射到 Phoenix 架构：生成器 = 流式 LLM 客户端；消费者 = 评估器或订阅处理器。设计上让生成器绝不使用start_as_current_span，改用start_span+ 在finally中手动span.end()，使清理不触碰 contextvars；
3. 应用层测试：现有测试（如test_playground_clients.py、评估器测试、chat_mutations/subscriptions 全链路带 trace 的测试）验证生成器侧修复能产出预期的 trace，且无 token 错误。

7. 设计决策总结

8. 参考资源

• Phoenix 代码：playground_clients.py、evaluators.py、subscriptions.py、chat_mutations.py
• 通用演示脚本：contextvars_async_gen_demo.py，仓库根目录运行：uv run python internal_docs/vignettes/otel-contextvars-async/contextvars_async_gen_demo.py
• 应用层测试：test_playground_clients.py，以及test_evaluators.py与完整 chat/subscriptions 链路（带 trace）的测试
• CPython 源码路径：Python/context.c、Python/compile.c、Python/ceval.c、Lib/asyncio/base_events.py、Lib/asyncio/tasks.py、Lib/asyncio/events.py
• 官方文档：《异步生成器函数》（语言参考）与《contextlib.aclosing》（标准库）中关于异步生成器提前退出后清理上下文错位的说明

• 可观测性
• AI 评测
• LLMOps
• AI 应用
• 人工智能