在业务中大规模集成 AI 大模型时，你是否遇到过这样的困境：用户反馈响应慢，但后台日志却一切正常；月度账单上的 Token 消耗远超预期，却找不到具体是哪个接口或哪个用户消耗的？传统的应用性能监控（APM）工具对 AI 调用的延迟、Token 消耗等关键指标往往无能为力，导致成本失控和体验下降成为“黑盒”。
本文将为你拆解一套完整的 AI 服务监控解决方案。我们将聚焦于两个核心可观测性指标：响应延迟与Token 消耗，并使用OpenTelemetry进行指标采集，最终将数据存储和展示在ClickHouse中。通过本篇实战指南，你将能搭建一个从数据采集、存储到可视化分析的完整监控链路，无论是评估模型性能、优化提示词工程，还是进行精细化的成本核算，都能做到心中有数。
1. 核心概念：为什么需要专门的 AI 监控？
在深入技术实现之前，我们首先要理解监控 AI 服务与传统 Web 服务的本质区别。
1.1 AI 服务的独特挑战
传统的 Web 服务监控主要关注请求 QPS、错误率、CPU/内存使用率、数据库查询耗时等。而一次 AI 模型调用（例如调用 OpenAI GPT-4 或本地部署的 Llama），其核心成本与性能体现在：
- Token 消耗：这是 AI 服务最直接的成本驱动因素。无论是输入（Prompt）还是输出（Completion），都按 Token 数量计费。监控每个请求的输入/输出 Token 数，是进行成本分摊、识别异常消耗（如提示词泄露导致长输出）、优化提示词效率的基础。
- 响应延迟：AI 模型的推理时间通常远长于简单的数据库查询。延迟包括网络传输、模型加载、推理计算等多个环节。监控 P50、P95、P99 分位的延迟，对于保障用户体验、设定合理的超时时间、评估不同模型或硬件的性能至关重要。
- 模型与参数：同一个服务可能调用不同的模型（如
gpt-3.5-turbo与gpt-4），或使用不同的参数（如temperature,max_tokens）。监控时需要区分这些维度，才能进行有效的对比分析。
1.2 监控架构概览
我们的目标是构建一个轻量、高效、可扩展的监控系统。整体架构如下：
[你的AI应用] --(发射指标)--> [OpenTelemetry Collector] --(写入)--> [ClickHouse] --(查询)--> [Grafana]- 数据采集层 (OpenTelemetry SDK)：集成到你的 AI 应用代码中，在每次调用 AI 模型时，记录耗时、Token 数等指标。
- 收集与转发层 (OpenTelemetry Collector)：接收来自多个应用实例的指标数据，进行聚合、批处理，并导出到指定的存储后端。
- 数据存储层 (ClickHouse)：一个高性能的列式数据库，特别适合存储和快速查询时序指标数据。
- 可视化层 (Grafana)：从 ClickHouse 中读取数据，绘制丰富的监控仪表盘。
接下来，我们将从环境准备开始，一步步实现这个架构。
2. 环境准备与版本说明
在开始动手之前，请确保你的开发环境满足以下要求。本文示例将使用 Python 作为 AI 应用的语言，但 OpenTelemetry 的概念是语言无关的。
2.1 基础软件环境
- 操作系统：Linux (Ubuntu 20.04/22.04)、macOS 或 WSL2。大部分命令在 Linux 环境下进行。
- Docker & Docker Compose：我们将使用容器化方式快速部署 OpenTelemetry Collector 和 ClickHouse。请确保已安装。
# 检查安装 docker --version docker-compose --version - Python：版本 3.8 及以上。我们将使用
openai库模拟 AI 调用。python3 --version pip3 --version
2.2 核心组件版本
为了确保兼容性，以下是本文演示所用的主要组件版本。你的实际环境可以略有不同，但建议保持大版本一致。
| 组件 | 版本 | 说明 |
|---|---|---|
| OpenTelemetry Python SDK | 1.24.0 | 用于在应用中埋点 |
| OpenTelemetry Collector | 0.104.0(Docker 镜像) | 指标收集与导出 |
| ClickHouse | 24.8.2-alpine(Docker 镜像) | 指标存储 |
| Grafana | 11.2.0(Docker 镜像) | 数据可视化 |
| OpenAI Python Client | 1.30.1 | 模拟 AI 调用 |
重要提示：OpenTelemetry 生态系统更新较快，配置方式可能随版本变化。本文的代码和配置基于上述版本测试通过，如果你的版本不同，请参考官方文档进行调整。
3. OpenTelemetry 与 ClickHouse 基础
3.1 OpenTelemetry 简介
OpenTelemetry (简称 OTel) 是一个云原生计算基金会 (CNCF) 下的项目，旨在提供一套统一的 API、SDK 和工具，用于采集、生成遥测数据（包括指标、链路追踪和日志）。它的核心优势在于标准化和供应商中立。
对于 AI 监控场景，我们主要使用其Metrics SDK。一个Meter工具可以创建各种指标，例如：
- Counter：单调递增的累计值，适合记录总请求数、总 Token 消耗量。
- Histogram：记录可聚合的数值分布，完美契合测量请求延迟、单次请求的 Token 数。
3.2 ClickHouse 为何适合监控数据
ClickHouse 是一个开源的列式 OLAP 数据库，以其惊人的查询速度著称。对于监控场景，它有如下优势：
- 高性能聚合：对时间序列数据的
GROUP BY、SUM、AVG等聚合查询极快。 - 高压缩比：列式存储和高效压缩算法，大幅降低存储成本。
- TTL (生存时间)：可以轻松为表设置数据自动过期策略，符合监控数据“近期热、远期冷”的特点。
- 丰富的表引擎：
MergeTree系列引擎（特别是SummingMergeTree、AggregatingMergeTree）是为聚合数据量身定做的。
我们将使用 OpenTelemetry Collector 的clickhouseexporter，将指标直接写入 ClickHouse 的特定表中。
4. 搭建监控基础设施：ClickHouse 与 Collector
我们首先使用 Docker Compose 搭建数据存储和收集层。
4.1 编写 Docker Compose 文件
创建一个项目目录ai-monitor-demo，并在其中创建docker-compose.yml文件。
# docker-compose.yml version: '3.8' services: clickhouse: image: clickhouse/clickhouse-server:24.8.2-alpine container_name: ai-monitor-clickhouse ports: - "8123:8123" # HTTP API 端口 - "9000:9000" # 原生TCP客户端端口 volumes: - ./clickhouse/data:/var/lib/clickhouse - ./clickhouse/config.xml:/etc/clickhouse-server/config.xml - ./clickhouse/users.xml:/etc/clickhouse-server/users.xml environment: - CLICKHOUSE_DB=otel - CLICKHOUSE_USER=admin - CLICKHOUSE_PASSWORD=admin123 ulimits: nproc: 65535 nofile: soft: 262144 hard: 262144 networks: - otel-network otel-collector: image: otel/opentelemetry-collector-contrib:0.104.0 container_name: ai-monitor-otel-collector command: ["--config=/etc/otel-collector-config.yaml"] volumes: - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml ports: - "4317:4317" # OTLP gRPC 接收端口 - "4318:4318" # OTLP HTTP 接收端口 - "8889:8889" # 健康检查/指标端口 - "13133:13133" # 健康检查扩展端口 depends_on: - clickhouse networks: - otel-network grafana: image: grafana/grafana:11.2.0 container_name: ai-monitor-grafana ports: - "3000:3000" environment: - GF_SECURITY_ADMIN_PASSWORD=admin123 volumes: - ./grafana/provisioning:/etc/grafana/provisioning - ./grafana/dashboards:/var/lib/grafana/dashboards depends_on: - clickhouse networks: - otel-network networks: otel-network: driver: bridge4.2 配置 OpenTelemetry Collector
创建otel-collector-config.yaml文件。这个配置定义了 Collector 如何接收指标（通过 OTLP），以及如何将其导出到 ClickHouse。
# otel-collector-config.yaml receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 exporters: debug: verbosity: detailed clickhouse: endpoint: tcp://clickhouse:9000?database=otel username: admin password: admin123 ttl: 720h # 数据保留30天 timeout: 5s logs_table_name: otel_logs traces_table_name: otel_traces metrics_table_name: otel_metrics # 针对指标表的额外配置 metrics: # 使用 TTL 并设置存储策略 ttl: 720h # 定义表结构，映射 OpenTelemetry 指标到 ClickHouse 列 table_columns: - name: ResourceAttributes type: Map(LowCardinality(String), String) - name: ScopeName type: LowCardinality(String) - name: ScopeVersion type: LowCardinality(String) - name: MetricName type: LowCardinality(String) - name: MetricDescription type: String - name: MetricUnit type: LowCardinality(String) - name: Attributes type: Map(LowCardinality(String), String) - name: StartTimeUnix type: UInt64 - name: TimeUnix type: UInt64 - name: Value type: Float64 - name: Flags type: UInt32 - name: HistogramCounts type: Array(UInt64) - name: HistogramBounds type: Array(Float64) - name: Exemplars type: String processors: batch: timeout: 5s send_batch_size: 1000 extensions: health_check: endpoint: 0.0.0.0:13133 pprof: endpoint: 0.0.0.0:1777 service: extensions: [pprof, health_check] pipelines: metrics: receivers: [otlp] processors: [batch] exporters: [debug, clickhouse] # debug 用于调试，生产可移除4.3 启动基础设施
在项目根目录下运行：
docker-compose up -d等待所有容器启动成功。你可以使用docker-compose logs -f查看日志。
验证服务：
- ClickHouse：访问
http://localhost:8123/play，使用用户名admin和密码admin123登录。执行SHOW DATABASES;应能看到otel数据库。 - Grafana：访问
http://localhost:3000，使用用户名admin和密码admin123登录。 - Collector：访问
http://localhost:13133应返回{"status":"Server available"}。
至此，监控的后端基础设施已就绪。
5. 在 Python AI 应用中集成监控
现在，我们编写一个简单的 Python 应用，模拟调用 AI 模型，并使用 OpenTelemetry 发送指标。
5.1 创建 Python 虚拟环境与依赖
在项目根目录外，创建一个新的应用目录ai-app。
mkdir ai-app && cd ai-app python3 -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate安装必要的 Python 包：
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-metrics pip install openai # 用于模拟AI调用5.2 编写核心监控与 AI 调用代码
创建文件app_with_monitoring.py：
# app_with_monitoring.py import time import random from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.sdk.resources import Resource # 1. 定义资源，标识你的服务 resource = Resource.create({ "service.name": "ai-text-generation-service", "service.version": "1.0.0", "deployment.environment": "demo", }) # 2. 配置指标导出到 OTLP Collector (HTTP) metric_exporter = OTLPMetricExporter( endpoint="http://localhost:4318/v1/metrics", # Collector 的 OTLP HTTP 端口 # 可选：添加认证头等 ) metric_reader = PeriodicExportingMetricReader(exporter=metric_exporter, export_interval_millis=5000) # 每5秒导出一次 # 3. 设置全局的 MeterProvider provider = MeterProvider( resource=resource, metric_readers=[metric_reader], ) metrics.set_meter_provider(provider) # 4. 创建 Meter meter = metrics.get_meter(__name__) # 5. 创建我们需要的指标 # Counter: 记录总请求数和总Token消耗 request_counter = meter.create_counter( name="ai.requests.total", description="Total number of AI model requests", unit="1", ) input_token_counter = meter.create_counter( name="ai.tokens.input.total", description="Total number of input tokens consumed", unit="1", ) output_token_counter = meter.create_counter( name="ai.tokens.output.total", description="Total number of output tokens consumed", unit="1", ) # Histogram: 记录请求延迟和每次请求的Token数分布 request_duration_histogram = meter.create_histogram( name="ai.request.duration", description="Duration of AI model requests", unit="ms", ) request_token_histogram = meter.create_histogram( name="ai.request.tokens.total", description="Total tokens (input+output) per request", unit="1", ) def simulate_ai_call(prompt: str, model: str = "gpt-3.5-turbo"): """ 模拟调用 AI 模型。 在实际项目中，这里应替换为真实的 OpenAI、Azure OpenAI 或本地模型的调用。 """ # 模拟网络和计算延迟 (50ms ~ 2000ms) latency_ms = random.randint(50, 2000) time.sleep(latency_ms / 1000.0) # 模拟 Token 计数：简单假设每个字符约等于 0.25 个 token input_tokens = int(len(prompt) * 0.25) + random.randint(1, 10) # 模拟生成长度不等的回复 output_length = random.randint(20, 200) output_tokens = int(output_length * 0.25) + random.randint(1, 20) # 模拟小概率失败 if random.random() < 0.05: # 5% 失败率 raise Exception("Simulated AI API failure") return { "content": "Simulated AI response with length " + str(output_length), "input_tokens": input_tokens, "output_tokens": output_tokens, "latency_ms": latency_ms, "model": model } def process_user_request(user_id: str, prompt: str, model: str = "gpt-3.5-turbo"): """ 处理用户请求，并记录监控指标。 """ start_time = time.time() attributes = { "user.id": user_id, "ai.model": model, "status.code": "200" # 默认成功 } try: # 调用 AI response = simulate_ai_call(prompt, model) duration_ms = (time.time() - start_time) * 1000 # 记录指标 request_counter.add(1, attributes) input_token_counter.add(response["input_tokens"], attributes) output_token_counter.add(response["output_tokens"], attributes) request_duration_histogram.record(duration_ms, attributes) total_tokens = response["input_tokens"] + response["output_tokens"] request_token_histogram.record(total_tokens, attributes) print(f"Request from {user_id} succeeded. Tokens: {total_tokens}, Latency: {duration_ms:.2f}ms") return response except Exception as e: duration_ms = (time.time() - start_time) * 1000 # 记录失败的请求，状态码标记为错误 error_attributes = attributes.copy() error_attributes["status.code"] = "500" request_counter.add(1, error_attributes) request_duration_histogram.record(duration_ms, error_attributes) print(f"Request from {user_id} failed: {e}") return None if __name__ == "__main__": print("Starting AI service with OpenTelemetry monitoring...") # 模拟连续处理一些请求 users = ["user_001", "user_002", "user_003", "user_004"] models = ["gpt-3.5-turbo", "gpt-4"] prompts = [ "Explain quantum computing in simple terms.", "Write a Python function to calculate Fibonacci sequence.", "What are the benefits of renewable energy?", "Summarize the history of the Internet." ] for i in range(20): # 模拟20个请求 user = random.choice(users) model = random.choice(models) prompt = random.choice(prompts) process_user_request(user, prompt, model) time.sleep(random.uniform(0.5, 2.0)) # 模拟随机请求间隔 print("Simulation finished. Metrics are being exported...") # 等待指标导出器完成最后的推送 time.sleep(10) print("Done.")5.3 运行应用并查看数据
- 确保
docker-compose服务仍在运行。 - 在
ai-app目录下，运行 Python 脚本：python app_with_monitoring.py - 观察控制台输出，会看到模拟的请求成功与失败信息。
- 登录 ClickHouse (
http://localhost:8123/play)，查询是否已收到指标数据：
你应该能看到USE otel; SELECT DISTINCT MetricName FROM otel_metrics ORDER BY MetricName;ai.requests.total,ai.request.duration等我们定义的指标名。 - 查询具体的指标数据：
此查询会显示最近几条请求延迟的直方图数据。SELECT toDateTime(TimeUnix/1000000000) as time, MetricName, Attributes['user.id'] as user, Attributes['ai.model'] as model, Value, HistogramBounds, HistogramCounts FROM otel_metrics WHERE MetricName = 'ai.request.duration' ORDER BY time DESC LIMIT 5;
6. 在 Grafana 中可视化监控数据
数据已进入 ClickHouse，现在我们在 Grafana 中创建仪表盘。
6.1 配置 ClickHouse 数据源
- 登录 Grafana (
http://localhost:3000)，默认账号admin/admin123。 - 点击左侧齿轮图标
Configuration->Data sources。 - 点击
Add data source，搜索并选择ClickHouse。 - 配置连接：
- Name:
ClickHouse-OTel - Host:
clickhouse:8123（注意：因为 Grafana 和 ClickHouse 在同一 Docker 网络otel-network下，所以可以用服务名） - Database:
otel - User:
admin - Password:
admin123 - Protocol:
HTTP
- Name:
- 点击
Save & test，应显示 “Data source is working”。
6.2 创建监控仪表盘
我们可以创建几个关键面板：
面板 1：请求速率与错误率
- 查询(请求总量)：
SELECT $timeSeries as t, count(*) as value FROM $table WHERE $timeFilter AND MetricName = 'ai.requests.total' GROUP BY t ORDER BY t - 查询(错误请求量，属性
status.code='500')：SELECT $timeSeries as t, count(*) as value FROM $table WHERE $timeFilter AND MetricName = 'ai.requests.total' AND Attributes['status.code'] = '500' GROUP BY t ORDER BY t - 可视化：使用
Stat或Time series图表。可以计算错误率：错误数 / 总数 * 100%。
面板 2：平均响应延迟与 P99 延迟
- 查询(平均延迟，需要利用直方图数据计算，这里简化查询平均值)：
注意：更精确的百分位数计算需要在查询时展开SELECT $timeSeries as t, avg(Value) as value FROM $table WHERE $timeFilter AND MetricName = 'ai.request.duration' AND Attributes['status.code'] = '200' -- 只看成功的请求 GROUP BY t ORDER BY tHistogramBounds和HistogramCounts列，或使用 ClickHouse 的quantile函数对Value进行估算。生产环境建议对直方图数据进行预聚合。
面板 3：Token 消耗趋势（按用户/模型）
- 查询(总输入 Token)：
SELECT $timeSeries as t, sum(Value) as value FROM $table WHERE $timeFilter AND MetricName = 'ai.tokens.input.total' GROUP BY t ORDER BY t - 查询(按模型分组)：
SELECT $timeSeries as t, Attributes['ai.model'] as metric, sum(Value) as value FROM $table WHERE $timeFilter AND MetricName = 'ai.tokens.input.total' GROUP BY t, metric ORDER BY t, metric - 可视化：使用
Time series图表，并开启Stack模式，可以清晰看到不同模型的 Token 消耗占比。
面板 4：单次请求 Token 数量分布
- 查询：
SELECT Value as tokens_per_request FROM $table WHERE $timeFilter AND MetricName = 'ai.request.tokens.total' AND Attributes['status.code'] = '200' - 可视化：使用
Histogram图表，可以直观看到大部分请求消耗的 Token 范围，有助于识别异常值（例如提示词泄露导致的长文本输出）。
将这些面板组合在一个仪表盘中，你就得到了一个专属的 AI 服务监控看板，可以实时观察服务的健康度、性能与成本。
7. 常见问题与排查思路
在搭建和使用过程中，你可能会遇到以下问题：
| 问题现象 | 可能原因 | 排查思路 |
|---|---|---|
Python 应用启动报错，提示opentelemetry-exporter-otlp相关错误 | 依赖版本不兼容或未安装 | 1. 检查pip list确认包已安装。2. 查看 OpenTelemetry Python SDK 和 Exporter 的版本兼容性，尽量使用较新且版本匹配的包。 |
| 应用运行后，ClickHouse 中查不到数据 | Collector 配置错误或网络不通 | 1. 检查 Collector 容器日志：docker-compose logs otel-collector。2. 确认 Python 应用中 endpoint指向正确的 Collector 地址和端口 (http://localhost:4318)。3. 在 Collector 配置中启用 debugexporter，查看是否收到数据。 |
| Grafana 中查询数据报错或为空 | 数据源配置错误或 SQL 查询语法问题 | 1. 在 Grafana 的Explore页面，使用配置好的 ClickHouse 数据源执行简单查询，如SELECT 1，测试连接。2. 检查 SQL 中的表名 ( otel_metrics)、字段名是否与 ClickHouse 中实际创建的表一致。3. 确认查询的时间范围 ( $timeFilter) 内有数据。 |
| 监控数据延迟很高 | Collector 的batch处理器配置或 MetricReader 导出间隔过长 | 1. 检查otel-collector-config.yaml中batch处理器的timeout（建议 5-10s）。2. 检查 Python 代码中 PeriodicExportingMetricReader的export_interval_millis（建议 5000-10000 ms）。3. 对于需要近实时监控的场景，可以适当缩短这些间隔，但会增加 Collector 负载。 |
| ClickHouse 磁盘空间增长过快 | 数据没有设置 TTL 或监控指标过于频繁 | 1. 确认 Collector 配置中ttl: 720h（30天）已生效。2. 可以在 ClickHouse 中为 otel_metrics表额外设置 TTL：ALTER TABLE otel_metrics MODIFY TTL TimeUnix + INTERVAL 30 DAY。3. 评估指标发射频率，非核心指标可以降低频率。 |
8. 最佳实践与工程建议
将监控系统投入生产环境时，需要考虑更多工程细节。
8.1 监控指标设计
- 遵循命名规范：使用点分隔的命名方式，如
ai.request.duration、business.order.value。添加前缀（如ai.）避免冲突。 - 精心设计属性 (Attributes)：属性是进行数据下钻 (drill-down) 分析的维度。像
user.id、ai.model、prompt.type、status.code都是非常有价值的属性。但注意，高基数字段（如直接使用用户ID）可能导致查询变慢，可以考虑使用哈希值或分组。 - 区分指标类型：
- Counter：用于只增不减的值（请求数、Token总数）。
- Histogram：用于记录分布（延迟、包大小、单个请求Token数）。
- Gauge：用于可增可减的瞬时值（并发请求数、内存使用量）。
8.2 性能与成本优化
- 采样与聚合：对于极高并发的服务，不是每个请求都需要记录完整的直方图。可以在 SDK 端或 Collector 端配置采样率，或使用
AggregatingMeterProvider在客户端进行预聚合。 - ClickHouse 表引擎优化：生产环境建议使用
AggregatingMergeTree或SummingMergeTree引擎来存储预聚合后的数据，而不是原始的指标数据，这能极大提升查询性能和降低存储成本。这通常需要在 Collector 或一个独立的聚合服务中完成。 - 控制数据粒度：根据需求决定数据存储的粒度。例如，原始数据保留7天，按小时聚合的数据保留30天，按天聚合的数据保留1年。
8.3 生产环境部署
- Collector 高可用：生产环境至少部署两个 Collector 实例，前端通过负载均衡器（如 Nginx）分发流量，避免单点故障。
- 安全：
- 为 ClickHouse 和 Grafana 配置强密码，并考虑网络隔离（如将 ClickHouse 置于内网，不暴露
8123端口到公网）。 - OTLP 端点可以考虑启用 TLS 加密传输。
- 为 ClickHouse 和 Grafana 配置强密码，并考虑网络隔离（如将 ClickHouse 置于内网，不暴露
- 资源限制：为 Docker 容器或 Pod 设置合理的 CPU 和内存限制，防止某个组件异常拖垮整个主机。
8.4 告警集成
监控的最终目的是发现问题并及时响应。在 Grafana 中，可以基于我们创建的仪表盘设置告警规则：
- 延迟告警：当
ai.request.duration的 P95 值超过 5 秒时触发。 - 错误率告警：当错误请求率 (
status.code='500'的请求占比) 连续 5 分钟超过 1% 时触发。 - Token 消耗异常告警：当某个用户的每小时 Token 消耗量突增 10 倍时触发，可能提示提示词被恶意利用或程序漏洞。
通过将 Grafana 告警连接到 Slack、钉钉、PagerDuty 等通知渠道，团队可以在第一时间获知服务异常。
从环境搭建、应用埋点、数据存储到可视化告警，我们完成了一个完整的 AI 服务可观测性闭环。这套方案的核心优势在于标准化和可扩展性——OpenTelemetry 让你未来可以无缝切换监控后端；ClickHouse 的高性能则确保了即使面对海量监控数据，查询也能快速响应。
