你无法优化你无法衡量的。Token 用量追踪、per-user/per-endpoint 成本拆解、异常调用检测——先看到数据，再谈优化。
| 工具 | 类型 | 免费层 | 特点 |
|---|---|---|---|
| Helicone | LLM 可观测性 | 有限免费 | 请求日志、成本追踪、缓存分析、延迟监控 |
| Langfuse | LLM 可观测性 | 开源免费 | 追踪、评估、Prompt 管理、成本分析 |
| OpenAI Usage API | 官方 | 免费 | 直接从 OpenAI 获取用量数据 |
| Anthropic Usage | 官方 | 免费 | Anthropic 控制台查看用量 |
| 自建监控 | 自定义 | 免费 | 完全控制、高度定制 |
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
@dataclass
class LLMCallRecord:
"""LLM 调用记录"""
timestamp: datetime
user_id: str
tenant_id: str
session_id: str
model: str
endpoint: str # chat/completions, embeddings, etc.
input_tokens: int
output_tokens: int
cached_tokens: int # 缓存命中的 token
cost: float # 美元
latency_ms: float
status: str # success, error, timeout
error_type: Optional[str] = None
# 可选的详细信息
request_id: Optional[str] = None
feature: Optional[str] = None # 功能标签：chat, search, summary, etc.
class CostMonitor:
"""LLM 成本监控器"""
# 各模型的价格表 ($/1M tokens)
PRICING = {
'gpt-5.5': {'input': 5.00, 'cached_input': 0.50, 'output': 30.00},
'gpt-5.4': {'input': 2.50, 'cached_input': 0.25, 'output': 15.00},
'gpt-5.4-mini': {'input': 0.75, 'cached_input': 0.075, 'output': 4.50},
'gpt-4o': {'input': 2.50, 'cached_input': 1.25, 'output': 10.00},
'gpt-4o-mini': {'input': 0.15, 'cached_input': 0.075, 'output': 0.60},
'claude-opus-4-20250514': {'input': 15.00, 'cached_input': 1.50, 'output': 75.00},
'claude-sonnet-4-20250514': {'input': 3.00, 'cached_input': 0.30, 'output': 15.00},
'claude-3-5-haiku-20241022': {'input': 0.80, 'cached_input': 0.08, 'output': 4.00},
'deepseek-chat': {'input': 0.07, 'cached_input': 0.014, 'output': 0.28},
}
def __init__(self, db_connection):
self.db = db_connection
def calculate_cost(self, model: str, input_tokens: int,
output_tokens: int, cached_tokens: int = 0) -> float:
"""计算单次调用成本"""
pricing = self.PRICING.get(model, {'input': 1.0, 'cached_input': 0.5, 'output': 5.0})
uncached_input = input_tokens - cached_tokens
input_cost = (uncached_input * pricing['input'] +
cached_tokens * pricing['cached_input']) / 1_000_000
output_cost = output_tokens * pricing['output'] / 1_000_000
return input_cost + output_cost
async def record(self, record: LLMCallRecord):
"""记录一次调用"""
# 写入时间序列数据库
await self.db.insert('llm_calls', {
'timestamp': record.timestamp,
'user_id': record.user_id,
'tenant_id': record.tenant_id,
'session_id': record.session_id,
'model': record.model,
'endpoint': record.endpoint,
'input_tokens': record.input_tokens,
'output_tokens': record.output_tokens,
'cached_tokens': record.cached_tokens,
'cost': record.cost,
'latency_ms': record.latency_ms,
'status': record.status,
'feature': record.feature,
})
async def get_daily_summary(self, date: str = None,
tenant_id: str = None) -> dict:
"""获取每日汇总"""
date = date or datetime.now().strftime('%Y-%m-%d')
query = """
SELECT
model,
COUNT(*) as call_count,
SUM(input_tokens) as total_input_tokens,
SUM(output_tokens) as total_output_tokens,
SUM(cached_tokens) as total_cached_tokens,
SUM(cost) as total_cost,
AVG(latency_ms) as avg_latency,
SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count
FROM llm_calls
WHERE DATE(timestamp) = %s
{tenant_filter}
GROUP BY model
ORDER BY total_cost DESC
"""
results = await self.db.execute(query, [date])
total_cost = sum(r['total_cost'] for r in results)
cache_savings = sum(
r['total_cached_tokens'] *
(self.PRICING.get(r['model'], {}).get('input', 1.0) -
self.PRICING.get(r['model'], {}).get('cached_input', 0.5)) / 1_000_000
for r in results
)
return {
'date': date,
'total_cost': total_cost,
'cache_savings': cache_savings,
'by_model': results,
'cost_without_cache': total_cost + cache_savings,
}
async def get_user_cost(self, user_id: str, days: int = 30) -> dict:
"""获取用户级别的成本"""
query = """
SELECT
DATE(timestamp) as date,
model,
SUM(cost) as daily_cost,
SUM(input_tokens + output_tokens) as daily_tokens,
COUNT(*) as daily_calls
FROM llm_calls
WHERE user_id = %s AND timestamp >= NOW() - INTERVAL '%s days'
GROUP BY DATE(timestamp), model
ORDER BY date DESC
"""
results = await self.db.execute(query, [user_id, days])
total_cost = sum(r['daily_cost'] for r in results)
return {
'user_id': user_id,
'period_days': days,
'total_cost': total_cost,
'daily_breakdown': results
}
# 中间件：自动记录每次 LLM 调用
class LLMMonitorMiddleware:
"""LLM 调用监控中间件"""
def __init__(self, monitor: CostMonitor):
self.monitor = monitor
async def wrap_call(self, func, **kwargs):
"""包装 LLM 调用，自动记录"""
start_time = datetime.now()
status = 'success'
error_type = None
try:
response = await func(**kwargs)
# 提取 token 用量
usage = getattr(response, 'usage', None)
if usage:
input_tokens = getattr(usage, 'prompt_tokens', 0)
output_tokens = getattr(usage, 'completion_tokens', 0)
cached_tokens = getattr(usage, 'prompt_tokens_details', {}).get('cached_tokens', 0)
else:
input_tokens = output_tokens = cached_tokens = 0
return response
except Exception as e:
status = 'error'
error_type = type(e).__name__
input_tokens = output_tokens = cached_tokens = 0
raise
finally:
latency = (datetime.now() - start_time).total_seconds() * 1000
cost = self.monitor.calculate_cost(
kwargs.get('model', 'unknown'),
input_tokens, output_tokens, cached_tokens
)
record = LLMCallRecord(
timestamp=start_time,
user_id=kwargs.get('user_id', 'unknown'),
tenant_id=kwargs.get('tenant_id', 'unknown'),
session_id=kwargs.get('session_id', 'unknown'),
model=kwargs.get('model', 'unknown'),
endpoint=kwargs.get('endpoint', 'chat'),
input_tokens=input_tokens,
output_tokens=output_tokens,
cached_tokens=cached_tokens,
cost=cost,
latency_ms=latency,
status=status,
error_type=error_type,
feature=kwargs.get('feature'),
)
await self.monitor.record(record)class AnomalyDetector:
"""异常调用检测器"""
def __init__(self, monitor: CostMonitor):
self.monitor = monitor
async def detect_anomalies(self) -> list[dict]:
"""检测异常调用模式"""
anomalies = []
# 1. 突增检测：某用户今日用量是均值 5 倍以上
spike_users = await self._detect_usage_spikes()
anomalies.extend(spike_users)
# 2. 异常高的单次调用成本
expensive_calls = await self._detect_expensive_calls()
anomalies.extend(expensive_calls)
# 3. 错误率飙升
error_spikes = await self._detect_error_spikes()
anomalies.extend(error_spikes)
# 4. 循环调用检测
loops = await self._detect_call_loops()
anomalies.extend(loops)
return anomalies
async def _detect_usage_spikes(self) -> list[dict]:
"""检测用量突增"""
# 比较今日 vs 过去7天日均
query = """
WITH daily_avg AS (
SELECT user_id, AVG(daily_cost) as avg_cost
FROM (
SELECT user_id, DATE(timestamp) as d, SUM(cost) as daily_cost
FROM llm_calls
WHERE timestamp >= NOW() - INTERVAL '7 days'
GROUP BY user_id, DATE(timestamp)
) sub
GROUP BY user_id
),
today_cost AS (
SELECT user_id, SUM(cost) as today_cost
FROM llm_calls
WHERE DATE(timestamp) = CURRENT_DATE
GROUP BY user_id
)
SELECT t.user_id, t.today_cost, d.avg_cost,
t.today_cost / NULLIF(d.avg_cost, 0) as spike_ratio
FROM today_cost t
JOIN daily_avg d ON t.user_id = d.user_id
WHERE t.today_cost > d.avg_cost * 5
ORDER BY spike_ratio DESC
"""
results = await self.monitor.db.execute(query)
return [{
'type': 'usage_spike',
'user_id': r['user_id'],
'today_cost': r['today_cost'],
'avg_cost': r['avg_cost'],
'spike_ratio': r['spike_ratio'],
'severity': 'critical' if r['spike_ratio'] > 10 else 'warning'
} for r in results]
async def _detect_expensive_calls(self) -> list[dict]:
"""检测异常昂贵的单次调用"""
query = """
SELECT * FROM llm_calls
WHERE DATE(timestamp) = CURRENT_DATE
AND cost > 1.0 -- 单次调用超过 $1
ORDER BY cost DESC
LIMIT 10
"""
results = await self.monitor.db.execute(query)
return [{
'type': 'expensive_call',
'user_id': r['user_id'],
'cost': r['cost'],
'model': r['model'],
'input_tokens': r['input_tokens'],
'severity': 'warning'
} for r in results]
async def _detect_call_loops(self) -> list[dict]:
"""检测循环调用（Agent 可能陷入循环）"""
query = """
SELECT user_id, session_id, COUNT(*) as call_count,
SUM(cost) as total_cost
FROM llm_calls
WHERE timestamp >= NOW() - INTERVAL '1 hour'
GROUP BY user_id, session_id
HAVING COUNT(*) > 50 -- 1小时内超过50次调用
ORDER BY call_count DESC
"""
results = await self.monitor.db.execute(query)
return [{
'type': 'call_loop',
'user_id': r['user_id'],
'session_id': r['session_id'],
'call_count': r['call_count'],
'total_cost': r['total_cost'],
'severity': 'critical'
} for r in results]# Helicone: 最简单的 LLM 可观测性方案
# 只需修改 base_url，无需改代码
import openai
# 标准方式
# client = openai.OpenAI(api_key="sk-...")
# Helicone 方式：只需改 base_url
client = openai.OpenAI(
api_key="sk-...",
base_url="https://oai.helicone.ai/v1", # 加这行
default_headers={
"Helicone-Auth": "sk-helicone-...", # Helicone API Key
"Helicone-Cache-Enabled": "true", # 启用缓存
"Helicone-User-Id": "user_123", # 用户 ID
"Helicone-Properties": json.dumps({ # 自定义标签
"tenant": "acme",
"feature": "chat"
})
}
)
# 之后所有 OpenAI 调用自动被 Helicone 记录
# 在 Helicone Dashboard 可以看到：
# - 每次请求的完整输入/输出
# - Token 用量和成本
# - 延迟分布
# - 缓存命中率
# - 按用户/模型/功能的成本拆解# Langfuse: 开源 LLM 可观测性平台
# pip install langfuse
from langfuse import Langfuse
langfuse = Langfuse(
public_key="pk-...",
secret_key="sk-...",
host="https://cloud.langfuse.com" # 或自部署地址
)
# 记录一次完整的 LLM 调用
trace = langfuse.trace(
name="chat_completion",
user_id="user_123",
session_id="session_456",
metadata={"tenant": "acme", "feature": "chat"}
)
# 记录 span（一个步骤）
generation = trace.generation(
name="llm_call",
model="gpt-4o",
input=messages,
output=response.choices[0].message.content,
usage={
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
},
metadata={"cost": calculated_cost}
)
# Langfuse Dashboard 提供：
# - 完整的 Trace 可视化
# - 成本追踪和预算告警
# - Prompt 版本管理
# - A/B 测试
# - 评估套件