数据截至 2026-08-18(上游 commit 36c7a7f6eca6)

自动埋点：一行 autolog 怎么把第三方 SDK 就地改造成埋点客户端

30 秒导读： 你写 mlflow.openai.autolog()，之后 client.chat.completions.create(...) 就自动出 trace——业务代码一行没改。这件事的实现是猴子补丁（monkey patch，运行期把别人的方法替换掉）：MLflow 把 SDK 类上的方法换成自己的"夹心函数"，夹心层负责开 span、调原函数、关 span。本章只讲补丁怎么打、怎么保证不炸用户程序、各家 SDK 怎么接；span 的数据结构见 01，span 出生之后怎么流到后端见 02。

1. 这章要解决的问题（零基础也能懂）

用起来什么样。 全部的用户侧代码就这么多：

import mlflow
from openai import OpenAI

mlflow.openai.autolog()  # 就这一行

resp = OpenAI().chat.completions.create(  # 这次调用会自动变成一棵 trace
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "hi"}],
)

难在哪。 你要拦的是别人家的代码：openai 这个包不是你写的，你没法在它的 create() 里加一句 start_span()。你能碰到的只有一个东西——Python 里类的方法就是类字典上的一个属性，可以被替换。

于是核心动作只有一个： 把 Completions.create 这个属性，换成一个长得一模一样、但里外多包了一层的新函数。

补丁前:  用户 ──→ Completions.create ──→ HTTP 请求

补丁后:  用户 ──→ safe_patch_function ──→ patched_call ──→ 原 create ──→ HTTP 请求
                        │                    │                 │
                        │                    ├─ 前：开 span     │
                        │                    └─ 后：关 span ←───┘
                        └─ 兜底：埋点炸了不许影响上面那条主线

真正的工程量在"兜底"那一行。 埋点是附加功能，用户的业务是主线。补丁代码自己出 bug 时，绝对不能把用户的 LLM 调用一起搞挂——这条铁律撑起了 safe_patch 一千多行里的绝大部分复杂度。

一句话直觉： 把 safe_patch 想成给别人的函数套一层保险丝。保险丝烧了（埋点出错）灯还得亮（原函数照常返回）；灯自己坏了（原函数抛异常）保险丝不背锅，异常原样抛给用户。

2. 顶层全景：三层结构

先看整体。从上到下三层，每层职责清楚：

第 1 层  开关与配置
  mlflow.autolog()  ─── post-import hook ──→ mlflow.<flavor>.autolog()
                                                     │
                                          @autologging_integration
                                                     │  把参数存进
                                                     ▼
                                        AUTOLOGGING_INTEGRATIONS  (全局 dict)
                                                     │
─────────────────────────────────────────────────────┼───────────────────
第 2 层  打补丁框架                                     │ 每次调用都回来读
  safe_patch(flavor, 目标类, 方法名, 补丁函数)           │ disable / silent / log_traces
        │                                             │
        ├─ gorilla.apply  把方法替换掉（可撤销）          │
        └─ 生成 safe_patch_function（异常隔离夹心层）─────┘
                                    │ 调用
─────────────────────────────────────┼──────────────────────────────────
第 3 层  各家集成的补丁函数              ▼
  函数式：patched_call（openai/anthropic/gemini/bedrock…）
  回调式：补一下构造函数，把 MlflowLangchainTracer 塞进人家的 callback 列表
                                    │ 产出
                                    ▼
                              LiveSpan → 见 02 章的管道

怎么读这张图： 竖着看是"配置 → 框架 → 集成"三层；第 1 层只在你调 autolog() 时跑一次，第 2、3 层在每次业务调用时跑。

各部件职责：

部件	干什么	在哪个文件
`mlflow.autolog()`	总开关，给所有 flavor 注册 import 钩子	`mlflow/tracking/fluent.py:3483`
`@autologging_integration`	每个 flavor 的 `autolog()` 都要戴的装饰器，负责存配置、加锁、撤旧补丁	`mlflow/utils/autologging_utils/__init__.py:385`
`AUTOLOGGING_INTEGRATIONS`	全局配置字典：flavor 名 → 该次 `autolog()` 的全部参数	`mlflow/utils/autologging_utils/__init__.py:62`
`safe_patch`	打补丁的唯一入口，产出异常隔离的夹心函数	`mlflow/utils/autologging_utils/safety.py:231`
`gorilla`	底层替换/还原属性的工具（MLflow vendored 的第三方库）	`mlflow/utils/gorilla.py:263`（`apply`）
补丁函数	每家集成自己写的 `patched_call`，真正开关 span	如 `mlflow/openai/autolog.py:257`

3. 通用打补丁框架：`safe_patch`

这一节是本章的核心。所有 flavor 都只通过 safe_patch 打补丁，没有例外。

3.1 它要解决的小问题

直接 Completions.create = my_wrapper 有三个致命缺陷：

撤不回来——原函数丢了，autolog(disable=True) 没法恢复。
不安全——包装层里任何一个 bug 都会顺着调用栈抛给用户，把人家的推理请求搞挂。
不像原函数——签名、docstring 全变了；很多框架（还有用户代码）会 inspect.signature() 原函数，一变就出错。

safe_patch 就是逐条解决这三点。

3.2 gorilla：可撤销的替换

MLflow 把 gorilla 这个小库 vendor 进了 mlflow/utils/gorilla.py，只用它的三个能力：

能力	符号	做法
打补丁	`gorilla.apply`（`mlflow/utils/gorilla.py:263`）	先把原属性另存为 `_gorilla_original_<name>`，再 `setattr` 新的
找原件	`gorilla.get_original_attribute`（`:562`）	沿 MRO 往上找，*优先找 `_gorilla_original_`**，保证多层继承里不会拿错
撤补丁	`gorilla.revert`（`:329`）	把 `_gorilla_original_<name>` 放回去（就地补丁）或直接删掉（继承来的方法）

命名模式定义在 mlflow/utils/gorilla.py:62（_ORIGINAL_NAME = "_gorilla_original_%s"）和 :65（_ACTIVE_PATCH）。safe_patch 固定用 allow_hit=True, store_hit=True 两个开关（safety.py:817），意思就是"允许覆盖已有属性，且必须把原件存下来"——store_hit 是可撤销的前提。

3.3 准备阶段：三种被补对象要分开处理

safe_patch（safety.py:231）在真正替换之前，先要搞清楚"我补的到底是个什么东西"。三种情况分三条路：

被补对象	判定	特殊处理
普通方法	默认	`original = original_fn`，直接用
`@property` 方法	`isinstance(original_fn, property)`（`safety.py:290`）	用 `original_fn.fget(self)` 拿到被 property 代理出来的那个函数，再调它（`safety.py:308-313`）
async 方法	`inspect.iscoroutinefunction(patch_function)`（`safety.py:268`）	走整套 `async` 复制版夹心层（`safety.py:524`）

property 这条路为什么麻烦。 A.f1 是 property 时，a1.f1 返回的是另一个函数，用户实际调的是 a1.f1(...)。所以补丁必须也做成 property：get_bound_safe_patch_fn（safety.py:681）先调一次 original_fn.fget(self) 做可用性检查（这样 hasattr 行为不变），再返回绑定好的夹心函数。property + async 的组合直接拒绝（safety.py:292 抛 MlflowException）。

async 判定有个反直觉的点（值得记住）： 判断依据是补丁函数是不是协程函数，而不是原函数。源码在 safety.py:265-268 特意留了注释解释——LangChain 那边存在"用同步补丁函数去补异步原函数"的既有用法，改成看原函数会破坏它。

签名伪装则由 update_wrapper_extended（safety.py:785）负责：它比 functools.update_wrapper 多做一件事——把 __signature__ 也抄过来，并且抄失败时只 debug 一句、不抛错（safety.py:798-801，注释里点名 TensorFlow 的 export_savedmodel 拒绝签名反射）。

3.4 夹心层内部：异常隔离怎么做到

safe_patch_function（safety.py:319）是被真正装到类上的那个函数。它的开头有一句全大写的警告注释（safety.py:332-334），大意是"这函数在 Databricks 运行时默认启用、处在关键路径上，改它出 bug 会直接搞挂用户的活儿"——可见这段代码的敏感度。

按执行顺序，它做五件事：

第一步：读配置决定要不要干活。 命中下面任一条件，就直接 return original(*args, **kwargs)，一个 span 都不产（safety.py:385-403）：

短路条件	含义
`active_session_failed`	当前 session 里已经有补丁炸过了，后续嵌套调用一律不再尝试
`autologging_is_disabled(flavor)`	用户传了 `disable=True`，或版本不兼容且开了 `disable_for_unsupported_versions`
`user_created_fluent_run_is_active and exclusive`	用户自己开了 run，且该集成是独占模式
`_AUTOLOGGING_GLOBALLY_DISABLED`	进程内被 `disable_autologging()` 上下文管理器全局关掉了

第二步：开一个 AutologgingSession（safety.py:419），见 3.5。

第三步：把"原函数"包装成一个带哨兵的 call_original（safety.py:439-475）。这是整套异常隔离的机关所在：

# 示意，非源码：夹心层用三个 nonlocal 变量当哨兵
original_has_been_called = False   # 补丁到底调没调原函数
original_result = None             # 原函数返回了什么
failed_during_original = False     # 异常是不是从原函数里冒出来的

def call_original(*a, **kw):
    nonlocal original_has_been_called, original_result
    try:
        original_has_been_called = True
        original_result = original(*a, **kw)   # 真正的 openai create()
        return original_result
    except Exception:
        nonlocal failed_during_original
        failed_during_original = True         # 打上"这是用户侧的锅"的标记
        raise

重点看 failed_during_original 这个标记：它是后面判断异常归属的唯一依据。真源码里这段拆成了 call_original_fn_with_event_logging（safety.py:422）和内层 _original_fn（safety.py:440）两层，多出来的部分是事件日志和测试期校验。

第四步：调用户写的补丁函数（safety.py:484）：patch_function(call_original, *args, **kwargs)。补丁函数的第一个参数永远是 original——这是所有集成必须遵守的约定。

第五步：按标记分流异常。 这就是那条铁律的落地（safety.py:489-522）：

patch_function 抛异常了
        │
        ├─ failed_during_original == True ?
        │      是 → raise（原封不动抛给用户；这是用户代码/网络/API 的错）
        │      否 → 吞掉，记进 patch_error，最后 _logger.warning 一句
        │
        └─ 之后无论如何都要给用户一个返回值：
               original_has_been_called ? 返回 original_result
                                        : 现在补调一次 original(*args, **kwargs)

最后那个 "补调" 分支（safety.py:508-512）特别值得注意：补丁函数的返回值是被丢弃的。用户拿到的永远是 original_result；如果补丁函数在调用原函数之前就炸了，夹心层会自己再去调一次原函数，保证用户的调用一定被执行。

一张表总结四种情形：

情形	用户看到什么	日志
一切正常	原函数返回值	无
原函数抛异常	同一个异常原样抛出	不记 patch error（认定是用户侧问题）
补丁开 span 时炸了（原函数还没调）	夹心层补调原函数，返回正常结果	`warning: Encountered unexpected error during {flavor} autologging`
补丁关 span 时炸了（原函数已调）	缓存的 `original_result`	同上

is_testing()（safety.py:198，看 MLFLOW_AUTOLOGGING_TESTING 环境变量）会推翻上面第 3、4 行：测试模式下补丁异常一律重抛，好让 CI 发现埋点自己的 bug。

async 版本是手抄的。 async_safe_patch_function（safety.py:524）逐行复制了同步版，只把上下文管理器和调用换成 async with / await。源码注释直说了这是刻意为之（safety.py:530-536）：宁可重复，也不敢为了去重而重构这段关键路径。

3.5 AutologgingSession：一次调用的"埋点作用域"

AutologgingSession（safety.py:737）就三个字段：integration、id（uuid）、state（running / succeeded / failed）。

_AutologgingSessionManager（safety.py:744）管理它，规则很简单：

start_session 是个 contextmanager（safety.py:749）：只有当前没有 session 时才新建；已有就复用。
退出时只有创建者才关闭（safety.py:760）——嵌套的内层调用不会把外层的 session 提前关掉。

这样一次顶层调用（比如 chain.invoke() 内部又触发了三次 openai.create）共享同一个 session。它的两个用途：

区分 run 归属：active_run() and not _AutologgingSessionManager.active_session() 用来判断"当前这个 run 是用户自己开的，还是 autolog 开的"（safety.py:377-379）。
失败熔断：某次补丁失败会把 session.state 置为 failed（safety.py:490），同 session 内后续调用直接跳过埋点，避免同一个 bug 刷屏。

一个要知道的边界： _session 是普通类属性（safety.py:745），不是 threading.local，也不是 ContextVar——它是进程级共享的。

3.6 撤补丁：`revert_patches`

打过的补丁全登记在 _AUTOLOGGING_PATCHES 这个 dict 里（safety.py:23，写入在 _store_patch，safety.py:824）。撤销时（revert_patches，safety.py:709）做两件事：

逐个 gorilla.revert(patch)，把原方法放回去。
跑一遍 _AUTOLOGGING_CLEANUP_CALLBACKS 里登记的清理回调（safety.py:723-728），且每个回调单独 try/except，一个失败不影响其他。

第 2 条是给"不是靠补丁接进来"的集成留的后门。典型用户是 OTel 集成：mlflow/otel/__init__.py:138 往里塞了一个 teardown_otel_processor，因为 span processor 不是补丁，gorilla.revert 管不着它。

3.7 `ExceptionSafe*`：传给别人的对象也得有保险丝

补丁常常要往原函数里塞新参数——最典型的就是往 LangChain 传一个 callback handler。这些对象后续会被第三方框架在它自己的流程里调用，一旦回调里抛异常，同样会搞挂主流程。

于是有一组"包一层 try/except"的工具：

工具	作用	位置
`exception_safe_function_for_class`	把单个函数包成吞异常版	`safety.py:36`
`picklable_exception_safe_function`	同上，但用 `functools.partial` 保持可 pickle	`safety.py:70`
`ExceptionSafeClass`	元类：把类上所有可调用属性挨个包一遍	`safety.py:111`
`ExceptionSafeAbstractClass`	同上，基于 `abc.ABCMeta`，给抽象基类的子类用	`safety.py:131`

为什么要两个元类？safety.py:113-130 的注释写得很清楚：ExceptionSafeClass 碰上 abc.ABC 的子类会报 TypeError: metaclass conflict，所以另造一个基于 ABCMeta 的版本。MlflowLangchainTracer 用的正是后者。

测试模式下还有一道强制检查：_validate_args（safety.py:1008）会断言"补丁塞给原函数的每一个新参数，要么是 exception-safe 函数，要么是 ExceptionSafeClass 的实例"（safety.py:1045-1060）。豁免名单 _VALIDATION_EXEMPT_ARGUMENTS（safety.py:918）很短，且每条都写了理由——比如 openai/anthropic 的 extra_headers 因为要注入 trace 头而必然与用户入参不同（safety.py:930-940）。

4. 配置注册表：一个全局 dict 管住所有开关

补丁已经装好了，但它每次执行都要知道"现在还该不该埋点"。这个信息来自一张全局表。

4.1 `AUTOLOGGING_INTEGRATIONS`

结构极简：{flavor 名: 该次 autolog() 的全部参数}（mlflow/utils/autologging_utils/__init__.py:62）。改它要拿 _autolog_conf_global_lock 这把可重入锁（:66），封装成装饰器 autologging_conf_lock（:71）。

读表有两个口子：

get_autologging_config(flavor, key, default)（:482）——单个配置项。
AutoLoggingConfig.init(flavor)（mlflow/utils/autologging_utils/config.py:23）——一次性取出 log_traces / log_models 等常用项打成 dataclass，补丁函数里更常用这个。

判断开关状态用 autologging_is_disabled(flavor)（:501），两条判定：显式 disable=True，或者"开了 disable_for_unsupported_versions 且当前 SDK 版本不在支持区间"。

4.2 `@autologging_integration` 装饰器

每个 flavor 的 autolog() 都被它包一层（:385）。它做的事按顺序：

autolog(*args, **kwargs) 被调用
   │
   ├─ 1. 拿全局锁（autologging_conf_lock）
   ├─ 2. 把「默认值 + 位置参数 + 关键字参数」合并，整个存进 AUTOLOGGING_INTEGRATIONS[name]   (:416-420)
   ├─ 3. revert_patches(name)  ← 先把上一次的补丁全撤掉，避免叠补丁          (:431)
   ├─ 4. disable=True 且 name != "mlflow" ? → return（函数体根本不执行！）    (:436)
   └─ 5. 设好 warning 行为，检查版本兼容，最后才调真正的 _autolog 函数体       (:463-465)

它还在入口处强制校验签名（validate_param_spec，:394）：autolog() 必须有 disable 和 silent 两个参数、且默认值都是 False，否则直接抛异常。这是一条编译期（导入期）契约——保证所有集成都能被统一开关。

第 4 步是一个必须知道的坑。 disable=True 时装饰器直接 return，被装饰的函数体一行都不跑。所以任何"关闭时要做的清理"都不能写在函数体里。各家的解法是拆成两个函数：对外的 autolog() 不带装饰器、负责清理，内部的 _autolog() 带装饰器、负责打补丁。

# 示意，非源码：这个拆法在多个 flavor 里反复出现
def autolog(disable=False, log_traces=True, ...):
    _autolog(disable=disable, log_traces=log_traces, ...)  # 存配置 + 打补丁
    if disable or not log_traces:
        remove_tracer()   # 关闭时的清理，必须写在装饰器外面
    else:
        set_tracer()

autolog.integration_name = FLAVOR_NAME   # mlflow.autolog() 靠这个属性认人


@autologging_integration(FLAVOR_NAME)
def _autolog(disable=False, log_traces=True, ...):
    ...

真实例子：mlflow/openai/autolog.py:33（对外）+ :120（内部）、mlflow/llama_index/autolog.py:7 + :47、mlflow/litellm/__init__.py:13 + :72、mlflow/agno/__init__.py:13 + :106。openai 那边的注释还留了一句 TODO，承认这个模式"不一致"、想找个统一解法（mlflow/openai/autolog.py:66-67）。

另一个副产品是 wrapped_autolog.integration_name = name（:471）——mlflow.autolog() 需要不导入模块就知道 flavor 名。

4.3 `mlflow.autolog()` 总开关：不导入、只挂钩

mlflow.autolog()（mlflow/tracking/fluent.py:3483）不会去 import 二十个 SDK——那样又慢又容易炸。它的做法是给每个候选库注册一个 post-import hook（fluent.py:3728-3729）：谁被 import 了，谁的 autolog() 才被调起来；而且钩子对已经导入的库会追溯生效。

候选库分两张表：

表	内容	位置
`LIBRARY_TO_AUTOLOG_MODULE`	传统 ML：sklearn、tensorflow、xgboost、transformers…	`fluent.py:3615`
`GENAI_LIBRARY_TO_AUTOLOG_MODULE`	GenAI：openai、anthropic、langchain、dspy、crewai、bedrock（键是 `boto3`）…	`fluent.py:3633`

两张表在普通环境下合并使用；在 Databricks Runtime 且 disable=False 时只用前一张（fluent.py:3655-3662），注释说明原因是该函数在 Databricks 里由系统自动调用，不想一次性把所有 GenAI 集成打开。

钩子回调 setup_autologging(module)（fluent.py:3682）里藏着一条优先级规则：用户手动调过的 flavor 配置，不能被 mlflow.autolog() 覆盖。实现靠一个标记键 AUTOLOGGING_CONF_KEY_IS_GLOBALLY_CONFIGURED（autologging_utils/__init__.py:59）：

配置里有这个键 → 上次是 mlflow.autolog() 设的 → 可以覆盖。
配置存在但没这个键 → 用户调过 mlflow.openai.autolog(...) → 直接 return，不动它（fluent.py:3699-3702）。

5. 两种集成范式

补丁框架是通用的，但"往哪儿打"分成两个流派。差别的根源在于：这个 SDK 自己有没有回调机制。

维度	函数式补丁	回调式集成
前提	SDK 没有回调钩子	SDK 自带 callback / handler 机制
补丁打在哪	业务方法本身（`create` / `run`）	只补构造函数，把自家 handler 塞进去
span 何时开关	补丁函数首尾	框架触发 `on_xxx_start` / `on_xxx_end`
父子关系怎么定	靠 OTel 上下文里的 active span	框架给的 `parent_run_id` + 上下文，双源仲裁
异常保护	`safe_patch` 夹心层	夹心层 + handler 类用 `ExceptionSafe*` 元类
代表	openai、anthropic、gemini、bedrock、crewai	langchain、dspy、llama_index、litellm

5.1 函数式：openai

打点清单。 _autolog()（mlflow/openai/autolog.py:118）就是一串 safe_patch 调用，一眼能数清补了哪些入口：

目标	方法	补丁函数	行
`ChatCompletions` / `Completions` / `Embeddings`	`create`	`patched_call`	`:133-134`
`ChatCompletions`（openai ≥ 1.92）	`parse`	`patched_call`	`:139`
三个 Async 版	`create`	`async_patched_call`	`:141-142`
`Images` / `AsyncImages`	`generate`	同上两种	`:147-148`
`Responses` / `AsyncResponses`	`create`、`parse`	同上两种	`:170-173`

每个 try/except ImportError 都对应一个"这个 API 只在某版本以上才有"的兼容分支——这是所有 flavor 的通用写法。

补丁函数骨架。 patched_call（:260）短得可以整段读懂它的结构：

# 示意，非源码：真实结构见 mlflow/openai/autolog.py:260-280
def patched_call(original, self, *args, **kwargs):
    config = AutoLoggingConfig.init(flavor_name="openai")   # 每次调用都重读配置
    run_id = mlflow.active_run().info.run_id if mlflow.active_run() else None

    if config.log_traces:
        span = _start_span(self, kwargs, run_id)   # 开 span
        _inject_tracing_headers(kwargs, span)      # 顺手往请求头塞 traceparent

    try:
        raw_result = original(self, *args, **kwargs)   # 调真家伙
    except Exception as e:
        if config.log_traces:
            _end_span_on_exception(span, e)   # 记异常事件 + status=ERROR
        raise                                  # 异常必须原样抛出

    if config.log_traces:
        _end_span_on_success(span, kwargs, raw_result, ...)
    return raw_result

async_patched_call（:283）是逐行对应的 await 版本——和 safe_patch 内部一样，选择了复制而不是抽象。

_start_span（:306）里有两个细节值得点出：

输入参数里除了 messages / input 之外的全部塞成 span 属性（:313），因为消息体本身走 inputs 字段。
run 关联是手动做的（:328-330）：autolog 自己创建的 run 不是 active run，不会被自动关联，所以直接往 trace metadata 写 SOURCE_RUN。

分布式追踪头注入。 _inject_tracing_headers（:527）从 span 里直接算出 W3C traceparent 塞进 extra_headers：

kwargs["extra_headers"] = tracing_headers | existing   # openai/autolog.py:531

注意合并方向：existing（用户自己传的）在右边，冲突时用户的值胜出。头本身由 _get_tracing_headers_from_span（mlflow/tracing/distributed/__init__.py:71）用 span 的 trace_id/span_id 直接拼字符串，不走 OTel 的全局上下文传播——因为这里要指向的是"刚开的这个 span"，不一定是当前 active span。这也正是 3.7 提到的那条 extra_headers 校验豁免的由来。

流式响应：把一串 chunk 收拢成一个 span。 这是函数式补丁里最麻烦的一块。流式调用返回的是迭代器，"调用结束"不是函数返回那一刻，而是迭代耗尽那一刻。

_end_span_on_success（:335）的解法是换掉迭代器：

result 是 Stream ?
    │
    是 → 把 result._iterator 换成一个包装生成器 (autolog.py:356)
    │        每 yield 一个 chunk：
    │           ├─ _add_span_event(span, i, chunk)  记成 span event
    │           └─ 存进 output 列表
    │        迭代结束后：_process_last_chunk(...)   ← 才真正关 span
    │
    否 → set_span_chat_attributes + span.end(outputs=result)  直接关

每个 chunk 记成一条 span event，名字是 mlflow.chunk.item.{index}、值是 JSON（_add_span_event，:544；常量在 mlflow/tracing/constant.py:196-197）。这样既保留了逐块时序，又不影响最终那个完整输出。

_process_last_chunk（:376）负责把碎片拼回一个完整对象，四条分支：

情形	处理	行
最后一块是 Responses API 的 `ResponseCompletedEvent`	直接用 `chunk.response`，服务端已经给了完整体	`:384-385`
一块都没有	`output = None`	`:386-387`
Responses API 流	`_reconstruct_response_from_stream`：只挑 `ResponseOutputItemDoneEvent`，拼成 `Response`	`:388-389` / `:497`
Chat Completions 流	`_reconstruct_completion_from_stream`：拼文本 + 从末块取元数据	`:390-392` / `:429`

_reconstruct_completion_from_stream（:429）的重建逻辑值得看一眼，它不是简单拼字符串：

先按 chunk.object 过滤掉非内容块（_filter_completion_stream_chunks，:414）。
老的 text completion 走遗留路径，直接返回拼好的字符串（:440-447）。
chat 路径把每块的 delta.content 拼成一条 assistant 消息，再用最后一块的 id / created / model / system_fingerprint 组装出一个货真价实的 ChatCompletion 对象（:486-494）——这样下游看到的结构与非流式完全一致。
content 是列表的情况单独处理（:460-469），注释指明是为了兼容 Databricks 的流式格式。
token 用量从最后一个带 usage 的块里反向找（_get_completion_stream_usage，:422），连 prompt_tokens_details.cached_tokens 都摘出来（:402-404）。

一个边界： 两个流包装生成器都在循环结束后引用循环变量 chunk（:354、:365）。如果一个 chunk 都没产出，chunk 未绑定会抛 UnboundLocalError；源码里没有针对空流的保护 (inferred：实际 API 极少返回零块流)。

5.2 回调式：langchain

LangChain 自带 BaseCallbackHandler 机制，MLflow 就不去补 invoke/ainvoke（那有几十个入口、还有 LCEL 的组合），而是只补一个构造函数。

注入点。 mlflow/langchain/autolog.py:50 补的是 BaseCallbackManager.__init__。补丁函数 _patched_callback_manager_init（:85）逻辑很短：

# 示意，非源码：见 mlflow/langchain/autolog.py:85-100
def _patched_callback_manager_init(original, self, *args, **kwargs):
    original(self, *args, **kwargs)                       # 1. 先让人家正常初始化
    if not AutoLoggingConfig.init(FLAVOR_NAME).log_traces:
        return                                            # 2. 关了就不注入
    for handler in self.inheritable_handlers:
        if isinstance(handler, MlflowLangchainTracer):
            return                                        # 3. 已有就别重复加
    self.add_handler(MlflowLangchainTracer(...), inherit=True)   # 4. 挂上

一旦挂成 inheritable handler，LangChain 自己会把它沿着 chain/graph 的每一层传下去——不用 MLflow 操心。

还得补两个补丁去堵漏。 光补 __init__ 不够，另外两个补丁都是踩坑踩出来的：

补丁	治什么病	位置
`BaseCallbackManager.merge`	`merge` 走 setter 而非构造函数，导致 `__init__` 里的去重判断失效，回调被加两遍	`mlflow/langchain/autolog.py:105`（`_patched_callback_manager_merge`）
`RunnableSequence.batch`	该方法按"步骤 × 条目"顺序在同一线程里交错执行，span 若挂进上下文会串台	`mlflow/langchain/autolog.py:139`（`_patched_runnable_sequence_batch`）

第二个的做法是临时把 _should_attach_span_to_context 这个 ContextVar 置成 False（autolog.py:155-160；变量定义在 mlflow/langchain/langchain_tracer.py:45），让这批 span 只靠 parent_run_id 建立父子关系，不碰 OTel 上下文。补丁函数里的注释画了那个交错执行顺序（autolog.py:143-149），是理解这个坑的最好材料。

Tracer 本体。 MlflowLangchainTracer（mlflow/langchain/langchain_tracer.py:48）的类声明本身就是一个知识点：

class MlflowLangchainTracer(BaseCallbackHandler, metaclass=ExceptionSafeAbstractClass):

元类保证了每一个 on_xxx 回调都被 try/except 包住——LangChain 在自己的流程里调这些回调，任何一个抛异常都会打断用户的 chain。类 docstring 直说了选 ExceptionSafeAbstractClass 的原因（:51-52）。

状态怎么存。 一个字典 self._run_span_mapping: {run_id: SpanWithToken}（:79）。构造函数上方有一句加粗的告诫（:71-72）：不要用实例变量保存单条 trace 的状态——同一个 tracer 实例会在多线程下同时服务多棵 trace，所有状态必须按 run_id 分桶。

回调与 span 的对应关系是机械的：

回调	动作	span_type	行
`on_chat_model_start`	开 span，把消息规范化成 `messages` 数组	`CHAT_MODEL`	`:309`
`on_llm_start`	开 span，输入是 prompt 列表	`LLM`	`:367`
`on_chain_start`	开 span；顺带把 `metadata.thread_id` 写成会话 ID	`CHAIN`	`:561`
`on_tool_start`	开 span；`input_str` 先试 `ast.literal_eval` 再兜底	`TOOL`	`:625`
`on_retriever_start`	开 span	`RETRIEVER`	`:680`
`on__end` / `on__error`	按 run_id 找回 span 并关掉；error 分支加异常事件 + `ERROR` 状态	—	如 `:596`、`:610`

最精彩的一段：父 span 怎么定。 _get_parent_span（:150）要在两个可能冲突的来源之间做仲裁：

     ┌─ 来源 A：MLflow 上下文里的 active span（get_current_active_span）
     └─ 来源 B：LangChain 给的 parent_run_id 对应的 span

  只有 A → 用 A          只有 B → 用 B          都没有 → 这是根 span
  两个都有且是同一个 → 用它
  两个都有但不同   → _resolve_parent_span 顺着 A 往上爬父链 (:179)
                        ├─ 爬到 B → 说明 A 在 B 里面，用 A（更近的父亲）
                        └─ 爬不到 → 用 B

为什么要这么绕？_resolve_parent_span 的 docstring（:179-227）给了两个只差一行缩进的例子：mlflow.start_span("parent") 写在 graph 外面时正确结构是 parent → tool → ChatOpenAI，写在 tool 内部时正确结构是 tool → parent → ChatOpenAI。光看 span 自身的元数据分不出这两种情况，只能爬树。 根源在注释里也写了（:151-157）：LangChain 大量用线程和 asyncio，ContextVar 经常传不下去，所以不能只信一个来源。

收尾。 _end_span（:248）用 try/finally 保证即使 span.end() 失败也要把 span 从上下文分离；detach 抛 ValueError（token 是在别的线程/协程里创建的）只记 debug 不上抛（:275-285）。flush()（:287）负责清理 LangChain 忘了触发 end 事件而泄漏的 span。

5.3 两种范式的共同套路

不管哪一派，翻开任何一个 flavor 的 autolog.py，都能看到同一套骨架：

顶上一个 FLAVOR_NAME 常量，贯穿 safe_patch 和配置查询。
autolog() 对外 / _autolog() 带装饰器（需要处理"关闭时清理"的才拆）。
每个 safe_patch 外面包 try/except ImportError，做版本/可选依赖兼容。
补丁函数签名固定 (original, self, *args, **kwargs)，同步/异步各一份。
结尾一句 _record_event(AutologgingEvent, {...}) 上报遥测。

6. 集成矩阵一览

同一套框架接了二十多家。下表只列落点机制，细节请直接按符号名 grep：

flavor	autolog 入口	落点机制	代表符号
openai	`mlflow/openai/autolog.py:33`	函数补丁（`create`/`parse`/`generate`）+ Agent SDK trace processor	`patched_call` / `async_patched_call`
anthropic	`mlflow/anthropic/__init__.py:17`	函数补丁 `Messages.create`；另补 Claude Agent SDK 的 `ClaudeSDKClient.__init__`	`patched_class_call` / `patched_claude_sdk_init`
gemini	`mlflow/gemini/__init__.py:18`	函数补丁，类方法和模块级函数都补	`patched_class_call` / `patched_module_call`
bedrock	`mlflow/bedrock/__init__.py:13`	两段式：先补 boto3 的 `ClientCreator.create_client`，拿到 client 后再补它的类	`patched_create_client` / `patch_bedrock_runtime_client`
crewai	`mlflow/crewai/__init__.py:25`	函数补丁，按"类路径 → 方法名"映射表批量打；表随 crewai 版本分叉	`_apply_patches` / `class_method_map`
autogen	`mlflow/autogen/__init__.py:24`	函数补丁，遍历 `BaseChatAgent.__subclasses__()` 挨个补	`patched_agent` / `patched_completion`
langchain	`mlflow/langchain/autolog.py:14`	回调注入（补 `BaseCallbackManager.__init__`）+ 两个补漏补丁	`MlflowLangchainTracer`
dspy	`mlflow/dspy/autolog.py:22`	混合：`dspy.settings.configure(callbacks=[...])` 注册回调，另补 `Teleprompter.compile` / `Evaluate.__call__`	`MlflowCallback` / `_patched_compile`
llama_index	`mlflow/llama_index/autolog.py:7`	纯官方 handler 注册，一个 `safe_patch` 都没有	`set_llama_index_tracer` / `MlflowSpanHandler`
litellm	`mlflow/litellm/__init__.py:13`	往 `litellm.success_callback` 追加 `"mlflow"`；另补线程池 `executor.submit` 让回调同步执行	`_append_mlflow_callbacks` / `_patched_submit`
agno	`mlflow/agno/__init__.py:13`	版本分叉：v2 用 OpenInference 的 OTel instrumentor，v1 走函数补丁	`_setup_otel_instrumentation` / `patched_class_call`
claude_code	`mlflow/claude_code/`	不走 safe_patch：CLI `mlflow autolog claude` 往 `.claude/settings.json` 装 hook，事后解析 transcript 造 span	`process_transcript` / `_create_llm_and_tool_spans`
mcp	`mlflow/mcp/`	不是 autolog 集成：把 MLflow CLI 命令暴露成 MCP 工具给 agent 用	`mlflow_mcp`

三个从表里读出来的结论：

绝大多数走函数补丁，因为多数 LLM SDK 没有回调机制。
有回调机制的一律优先用回调（langchain / dspy / llama_index / litellm），补丁只用来"把自己塞进去"或堵漏。
末两行是提醒：mlflow/ 下有 autolog.py 不等于走这套框架，claude_code 是外部 hook + 日志解析，mcp 压根是另一件事。

7. span_type 语义映射与 GenAI 语义约定

补丁开 span 时要回答一个问题：这次调用在语义上属于哪一类？ 答案就是 span_type。

7.1 SpanType 常量

一共 15 个，定义在 mlflow/entities/span.py:52（class SpanType）。注释特意说明没有用 enum，因为要允许自定义字符串：

分组	取值
模型调用	`LLM`、`CHAT_MODEL`、`EMBEDDING`、`RERANKER`
编排	`CHAIN`、`AGENT`、`WORKFLOW`、`TASK`
外部动作	`TOOL`、`RETRIEVER`、`MEMORY`
加工与评判	`PARSER`、`GUARDRAIL`、`EVALUATOR`
兜底	`UNKNOWN`

7.2 openai 怎么映射

_get_span_type(task)（mlflow/openai/autolog.py:173）是一张"资源类 → span 类型"的字典：

OpenAI 资源类	span_type
`ChatCompletions` / `AsyncChatCompletions` / Beta 版 / `Responses`	`CHAT_MODEL`
`Completions` / `AsyncCompletions`	`LLM`
`Embeddings` / `AsyncEmbeddings`	`EMBEDDING`
`Images` / `AsyncImages`	`TOOL`
都不匹配	`UNKNOWN`

巧的是最后的查表方式（:222-227）：不是 span_type_mapping[task] 直接取，而是遍历字典用 issubclass 判断。注释说明了原因——第三方包装类（例如 Databricks 自己的 ChatCompletions 子类）也要能落到正确类型上。多花一次 O(n) 遍历，换来对下游生态的兼容。

LangChain 那边不需要映射表：回调名本身就是语义，on_tool_start 就开 TOOL span（见 5.2 的表）。

7.3 转成 OpenTelemetry GenAI 语义约定

MLflow 的 span 属性是自家格式。要导给遵循 OTel GenAI 语义约定的后端，就得翻译一次。

机制是"标记 + 分派"两步：

第一步，埋点时打上格式标记。 openai 的 _start_span 给 CHAT_MODEL / LLM 类型的 span 写 MESSAGE_FORMAT = "openai"（mlflow/openai/autolog.py:311-312）；LangChain tracer 在 on_chat_model_start 和 on_llm_start 两处各写一次 "langchain"（mlflow/langchain/langchain_tracer.py:325、:382）。

第二步，导出时按标记选转换器。 _get_converter(message_format, inputs)（mlflow/tracing/export/genai_semconv/translator.py:133）：

message_format == "openai" ?
     ├─ inputs 里有 "input" 键 → OpenAIResponsesConverter   (Responses API)
     └─ 否则                   → OpenAIChatCompletionConverter
"anthropic" → AnthropicConverter
"gemini"    → GeminiConverter
"bedrock"   → BedrockConverseConverter
其他（含 "langchain"、None）→ 兜底也用 OpenAIChatCompletionConverter

转换器长什么样。 抽象基类 GenAiSemconvConverter（mlflow/tracing/export/genai_semconv/converter.py:28）只强制两个抽象方法：convert_inputs 和 convert_outputs；其余（system 指令、请求参数、响应属性）都有默认实现，子类按需覆盖。

mlflow/openai/genai_semconv_converter.py 里两个实现的分工：

类	管什么	行
`OpenAIChatCompletionConverter`	Chat Completions 格式；文件头注释说 Groq / Bedrock 也复用它	`:16`
`OpenAIResponsesConverter`	Responses API 格式，输入输出都是 item 列表	`:64`

转换的核心是把各家的消息体统一成 {role, parts[]} 结构（_convert_message，:161）。几个具体动作：

system 消息从 messages 里摘出去，单独走 convert_system_instructions（:23），因为语义约定把系统指令列为独立字段。
内容块按类型分流（_convert_content，:184）：文本 → {type: "text"}；data: 开头的图片 URL 解析成 {type: "blob"} 带 mime 和 base64，普通 URL 则是 {type: "uri"}（_convert_image_url，:215）；音频转 blob。
工具调用转成 {type: "tool_call", id, name, arguments}（_convert_tool_call，:229），参数字符串尽力 json.loads、失败就原样保留（_parse_tool_arguments，:249）。
工具定义从 OpenAI 的嵌套 {type, function:{...}} 拍平成语义约定要的平铺形式（_flatten_tools，:256）。
有个兜底很有意思（:167-173）：gpt-4o-audio-preview 把音频回复放在独立的 audio 字段而不是 content，导致 parts 为空；这时退而取 audio.transcript 当文本，免得导出一条空消息。

8. 巧妙之处（可借鉴）

按"能带走的程度"排序：

用哨兵变量把"谁的异常"分清楚。 failed_during_original 这一个布尔量（safety.py:433-434）就把"用户侧异常"和"埋点侧异常"彻底分开，从而支持"前者照抛、后者降级"。任何做无侵入拦截的库都要面对这个问题，这是最轻的解法。
补丁函数的返回值不算数。 用户拿到的永远是 original_result；补丁没调原函数的话夹心层自己补调（safety.py:508-512）。这条设计让"补丁写错了"最坏也只是丢埋点，不会丢结果。
配置存全量、读时查。 autologging_integration 把 autolog() 的默认值 + 实参一起存进全局 dict（:411-420），补丁函数每次执行都重读。好处是运行中改配置立即生效，坏处是每次调用多几次 dict 查找。
打新补丁之前先撤旧补丁。 revert_patches(name) 在装饰器里无条件先跑（:431），彻底杜绝了反复调 autolog() 导致补丁叠罗汉、一次调用产生 N 层 span 的问题。
靠 post-import hook 避免导入二十个 SDK。 mlflow.autolog() 不 import 任何第三方库，只挂钩（fluent.py:3728），而且对已导入的库追溯生效。
流式响应用"换掉迭代器"来延后关 span（mlflow/openai/autolog.py:351），而不是去猜什么时候结束；同时每块记一条 event，时序和汇总两不误。
父 span 冲突时爬树仲裁（langchain_tracer.py:179）。承认"元数据不足以判断"，那就老老实实走一遍父链——比拍脑袋定优先级正确得多。
不敢重构的地方就明说。 async 版夹心层是同步版的逐行拷贝，注释写明"刻意不抽象，因为这在 Databricks 默认开启、是关键路径"（safety.py:530-536）。这是很诚实的工程判断记录。

9. 边界与局限

只补名字，不补引用。 gorilla 改的是类属性。如果用户在 autolog() 之前就把方法取出来存成了变量（f = client.chat.completions.create），那个引用指向的还是原函数，补丁对它无效 (inferred)。
session 是进程级的。 _AutologgingSessionManager._session 是普通类属性（safety.py:745），不是 thread-local。多线程并发调用时 session 的归属语义比较模糊。
disable=True 时函数体不执行（:436），这条已经逼出了多个 flavor 的 autolog / _autolog 拆分；源码自己也在 TODO 里承认这个模式不统一（mlflow/openai/autolog.py:66-67）。
async property 不支持，直接抛异常（safety.py:292）；manage_run=True 也不支持 async（safety.py:272）。
ExceptionSafeClass 管不到 classmethod / staticmethod，docstring 明说了原因是它们不总是可调用对象（safety.py:97-99）。
版本兼容只卡下限。 _check_and_log_warning_for_unsupported_package_versions（autologging_utils/__init__.py:354）只检查最低版本，注释解释：ml-package-versions.yml 里的上限跟不上上游发版节奏，跨版本测试实际跑的是最新版。
空流未加保护：流包装生成器在零 chunk 时会引用未绑定的 chunk（mlflow/openai/autolog.py:351）。
LangChain 的 run_tracer_inline 有个小陷阱：get_autologging_config(FLAVOR_NAME, "run_tracer_inline", True)（mlflow/langchain/autolog.py:100）传的兜底值是 True，但 autolog() 签名里默认是 False（:20），而装饰器会把所有参数默认值都写进配置表（autologging_utils/__init__.py:411-420），所以那个 True 实际永远用不上 (inferred)。

10. 代码地图

主题	文件	符号名
打补丁入口	`mlflow/utils/autologging_utils/safety.py`	`safe_patch`
异常隔离夹心层	`mlflow/utils/autologging_utils/safety.py`	`safe_patch_function` / `call_original`
async 夹心层	`mlflow/utils/autologging_utils/safety.py`	`async_safe_patch_function`
埋点会话	`mlflow/utils/autologging_utils/safety.py`	`AutologgingSession` / `_AutologgingSessionManager`
补丁登记与撤销	`mlflow/utils/autologging_utils/safety.py`	`_wrap_patch` / `_store_patch` / `revert_patches`
回调安全元类	`mlflow/utils/autologging_utils/safety.py`	`ExceptionSafeClass` / `ExceptionSafeAbstractClass`
测试期参数校验	`mlflow/utils/autologging_utils/safety.py`	`_validate_args` / `_VALIDATION_EXEMPT_ARGUMENTS`
签名伪装	`mlflow/utils/autologging_utils/safety.py`	`update_wrapper_extended`
底层属性替换	`mlflow/utils/gorilla.py`	`apply` / `revert` / `get_original_attribute`
配置注册表	`mlflow/utils/autologging_utils/__init__.py`	`AUTOLOGGING_INTEGRATIONS` / `autologging_integration`
配置读取	`mlflow/utils/autologging_utils/__init__.py`	`get_autologging_config` / `autologging_is_disabled`
配置 dataclass	`mlflow/utils/autologging_utils/config.py`	`AutoLoggingConfig`
全局总开关	`mlflow/tracking/fluent.py`	`autolog` / `setup_autologging` / `GENAI_LIBRARY_TO_AUTOLOG_MODULE`
函数式集成范例	`mlflow/openai/autolog.py`	`_autolog` / `patched_call` / `async_patched_call`
流式重建	`mlflow/openai/autolog.py`	`_process_last_chunk` / `_reconstruct_completion_from_stream` / `_reconstruct_response_from_stream`
追踪头注入	`mlflow/openai/autolog.py`	`_inject_tracing_headers`
span 类型映射	`mlflow/openai/autolog.py`	`_get_span_type`
回调式集成范例	`mlflow/langchain/autolog.py`	`_patched_callback_manager_init` / `_patched_callback_manager_merge`
LangChain tracer	`mlflow/langchain/langchain_tracer.py`	`MlflowLangchainTracer` / `_get_parent_span` / `_resolve_parent_span`
span 类型常量	`mlflow/entities/span.py`	`SpanType`
语义约定转换器	`mlflow/openai/genai_semconv_converter.py`	`OpenAIChatCompletionConverter` / `OpenAIResponsesConverter`
转换器分派	`mlflow/tracing/export/genai_semconv/translator.py`	`_get_converter`
转换器基类	`mlflow/tracing/export/genai_semconv/converter.py`	`GenAiSemconvConverter`

继续读： span 对象本身长什么样、@mlflow.trace 手动埋点怎么用 → 01 · 追踪数据模型；本章产出的 LiveSpan 之后怎么被处理、导出、落盘 → 02 · 追踪运行时；这些 trace 怎么被拿去评估 → 04 · 评估引擎。

1. 这章要解决的问题（零基础也能懂）​

2. 顶层全景：三层结构​

3. 通用打补丁框架：safe_patch​

3.1 它要解决的小问题​

3.2 gorilla：可撤销的替换​

3.3 准备阶段：三种被补对象要分开处理​

3.4 夹心层内部：异常隔离怎么做到​

3.5 AutologgingSession：一次调用的"埋点作用域"​

3.6 撤补丁：revert_patches​

3.7 ExceptionSafe*：传给别人的对象也得有保险丝​

4. 配置注册表：一个全局 dict 管住所有开关​

4.1 AUTOLOGGING_INTEGRATIONS​

4.2 @autologging_integration 装饰器​

4.3 mlflow.autolog() 总开关：不导入、只挂钩​

5. 两种集成范式​

5.1 函数式：openai​

5.2 回调式：langchain​

5.3 两种范式的共同套路​

6. 集成矩阵一览​

7. span_type 语义映射与 GenAI 语义约定​

7.1 SpanType 常量​

7.2 openai 怎么映射​

7.3 转成 OpenTelemetry GenAI 语义约定​

8. 巧妙之处（可借鉴）​

9. 边界与局限​

10. 代码地图​