我们的生产环境运行着一组基于 Hugging Face Transformers 的模型服务，主要通过 FastAPI 暴露为 RESTful API。最初，我们仅使用 Datadog APM 进行基本的请求延迟和错误率监控。随着业务规模扩大，这种方案的短板愈发明显。我们需要的不仅仅是服务是否“活着”，而是模型推理的内部行为指标：例如，每个请求的 token 生成速率（tokens/sec）、首个 token 的延迟（TTFT）、输入与输出的 token 数量、甚至是模型输出的情感分布漂移。这些是典型的高基数（High Cardinality）指标，model_id, model_version, user_id 等标签的组合会产生海量的唯一时间序列。
定义复杂技术问题：LLM 推理服务的可观测性鸿沟
一个典型的推理服务代码片段如下：
# main.py
import os
import time
import logging
import asyncio
from fastapi import FastAPI, Request, HTTPException
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
from pydantic import BaseModel
from typing import Dict, Any
# --- 日志配置 ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# --- 模型加载 ---
MODEL_ID = os.environ.get("MODEL_ID", "distilgpt2")
try:
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForCausalLM.from_pretrained(MODEL_ID)
# 在生产环境中，应考虑使用更优化的推理引擎，如 vLLM 或 TGI
generator = pipeline('text-generation', model=model, tokenizer=tokenizer, device=0)
logger.info(f"模型 {MODEL_ID} 加载成功。")
except Exception as e:
logger.error(f"模型加载失败: {e}", exc_info=True)
raise RuntimeError("无法初始化模型服务") from e
app = FastAPI()
class InferenceRequest(BaseModel):
prompt: str
max_new_tokens: int = 50
# 这是一个简化的遥测客户端，实际项目中会是一个独立的、健壮的模块
# 稍后我们将详细探讨其实现
from telemetry_client import TelemetryClient
telemetry_client = TelemetryClient()
@app.post("/generate")
async def generate(request: InferenceRequest, http_request: Request):
request_id = http_request.headers.get("X-Request-ID", "unknown")
try:
start_time = time.perf_counter()
# 在真实项目中，这里会有更复杂的逻辑来处理流式生成和首个 token 延迟
# 为简化，我们一次性生成
outputs = generator(
request.prompt,
max_new_tokens=request.max_new_tokens,
num_return_sequences=1
)
end_time = time.perf_counter()
generated_text = outputs[0]['generated_text']
# --- 核心遥测数据采集 ---
# 这些指标是标准 APM 工具无法提供的
input_tokens = len(tokenizer.encode(request.prompt))
output_tokens = len(tokenizer.encode(generated_text)) - input_tokens
inference_duration_ms = (end_time - start_time) * 1000
tokens_per_sec = (output_tokens / (inference_duration_ms / 1000)) if inference_duration_ms > 0 else 0
metrics_payload = {
"measurement": "inference_performance",
"tags": {
"model_id": MODEL_ID,
"model_version": "v1.0.0", # 应从配置或模型元数据中获取
"client_id": http_request.headers.get("X-Client-ID", "default"),
},
"fields": {
"inference_duration_ms": inference_duration_ms,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"tokens_per_sec": tokens_per_sec,
"ttft_ms": -1, # 首token延迟，非流式为-1
}
}
# 异步发送，避免阻塞请求响应
asyncio.create_task(telemetry_client.emit_metrics(metrics_payload))
return {"response": generated_text}
except Exception as e:
logger.error(f"推理请求失败 (Request ID: {request_id}): {e}", exc_info=True)
# 发送错误指标
error_payload = {
"measurement": "inference_errors",
"tags": {"model_id": MODEL_ID},
"fields": {"count": 1}
}
asyncio.create_task(telemetry_client.emit_metrics(error_payload))
raise HTTPException(status_code=500, detail="模型推理时发生内部错误")
问题在于 telemetry_client 应该将数据发送到哪里。这引出了我们的第一个架构决策点。
方案A：完全依赖 Datadog Custom Metrics
最直接的方案是利用 Datadog 的 Custom Metrics API。
优势分析:
- 统一平台: 所有指标、日志、APM 数据都在一个地方，便于关联分析。运维团队无需学习新工具。
- 开箱即用: Datadog Agent (DogStatsD) 提供了非常成熟的客户端库，可以轻松地从 Python 应用发送指标，无需我们自己管理数据摄入的端点。
- 强大的告警和仪表盘: Datadog 在告警和可视化方面非常成熟，可以快速搭建监控仪表盘并设置复杂的告警规则。
劣势分析:
- 成本问题: Datadog Custom Metrics 的定价模型基于不同的时间序列数量。对于我们这种场景，
model_id,model_version,client_id等标签的笛卡尔积会迅速产生数万甚至数十万个唯一的时间序列。这会导致成本失控，尤其是当我们想要增加更多维度（如A/B_test_group）时。 - 查询灵活性受限: Datadog 的查询语言（Metrics Query）虽然强大，但在处理复杂的、探索性的分析时，不如 SQL 灵活。例如，要分析“所有版本号以
v2.开头且client_id为corp-A的模型的 P99tokens_per_sec与input_tokens数量的关联性”，用 SQL 表达会更直观、更强大。 - 数据保留策略: 长期的数据保留在 Datadog 中成本高昂。而对于模型性能漂移分析，我们可能需要回溯数月甚至一年的原始数据，进行复杂的聚合和统计。
在真实项目中，成本是决定技术选型的一个关键因素。将所有高基数遥测数据直接灌入 Datadog 是一条通往巨额账单的捷径。
方案B：引入 TimescaleDB 的混合观测架构
该方案的核心思想是数据分层。
- 热路径 (Hot Path): 关键的、低基数的健康指标（如总请求数、错误率、平均延迟）和服务状态，仍然发送到 Datadog。这些是触发紧急告警、用于 SRE On-Call 的核心指标。
- 温/冷路径 (Warm/Cold Path): 所有高基数的、详细的推理遥测数据（如上文代码中采集的
inference_performance），被发送到一个我们自己管理的 TimescaleDB 实例中。
优势分析:
- 成本效益: TimescaleDB 是基于 PostgreSQL 的开源时序数据库。我们可以将其部署在自己的基础设施或云上，存储成本远低于 Datadog Custom Metrics。对于 TB 级别的遥测数据，成本优势极其显著。
- 查询的终极灵活性: 我们可以利用完整的 SQL 功能，包括窗口函数、JOIN、公用表表达式（CTE）等，进行深度的数据挖掘和分析。这对于数据科学家和模型开发者来说是巨大的福音。
- 灵活的数据保留和降采样: TimescaleDB 内置了强大的数据生命周期管理功能。我们可以轻松设置策略，例如：保留原始数据 30 天，然后将其降采样为每小时的均值/最大值/P95 并保留一年，最后自动删除旧数据。这一切都可以通过简单的 SQL 命令完成。
- 与 Datadog 的集成: 这不是一个非此即彼的选择。我们可以构建一个桥梁：通过一个自定义的 Datadog Agent Check，定期查询 TimescaleDB 中的聚合数据，然后将这些低基数的聚合结果作为 Custom Metrics 发送给 Datadog。
劣势分析:
- 架构复杂度增加: 我们需要引入并维护一个新的数据库组件（TimescaleDB）。这包括部署、备份、监控、扩容等运维工作。
- 自建可视化和告警: 对于存储在 TimescaleDB 中的数据，我们需要使用 Grafana 等工具自建仪表盘。虽然 Grafana 功能强大，但相比 Datadog，需要更多的配置工作。告警也需要依赖 Grafana Alerting 或其他外部系统。
- 开发成本: 需要开发一个遥测数据收集器，以及前面提到的 Datadog Agent Check。
最终选择与理由：拥抱混合架构
经过权衡，我们选择了方案B。理由如下：
模型的性能和行为分析是我们业务的核心需求，而不是一个可有可无的附加项。牺牲查询灵活性和数据深度来换取暂时的便利是不可接受的。成本是另一个决定性因素，方案A的长期成本增长曲线是不可持续的。
虽然方案B引入了运维复杂性，但这种复杂性是可控的。TimescaleDB 的成熟度和基于 PostgreSQL 的生态使其运维相对简单。通过将聚合数据回传给 Datadog，我们保留了统一告警平台的好处，实现了两全其美。
核心实现概览
下面是整个混合架构的核心代码和配置。
1. 整体架构图
使用 Mermaid.js 描绘数据流：
graph TD
A[Hugging Face Service on K8s Pods] -- gRPC/HTTP --> B(Telemetry Collector);
B -- Batch Insert --> C[TimescaleDB];
subgraph Datadog Ecosystem
D[Datadog Agent]
end
subgraph Self-Managed
C
B
end
D -- Custom Check (Python) --> C;
D -- Aggregated Metrics --> E[Datadog Platform];
A -- Standard APM/Logs via Agent --> E;
F[Grafana] -- SQL Queries --> C;
G[Data Scientist/ML Engineer] -- Ad-hoc SQL --> C;
H[SRE/On-Call] -- Dashboards/Alerts --> E;
style C fill:#f9f,stroke:#333,stroke-width:2px
style B fill:#bbf,stroke:#333,stroke-width:2px
2. TimescaleDB Schema 设计
数据库表的设计至关重要。我们利用 TimescaleDB 的 hypertables 和 JSONB 类型。
-- 连接到你的数据库后执行
-- 1. 确保 timescaledb 扩展已创建
CREATE EXTENSION IF NOT EXISTS timescaledb;
-- 2. 创建主表用于存储详细的推理遥测数据
CREATE TABLE inference_telemetry (
time TIMESTAMPTZ NOT NULL,
model_id TEXT NOT NULL,
model_version TEXT NOT NULL,
client_id TEXT,
-- 使用 tags 字段存储其他高基数标签，便于扩展
tags JSONB,
-- 核心性能指标
inference_duration_ms DOUBLE PRECISION,
input_tokens INT,
output_tokens INT,
tokens_per_sec DOUBLE PRECISION,
ttft_ms DOUBLE PRECISION
);
-- 3. 将其转换成 Hypertables，这是 TimescaleDB 的核心
-- 按 'time' 列分区，每个 chunk 存储 1 天的数据
SELECT create_hypertable('inference_telemetry', 'time', chunk_time_interval => INTERVAL '1 day');
-- 4. 为常用查询字段创建索引，特别是标签
CREATE INDEX ON inference_telemetry (model_id, time DESC);
CREATE INDEX ON inference_telemetry (client_id, time DESC);
-- 使用 GIN 索引加速对 JSONB 标签的查询
CREATE INDEX ON inference_telemetry USING GIN (tags);
-- 5. 设置数据保留策略：自动删除超过90天的原始数据
SELECT add_retention_policy('inference_telemetry', INTERVAL '90 days');
-- 6. (可选) 设置连续聚合，用于性能优化
-- 创建一个物化视图，每小时计算一次 P95/P99 延迟
CREATE MATERIALIZED VIEW inference_telemetry_hourly
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 hour', time) AS bucket,
model_id,
client_id,
approx_percentile(0.95, percentile_agg(inference_duration_ms)) as p95_duration_ms,
approx_percentile(0.99, percentile_agg(inference_duration_ms)) as p99_duration_ms,
avg(tokens_per_sec) as avg_tokens_per_sec,
sum(output_tokens) as total_output_tokens,
count(*) as request_count
FROM inference_telemetry
GROUP BY bucket, model_id, client_id;
-- 设置策略，让物化视图每30分钟刷新一次
SELECT add_continuous_aggregate_policy('inference_telemetry_hourly',
start_offset => INTERVAL '1 hour',
end_offset => INTERVAL '10 minutes',
schedule_interval => INTERVAL '30 minutes');
这里的关键点是使用 create_hypertable，它会自动在后台对数据按时间进行分区（chunking），使得时间范围查询和数据删除操作极其高效。连续聚合（Continuous Aggregates）是另一个杀手级特性，它能自动维护聚合数据，避免在高层仪表盘上进行昂贵的实时计算。
3. 遥测收集器与客户端 (telemetry_client.py)
为了解耦和提高性能，我们不直接在 FastAPI 服务中写入数据库，而是通过一个轻量级的异步客户端将数据发送到一个独立的收集器服务。这里为了简化，我们展示一个直接写入数据库的异步客户端实现。在真实生产中，中间应该有一个消息队列（如 Kafka 或 aio-pika）来缓冲峰值流量。
# telemetry_client.py
import os
import logging
import asyncio
import asyncpg
from typing import Dict, Any
logger = logging.getLogger(__name__)
# 从环境变量获取配置，这是生产实践
DB_USER = os.environ.get("TSDB_USER", "postgres")
DB_PASS = os.environ.get("TSDB_PASS", "password")
DB_HOST = os.environ.get("TSDB_HOST", "localhost")
DB_NAME = os.environ.get("TSDB_NAME", "metrics")
DSN = f"postgres://{DB_USER}:{DB_PASS}@{DB_HOST}/{DB_NAME}"
class TelemetryClient:
def __init__(self, dsn: str = DSN, pool_size: int = 10):
self._dsn = dsn
self._pool = None
self._pool_size = pool_size
self._lock = asyncio.Lock()
async def _get_pool(self):
"""延迟初始化连接池，确保在事件循环中创建。"""
if self._pool is None:
async with self._lock:
if self._pool is None:
try:
self._pool = await asyncpg.create_pool(
self._dsn,
min_size=1,
max_size=self._pool_size
)
logger.info("TimescaleDB 连接池已创建。")
except Exception as e:
logger.error(f"创建 TimescaleDB 连接池失败: {e}", exc_info=True)
raise
return self._pool
async def emit_metrics(self, payload: Dict[str, Any]):
"""
异步发送指标到 TimescaleDB。
在生产环境中，这里应该有重试和死信队列逻辑。
"""
try:
pool = await self._get_pool()
async with pool.acquire() as connection:
# 假设 payload['measurement'] 对应表名
table_name = payload.get("measurement", "inference_telemetry")
if table_name != "inference_telemetry":
logger.warning(f"不支持的 measurement: {table_name}")
return
tags_data = payload.get("tags", {})
fields_data = payload.get("fields", {})
await connection.execute("""
INSERT INTO inference_telemetry (
time, model_id, model_version, client_id, tags,
inference_duration_ms, input_tokens, output_tokens,
tokens_per_sec, ttft_ms
) VALUES (
NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9
)
""",
tags_data.get("model_id"),
tags_data.get("model_version"),
tags_data.get("client_id"),
None, # 暂不使用 tags JSONB 字段
fields_data.get("inference_duration_ms"),
fields_data.get("input_tokens"),
fields_data.get("output_tokens"),
fields_data.get("tokens_per_sec"),
fields_data.get("ttft_ms")
)
except Exception as e:
# 这里的错误处理至关重要，不能因为遥测失败而影响主服务
logger.error(f"发送遥测数据失败: {e}", exc_info=False) # 在高流量下避免 exc_info4. Datadog 自定义 Agent Check
这是连接两个世界的桥梁。我们编写一个 Python 类，继承自 Datadog Agent 的 AgentCheck。此代码将被部署到 Datadog Agent 的 checks.d 目录中。
timescaledb_bridge.py
# /etc/datadog-agent/checks.d/timescaledb_bridge.py
import psycopg2
from datadog_checks.base import AgentCheck
class TimescaleDBBridgeCheck(AgentCheck):
def check(self, instance):
db_host = instance.get('host')
db_port = instance.get('port', 5432)
db_user = instance.get('username')
db_password = instance.get('password')
db_name = instance.get('dbname')
# 标签会附加到所有从这个 check 发送的指标上
custom_tags = instance.get('tags', [])
connection = None
try:
connection = psycopg2.connect(
host=db_host,
port=db_port,
user=db_user,
password=db_password,
dbname=db_name,
connect_timeout=5
)
self.log.info("成功连接到 TimescaleDB")
with connection.cursor() as cursor:
# --- 查询1: 获取每分钟每个模型的 P95 延迟和请求数 ---
# 查询过去5分钟的数据，避免因时钟偏差错过数据
cursor.execute("""
SELECT
model_id,
approx_percentile(0.95, percentile_agg(inference_duration_ms)),
count(*)
FROM inference_telemetry
WHERE time > NOW() - INTERVAL '5 minutes'
GROUP BY model_id;
""")
for row in cursor.fetchall():
model_id, p95_latency, count = row
if model_id and p95_latency is not None:
metric_tags = custom_tags + [f"model_id:{model_id}"]
self.gauge('llm.inference.p95_latency_ms', float(p95_latency), tags=metric_tags)
self.gauge('llm.inference.request_count_min', float(count) / 5.0, tags=metric_tags) # 转换为每分钟速率
# --- 查询2: 获取总的错误率 ---
# 这是一个假设的错误表
# cursor.execute("...")
self.service_check('timescaledb.can_connect', AgentCheck.OK, tags=custom_tags)
except Exception as e:
self.log.error(f"连接或查询 TimescaleDB 失败: {e}")
self.service_check('timescaledb.can_connect', AgentCheck.CRITICAL, message=str(e), tags=custom_tags)
finally:
if connection:
connection.close()
timescaledb_bridge.yaml
# /etc/datadog-agent/conf.d/timescaledb_bridge.d/conf.yaml
instances:
- host: 'your_timescaledb_host'
port: 5432
username: 'datadog_reader' # 最佳实践：创建一个只读用户
password: 'your_read_only_password'
dbname: 'metrics'
# 这些标签将附加到所有指标上
tags:
- 'env:production'
- 'service:llm-observability-bridge'
# 日志配置（可选）
logs:
- type: file
path: /var/log/datadog/agent.log
source: datadog-agent部署后，Datadog Agent 会每隔15秒（默认）执行 check 方法，查询 TimescaleDB，并将结果 llm.inference.p95_latency_ms 和 llm.inference.request_count_min 作为 Custom Metrics 发送到 Datadog。这样，我们就可以在 Datadog 中为这些关键聚合指标设置告警，而无需为数百万个原始时间序列付费。
架构的扩展性与局限性
此架构并非一劳永逸。一个明显的局限性是，它在应用和数据库之间引入了一个自定义的数据通路，这意味着日志、指标和追踪的自动关联性不如单一平台（如纯 Datadog）来得直接。要实现从 Datadog 仪表盘下钻到 TimescaleDB 中的特定原始数据，需要开发额外的前端或使用 Grafana 链接。
扩展路径是清晰的。我们可以利用 TimescaleDB 的强大 SQL 能力进行更复杂的模型行为分析。例如，定期运行一个任务，分析模型输出的文本（如果合规允许存储），计算词汇多样性、重复率或情感得分，并将这些“模型质量”指标存回数据库。这为我们提供了一个强大的平台来检测模型漂移，而不仅仅是性能衰退。
另一个方向是优化遥测收集器，将其构建为一个高可用的独立服务，使用 gRPC 协议，并引入批处理和压缩机制，以应对更大规模的推理集群。同时，对 TimescaleDB 本身，随着数据量的增长，需要考虑多节点部署、读写分离等更高级的数据库架构。
