AI 工程化绕不开的三个关键词最近一次技术访谈里 Charity Majors 又把这几个老问题重新摆到了桌面上Determinism、Instrumentation还有一句很形象的“Eating Your Broccoli”。如果你正在做 AI 应用落地或者负责模型上线后的稳定性这篇文章值得认真读完。我会结合可观测性工程的实际经验把这几件事拆开讲清楚并且给出可以直接参考的代码示例和排查思路。先说一个很多团队都遇到过的场景。模型在离线评测时 AUC 很漂亮推理延迟也达标可是一上生产用户反馈就开始变得奇怪同一个问题上午回答正常下午就开始胡说同一个 Prompt在这个实例上返回结果在另一个实例上就完全不一样。排查了半天发现既不是模型参数被改也不是数据管道中断最后定位到的是一个毫秒级的时间戳差异或者一次哈希遍历顺序的不稳定。这种问题就是典型的确定性缺失。AI 系统本身的随机性、并行调度、分布式状态都会让“同样的输入得到同样输出”变成一件很难的事情。但这恰恰是工程化必须解决的问题。Charity Majors 的观点非常直接可观测性不是简单的日志收集和看板展示它决定了一个系统能不能被理解而确定性决定了一个系统能不能被信任。如果你连一次请求内部发生了什么都无法复现和追溯那就谈不上对系统做任何有效的干预。1. 背景与核心概念1.1 Charity Majors 是谁为什么她的观点值得关注Charity Majors 是 Honeycomb 的联合创始人和 CTO也是可观测性领域非常有影响力的工程领导者。她早年在 Facebook、Parse 等公司负责基础设施和数据库团队后来把“Production Engineering”和“Observability”这两个概念在工程社区里推到了一个新的高度。她最出名的主张之一是日志不够用指标不够用追踪也不够用三者必须结合而且要围绕事件和请求上下文来组织。在 AI 这个议题上她的立场其实很朴素也很硬核AI 系统首先是分布式系统其次才谈得上智能。任何分布式系统该有的确定性、可观测性、故障恢复能力AI 系统一样都不能少。她甚至说很多团队对模型训练阶段的指标非常敏感但对推理阶段的可观测性几乎漠不关心这是一种典型的工程失衡。这个观点放到今天的 AI 工程实践里非常应景。大量团队正在从“能跑通模型”走向“能稳定运行模型服务”过去靠 Jupyter Notebook 和离线评测就能交付的方式已经不够了生产环境需要的是可以用指标数据说话、可以定位到具体请求链路的系统。1.2 DeterminismAI 系统里的确定性为什么重要确定性指的是在给定相同输入和相同初始状态的条件下系统应该产生一致的输出。传统软件工程里确定性通常被认为是理所当然的同一个函数传入同样的参数返回值应该一致。但在 AI 系统里这个前提很容易被打破。比如深度学习推理时GPU 上的浮点运算顺序可能因为并行策略不同而产生微小差异比如模型服务在多副本部署时每个副本的权重加载顺序不同可能导致结果不一致再比如使用浮点数累加时线程调度顺序不同最终结果也会出现细微偏差。这些问题在单次请求里几乎无法被感知但一旦你开始做自动化测试、回归对比、A/B 实验或者需要审计某一次线上事故时确定性的缺失就会变成巨大的障碍。你无法回答一个最基本的问题这次异常是模型本身的问题还是系统的随机性造成的1.3 Instrumentation插桩与可观测性的真正含义Instrumentation 翻译成中文通常是“插桩”或“埋点”但它远不止在代码里加几行日志那么简单。真正的可观测性要求系统内部的状态变化、外部请求的行为轨迹、资源消耗的关联关系都必须能够被收集、关联和查询。传统监控回答的是“这个服务挂了没有”可观测性回答的是“这个服务为什么挂了挂在了哪个环节”。对于 AI 系统来说可观测性要回答的问题更复杂这次推理为什么耗时 800ms是哪一层模型计算拖慢了速度这一次输出为什么不符合预期是输入特征分布变了还是模型版本被意外切换了要做好 AI 系统的插桩你需要对请求链路做端到端的追踪。从 API 入口开始到预处理、特征工程、模型推理、后处理、返回结果每一步都需要记录关键事件和上下文信息。只有这样当某一次线上请求出现问题时你才能把链路拉出来精确地回放和分析。2. 环境准备与版本说明在进入具体实践之前先统一一下环境。本文示例会使用 Python 和 OpenTelemetry这是目前可观测性领域比较通用的技术栈。版本建议以你的项目实际情况为准本文重点演示的是配置思路和代码结构。操作系统Linux / macOS / Windows 均可 Python3.8 或以上 OpenTelemetry SDK建议使用 1.20 以上版本 模型推理框架PyTorch / TensorFlow / ONNX Runtime 均可如果你使用的是 Java 或 Go 技术栈可观测性的核心思路完全一致只是 SDK 接入方式有些差别。本文的示例代码主要用于说明概念你可以根据自己的技术栈做迁移。3. 核心原理拆解AI 可观测性的三大支柱3.1 分布式追踪把一次请求的完整旅程串起来分布式追踪是理解 AI 系统行为的基础。它把一个用户请求从进入 API Gateway 开始到调用预处理服务、模型推理服务、后处理服务再到最后返回响应的整个过程串成一条链路。在 AI 场景里追踪的粒度需要比传统业务系统更细。一次模型推理内部往往包含多个张量操作、多次内存拷贝、多次算子调度这些在传统 RPC 追踪里通常不会单独记录但在 AI 可观测性里它们恰恰是性能分析的富矿。比如一个典型的文本分类服务完整追踪链路可能长这样User Request - API Gateway - Preprocessing Service - Tokenizer - Feature Encoding - Model Inference Service - Input Tensor Preparation - Forward Pass - Softmax - Postprocessing Service - Label Mapping - Response Formatting - User Response每一段都可以记录耗时、状态、错误信息、关键参数值。当用户反馈“某个请求特别慢”时你可以在追踪系统里找到对应的 trace ID然后逐段定位瓶颈。3.2 结构化日志给每一次推理建立完整档案日志是传统排查最重要的手段在 AI 系统里依然如此。但重要的一步升级是从非结构化文本日志变为结构化 JSON 日志。每一行日志都应该包含足够多的上下文而不是孤零零的一行报错信息。一个典型的 AI 推理日志应该包含以下字段{ timestamp: 2025-06-18T10:15:30.123456Z, service: model-inference, trace_id: abc123def456, model_version: v2.3.1, input_hash: a1b2c3d4e5f6, latency_ms: 812, gpu_utilization: 87.5, prediction: positive, confidence: 0.987, error: null }有了这种结构化日志你就可以用 LogQL、SQL 或者其他查询语言在日志系统里做多维度的聚合分析。比如查询最近一小时所有 confidence 低于 0.6 的请求占比或者按模型版本分组统计平均延迟。3.3 指标与仪表盘从宏观视角把握系统健康度指标负责回答“系统整体状态如何”的问题。在 AI 推理服务里核心指标可以分为三类延迟类p50、p95、p99 推理延迟端到端请求延迟。吞吐类QPS、并发请求数、GPU 利用率、显存占用。质量类预测置信度分布、拒答率、超时率、错误率。质量类指标经常被团队忽略但它恰恰是 AI 服务最需要关注的。传统 Web 服务关注 5xx 错误率就够了AI 服务还需要关注“模型输出了不合理结果但 HTTP 状态码仍然是 200”的情况。这类问题只能通过置信度分布、输入输出特征监控等质量指标来发现。指标和追踪需要配合使用。指标负责发现异常追踪负责定位异常。比如你的服务级别指标显示 p99 延迟突然从 400ms 涨到 1.2s这时候你需要通过对该时间窗口内的请求做追踪采样找到究竟是哪个服务或者哪个环节劣化了。4. 完整实战案例为 AI 推理服务接入可观测性下面我们用一个完整的示例演示如何为一个基于 FastAPI 和 PyTorch 的文本分类服务接入 OpenTelemetry 和结构化日志。4.1 创建项目结构ai_observability_demo/ ├── app.py # FastAPI 应用入口 ├── model.py # 模型加载与推理逻辑 ├── tracing.py # OpenTelemetry 配置 ├── logging_config.py # 结构化日志配置 ├── requirements.txt # 依赖清单 └── docker-compose.yml # 可选本地部署 Jaeger 和 Prometheus4.2 添加依赖# requirements.txt fastapi0.111.0 uvicorn0.30.1 torch2.3.0 transformers4.41.2 opentelemetry-api1.25.0 opentelemetry-sdk1.25.0 opentelemetry-instrumentation-fastapi0.46b0 opentelemetry-exporter-otlp1.25.0 python-json-logger2.0.7这里用 FastAPI 作为示例框架因为它自带 OpenAPI 文档调试方便。你可以根据项目实际情况替换为 Flask、Django 或其他框架。4.3 配置 OpenTelemetry 追踪先创建一个 tracing.py 文件负责初始化 OpenTelemetry SDK 并接入 FastAPI 的中间件。# tracing.py from opentelemetry import trace from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor def setup_tracing(app, service_name: str ai-inference-service): # 1. 配置服务名称和资源标签 resource Resource.create({ service.name: service_name, service.version: v2.3.1, }) # 2. 创建 TracerProvider并添加 OTLP Exporter provider TracerProvider(resourceresource) exporter OTLPSpanExporter(endpointhttp://localhost:4317, insecureTrue) processor BatchSpanProcessor(exporter) provider.add_span_processor(processor) # 3. 设置全局 TracerProvider trace.set_tracer_provider(provider) # 4. 自动采集 FastAPI 请求链路 FastAPIInstrumentor.instrument_app(app, tracer_providerprovider)这段代码的核心思想是应用启动时把 FastAPI 应用的所有 HTTP 请求自动接入追踪系统每个请求都会生成一条 trace并且发送到 OTLP Collector 或 Jaeger。4.4 配置结构化日志接下来配置结构化 JSON 日志。Python 的 logging 模块可以通过自定义 Formatter 实现。# logging_config.py import json import logging import sys from datetime import datetime, timezone class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) - str: log_entry { timestamp: datetime.now(timezone.utc).isoformat(), level: record.levelname, logger: record.name, message: record.getMessage(), } # 自动附加 trace_id 和 span_id if hasattr(record, trace_id) and record.trace_id: log_entry[trace_id] record.trace_id if hasattr(record, span_id) and record.span_id: log_entry[span_id] record.span_id # 附加附加字段 for key, value in record.__dict__.items(): if key not in log_entry and not key.startswith(_) and key not in [ args, asctime, created, exc_info, exc_text, filename, funcName, levelname, levelno, lineno, message, module, msecs, msg, name, pathname, process, processName, relativeCreated, stack_info, thread, threadName, taskName ]: log_entry[key] value return json.dumps(log_entry, ensure_asciiFalse) def setup_logging(): handler logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) root_logger logging.getLogger() root_logger.handlers [handler] root_logger.setLevel(logging.INFO) return root_logger注意这里为了在日志中自动附带 trace_id需要在记录日志时把 trace 上下文注入进去。可以写一个简单的辅助函数# app.py 中的日志辅助函数 import logging from opentelemetry import trace logger logging.getLogger(ai-inference) def log_with_trace(level: str, message: str, **kwargs): span_context trace.get_current_span().get_span_context() extra { trace_id: format(span_context.trace_id, 032x) if span_context.is_valid else None, span_id: format(span_context.span_id, 016x) if span_context.is_valid else None, } extra.update(kwargs) if level info: logger.info(message, extraextra) elif level error: logger.error(message, extraextra) elif level warning: logger.warning(message, extraextra)这样日志和追踪就通过 trace_id 关联起来了。你在看日志时可以直接跳转到对应的追踪链路反之亦然。4.5 编写模型推理服务现在我们写一个简单的文本分类服务模拟真实推理场景。# model.py import torch import torch.nn.functional as F from transformers import AutoModelForSequenceClassification, AutoTokenizer class TextClassifier: def __init__(self, model_name: str bert-base-uncased): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() def predict(self, text: str) - dict: # 1. Tokenize inputs self.tokenizer(text, return_tensorspt, truncationTrue, max_length512) # 2. 推理 with torch.no_grad(): outputs self.model(**inputs) # 3. 计算概率 probabilities F.softmax(outputs.logits, dim-1) confidence, prediction torch.max(probabilities, dim-1) return { prediction: int(prediction.item()), confidence: float(confidence.item()), }在 app.py 里接入追踪和日志# app.py import time import logging from fastapi import FastAPI, Request from pydantic import BaseModel from model import TextClassifier from tracing import setup_tracing from logging_config import setup_logging from app_logger import log_with_trace logger setup_logging() classifier TextClassifier() app FastAPI(titleAI Inference Service) setup_tracing(app) class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): prediction: int confidence: float latency_ms: float app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest, http_request: Request): start_time time.time() # 使用 OpenTelemetry 创建自定义 Span tracer trace.get_tracer(ai-inference) with tracer.start_as_current_span(model-inference) as span: span.set_attribute(input.length, len(request.text)) result classifier.predict(request.text) span.set_attribute(prediction, result[prediction]) span.set_attribute(confidence, result[confidence]) latency (time.time() - start_time) * 1000 log_with_trace( info, model inference completed, input_previewrequest.text[:50], predictionresult[prediction], confidenceresult[confidence], latency_msround(latency, 2), ) return PredictResponse( predictionresult[prediction], confidenceresult[confidence], latency_msround(latency, 2), ) app.get(/health) async def health_check(): return {status: ok}4.6 运行与验证启动服务uvicorn app:app --host 0.0.0.0 --port 8000在另一个终端发送请求curl -X POST http://localhost:8000/predict \ -H Content-Type: application/json \ -d {text: This movie is amazing!}正常情况下你会收到类似响应{ prediction: 1, confidence: 0.987, latency_ms: 812.35 }同时服务端会输出结构化 JSON 日志。如果本地部署了 Jaeger打开 http://localhost:16686就能看到完整的请求链路。5. 常见问题与排查思路5.1 trace_id 在日志中显示为 None问题现象常见原因解决思路日志里 trace_id 和 span_id 都是 None没有在日志调用时注入当前 span 上下文使用trace.get_current_span().get_span_context()获取上下文并在 logging 时通过 extra 传递日志和追踪系统无法关联日志格式没有包含 trace_id统一日志 Formatter保证所有 handler 都输出 trace_id 字段部分请求没有生成 traceOpenTelemetry Instrumentor 没有覆盖到异步接口检查 FastAPIInstrumentor 是否正确初始化并确认服务启动时完成了 setup_tracing()5.2 模型推理结果不确定问题现象常见原因解决思路多副本部署后输出不一致模型权重加载顺序或批处理顺序不一致为推理服务固定随机种子使用确定性算法使用 ONNX Runtime 并开启确定性模式相同输入在不同 GPU 上结果不同GPU 并行算子浮点累加顺序不同对关键输出使用容差断言记录推理运行的硬件信息和环境指纹重启后结果变化动态图模型的初始化存在随机性开启torch.manual_seed()、torch.backends.cudnn.deterministic True5.3 高并发下日志出现乱码或缺失问题现象常见原因解决思路多线程写日志时出现交错多个线程共享同一个 Handler日志写入不是原子操作使用 QueueHandler QueueListener 实现异步日志日志丢失异步日志队列溢出被丢弃监控队列大小合理配置queue_size和max_buffer_size日志延迟过高同步日志 I/O 阻塞改为异步日志并使用批量写入6. 最佳实践与工程建议6.1 从第一天就设计可观测性而不是事后补救很多团队是线上出了问题才开始接日志和监控这种“亡羊补牢”的做法在 AI 系统里代价极高。模型推理问题往往不是单点故障而是数据分布、模型版本、基础设施三者的综合结果。你之前没有采集的数据事后是补不回来的。建议在服务初始化阶段就完成三件事第一接入分布式追踪确保每个请求都有 trace_id第二配置结构化日志统一日志格式和关键字段第三建立核心指标看板覆盖延迟、吞吐、质量三个维度。6.2 确定性设计要贯穿整个推理链路确定性不是某个函数可以单独保证的它需要从数据输入、模型加载、推理计算到输出解析的完整链路共同配合。具体来说可以从以下几个层面入手数据层面对输入文本做标准化处理包括 Unicode 归一化、大小写转换、特殊字符处理。模型层面固定随机种子关闭 Dropout 等训练阶段才启用的算子使用确定性推理模式。环境层面记录推理环境的 Python 版本、依赖版本、GPU 型号、驱动版本方便复现问题。部署层面使用容器镜像锁定完整运行环境避免不同节点间环境漂移。6.3 充分利用 Trace 与 Log 的关联把日志和追踪关联起来是排查效率倍增的关键。每次记录日志时都要把当前的 trace_id 带上。这样当你从日志系统里发现一条异常日志时可以一键跳转到对应的完整请求链路反过来当你在追踪系统里发现某个耗时异常的 Span 时也可以看到这个 Span 打印的所有日志。在 OpenTelemetry 生态里这个能力通过 Baggage 和 Span Context 传递实现。你需要确保日志 Formatter 能自动提取当前 Span 的 trace_id 和 span_id。任何下游服务或异步任务都保持上下文传播。离线任务也要有独立的 trace_id方便和在线请求做关联分析。6.4 对模型版本做全生命周期追踪很多团队只会记录当前部署的模型版本但缺乏版本变更历史。这会导致一个问题当线上指标突然劣化时你无法快速判断是不是模型版本变更引起的。推荐的做法是每次模型发布时生成一个不可变的版本号并写入镜像标签。在推理日志里记录模型版本号和推理时间戳。定期对同一输入的多个版本做对比评测记录最佳版本。有了这些信息当线上出现异常时你可以通过时间轴快速关联这个时间点的模型版本是什么代码版本是什么依赖版本是什么数据管道版本是什么7. 延伸到 AI 工程实践的下一个阶段可观测性不是一次性工程它是一个持续演进的能力。随着你的 AI 系统从单模型走向多模型编排从离线批处理走向实时在线推理从单机房走向多区域部署可观测性覆盖的场景会越来越广也会越来越复杂。现在不少团队开始把可观测性能力与模型评估、数据质量监控、AI Agent 追踪结合起来。比如在 Agent 应用中一次用户请求可能触发多个工具调用、多次模型推理、多轮上下文管理传统的单请求追踪已经不够需要更细粒度的 Span 来记录每一步工具调用的输入输出、模型消耗的 Token 数量、上下文窗口的使用情况。这些方向都值得继续深入。在做 AI 工程化的过程中建议先建立起“确定性 可观测性”的基本框架再逐步扩展功能。每次线上事故排查都是一次对可观测性体系的检验。你会发现真正能快速定位问题的团队靠的不是运气而是平时积累的插桩覆盖度和数据关联能力。如果你正准备为自己的推理服务接入可观测性可以从最小闭环开始先接入分布式追踪和结构化日志确认 trace_id 能贯穿整条链路再逐步增加指标和看板。这个基础打牢之后你的 AI 系统会让你睡得踏实很多。如果你对 OpenTelemetry 的接入配置或者 AI Agent 追踪怎么做有疑问可以继续往下深入也可以在评论区聊聊你踩过的坑。