LLM Observability: мониторинг и анализ
opensourceaillmobservabilitymonitoringit
Введение: почему LLM требует особого мониторинга
Проблема
Traditional monitoring:
HTTP request → Response → Log
Metrics: latency, error rate, throughput
LLM monitoring:
Prompt → [Context + Model] → Token-by-token Output
Metrics: latency, error rate, token usage, quality, cost
✗ Non-deterministic outputs
✗ Token-level billing
✗ Quality is subjective
✗ Prompt injection attacks
✗ Context window limits
Факт: LLM API calls can cost 10-100x more than traditional API calls when unmonitored.
Что такое LLM Observability?
Core Concepts
LLM Observability = Monitor + Trace + Analyze
Three pillars:
1. Tracing: track every LLM call end-to-end
2. Metrics: measure performance and quality
3. Logging: store prompts and responses for analysis
LLM Call Lifecycle
User: "What is machine learning?"
↓
1. Gateway: authenticate, rate limit
↓
2. Router: select model (gpt-4, claude, local)
↓
3. Prompt builder: add system prompt + context
↓
4. LLM API: stream tokens
↓
5. Post-processor: filter, format
↓
6. Cache: store result
↓
7. Response: return to user
Уровень 1: Tracing
Distributed Tracing
Trace = sequence of spans
Span structure:
{
"trace_id": "abc-123",
"span_id": "span-456",
"operation": "llm.call",
"model": "gpt-4",
"input_tokens": 150,
"output_tokens": 320,
"duration_ms": 2400,
"cost_usd": 0.45,
"children": [...]
}
Trace example:
Trace: user-query-789
├─ Span: api-gateway (5ms)
├─ Span: prompt-builder (2ms)
├─ Span: llm-call (2400ms)
│ ├─ Token 1: 12ms
│ ├─ Token 2: 8ms
│ └─ ...
└─ Span: post-process (3ms)
OpenTelemetry for LLM
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
tracer = trace.get_tracer("llm-service")
def call_llm(user_prompt: str, model: str):
with tracer.start_as_current_span("llm.call") as span:
# Set attributes
span.set_attribute("llm.model", model)
span.set_attribute("llm.prompt_length", len(user_prompt))
# Record prompt
span.add_event("prompt", {
"message": user_prompt,
"role": "user"
})
# Call LLM
response = llm.generate(user_prompt, model=model)
# Record response
span.add_event("response", {
"content": response.text,
"tokens": len(response.tokens)
})
# Set cost
cost = calculate_cost(model, response)
span.set_attribute("llm.cost_usd", cost)
return response
LangSmith Tracing
from langsmith import traceable, Client
client = Client()
@traceable(run_type="llm")
def call_gpt4(message: str) -> str:
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": message}]
)
return response.choices[0].message.content
# Automatically traced in LangSmith
result = call_gpt4("Hello, world!")
Уровень 2: Metrics
Key LLM Metrics
Performance metrics:
- Time to first token (TTFT)
- Tokens per second (TPS)
- Total latency (p50, p95, p99)
- Time per output token
Quality metrics:
- Token diversity (entropy)
- Response length distribution
- Error rate by error type
- Context window utilization
Cost metrics:
- Cost per query
- Cost per model
- Cost per user
- Total spend
Usage metrics:
- Requests per minute
- Token usage (input vs output)
- Model distribution
- Cache hit rate
Prometheus Integration
from prometheus_client import Histogram, Counter, Gauge
import time
# Metrics
request_latency = Histogram(
'llm_request_latency_seconds',
'LLM request latency',
['model', 'endpoint']
)
token throughput = Histogram(
'llm_tokens_per_second',
'Tokens per second',
['model']
)
request_count = Counter(
'llm_request_total',
'Total LLM requests',
['model', 'status']
)
cache_hits = Counter(
'llm_cache_hits_total',
'Cache hits'
)
@request_latency.time()
def monitored_call(prompt: str, model: str):
start = time.time()
# Check cache
cache_key = hash(prompt)
if cache.get(cache_key):
cache_hits.inc()
return cache.get(cache_key)
# Call LLM
response = llm.generate(prompt, model=model)
# Record tokens/sec
duration = time.time() - start
tps = len(response.tokens) / duration
token_throughput.labels(model=model).observe(tps)
# Increment counter
request_count.labels(model=model, status="success").inc()
# Cache result
cache.set(cache_key, response)
return response
Grafana Dashboard
Dashboard panels:
1. Request rate (req/s) — line chart
2. Latency p50/p95/p99 — line chart
3. Tokens per second — bar chart
4. Cost per hour — gauge
5. Model distribution — pie chart
6. Error rate — line chart
7. Cache hit rate — gauge
8. Token usage (input/output) — stacked bar
Уровень 3: Prompt Logging
Structured Logging
import json
import logging
logger = logging.getLogger("llm-observability")
def log_llm_call(prompt: str, response: str,
model: str, metadata: dict):
logger.info(json.dumps({
"event": "llm_call",
"timestamp": datetime.utcnow().isoformat(),
"model": model,
"prompt": {
"system": metadata.get("system_prompt", ""),
"user": prompt,
"messages_count": len(metadata.get("messages", []))
},
"response": {
"content": response,
"tokens": metadata.get("output_tokens", 0),
"finish_reason": metadata.get("finish_reason")
},
"performance": {
"latency_ms": metadata.get("latency_ms"),
"ttft_ms": metadata.get("ttft_ms"),
"tokens_per_second": metadata.get("tps")
},
"cost": {
"input_tokens": metadata.get("input_tokens"),
"output_tokens": metadata.get("output_tokens"),
"total_usd": metadata.get("cost_usd")
},
"metadata": metadata
}))
Prompt Versioning
Prompt versions:
v1.0: "Answer the following question..."
v1.1: "Answer the following question concisely..."
v2.0: "You are a helpful assistant. Answer..."
Track which prompt version produced which result:
{
"prompt_version": "v2.0",
"prompt_hash": "abc123",
"changes": "Added role definition",
"impact": {
"avg_response_length": -15%,
"quality_score": +0.12
}
}
PII Redaction
import re
def redact_pii(text: str) -> str:
"""Remove PII from logs"""
# Email
text = re.sub(
r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
'[EMAIL]', text
)
# Phone
text = re.sub(
r'\b\d{3}[-.]?\d{4}\b',
'[PHONE]', text
)
# Credit card
text = re.sub(
r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b',
'[CC]', text
)
# SSN
text = re.sub(
r'\b\d{3}-\d{2}-\d{4}\b',
'[SSN]', text
)
return text
Уровень 4: Quality Monitoring
Automated Quality Checks
class QualityChecker:
def __init__(self, llm):
self.llm = llm
def check(self, prompt: str, response: str) -> dict:
return {
"toxicity": self._check_toxicity(response),
"relevance": self._check_relevance(prompt, response),
"factual_consistency": self._check_facts(prompt, response),
"grammar": self._check_grammar(response),
"length": self._check_length(response),
"format": self._check_format(response)
}
def _check_relevance(self, prompt: str, response: str) -> float:
"""Score 0-1: how relevant is the response?"""
prompt = f"""
Rate the relevance of this response (0-10):
Question: {prompt}
Response: {response}
"""
score = self.llm.generate(prompt)
return self._parse_score(score)
Feedback Collection
class FeedbackCollector:
def __init__(self, db):
self.db = db
def record(self, trace_id: str, feedback: str,
rating: int = None):
"""Record user feedback"""
self.db.insert({
"trace_id": trace_id,
"feedback_type": feedback, # "thumbs_up", "thumbs_down"
"rating": rating, # 1-5
"timestamp": datetime.utcnow(),
"comment": feedback.get("comment")
})
def get_negative_feedback(self) -> list[dict]:
"""Get all negative feedback for review"""
return self.db.query({
"feedback_type": "thumbs_down",
"sort": "-timestamp",
"limit": 100
})
Drift Detection
class DriftDetector:
def __init__(self, baseline_metrics: dict):
self.baseline = baseline_metrics
def check(self, current_metrics: dict) -> dict:
"""Detect significant drift from baseline"""
drift = {}
for metric, baseline_value in self.baseline.items():
current_value = current_metrics.get(metric)
if current_value is None:
continue
change = abs(
current_value - baseline_value
) / baseline_value
if change > 0.1: # 10% threshold
drift[metric] = {
"baseline": baseline_value,
"current": current_value,
"change_pct": change * 100
}
return drift
Уровень 5: Cost Tracking
Token-level Cost Calculation
MODEL_PRICES = {
"gpt-4": {
"input_per_1k": 0.03,
"output_per_1k": 0.06
},
"gpt-3.5-turbo": {
"input_per_1k": 0.0015,
"output_per_1k": 0.002
},
"claude-3-opus": {
"input_per_1k": 0.015,
"output_per_1k": 0.075
}
}
def calculate_cost(model: str, input_tokens: int,
output_tokens: int) -> float:
prices = MODEL_PRICES.get(model, MODEL_PRICES["gpt-3.5-turbo"])
input_cost = (input_tokens / 1000) * prices["input_per_1k"]
output_cost = (output_tokens / 1000) * prices["output_per_1k"]
return input_cost + output_cost
# Example:
# gpt-4, 500 input tokens, 1000 output tokens
# cost = (500/1000) * 0.03 + (1000/1000) * 0.06
# = 0.015 + 0.06 = $0.075
Cost Dashboard
Cost breakdown:
┌─────────────────────────────────────┐
│ Total today: $127.45 │
│ Total this month: $3,421.80 │
├─────────────────────────────────────┤
│ By model: │
│ GPT-4: $2,100 (61%) │
│ GPT-3.5: $890 (26%) │
│ Claude: $432 (13%) │
├─────────────────────────────────────┤
│ By team: │
│ Engineering: $1,800 (53%) │
│ Marketing: $920 (27%) │
│ Support: $702 (20%) │
├─────────────────────────────────────┤
│ Top queries by cost: │
│ 1. "Summarize this document" $45 │
│ 2. "Write email" $32 │
│ 3. "Code review" $28 │
└─────────────────────────────────────┘
Cost Optimization
class CostOptimizer:
def __init__(self, models: dict):
self.models = models # cheap and powerful
def route(self, query: str) -> str:
"""Route to cheapest appropriate model"""
if self._is_simple(query):
return "gpt-3.5-turbo" # Cheap
if self._is_complex(query):
return "gpt-4" # Powerful
if self._is_creative(query):
return "claude-3" # Creative
return "gpt-3.5-turbo" # Default to cheap
def _is_simple(self, query: str) -> bool:
"""Simple factual questions"""
return len(query.split()) < 20
def _is_complex(self, query: str) -> bool:
"""Complex reasoning required"""
return any(word in query.lower()
for word in ["analyze", "compare", "reason"])
Уровень 6: Alerting
Alert Rules
Performance alerts:
- Latency p99 > 10s for 5 min
- TTFT > 5s for 5 min
- Token throughput < 10 tok/s
Error alerts:
- Error rate > 5% for 5 min
- Rate limit errors > 10/min
- Model unavailable
Quality alerts:
- Negative feedback rate > 20%
- Toxicity score > 0.1
- Drift detected
Cost alerts:
- Daily spend > $500
- Hourly spend > $50
- Unexpected spike > 2x baseline
Alert Configuration
ALERT_RULES = [
{
"name": "high_latency",
"metric": "llm_request_latency_seconds",
"condition": "histogram_quantile(0.99, rate(...)) > 10",
"for": "5m",
"severity": "warning",
"channels": ["slack", "pagerduty"]
},
{
"name": "high_error_rate",
"metric": "llm_request_total",
"condition": "rate(status:error) / rate(status:all) > 0.05",
"for": "5m",
"severity": "critical",
"channels": ["slack", "pagerduty", "email"]
},
{
"name": "cost_spike",
"metric": "llm_cost_total",
"condition": "rate(...) > 50",
"for": "1h",
"severity": "warning",
"channels": ["slack", "email"]
}
]
Уровень 7: Debugging
Trace Analysis
Debugging slow responses:
Trace: query-123 (5200ms total)
├─ api-gateway: 5ms ✓
├─ prompt-builder: 2ms ✓
├─ llm-call: 5180ms ✗ SLOW
│ ├─ queue_time: 200ms
│ ├─ ttft: 3200ms ✗ VERY SLOW
│ ├─ streaming: 1980ms (156 tok/s)
│ └─ post-process: 0ms
└─ cache-store: 10ms ✓
Root cause: TTFT 3200ms → model warmup or queue
---
## Заключение
LLM observability is essential for production systems. Without proper monitoring, you're flying blind — spending money on inefficient calls, serving low-quality responses, and not knowing until users complain.
**Key takeaways:**
- Implement tracing from day one
- Track cost per query
- Monitor quality metrics
- Set up alerts for anomalies
- Log everything (with PII redaction)
**Tools to get started:**
- LangSmith for tracing
- Prometheus + Grafana for metrics
- OpenTelemetry for distributed tracing
- Custom quality checkers
---
## Ресурсы
- [OpenTelemetry Documentation](https://opentelemetry.io/)
- [LangSmith Documentation](https://docs.smith.langchain.com/)
- [Prometheus Documentation](https://prometheus.io/docs/)