AI API 网关日志分析实战：OpenTelemetry + ClickHouse 成本优化完整指南

作为一名长期在生产环境运行 AI 应用的工程师，我深知 API 成本控制的重要性。让我先算一笔真实的账：假设你的应用每月消耗 100 万 output token，GPT-4.1 官方定价 $8/MTok 需要 $8000，Claude Sonnet 4.5 官方 $15/MTok 需要 $15000，而 HolySheep AI 按 ¥1=$1 结算（官方汇率 ¥7.3=$1），同样的调用量只需 ¥800 和 ¥1500。这意味着什么？每月节省超过 85% 的成本。这不是我编造的营销数字，而是实打实的汇率差带来的价值。今天我要分享的是如何通过 OpenTelemetry + ClickHouse 构建企业级日志分析系统，让你不仅能看见每一分钱的流向，还能找到优化空间。

为什么需要专业日志分析架构

我在项目中见过太多团队用最简单的 MySQL 记录 API 调用，结果在日志量超过 100 万条/天后查询直接超时。更糟糕的是，他们根本无法回答"这个月 DeepSeek V3.2 的平均延迟是多少"这样的基础问题。HolySheep AI 的日志虽然已经非常完善，但在对接多个模型供应商时，你需要一个统一的中转层来聚合所有数据。这就是 OpenTelemetry + ClickHouse 组合的核心价值：OpenTelemetry 提供标准的链路追踪能力，ClickHouse 提供亿级日志的秒级聚合查询。

架构设计概览

整体架构分为三个核心组件：OpenTelemetry SDK 负责在应用层埋点采集，OTEL Collector 作为中间件收集并转发数据，ClickHouse 作为存储和查询引擎。整个数据流延迟在 50ms 以内（国内直连实测），完全满足实时监控需求。

依赖安装与基础配置

首先安装所有必要的 Python 依赖包：

pip install opentelemetry-api opentelemetry-sdk \
    opentelemetry-exporter-otlp \
    opentelemetry-instrumentation-fastapi \
    opentelemetry-instrumentation-requests \
    opentelemetry-instrumentation-httpx \
    clickhouse-driver fastapi uvicorn pydantic \
    python-dotenv aiohttp

接下来创建项目配置文件 config.py：

import os
from dotenv import load_dotenv

load_dotenv()

HolySheep API 配置（国内直连 <50ms）
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

ClickHouse 连接配置
CLICKHOUSE_HOST = os.getenv("CLICKHOUSE_HOST", "localhost")
CLICKHOUSE_PORT = int(os.getenv("CLICKHOUSE_PORT", "9000"))
CLICKHOUSE_DATABASE = "ai_gateway_logs"
CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")

OpenTelemetry 配置
OTEL_ENDPOINT = os.getenv("OTEL_ENDPOINT", "http://localhost:4317")
SERVICE_NAME = "ai-api-gateway"
SERVICE_VERSION = "1.0.0"

采样配置
SAMPLE_RATE = float(os.getenv("SAMPLE_RATE", "1.0"))  # 1.0 = 100% 采样

OpenTelemetry SDK 初始化配置

这是整个链路追踪的核心模块。我建议将采样率设置在 10%-50% 之间，以平衡数据完整性和存储成本：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
import os

def setup_telemetry():
    """初始化 OpenTelemetry SDK"""
    
    # 创建资源信息
    resource = Resource(attributes={
        SERVICE_NAME: os.getenv("SERVICE_NAME", "ai-api-gateway"),
        SERVICE_VERSION: "1.0.0",
        "deployment.environment": os.getenv("ENV", "production"),
        "host.name": os.getenv("HOSTNAME", "gateway-01")
    })
    
    # 配置采样率（生产环境建议 0.3-0.5）
    sampler = TraceIdRatioBased(float(os.getenv("SAMPLE_RATE", "0.5")))
    
    # 创建 TracerProvider
    provider = TracerProvider(resource=resource, sampler=sampler)
    
    # 配置 OTLP 导出器（发送到 Collector）
    otlp_exporter = OTLPSpanExporter(
        endpoint=os.getenv("OTEL_ENDPOINT", "http://localhost:4317"),
        insecure=True
    )
    
    # 使用批量导出器，避免高频调用影响性能
    span_processor = BatchSpanProcessor(otlp_exporter)
    provider.add_span_processor(span_processor)
    
    # 设置全局 TracerProvider
    trace.set_tracer_provider(provider)
    
    return trace.get_tracer(__name__)

初始化并导出全局 tracer
tracer = setup_telemetry()

ClickHouse 表结构设计与分区策略

ClickHouse 的列式存储天然适合日志分析场景。我设计了按天分区的物化视图，既能快速清理过期数据，又能在查询时大幅减少扫描范围：

-- 创建数据库
CREATE DATABASE IF NOT EXISTS ai_gateway_logs ON CLUSTER default;

-- 主表：存储所有 API 调用原始日志
CREATE TABLE IF NOT EXISTS ai_gateway_logs.api_calls
(
    timestamp DateTime DEFAULT now(),
    request_id String,
    trace_id String,
    span_id String,
    
    -- 模型信息
    provider Enum8('openai' = 1, 'anthropic' = 2, 'google' = 3, 'deepseek' = 4, 'holysheep' = 5),
    model String,
    
    -- Token 统计
    prompt_tokens UInt32,
    completion_tokens UInt32,
    total_tokens UInt32 AS prompt_tokens + completion_tokens,
    
    -- 性能指标（毫秒）
    latency_ms UInt32,
    time_to_first_token_ms Nullable(UInt32),
    
    -- 请求详情
    request_url String,
    request_method String,
    http_status UInt16,
    
    -- 错误信息
    error_code Nullable(String),
    error_message Nullable(String),
    
    -- 成本计算
    input_cost_usd Float64,
    output_cost_usd Float64,
    total_cost_usd Float64,
    
    -- 额外维度
    user_id Nullable(String),
    api_key_hash String,
    ip_address Nullable(String),
    
    -- 元数据
    metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (provider, model, timestamp)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

-- 物化视图：按分钟聚合的性能统计
CREATE MATERIALIZED VIEW IF NOT EXISTS ai_gateway_logs.api_calls_minute_agg
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (provider, model, timestamp, http_status)
AS SELECT
    toStartOfMinute(timestamp) AS timestamp,
    provider,
    model,
    http_status,
    
    count() AS request_count,
    sum(prompt_tokens) AS total_prompt_tokens,
    sum(completion_tokens) AS total_completion_tokens,
    sum(total_tokens) AS total_tokens,
    
    avg(latency_ms) AS avg_latency_ms,
    quantile(0.5)(latency_ms) AS p50_latency_ms,
    quantile(0.95)(latency_ms) AS p95_latency_ms,
    quantile(0.99)(latency_ms) AS p99_latency_ms,
    
    avg(input_cost_usd) AS avg_input_cost,
    avg(output_cost_usd) AS avg_output_cost,
    sum(total_cost_usd) AS total_cost_usd,
    
    countIf(http_status >= 400) AS error_count
FROM ai_gateway_logs.api_calls
GROUP BY timestamp, provider, model, http_status;

-- 物化视图：按小时聚合的成本统计
CREATE MATERIALIZED VIEW IF NOT EXISTS ai_gateway_logs.api_calls_hourly_cost
ENGINE = SummingMergeTree()
ORDER BY (provider, model, timestamp)
AS SELECT
    toStartOfHour(timestamp) AS timestamp,
    provider,
    model,
    
    sum(total_cost_usd) AS total_cost_usd,
    sum(total_tokens) AS total_tokens,
    count() AS request_count
FROM ai_gateway_logs.api_calls
GROUP BY timestamp, provider, model;

ClickHouse 日志写入器实现

我实现了支持批量写入和断线重连的日志写入器，这是保证数据不丢失的关键：

from clickhouse_driver import Client
from typing import List, Dict, Any
from datetime import datetime
import logging
import json

logger = logging.getLogger(__name__)

class ClickHouseLogger:
    """ClickHouse 日志写入器，支持批量写入和缓冲"""
    
    def __init__(self, config: dict):
        self.host = config.get("host", "localhost")
        self.port = config.get("port", 9000)
        self.database = config.get("database", "ai_gateway_logs")
        self.user = config.get("user", "default")
        self.password = config.get("password", "")
        
        self._client = None
        self._buffer: List[Dict] = []
        self._buffer_size = 500  # 批量写入阈值
        self._flush_interval = 5  # 最多等待 5 秒
        
        # 初始化时创建表
        self._ensure_tables()
    
    @property
    def client(self) -> Client:
        if self._client is None:
            self._client = Client(
                host=self.host,
                port=self.port,
                database=self.database,
                user=self.user,
                password=self.password,
                connection_timeout=10,
                send_receive_timeout=30,
                max_execution_time=60
            )
        return self._client
    
    def _ensure_tables(self):
        """确保必要的表存在"""
        try:
            self.client.execute("""
                CREATE DATABASE IF NOT EXISTS ai_gateway_logs
            """)
            logger.info("ClickHouse database ensured")
        except Exception as e:
            logger.warning(f"Table creation check failed: {e}")
    
    def write_span(self, span_data: Dict[str, Any]):
        """写入单个 span 数据"""
        self._buffer.append(span_data)
        
        if len(self._buffer) >= self._buffer_size:
            self.flush()
    
    def write_spans(self, spans: List[Dict[str, Any]]):
        """批量写入多个 span"""
        self._buffer.extend(spans)
        
        if len(self._buffer) >= self._buffer_size:
            self.flush()
    
    def flush(self):
        """强制刷新缓冲区"""
        if not self._buffer:
            return
        
        try:
            # 构建批量插入语句
            columns = [
                'timestamp', 'request_id', 'trace_id', 'span_id',
                'provider', 'model', 'prompt_tokens', 'completion_tokens',
                'latency_ms', 'http_status', 'error_code', 'error_message',
                'input_cost_usd', 'output_cost_usd', 'total_cost_usd',
                'api_key_hash', 'ip_address'
            ]
            
            values = []
            for record in self._buffer:
                values.append((
                    record.get('timestamp', datetime.now()),
                    record.get('request_id', ''),
                    record.get('trace_id', ''),
                    record.get('span_id', ''),
                    record.get('provider', 'holysheep'),
                    record.get('model', ''),
                    record.get('prompt_tokens', 0),
                    record.get('completion_tokens', 0),
                    record.get('latency_ms', 0),
                    record.get('http_status', 200),
                    record.get('error_code'),
                    record.get('error_message'),
                    record.get('input_cost_usd', 0.0),
                    record.get('output_cost_usd', 0.0),
                    record.get('total_cost_usd', 0.0),
                    record.get('api_key_hash', ''),
                    record.get('ip_address')
                ))
            
            self.client.execute(
                f"INSERT INTO api_calls ({', '.join(columns)}) VALUES",
                values
            )
            
            logger.info(f"Flushed {len(self._buffer)} records to ClickHouse")
            self._buffer = []
            
        except Exception as e:
            logger.error(f"Failed to flush to ClickHouse: {e}")
            # 保留缓冲区，等待重试
            raise
    
    def __del__(self):
        """析构时确保数据刷新"""
        if self._buffer:
            try:
                self.flush()
            except:
                pass

全局日志写入器实例
ch_logger = ClickHouseLogger({
    "host": "your-clickhouse-host",
    "port": 9000,
    "database": "ai_gateway_logs",
    "user": "default",
    "password": ""
})

集成 HolySheep API 的网关实现

这里展示如何将 OpenTelemetry 追踪与 HolySheep AI 网关无缝集成。核心思路是在请求前后自动记录延迟、token 消耗和成本：

import httpx
import hashlib
import time
import uuid
from datetime import datetime
from typing import Optional, Dict, Any
from contextlib import asynccontextmanager

定价配置（$/MTok）- 2026年主流模型价格
MODEL_PRICING = {
    "gpt-4.1": {"input": 2.0, "output": 8.0},
    "gpt-4.1-mini": {"input": 0.3, "output": 1.2},
    "claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
    "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
    "gemini-2.5-flash": {"input": 0.125, "output": 2.50},
    "deepseek-v3.2": {"input": 0.1, "output": 0.42},
}

class AIServiceGateway:
    """AI 服务网关 - 集成 OpenTelemetry 追踪"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:16]
        
        # HTTP 客户端（连接池复用）
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(60.0, connect=10.0),
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
        )
        
        # 获取 tracer
        from opentelemetry import trace
        self.tracer = trace.get_tracer(__name__)
    
    async def chat_completion(
        self,
        model: str,
        messages: list,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        user_id: Optional[str] = None,
        ip_address: Optional[str] = None
    ) -> Dict[str, Any]:
        """调用 chat completion 并自动追踪"""
        
        request_id = str(uuid.uuid4())
        trace_id = format(uuid.uuid4().int, '032x')
        span_id = format(uuid.uuid4().int % (10**16), '016x')
        
        start_time = time.perf_counter()
        
        with self.tracer.start_as_current_span(
            f"chat.{model}",
            attributes={
                "request.id": request_id,
                "request.model": model,
                "request.message_count": len(messages),
                "request.temperature": temperature,
            }
        ) as span:
            try:
                # 构建请求
                payload = {
                    "model": model,
                    "messages": messages,
                    "temperature": temperature,
                }
                if max_tokens:
                    payload["max_tokens"] = max_tokens
                
                headers = {
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                    "X-Request-ID": request_id,
                    "X-Trace-ID": trace_id,
                }
                
                # 发送请求
                response = await self.client.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    headers=headers
                )
                
                elapsed_ms = int((time.perf_counter() - start_time) * 1000)
                
                # 解析响应
                if response.status_code == 200:
                    result = response.json()
                    usage = result.get("usage", {})
                    prompt_tokens = usage.get("prompt_tokens", 0)
                    completion_tokens = usage.get("completion_tokens", 0)
                    
                    # 计算成本
                    pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
                    input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
                    output_cost = (completion_tokens / 1_000_000) * pricing["output"]
                    total_cost = input
相关资源
📚 AI API 技术文章库
💰 查看价格
📖 开发者文档
🚀 免费注册
相关文章
Hybrid Search 混合检索实战：从报错到落地的完整指南
MCP Sampling 功能详解：让工具主动请求 LLM 推理
马来西亚开发者 AI API 本地化接入指南：2026 最新选型方案