Agent实时评测与最佳实践

┌─────────────────────────────────────────────────────────────────────┐
│                        Agent评测演进路线                             │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  阶段1: 离线评测          阶段2: 混合评测          阶段3: 实时评测    │
│  (2023-2024初)           (2024-现在) ◀── 当前     (未来趋势)        │
│                                                                     │
│  • 静态测试集             • 离线+在线结合          • 全链路实时监控   │
│  • 批量运行评估           • 生产环境采样           • 自动异常检测     │
│  • 人工分析结果           • 部分自动化监控         • 智能告警         │
│  • 周期性评测             • 关键指标追踪           • 自适应优化       │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# Langfuse 实时追踪示例
from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://cloud.langfuse.com"# 或自托管地址
)

@observe()  # 自动追踪函数执行
def my_agent(query: str):
    # Agent逻辑
    result = llm.invoke(query)
    return result

# 实时追踪数据自动上报
result = my_agent("用户查询")

# 查看追踪数据
traces = langfuse.get_traces()
for trace in traces:
    print(f"延迟: {trace.latency}ms")
    print(f"Token: {trace.total_tokens}")
    print(f"成本: ${trace.total_cost}")

实时监控:
  -全链路Trace追踪
-延迟/Token/成本实时统计
-异常检测和告警

在线评估:
-生产数据采样评估
-LLM-as-Judge自动评分
-人工反馈收集

实验管理:
-A/B测试支持
-Prompt版本管理
-回归测试

import os
from langsmith import Client
from langchain.callbacks import LangChainTracer

# 1. 启用生产环境追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-agent"

# 2. 配置采样率（控制成本）
tracer = LangChainTracer(
    project_name="production-agent",
    # 生产环境建议10-20%采样率
    sample_rate=0.1
)

# 3. 设置在线评估器
client = Client()

# 定义在线评估规则
online_evaluator = client.create_online_evaluator(
    name="production-quality-check",
    evaluator_type="llm",
    criteria={
        "relevance": "回答是否与问题相关",
        "helpfulness": "回答是否有帮助",
        "safety": "回答是否安全合规"
    },
    sample_rate=0.05# 5%的请求进行LLM评估
)

┌─────────────────────────────────────────────────────────────────────┐
│                         生产环境                                     │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐          │
│  │ Agent 1 │    │ Agent 2 │    │ Agent 3 │    │ Agent N │          │
│  └────┬────┘    └────┬────┘    └────┬────┘    └────┬────┘          │
│       │              │              │              │                │
│       └──────────────┴──────────────┴──────────────┘                │
│                              │                                       │
│                              ▼                                       │
│                    ┌─────────────────┐                              │
│                    │  Trace Collector │  ◀── 异步收集，不阻塞主流程  │
│                    │  (采样 + 缓冲)   │                              │
│                    └────────┬────────┘                              │
└─────────────────────────────┼───────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│                       评测平台                                       │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐     │
│  │   Trace Store   │  │  Metric Store   │  │  Evaluation     │     │
│  │   (时序数据库)   │  │  (Prometheus)   │  │  Engine         │     │
│  └────────┬────────┘  └────────┬────────┘  └────────┬────────┘     │
│           │                    │                    │               │
│           └────────────────────┴────────────────────┘               │
│                              │                                       │
│                              ▼                                       │
│                    ┌─────────────────┐                              │
│                    │   Dashboard     │                              │
│                    │   + Alerting    │                              │
│                    └─────────────────┘                              │
└─────────────────────────────────────────────────────────────────────┘

class RealtimeMetrics:
    """实时监控指标"""
    
    # 性能指标
    latency_p50: float      # 延迟P50
    latency_p95: float      # 延迟P95
    latency_p99: float      # 延迟P99
    throughput: float       # 吞吐量(QPS)
    
    # 成本指标
    tokens_per_request: float   # 每请求Token数
    cost_per_request: float     # 每请求成本
    daily_cost: float           # 日成本
    
    # 质量指标（采样评估）
    success_rate: float         # 成功率
    error_rate: float           # 错误率
    hallucination_rate: float   # 幻觉率（采样）
    relevance_score: float      # 相关性得分（采样）
    
    # 工具使用指标
    tool_call_rate: float       # 工具调用率
    tool_success_rate: float    # 工具成功率
    avg_tool_calls: float       # 平均工具调用次数

# 告警规则配置
alerts:
-name:high_latency
    condition:latency_p95>5000# 5秒
    severity:warning
    message:"Agent响应延迟过高"
    
-name:high_error_rate
    condition:error_rate>0.05# 5%
    severity:critical
    message:"Agent错误率异常"
    
-name:cost_spike
    condition:hourly_cost>daily_budget/12
    severity:warning
    message:"成本超出预期"
    
-name:tool_failure
    condition:tool_success_rate<0.9
    severity:warning
    message:"工具调用成功率下降"

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse()

@observe(as_type="generation")
def customer_service_agent(query: str, user_id: str):
    """客服Agent主函数"""
    
    # 1. 记录用户信息（脱敏）
    langfuse_context.update_current_observation(
        metadata={"user_segment": get_user_segment(user_id)}
    )
    
    # 2. Agent处理
    response = agent.run(query)
    
    # 3. 记录输出质量评分（异步）
    langfuse_context.score_current_observation(
        name="auto_quality",
        value=quick_quality_check(query, response),
        comment="自动质量检查"
    )
    
    return response

# 配置采样评估（5%请求进行深度评估）
@langfuse.on_trace(sample_rate=0.05)
def deep_evaluation(trace):
    """深度质量评估"""
    scores = {
        "relevance": evaluate_relevance(trace),
        "helpfulness": evaluate_helpfulness(trace),
        "safety": evaluate_safety(trace)
    }
    for name, score in scores.items():
        langfuse.score(trace_id=trace.id, name=name, value=score)

dashboard:
  panels:
    -title:"实时请求量"
      type:timeseries
      query:rate(agent_requests_total[5m])
      
    -title:"延迟分布"
      type:heatmap
      query:agent_latency_seconds
      
    -title:"质量得分趋势"
      type:timeseries
      query:avg(agent_quality_score)by(score_type)
      
    -title:"错误分布"
      type:pie
      query:sum(agent_errors_total)by(error_type)

from langsmith import Client
import random

client = Client()

def agent_with_ab_test(query: str):
    """带A/B测试的Agent"""
    
    # 随机分配实验组
    variant = "A"if random.random() < 0.5else"B"
    
    # 使用对应版本的Prompt
    prompt = get_prompt_version(variant)
    
    # 执行并记录
    with client.trace(
        name="ab_test_agent",
        metadata={"variant": variant}
    ) as trace:
        response = llm.invoke(prompt.format(query=query))
        
        # 记录实验标签
        trace.update(tags=[f"variant:{variant}"])
        
    return response

# 分析A/B测试结果
def analyze_ab_test():
    """分析A/B测试结果"""
    
    # 获取两组数据
    variant_a = client.list_runs(
        project_name="production",
        filter='has(tags, "variant:A")',
        start_time=datetime.now() - timedelta(days=7)
    )
    
    variant_b = client.list_runs(
        project_name="production", 
        filter='has(tags, "variant:B")',
        start_time=datetime.now() - timedelta(days=7)
    )
    
    # 计算指标对比
    metrics_a = calculate_metrics(variant_a)
    metrics_b = calculate_metrics(variant_b)
    
    return {
        "variant_a": metrics_a,
        "variant_b": metrics_b,
        "winner": "A"if metrics_a["score"] > metrics_b["score"] else"B"
    }

from dataclasses import dataclass
from typing import List
import statistics

@dataclass
class AnomalyDetector:
    """异常检测器"""
    
    window_size: int = 100# 滑动窗口大小
    threshold_std: float = 3.0# 标准差阈值
    
    def __init__(self):
        self.latency_history: List[float] = []
        self.error_history: List[bool] = []
    
    def check_latency_anomaly(self, latency: float) -> bool:
        """检测延迟异常"""
        self.latency_history.append(latency)
        if len(self.latency_history) > self.window_size:
            self.latency_history.pop(0)
        
        if len(self.latency_history) < 10:
            returnFalse
            
        mean = statistics.mean(self.latency_history[:-1])
        std = statistics.stdev(self.latency_history[:-1])
        
        return latency > mean + self.threshold_std * std
    
    def check_error_spike(self) -> bool:
        """检测错误率飙升"""
        if len(self.error_history) < self.window_size:
            returnFalse
            
        recent_error_rate = sum(self.error_history[-20:]) / 20
        baseline_error_rate = sum(self.error_history[:-20]) / (len(self.error_history) - 20)
        
        return recent_error_rate > baseline_error_rate * 2

# 集成到监控流程
detector = AnomalyDetector()

def monitor_request(trace):
    """监控单次请求"""
    
    # 检测延迟异常
    if detector.check_latency_anomaly(trace.latency):
        send_alert(
            level="warning",
            message=f"延迟异常: {trace.latency}ms",
            trace_id=trace.id
        )
    
    # 检测错误率飙升
    detector.error_history.append(trace.error isnotNone)
    if detector.check_error_spike():
        send_alert(
            level="critical",
            message="错误率异常飙升",
            recent_errors=get_recent_errors()
        )

Phase 1: 基础可观测性（1-2周）
├── 接入Trace追踪（Langfuse/LangSmith）
├── 配置基础指标收集
├── 搭建监控仪表板
└── 设置基础告警

Phase 2: 质量评估（2-4周）
├── 实现采样评估机制
├── 配置LLM-as-Judge评估
├── 建立质量基线
└── 优化采样策略

Phase 3: 高级功能（4-8周）
├── A/B测试框架
├── 异常检测系统
├── 自动化回归测试
└── 成本优化分析

Phase 4: 持续优化（持续）
├── 评估指标迭代
├── 告警规则调优
├── 评估效率优化
└── 新场景覆盖