# OpenTelemetry GenAI 语义约定 - Token追踪

# Span属性定义
gen_ai.system        # LLM提供商: openai, anthropic
gen_ai.request.model   # 模型名称: gpt-4, claude-3
gen_ai.usage.input_tokens   # 输入Token数
gen_ai.usage.output_tokens  # 输出Token数
gen_ai.usage.total_tokens   # 总Token数
gen_ai.response.finish_reasons # 生成停止原因

# LangSmith成本告警规则配置

# 告警规则1：小时成本超限
Name:        Hourly Cost Alert
Metric:      total_cost (sum)
Threshold:   > $10.00
Window:      60 minutes
Filter:      tags contains env:production
Severity:    Medium

# 告警规则2：平均Token数异常
Name:        Avg Tokens per Run
Metric:      total_tokens (mean)
Threshold:   > 4000
Window:      15 minutes
Filter:      tags contains env:production AND name = "agent-chain"
Severity:    Low

# 告警规则3：预算消耗百分比
# 建议设置: 50%, 80%, 100% 三档
Name:        Budget 80% Consumed
Metric:      cost_percentage
Threshold:   > 0.80
Window:      Daily rolling
Severity:    High

# 启动vLLM并暴露metrics端点
vllm serve mistralai/Mistral-7B-v0.1 \
    --max-model-len 2048 \
    --disable-log-requests

# 访问 http://localhost:8000/metrics 查看原始Prometheus指标

# Prometheus配置 - prometheus.yml
scrape_configs:
  - job_name: 'vllm'
    static_configs:
      - targets: ['vllm-server:8000']
    metrics_path: '/metrics'
    scrape_interval: 10s

# Grafana PromQL查询 - 延迟指标

# P95 端到端延迟
histogram_quantile(0.95, 
  sum by (model_name, le) (
    rate(vllm:request_latency_seconds_bucket[5m])
  )
)

# P99 推理延迟
histogram_quantile(0.99, 
  sum by (batch_size) (
    rate(vllm:inference_latency_seconds_bucket{model_name="example"}[5m])
  )
)

# 每Token平均延迟
sum(rate(vllm:time_per_output_token_seconds_sum[5m])) by (model_name) /
sum(rate(vllm:time_per_output_token_seconds_count[5m])) by (model_name)

# LangSmith错误率告警配置

# 告警规则：错误率超过5%
Name:        Error Rate Alert
Metric:      error_rate
Aggregation: Percentage
Threshold:   > 0.05  # 5%
Window:      10 minutes
Filter:      tags contains env:production
Severity:    High

# 过滤特定错误类型
Additional Filter: name contains "llm" AND error_type = "RateLimitExceeded"
Note: 监控特定端点的限流问题

# Python 熔断器实现示例
import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN
    
    def call(self, func, *args, **kwargs):
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'HALF_OPEN'
            else:
                raise Exception("Circuit is OPEN")
        
        try:
            result = func(*args, **kwargs)
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = 'OPEN'
            raise

# 使用示例
breaker = CircuitBreaker(failure_threshold=5, timeout=60)
result = breaker.call(llm_api_call, prompt)

# Python 重试处理器实现
import asyncio
import random

class RetryHandler:
    def __init__(self, max_retries=3, base_delay=1):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def retry_with_backoff(self, func, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return await func(*args, **kwargs)
            except RateLimitError:
                if attempt == self.max_retries - 1:
                    raise
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(delay)
            except AuthenticationError:
                raise  # 不重试认证错误
            except NetworkError:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(self.base_delay * random(1, 3))

# OpenTelemetry GenAI 语义约定 - Span事件

# Span事件名称和触发时机
gen_ai.system.message    # LLM调用前 - 系统消息
gen_ai.user.message      # LLM调用前 - 用户消息
gen_ai.assistant.message # LLM调用后 - 助手响应
gen_ai.tool.message      # 工具调用后 - 工具结果

# 事件属性
gen_ai.prompt.content    # 提示词内容
gen_ai.completion.content # 补全内容
gen_ai.prompt.role       # 角色 (system/user/assistant)

# 安装 OpenTelemetry instrumentation
pip install opentelemetry-instrumentation-openai
pip install opentelemetry-instrumentation-anthropic
pip install opentelemetry-instrumentation-langchain
pip install opentelemetry-instrumentation-llamaindex

# Python代码集成
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# 配置TracerProvider
resource = Resource.create({
    "service.name": "my-llm-app",
    "deployment.environment": "production"
})

provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

# 设置OTLP导出器
exporter = OTLPSpanExporter(
    endpoint="http://collector:4318/v1/traces"
)
provider.add_span_processor(BatchSpanProcessor(exporter))

# Python采样器配置
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased

# 采样10%的根Span，子Span继承决策
sampler = ParentBased(root=TraceIdRatioBased(0.10))

provider = TracerProvider(sampler=sampler, resource=resource)

# OpenTelemetry Collector配置 - PII保护
# otel-collector-config.yaml
processors:
  transform:
    trace_statements:
      - context: spanevent
        statements:
          # 删除提示词和补全内容
          - delete_matching_keys(attributes, "gen_ai.prompt.content")
          - delete_matching_keys(attributes, "gen_ai.completion.content")
          
# 或使用哈希进行关联（不暴露内容）
attributes:
  actions:
    - key: gen_ai.prompt.content
      action: hash
      hash_salt: ${env:HASH_SALT}

# LangSmith实验配置 - 比较提示词变体

# 创建实验
Experiment Name: prompt-v2-vs-v3
Variants:
  - Control: prompt_version = "v2"
  - Treatment: prompt_version = "v3"

# 对比指标
Primary Metrics:
  - feedback_score (mean)
  - latency_p95
  - total_cost
  
Statistical Significance:
  Min Sample Size: 100 runs per variant
  Confidence Level: 95%

# LangSmith Python SDK 配置
from langchain_core.tracers import LangSmithTracer
from langchain_core.runnables import RunnableConfig

# 追踪配置
@traceable(
    project_name="production-agent",
    tags=["env:production", "feature:qa"],
    metadata={
        "model": "gpt-4o",
        "version": "v2.1"
    }
)
def answer_question(question: str) -> str:
    # 业务逻辑
    return response

# LangChain Chain 调用配置
chain.invoke(
    {"question": user_question},
    config={
        "tags": ["feature:qa", "env:production"],
        "metadata": {
            "user_id": user_id,
            "session_id": session_id
        }
    }
)

# 规则：每个生产调用至少需要 env:production 和 feature:* 标签

# LangSmith 生产告警配置 - 四个核心规则

# ============================================
# 告警1：P95延迟SLA违规
# ============================================
Name:        P95 Latency > 5s
Metric:      run_latency_ms (p95)
Threshold:   > 5000 ms
Window:      5 minutes
Filter:      tags contains env:production
Severity:    Critical

# ============================================
# 告警2：错误率飙升
# ============================================
Name:        Error Rate > 5%
Metric:      error_rate
Threshold:   > 0.05
Window:      10 minutes
Filter:      tags contains env:production
Severity:    High

# ============================================
# 告警3：成本异常
# ============================================
Name:        Hourly Cost > $10
Metric:      total_cost (sum)
Threshold:   > $10.00
Window:      60 minutes
Filter:      tags contains env:production
Severity:    Medium

# ============================================
# 告警4：Token预算超限
# ============================================
Name:        Avg Tokens per Run > 4000
Metric:      total_tokens (mean)
Threshold:   > 4000
Window:      15 minutes
Filter:      tags contains env:production AND name = "answer-question"
Severity:    Low

# LangSmith Webhook 配置 - Slack通知
import httpx

LANGSMITH_API_KEY = "ls__your_key_here"
PROJECT_ID = "your-project-id"

webhook_payload = {
    "name": "slack-production-alerts",
    "type": "webhook",
    "config": {
        "url": "https://hooks.slack.com/services/T00/B00/xxx",
        "headers": {"Content-Type": "application/json"}
    }
}

response = httpx.post(
    f"https://api.smith.langchain.com/api/v1/projects/{PROJECT_ID}/alert-channels",
    headers={"x-api-key": LANGSMITH_API_KEY},
    json=webhook_payload
)
print(response.json())

# Slack 告警消息格式
🚨 [Critical] P95 Latency > 5s

Project: my-llm-app
Value: 7,340ms (threshold: 5,000ms)
Window: 2026-03-09 14:05 → 14:10 UTC
View runs: https://smith.langchain.com/...

# 用户反馈追踪
from langsmith import Client

client = Client()

def record_user_feedback(run_id: str, score: int, comment: str = ""):
    # score: 1 = positive, 0 = negative
    client.create_feedback(
        run_id=run_id,
        key="user_thumbs",
        score=score,
        comment=comment,
        source_info={"source": "webapp", "version": "v2.1"}
    )

# API端点调用
@app.post("/feedback")
async def submit_feedback(run_id: str, rating: int):
    record_user_feedback(run_id=run_id, score=rating)
    return {"status": "ok"}

# Grafana Dashboard JSON 配置片段
{
  "title": "LLM Application Monitoring",
  "panels": [
    {
      "title": "P95 Latency",
      "type": "timeseries",
      "datasource": {"type": "prometheus", "uid": "${DS_PROMETHEUS}"},
      "expr": "histogram_quantile(0.95, sum(rate(llm_latency_seconds_bucket[5m])) by (le))"
    },
    {
      "title": "Token Usage",
      "type": "timeseries",
      "datasource": {"type": "prometheus"},
      "expr": "sum(rate(gen_ai_usage_input_tokens[5m]))"
    },
    {
      "title": "Error Rate",
      "type": "stat",
      "datasource": {"type": "prometheus"},
      "expr": "sum(rate(llm_errors_total[5m])) / sum(rate(llm_requests_total[5m]))"
    }
  ]
}