自动埋点机制:wrapt 猴补丁如何拦住每一次 LLM 调用

30 秒导读: 你写 client.chat.completions.create(...),没加任何埋点代码,可 OpenLLMetry 却精确记下了这次调用的模型、参数、耗时、token 数。秘密只有一句话:在 import openai 之后, 用 wrapt 把库里的 Completions.create 方法悄悄换成一个自己的函数(wrapper),你以为在调 OpenAI,其实先经过了它。本章讲清这条"偷梁换柱"的链路——从总调度到单个 wrapper 的开 span。

本章接在 01-sdk-bootstrap-pipeline.md(SDK 怎么启动、怎么把 span 导出去)之后:启动流程的最后一步,就是调用本章的主角 init_instrumentations。至于 span 上 具体记了哪些字段、字段怎么命名,是 03-semantic-conventions.md 的事;而用户手动给自己的函数加 span(@workflow/@task),见 04-manual-decorators.md。本章只讲"自动"这一半。

1. 这是什么(零基础也能懂)

一句话定义: 自动埋点(auto-instrumentation)= 不让你改一行业务代码,就能给第三方库的关键函数装上"探针"。

它要解决的问题: 传统埋点要你自己在每个调用点前后写 span = start_span(); ...; span.end() ——几十个调用点就是几十处重复,还容易漏。OpenLLMetry 支持 30 多个库(OpenAI、Anthropic、 LangChain、各种向量库……,完整清单见 Instruments 枚举),不可能要求用户逐个手工插桩。

它怎么用起来: 用户只写一行初始化,之后照常调 SDK,埋点就自动发生了。

# 示意,非源码:用户视角——埋点是"隐形"的
from traceloop.sdk import Traceloop
Traceloop.init(app_name="my-app")          # ← 只有这一行是 OpenLLMetry 的

from openai import OpenAI
client = OpenAI()
client.chat.completions.create(            # ← 照常调,没有任何埋点代码
    model="gpt-4o",
    messages=[{"role": "user", "content": "hi"}],
)
# 但这次调用已经被记成了一个名为 "openai.chat" 的 span

一句话直觉: 把它想成门口的门禁。你要进屋(调 OpenAI),必须先刷卡——但门禁不是你装的, 是 Traceloop.init() 在你 import 完库之后,趁你没注意,把"直接进屋的门"换成了"先刷卡再进屋的门"。你走的还是同一个门框,只是中间多了一道记录。

2. 顶层全景(它大概怎么转)

自动埋点分两个时刻,别混:

时刻	发生什么	谁干的
启动时(一次性)	遍历所有支持的库,凡是装了的就把它的目标函数替换掉	`init_instrumentations` → 各 `init_*_instrumentor` → `wrap_function_wrapper`
每次调用时	被替换后的函数(wrapper)先开 span、记参数,再调真函数,最后记结果、关 span	`chat_wrapper` / `ChatStream` 等

怎么读下面这张图: 上半段是"启动时"的一次性布线,下半段是"每次调用"走的路。关键的一步是中间那条 wrap_function_wrapper(...)——布完线之后,库函数就永久指向 wrapper 了。

【启动时·布线一次】
用户                          OpenLLMetry
────                          ───────────
Traceloop.init()  ─────►  init_instrumentations()          tracing.py:495
                             │ instruments = 全集 - block   (差集)
                             │ for 每个 Instrument:
                             │    init_<lib>_instrumentor()
                             ▼
                          is_package_installed("openai")?   package_check.py:14
                             │ 装了才继续
                             ▼
                          OpenAIInstrumentor().instrument() openai/__init__.py:12
                             │ is_openai_v1()? → 选 V1 分支
                             ▼
                          wrapt.wrap_function_wrapper(       v1/__init__.py:131
                            "openai.resources.chat.completions",
                            "Completions.create", chat_wrapper(...))
                             │
                             ▼
                    ★ 库里的 Completions.create 现在指向 chat_wrapper ★

【每次调用·走 wrapper】
client.chat.completions.create(...)                        ← 实际执行 chat_wrapper
        │
        └─► 开 span → 记请求 → 调真函数 → 记响应 → 关 span   chat_wrappers.py:71

部件一句话职责:

部件	干什么	文件:符号
`Instruments`	支持的库清单(枚举)	`instruments.py:4` `Instruments`
`init_instrumentations`	总调度:遍历清单、算 block/allow 差集、逐个装	`tracing/tracing.py:495`
`init_openai_instrumentor`	单库开关:装了才 import + `instrument()`	`tracing/tracing.py:644`
`OpenAIInstrumentor`	标准 OTel instrumentor;按版本分发	`openai/__init__.py:12`
`OpenAIV1Instrumentor`	真正调 wrapt 替换函数、建 meter	`openai/v1/__init__.py:52`
`chat_wrapper`	每次 chat 调用的探针:开 span、记数据	`openai/shared/chat_wrappers.py:71`
`ChatStream`	流式响应的代理,边吐 chunk 边累积	`openai/shared/chat_wrappers.py:652`

3. 核心原理

3.1 wrapt 猴补丁:偷梁换柱的最小形态

要解决的小问题: 怎么在不动用户代码、甚至不动 openai 库源码的前提下,让 client.chat.completions.create(...) 每次都先经过我?

思路: Python 的方法是可以运行时替换的对象。只要在库被 import 后,把 Completions.create 这个属性指向"我的函数",之后所有调用就都走我的函数了。这类"运行期把别人的函数换成自己的"手法俗称猴补丁(monkeypatch)。OpenLLMetry 用 wrapt 库来做,好处是 wrapt 能保留原函数的签名、self 绑定、__wrapped__ 引用,替换得很干净。

原理演示:

# 示意,非源码:wrapt 猴补丁的最小骨架
import wrapt

def my_wrapper(wrapped, instance, args, kwargs):
    #   wrapped  = 被包住的原函数(这里是真正的 Completions.create)
    #   instance = 方法的 self(那个 Completions 对象)
    #   args/kwargs = 调用时传进来的参数
    print("调用前:开一个 span,记下 messages/model")   # ← 埋点:记请求
    result = wrapped(*args, **kwargs)                    # ← 原封不动调真函数
    print("调用后:记下 usage,关掉 span")               # ← 埋点:记响应
    return result

wrapt.wrap_function_wrapper(
    "openai.resources.chat.completions",  # 目标模块(用"字符串路径",wrapt 会延迟 import)
    "Completions.create",                  # 目标方法,写成 "类.方法"
    my_wrapper,                            # 替换成它
)
# 此后 client.chat.completions.create(...) 实际先跑 my_wrapper

重点看: wrap_function_wrapper 的前两个参数是字符串——模块路径 + 类.方法。传字符串而不是对象,是为了让 wrapt 能在合适时机自己去定位那个方法;这也解释了为什么埋点必须发生在 "库能被 import 到"之后。真正的 wrapper 签名永远是这四件套 (wrapped, instance, args, kwargs), 这是 wrapt 规定的接口。

真实实现: 库里同样一句(节选),把 chat 的 create 换成 chat_wrapper:

# openai/v1/__init__.py:131  OpenAIV1Instrumentor._instrument
wrap_function_wrapper(
    "openai.resources.chat.completions",
    "Completions.create",
    chat_wrapper(tracer, tokens_histogram, choice_counter, ...),
)

同一个 _instrument 方法里,还对 AsyncCompletions.create(异步)、Completions.parse (结构化输出)、Embeddings.create、Images.generate、Responses / Assistants / Realtime 等 一整批方法各来一次 wrap_function_wrapper,见 openai/v1/__init__.py:131-355。可选/新版本才有的方法用 self._try_wrap(openai/v1/__init__.py:56)——它把 AttributeError / ModuleNotFoundError 吞掉,这样旧版本 openai 里不存在的方法不会让整个埋点崩掉。

3.2 总调度 `init_instrumentations`:遍历清单 + block/allow 差集

要解决的小问题: 30 多个库,用户机器上只装了其中几个;还想允许用户"只开这几个"或"屏蔽这几个"。

思路: 维护一个全量清单(Instruments 枚举),启动时:先按用户的 instruments(白名单, 默认=全集)和 block_instruments(黑名单)取差集,再逐个尝试安装;每个库自己再判断"装没装"。

真实实现——先算集合:

# tracing/tracing.py:495  init_instrumentations
block_instruments = block_instruments or set()
# 显式判 None:空集合是"合法的空白名单",不能当成"没传"
instruments = instruments if instruments is not None else set(Instruments)
# 去掉被显式屏蔽的
instruments = instruments - block_instruments

再逐个分发(节选):这是一大段 if/elif,每个分支对应一个库,调用它的 init_*_instrumentor();只要有任意一个成功,instrument_set 就置 True:

# tracing/tracing.py:510  init_instrumentations
for instrument in instruments:
    if instrument == Instruments.OPENAI:
        if init_openai_instrumentor(should_enrich_metrics, base64_image_uploader, use_attributes):
            instrument_set = True
    elif instrument == Instruments.ANTHROPIC:
        if init_anthropic_instrumentor(should_enrich_metrics, base64_image_uploader, use_attributes):
            instrument_set = True
    # … 其余 30+ 个库 …
    else:
        print(Fore.RED + f"Warning: {instrument} instrumentation does not exist.")

关键细节:

若一圈下来 instrument_set 仍为 False(一个库都没装成),会打印一条红色告警提示"没有有效 instrument",见 tracing/tracing.py:632。这是用户最常见的坑:装了 SDK 但没装目标库。
init_instrumentations 的调用方是 SDK 启动流程(见 01-sdk-bootstrap-pipeline.md),在 tracing/tracing.py:171, 把这几样传进来:should_enrich_metrics(要不要多算 token 等富指标)、 image_uploader.aupload_base64_image(把请求里的 base64 图片上传、在 span 里换成 URL 的上传器)、instruments / block_instruments(白/黑名单)、use_attributes(内容记成 span 属性还是 event,详见第 3 章)。

3.3 单库开关 `init_*_instrumentor`:装了才装

要解决的小问题: 用户没装 anthropic,却在清单里遍历到了它——不能因此报错。

思路: 每个 init_*_instrumentor 都是"先探测、再 import、外裹 try"的三段式。以 OpenAI 为例:

# tracing/tracing.py:644  init_openai_instrumentor
def init_openai_instrumentor(should_enrich_metrics, base64_image_uploader, use_attributes=True):
    try:
        if is_package_installed("openai"):                       # ① 装了吗?
            from opentelemetry.instrumentation.openai import OpenAIInstrumentor  # ② 装了才 import
            instrumentor = OpenAIInstrumentor(
                enrich_assistant=should_enrich_metrics,
                get_common_metrics_attributes=metrics_common_attributes,
                upload_base64_image=base64_image_uploader,
                use_attributes=use_attributes,
            )
            if not instrumentor.is_instrumented_by_opentelemetry:  # ③ 没被装过才装,防重复
                instrumentor.instrument()
            return True
    except Exception as e:                                        # ④ 任何异常只记日志,不炸整个进程
        logging.error(f"Error initializing OpenAI instrumentor: {e}")
    return False

三个要点:

is_package_installed("openai")——查 importlib.metadata 列出的已装发行包集合(启动时缓存一份),见 utils/package_check.py:14。没装就跳过,连 import 都不做。
配置通过构造参数注入:should_enrich_metrics、上传器、use_attributes 都在这里塞进 instrumentor。init_anthropic_instrumentor(tracing/tracing.py:668)几乎一模一样,只是参数名是 enrich_token_usage 而非 enrich_assistant——各库按自己的语义命名。
全程 try/except:某个库的埋点初始化失败,只写一条 logging.error,绝不让 Traceloop.init() 抛异常。这是"可观测性不该拖垮业务"的基本原则。

3.4 `OpenAIInstrumentor`:标准 OTel 外壳 + 版本分发

思路: OpenLLMetry 的每个库埋点都是一个标准的 OpenTelemetry BaseInstrumentor 子类——这样它既能被 OpenLLMetry 的 SDK 驱动,也能被任何原生 OTel 生态单独使用。BaseInstrumentor.instrument() 是父类提供的入口,子类只需实现 _instrument。

OpenAIInstrumentor 的 _instrument 只做一件事:按 openai 库的大版本分发。

# openai/__init__.py:57  OpenAIInstrumentor._instrument
def _instrument(self, **kwargs):
    if is_openai_v1():
        from opentelemetry.instrumentation.openai.v1 import OpenAIV1Instrumentor
        OpenAIV1Instrumentor().instrument(**kwargs)
    else:
        from opentelemetry.instrumentation.openai.v0 import OpenAIV0Instrumentor
        OpenAIV0Instrumentor().instrument(**kwargs)

is_openai_v1() 就是把已装的 openai 版本号和 1.0.0 比一下(utils.py:21)——因为 openai 库 1.0 前后 API 结构天差地别(v0 是 openai.ChatCompletion.create,v1 是 openai.resources.chat.completions.Completions.create),wrapt 要替换的目标路径完全不同,所以必须分两套 instrumentor。

构造 OpenAIInstrumentor 时传入的配置,会被写进一个进程级全局 Config (openai/shared/config.py:6):Config.enrich_assistant、Config.upload_base64_image、 Config.use_legacy_attributes 等。wrapper 运行时从这个全局读配置,而不是层层透传参数。

3.5 `chat_wrapper`:一次调用里,探针到底做了什么

这是自动埋点"每次调用"那一半的核心。先解决一个工程细节:wrapt 只给 wrapper 四个参数 (wrapped, instance, args, kwargs),可 chat_wrapper 还需要 tracer、一堆 meter,怎么办?

答案是柯里化(currying):chat_wrapper 被 @_with_chat_telemetry_wrapper 装饰 (utils.py:86),这个装饰器把它变成一个"工厂"——先用 tracer/meter 调一次,得到一个真正符合 wrapt 四参签名的内层 wrapper,再交给 wrap_function_wrapper。

# 示意,非源码:柯里化如何把"多参函数"变成"wrapt 四参 wrapper"
def _with_chat_telemetry_wrapper(func):
    def factory(tracer, token_counter, ...):        # 先收 tracer/meter
        def wrapper(wrapped, instance, args, kwargs):  # 这才是 wrapt 要的签名
            return func(tracer, token_counter, ..., wrapped, instance, args, kwargs)
        return wrapper
    return factory
# 所以 wrap_function_wrapper(..., chat_wrapper(tracer, meters...)) 传进去的是内层 wrapper

真实的 wrapper 主干(chat_wrappers.py:71),按顺序做这几件事:

抑制开关:若上下文里设了 _SUPPRESS_INSTRUMENTATION_KEY 等,直接 return wrapped(*args, **kwargs) 跳过埋点(chat_wrappers.py:85)——用于避免嵌套/重复埋点。

手动开 span(不是 with,因为流式响应要等生成器耗尽才能关):

# chat_wrappers.py:91
span = tracer.start_span(
    SPAN_NAME,                     # "openai.chat"
    kind=SpanKind.CLIENT,
    attributes={GenAIAttributes.GEN_AI_OPERATION_NAME: OPERATION_NAME.value},
)

抓请求参数:_handle_request(span, kwargs, instance)(chat_wrappers.py:271)把 model、temperature、max_tokens、messages 等从 kwargs 里读出来写进 span。底层靠 _set_request_attributes(shared/__init__.py:137)+ _set_span_attribute(空值自动跳过, shared/__init__.py:35)。具体记哪些字段、叫什么名,是第 3 章的内容。
调真函数并计时:
```
# chat_wrappers.py:100
start_time = time.time()
response = wrapped(*args, **kwargs)   # ← 真正打到 OpenAI
end_time = time.time()
```
若这里抛异常:记 error.type、span.record_exception(e)、置 ERROR 状态、span.end(),再把异常原样 re-raise(chat_wrappers.py:104-122)——埋点不吞业务异常。
分流式 / 非流式:
- 非流式:_handle_response(...) 写 usage、choices、reasoning tokens,然后 span.end() 返回原始 response(chat_wrappers.py:155)。
- 流式(is_streaming_response(response) 为真,shared/__init__.py:384):不关 span, 而是把响应包进 ChatStream 代理返回(chat_wrappers.py:127)。

异步版 achat_wrapper(chat_wrappers.py:171)结构完全对称,只是 await wrapped(...)。

注:模块里另有一个 shared/span_utils.py,当前是空文件(占位)。真正设置 span 属性的辅助函数都在 shared/__init__.py(_set_request_attributes / _set_response_attributes / _set_span_attribute 等),别被文件名误导。

3.6 流式的难点:`ChatStream` 怎么边吐边记

要解决的小问题: 流式响应是个生成器,内容一块块(chunk)吐出来,调用返回那一刻根本还没有完整结果,也没有 token 统计。可 span 又不能一直开着不关。怎么办?

思路: 用一个透明代理把原始 stream 包起来——用户照常 for chunk in stream,但每取一块, 代理就顺手做三件事:发一个 chunk 事件、(首块时)记 time-to-first-token、把这块累加进 "完整响应";等生成器耗尽(StopIteration),再一次性写结果、关 span。

ChatStream 继承 wrapt.ObjectProxy(chat_wrappers.py:652),所以它"长得就像"原 stream。核心在 __next__:

# chat_wrappers.py:734  ChatStream.__next__
def __next__(self):
    try:
        chunk = self.__wrapped__.__next__()      # 从真 stream 取下一块
    except Exception as e:
        if isinstance(e, StopIteration):
            self._process_complete_response()     # 吐完了 → 结算并关 span
        else:
            self._ensure_cleanup()                # 出错也要收尾、关 span
            ...
        raise
    else:
        self._process_item(chunk)                 # 处理这一块
        return chunk                              # 原样交还给用户

_process_item(chat_wrappers.py:766)干两件关键事:

给 span 加一个 GEN_AI_CONTENT_COMPLETION_CHUNK 事件;首块时记录 time-to-first-token(首 token 延迟 = 现在 − 请求发出时刻),写进 streaming_time_to_first_token 直方图——这是衡量"模型多久开始回话"的关键指标。
_accumulate_stream_items(item, self._complete_response)(chat_wrappers.py:1183):把这块 delta 累加进 complete_response——拼接 content 文本、逐段拼 tool_calls 的 name/arguments、从最后几块里捞 usage。等于在客户端把流式碎片重新拼回一条完整消息。

生成器耗尽时,_process_complete_response(chat_wrappers.py:790)结算 token/choice/duration 指标、记 time-to-generate(首 token 到结束的时长)、把拼好的响应写进 span、span.end()。此外还有 __del__ / __exit__ / _ensure_cleanup 兜底(chat_wrappers.py:697、836),保证即便用户没读完流、或对象被 GC,span 也一定会被关掉、指标尽量补记——绝不泄漏一个开着的 span。

4. 巧妙之处(可借鉴的技术)

巧在哪	说明	位置
传字符串路径给 wrapt	`"openai.resources.chat.completions"` 让 wrapt 延迟定位,无需先持有对象	`v1/__init__.py:131` `wrap_function_wrapper`
`_try_wrap` 吞 Attribute/ModuleNotFound	新版本才有的方法(Responses/Realtime)在旧库上缺失也不炸	`v1/__init__.py:56` `_try_wrap`
柯里化适配 wrapt 四参签名	用装饰器把"要 tracer/meter 的函数"变成合法 wrapper	`utils.py:86` `_with_chat_telemetry_wrapper`
手动 start_span 而非 with	流式响应寿命超出函数返回,span 必须能"跨返回"存活	`chat_wrappers.py:91`
ObjectProxy 做透明流代理	用户 API 完全不变,埋点却嵌在 `__next__` 里	`chat_wrappers.py:652` `ChatStream`
多重收尾兜底	`__del__`/`__exit__`/`_ensure_cleanup` + 线程锁,保证 span 必关	`chat_wrappers.py:836` `_ensure_cleanup`
埋点异常不外泄	`init_*` 与 `dont_throw` 装饰器都只记日志	`utils.py:132` `dont_throw`

5. 边界与局限(诚实)

必须在库被 import、且方法未被调用前完成布线。 猴补丁替换的是"类上的方法";极端场景下若某处已提前抓住了原函数引用,替换未必生效。常规用法(先 Traceloop.init() 再 import/用 openai)没问题。
强依赖上游 API 形状。 目标路径(openai.resources.chat.completions、Completions.create) 是硬编码字符串;OpenAI 库大改结构就得跟着改——这正是要 v0/v1 两套、并用 _try_wrap 容错的原因。
本章不覆盖的部分: span 上字段的命名与取舍(GenAI 语义约定)在 03-semantic-conventions.md;用户手写的 workflow/task/agent/tool 装饰器与上下文传播在 04-manual-decorators.md;启动、导出管道在 01-sdk-bootstrap-pipeline.md。全景与阅读顺序见 index.md。

6. 代码地图(导航索引)

主题	文件路径	符号名
支持的库清单(枚举)	`packages/traceloop-sdk/traceloop/sdk/instruments.py`	`Instruments`
总调度:遍历+差集+分发	`packages/traceloop-sdk/traceloop/sdk/tracing/tracing.py:495`	`init_instrumentations`
调用总调度的地方(启动流程)	`packages/traceloop-sdk/traceloop/sdk/tracing/tracing.py:171`	`init_instrumentations(...)`
单库开关(OpenAI)	`packages/traceloop-sdk/traceloop/sdk/tracing/tracing.py:644`	`init_openai_instrumentor`
单库开关(Anthropic)	`packages/traceloop-sdk/traceloop/sdk/tracing/tracing.py:668`	`init_anthropic_instrumentor`
包是否已安装	`packages/traceloop-sdk/traceloop/sdk/utils/package_check.py:14`	`is_package_installed`
标准 instrumentor + 版本分发	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/__init__.py:12`	`OpenAIInstrumentor`
版本判定	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py:21`	`is_openai_v1`
v1 埋点:wrapt 替换 + 建 meter	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py:72`	`OpenAIV1Instrumentor._instrument`
可选方法容错替换	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py:56`	`OpenAIV1Instrumentor._try_wrap`
chat 探针(每次调用)	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py:71`	`chat_wrapper`
柯里化适配器	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py:86`	`_with_chat_telemetry_wrapper`
写请求属性	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/__init__.py:137`	`_set_request_attributes`
是否流式	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/__init__.py:384`	`is_streaming_response`
流式代理	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py:652`	`ChatStream`
流式碎片累加	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py:1183`	`_accumulate_stream_items`
全局配置载体	`packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/config.py:6`	`Config`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

3. 核心原理​

3.1 wrapt 猴补丁:偷梁换柱的最小形态​

3.2 总调度 init_instrumentations:遍历清单 + block/allow 差集​

3.3 单库开关 init_*_instrumentor:装了才装​

3.4 OpenAIInstrumentor:标准 OTel 外壳 + 版本分发​

3.5 chat_wrapper:一次调用里,探针到底做了什么​

3.6 流式的难点:ChatStream 怎么边吐边记​

4. 巧妙之处(可借鉴的技术)​

5. 边界与局限(诚实)​

6. 代码地图(导航索引)​