用户请求
  └── Query Rewrite (LLM Call #1)
       └── Vector Search (Embedding Call + DB Query)
            └── Rerank (LLM Call #2 or Cross-Encoder)
                 └── Context Assembly
                      └── Main Generation (LLM Call #3, Stream)
                           └── Tool Calls (LLM Call #4, if needed)
                                └── Tool Execution (HTTP Call)
                                     └── Final Generation (LLM Call #5)
                                          └── Safety Check (LLM Call #6)
                                               └── Response

# docker-compose.yml
version: "3.9"
services:
  langfuse-db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: langfuse
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: langfuse
    volumes:
      - langfuse_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langfuse"]
      interval: 5s
      timeout: 5s
      retries: 5

  langfuse-server:
    image: langfuse/langfuse:latest
    ports:
      - "3000:3000"
    environment:
      DATABASE_URL: postgresql://langfuse:${DB_PASSWORD}@langfuse-db:5432/langfuse
      NEXTAUTH_SECRET: ${NEXTAUTH_SECRET}
      NEXTAUTH_URL: http://localhost:3000
      SALT: ${SALT}
    depends_on:
      langfuse-db:
        condition: service_healthy

volumes:
  langfuse_data:

# 获取 LangFuse API Key 后，配置环境变量
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_HOST="http://localhost:3000"

from langfuse.callback import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

langfuse_handler = CallbackHandler(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="http://localhost:3000"
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)
response = llm.invoke(
    [HumanMessage(content="解释 Transformer 中的注意力机制")],
    config={"callbacks": [langfuse_handler]}
)

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="http://localhost:3000"
)

@observe(as_type="generation")
def call_llm(prompt: str) -> str:
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return response.choices[0].message.content

@observe()
def rag_pipeline(query: str) -> str:
    docs = retrieve_documents(query)
    context = "\n".join([d.page_content for d in docs])
    prompt = f"基于以下上下文回答问题：\n{context}\n\n问题：{query}"
    return call_llm(prompt)

result = rag_pipeline("什么是 RAG？")
langfuse.flush()

from langfuse import Langfuse

langfuse = Langfuse()

dataset = langfuse.create_dataset(name="rag-qa-evaluation-v2")

experiment_items = [
    {
        "input": {"query": "什么是 MCP 协议？", "context_docs": ["mcp_spec_v1.md"]},
        "expected_output": "MCP 是 Model Context Protocol 的缩写...",
        "metadata": {"difficulty": "easy", "category": "definition"}
    },
    {
        "input": {"query": "比较 LangChain 和 LlamaIndex 的差异", "context_docs": ["comparison.md"]},
        "expected_output": "LangChain 侧重于 Agent 编排...",
        "metadata": {"difficulty": "medium", "category": "comparison"}
    }
]

for item in experiment_items:
    langfuse.create_dataset_item(
        dataset_name="rag-qa-evaluation-v2",
        input=item["input"],
        expected_output=item["expected_output"],
        metadata=item["metadata"]
    )

from prometheus_client import Counter, Histogram, Gauge, start_http_server

LLM_TOKEN_USAGE = Counter(
    "llm_token_usage_total",
    "Total token consumption",
    ["model", "token_type", "team", "project"]
)

LLM_REQUEST_DURATION = Histogram(
    "llm_request_duration_seconds",
    "LLM request latency",
    ["model", "endpoint"],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0]
)

TTFT_LATENCY = Histogram(
    "llm_ttft_seconds",
    "Time to first token",
    ["model"],
    buckets=[0.1, 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 3.0]
)

LLM_ERROR_TOTAL = Counter(
    "llm_errors_total",
    "LLM errors by type",
    ["model", "error_type"]
)

HALLUCINATION_RATE = Gauge(
    "llm_hallucination_rate",
    "Hallucination rate (rolling 1h)",
    ["model", "task_type"]
)

CACHE_HIT_RATIO = Gauge(
    "llm_cache_hit_ratio",
    "Semantic cache hit ratio",
    ["model"]
)

COST_DOLLARS = Counter(
    "llm_cost_dollars_total",
    "Estimated cost in USD",
    ["model", "team"]
)

start_http_server(8000)

# Panel: Token 消耗速率（按模型分组）
- expr: rate(llm_token_usage_total[5m])
  legend: "{{model}} - {{token_type}}"
  title: "Token 消耗速率"

# Panel: P95 延迟趋势
- expr: histogram_quantile(0.95, rate(llm_request_duration_seconds_bucket[5m]))
  legend: "{{model}}"
  title: "P95 延迟"

# Panel: TTFT P95
- expr: histogram_quantile(0.95, rate(llm_ttft_seconds_bucket[5m]))
  legend: "{{model}}"
  title: "Time to First Token P95"

# Panel: 幻觉率（1 小时滚动窗口）
- expr: llm_hallucination_rate
  legend: "{{model}} - {{task_type}}"
  title: "幻觉率"

# Panel: 缓存命中率
- expr: llm_cache_hit_ratio
  legend: "{{model}}"
  title: "语义缓存命中率"

# Panel: 每日成本趋势
- expr: increase(llm_cost_dollars_total[24h])
  legend: "{{model}} - {{team}}"
  title: "每日成本 (USD)"

# prometheus_ai_rules.yml
groups:
  - name: ai_application_alerts
    rules:
      - alert: HighTTFT
        expr: histogram_quantile(0.95, rate(llm_ttft_seconds_bucket[5m])) > 1.5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "TTFT P95 超过 1.5 秒"
          description: "模型 {{ $labels.model }} 的 TTFT P95 持续超过 1.5 秒，可能影响用户体验。"

      - alert: TokenBudgetExceeded
        expr: increase(llm_cost_dollars_total[1h]) > 50
        for: 0m
        labels:
          severity: critical
        annotations:
          summary: "1 小时 Token 成本超过 $50"
          description: "团队 {{ $labels.team }} 在过去 1 小时内 Token 消耗成本达到 ${{ $value }}。"

      - alert: HighHallucinationRate
        expr: llm_hallucination_rate > 0.15
        for: 30m
        labels:
          severity: critical
        annotations:
          summary: "幻觉率超过 15%"
          description: "模型 {{ $labels.model }} 的幻觉率达到 {{ $value | humanizePercentage }}，超过安全阈值。"

      - alert: LLMHighErrorRate
        expr: rate(llm_errors_total[5m]) > 0.1
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "LLM 错误率偏高"
          description: "模型 {{ $labels.model }} 的错误率 {{ $value | humanizePercentage }}。"

      - alert: CacheHitRateDropped
        expr: llm_cache_hit_ratio < 0.2
        for: 1h
        labels:
          severity: info
        annotations:
          summary: "缓存命中率低于 20%"
          description: "语义缓存命中率降至 {{ $value | humanizePercentage }}，请检查缓存策略。"

class TokenBudgetManager:
    def __init__(self, config: dict):
        self.limits = {
            "per_request": config.get("per_request", 8000),
            "per_user_daily": config.get("per_user_daily", 50000),
            "per_team_daily": config.get("per_team_daily", 1000000),
            "global_hourly": config.get("global_hourly", 5000000),
        }
        self.usage_tracker = UsageTracker()

    def check_budget(self, scope: str, entity_id: str, estimated_tokens: int) -> bool:
        limit = self.limits.get(scope, float("inf"))
        current_usage = self.usage_tracker.get_usage(scope, entity_id)
        if current_usage + estimated_tokens > limit:
            return False
        return True

    def record_usage(self, scope: str, entity_id: str, input_tokens: int,
                     output_tokens: int, model: str):
        total = input_tokens + output_tokens
        self.usage_tracker.record(scope, entity_id, total)
        cost = self._calculate_cost(model, input_tokens, output_tokens)
        self.usage_tracker.record_cost(scope, entity_id, cost)

    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
            "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
            "deepseek-v3": {"input": 0.27, "output": 1.10},
        }
        rates = pricing.get(model, {"input": 1.0, "output": 3.0})
        return (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000

import json
import time
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class AILogEntry:
    timestamp: float
    trace_id: str
    span_id: str
    parent_span_id: Optional[str]
    level: str
    event_type: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    latency_ms: float
    ttft_ms: Optional[float]
    stop_reason: str
    team: str
    project: str
    user_id: str
    error: Optional[str] = None
    cache_hit: bool = False
    safety_triggered: bool = False
    metadata: Optional[dict] = None

def log_llm_call(entry: AILogEntry):
    safe_entry = asdict(entry)
    safe_entry["input_tokens"] = entry.input_tokens
    safe_entry["cost_usd"] = calculate_cost(
        entry.model, entry.input_tokens, entry.output_tokens
    )
    print(json.dumps(safe_entry, ensure_ascii=False))

# 查询特定模型的错误日志
{app="ai-gateway"} |= "error" | json | model="gpt-4o" | level="error"

# 查询 Token 消耗异常高的请求
{app="ai-gateway"} | json | total_tokens > 10000

# 查询幻觉检测触发的日志
{app="ai-gateway"} | json | hallucination_detected=true

# 按团队统计 1 小时内的 Token 消耗
sum by (team) (
  sum_over_time({app="ai-gateway"} | json | unwrap total_tokens [1h])
)

# escalation-policy.yml
escalation_policies:
  - name: "AI 应用告警升级"
    steps:
      - delay: 0m
        notify:
          - type: slack
            channel: "#ai-alerts"
      - delay: 15m
        notify:
          - type: pagerduty
            service: "ai-production"
      - delay: 30m
        notify:
          - type: phone
            targets:
              - "team-lead"
      - delay: 60m
        notify:
          - type: phone
            targets:
              - "engineering-director"
            repeat: 15m

┌──────────────────────────────────────────────────────────────────┐
│                     用户请求层 (User Requests)                     │
│          Web / Mobile / API / CLI / Embedded Widget               │
└──────────────────────────┬───────────────────────────────────────┘
                           │
┌──────────────────────────▼───────────────────────────────────────┐
│                    API Gateway / Load Balancer                     │
│              请求路由 · 限流 · 鉴权 · Token 预算检查                  │
└──────────────────────────┬───────────────────────────────────────┘
                           │
┌──────────────────────────▼───────────────────────────────────────┐
│                   AI 应用服务层 (App Server)                       │
│                                                                   │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐        │
│  │   RAG    │  │  Agent   │  │  Safety  │  │  Budget  │        │
│  │ Pipeline │  │ Executor │  │  Filter  │  │ Manager  │        │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘        │
│       │              │              │              │               │
│  ┌────▼──────────────▼──────────────▼──────────────▼────┐        │
│  │            Observability SDK (LangFuse/OpenTelemetry) │        │
│  │       Trace Collection · Metric Emission · Log Emit   │        │
│  └──────┬────────────────┬────────────────┬──────────────┘        │
└─────────┼────────────────┼────────────────┼──────────────────────┘
          │                │                │
    ┌─────▼─────┐   ┌─────▼─────┐   ┌─────▼─────┐
    │  LangFuse │   │ Prometheus│   │   Loki    │
    │  (Traces  │   │ + Grafana │   │  (Logs)   │
    │  & Eval)  │   │ (Metrics) │   │           │
    └─────┬─────┘   └─────┬─────┘   └─────┬─────┘
          │                │                │
          │         ┌──────▼──────┐         │
          │         │ Alert Rules │         │
          │         │ (Prometheus │         │
          │         │   Alerting) │         │
          │         └──────┬──────┘         │
          │                │                │
    ┌─────▼────────────────▼────────────────▼─────┐
    │              告警通知分发                       │
    │   Slack · 飞书 · PagerDuty · 邮件 · 短信       │
    └─────────────────────────────────────────────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
    ┌─────▼─────┐   ┌─────▼─────┐   ┌─────▼─────┐
    │ GPT-4o    │   │ Claude    │   │ DeepSeek  │
    │ (Primary) │   │ (Fallback)│   │ (Budget)  │
    └───────────┘   └───────────┘   └───────────┘