前言：Demo 到生产的鸿沟你做了一个 LLM 应用的 Demo：# Demo 版本：100 行代码搞定 response = openai.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": user_input}] ) print(response.choices[0].message.content)但生产环境需要考虑：mindmap root((生产环境考量)) 性能 延迟 吞吐量 并发 可靠性 错误处理 重试机制 降级方案 成本 Token 优化 模型选择 缓存策略 安全 输入输出过滤 权限控制 审计日志 可观测 监控 日志 追踪一、整体架构设计1.1 典型 LLM 应用架构flowchart TB subgraph 客户端 Web[Web 应用] Mobile[移动应用] API[API 调用] end subgraph 网关层 LB[负载均衡] Gateway[API Gateway] Auth[认证授权] RateLimit[限流] end subgraph 业务层 Router[请求路由] InputProc[输入处理] OutputProc[输出处理] Cache[缓存层] end subgraph LLM层 Orchestrator[编排层] LLM1[GPT-4] LLM2[Claude] LLM3[本地模型] RAG[RAG 服务] Agent[Agent 服务] end subgraph 数据层 VectorDB[(向量数据库)] RelDB[(关系数据库)] Redis[(Redis)] MQ[消息队列] end subgraph 可观测性 Log[日志] Metrics[指标] Trace[追踪] end Web --> LB Mobile --> LB API --> LB LB --> Gateway Gateway --> Auth Auth --> RateLimit RateLimit --> Router Router --> InputProc InputProc --> Cache Cache --> Orchestrator Orchestrator --> LLM1 Orchestrator --> LLM2 Orchestrator --> LLM3 Orchestrator --> RAG Orchestrator --> Agent RAG --> VectorDB OutputProc --> Cache Router -.-> Log Orchestrator -.-> Metrics LLM1 -.-> Trace1.2 核心组件说明组件作用推荐技术API Gateway认证、限流、路由Kong, Nginx, AWS API Gateway编排层LLM 调用逻辑LangChain, 自研缓存层减少重复调用Redis, Memcached向量数据库RAG 检索Milvus, Pinecone, Chroma消息队列异步处理Kafka, RabbitMQ, Redis可观测性监控追踪Prometheus, Grafana, Jaeger1.3 架构模式选择graph TB subgraph 同步模式 S1[请求] --> S2[处理] --> S3[响应] S3 -->|延迟高| S4[用户等待] end subgraph 异步模式 A1[请求] --> A2[入队] A2 --> A3[立即返回任务ID] A4[Worker处理] --> A5[结果存储] A6[轮询/回调获取结果] end subgraph 流式模式 T1[请求] --> T2[流式处理] T2 --> T3[持续返回] T3 --> T4[用户实时看到] end style T4 fill:#4ecdc4选择建议：同步模式：简单请求，延迟要求不高异步模式：复杂任务，可以接受等待流式模式：对话场景，提升用户体验二、性能优化2.1 延迟优化import asyncio import time from typing import List from openai import AsyncOpenAI class LatencyOptimizer: """延迟优化器""" def __init__(self): self.client = AsyncOpenAI() async def parallel_calls(self, prompts: List[str]) -> List[str]: """并行调用多个 LLM 请求""" tasks = [ self.client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": p}] ) for p in prompts ] responses = await asyncio.gather(*tasks) return [r.choices[0].message.content for r in responses] async def streaming_response(self, prompt: str): """流式响应""" stream = await self.client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], stream=True, ) async for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content async def speculative_execution(self, prompt: str): """投机执行：同时调用快慢模型""" fast_task = self.client.chat.completions.create( model="gpt-3.5-turbo", # 快但质量一般 messages=[{"role": "user", "content": prompt}] ) slow_task = self.client.chat.completions.create( model="gpt-4", # 慢但质量好 messages=[{"role": "user", "content": prompt}] ) # 先返回快的结果，后台继续等慢的 done, pending = await asyncio.wait( [fast_task, slow_task], return_when=asyncio.FIRST_COMPLETED ) fast_result = done.pop().result() # 返回快的结果，同时安排后续替换 return { "initial": fast_result.choices[0].message.content, "pending_upgrade": pending, }2.2 缓存策略import hashlib import json from typing import Optional import redis class LLMCache: """LLM 响应缓存""" def __init__(self, redis_url: str = "redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.default_ttl = 3600 # 1小时 def _get_cache_key(self, prompt: str, model: str, params: dict) -> str: """生成缓存键""" # 将请求参数哈希 content = json.dumps({ "prompt": prompt, "model": model, "params": params, }, sort_keys=True) return f"llm:cache:{hashlib.md5(content.encode()).hexdigest()}" def get(self, prompt: str, model: str, params: dict) -> Optional[str]: """获取缓存""" key = self._get_cache_key(prompt, model, params) cached = self.redis.get(key) if cached: return json.loads(cached)["response"] return None def set(self, prompt: str, model: str, params: dict, response: str, ttl: int = None): """设置缓存""" key = self._get_cache_key(prompt, model, params) value = json.dumps({"response": response}) self.redis.setex(key, ttl or self.default_ttl, value) def get_or_call(self, prompt: str, model: str, params: dict, llm_func): """获取缓存或调用 LLM""" cached = self.get(prompt, model, params) if cached: return cached, True # 返回缓存命中标志 response = llm_func(prompt, model, params) self.set(prompt, model, params, response) return response, False class SemanticCache: """语义缓存：相似问题返回相同答案""" def __init__(self, embedding_model, similarity_threshold: float = 0.95): self.embedding_model = embedding_model self.threshold = similarity_threshold self.cache = {} # 实际应用中用向量数据库 def get(self, query: str) -> Optional[str]: """语义匹配获取缓存""" query_embedding = self.embedding_model.encode(query) for cached_query, (cached_embedding, response) in self.cache.items(): similarity = self._cosine_similarity(query_embedding, cached_embedding) if similarity > self.threshold: return response return None def set(self, query: str, response: str): """设置语义缓存""" embedding = self.embedding_model.encode(query) self.cache[query] = (embedding, response)2.3 批处理优化import asyncio from typing import List, Dict from dataclasses import dataclass from collections import defaultdict @dataclass class BatchRequest: """批处理请求""" request_id: str prompt: str future: asyncio.Future class BatchProcessor: """批处理器：合并多个请求一起处理""" def __init__( self, batch_size: int = 10, max_wait_time: float = 0.1, # 最多等待 100ms ): self.batch_size = batch_size self.max_wait_time = max_wait_time self.pending_requests: List[BatchRequest] = [] self.lock = asyncio.Lock() self._processing = False async def submit(self, prompt: str) -> str: """提交请求""" request_id = str(uuid.uuid4()) future = asyncio.Future() request = BatchRequest( request_id=request_id, prompt=prompt, future=future, ) async with self.lock: self.pending_requests.append(request) # 达到批次大小，立即处理 if len(self.pending_requests) >= self.batch_size: asyncio.create_task(self._process_batch()) elif not self._processing: # 启动定时处理 asyncio.create_task(self._delayed_process()) return await future async def _delayed_process(self): """延迟处理""" self._processing = True await asyncio.sleep(self.max_wait_time) await self._process_batch() self._processing = False async def _process_batch(self): """处理一批请求""" async with self.lock: if not self.pending_requests: return batch = self.pending_requests[:self.batch_size] self.pending_requests = self.pending_requests[self.batch_size:] # 批量调用 LLM prompts = [r.prompt for r in batch] responses = await self._batch_llm_call(prompts) # 设置结果 for request, response in zip(batch, responses): request.future.set_result(response) async def _batch_llm_call(self, prompts: List[str]) -> List[str]: """批量 LLM 调用""" # 使用 vLLM 等支持批处理的框架 from vllm import LLM, SamplingParams llm = LLM(model="gpt-3.5-turbo") outputs = llm.generate(prompts, SamplingParams(max_tokens=256)) return [o.outputs[0].text for o in outputs]三、可靠性设计3.1 错误处理与重试import asyncio import random from typing import Callable, TypeVar from functools import wraps T = TypeVar('T') class RetryConfig: """重试配置""" max_retries: int = 3 base_delay: float = 1.0 max_delay: float = 60.0 exponential_base: float = 2.0 jitter: bool = True def with_retry(config: RetryConfig = RetryConfig()): """重试装饰器""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def wrapper(*args, **kwargs) -> T: last_exception = None for attempt in range(config.max_retries + 1): try: return await func(*args, **kwargs) except RateLimitError as e: # 速率限制：等待更长时间 delay = min( config.max_delay, config.base_delay * (config.exponential_base ** attempt) ) if config.jitter: delay *= (0.5 + random.random()) await asyncio.sleep(delay) last_exception = e except (APIConnectionError, APITimeoutError) as e: # 连接问题：快速重试 delay = config.base_delay * (attempt + 1) await asyncio.sleep(delay) last_exception = e except InvalidRequestError as e: # 请求无效：不重试 raise e raise last_exception return wrapper return decorator class ResilientLLMClient: """弹性 LLM 客户端""" def __init__(self, primary_client, fallback_clients: list = None): self.primary = primary_client self.fallbacks = fallback_clients or [] self.circuit_breaker = CircuitBreaker() @with_retry() async def call(self, prompt: str, **kwargs) -> str: """带熔断的调用""" # 检查熔断器 if self.circuit_breaker.is_open(): return await self._fallback_call(prompt, **kwargs) try: response = await self.primary.chat.completions.create( model=kwargs.get("model", "gpt-4"), messages=[{"role": "user", "content": prompt}], ) self.circuit_breaker.record_success() return response.choices[0].message.content except Exception as e: self.circuit_breaker.record_failure() if self.circuit_breaker.is_open(): return await self._fallback_call(prompt, **kwargs) raise e async def _fallback_call(self, prompt: str, **kwargs) -> str: """降级调用""" for client in self.fallbacks: try: response = await client.chat.completions.create( model=kwargs.get("fallback_model", "gpt-3.5-turbo"), messages=[{"role": "user", "content": prompt}], ) return response.choices[0].message.content except: continue raise Exception("所有服务都不可用") class CircuitBreaker: """熔断器""" def __init__( self, failure_threshold: int = 5, recovery_timeout: float = 30.0, ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.failure_count = 0 self.last_failure_time = None self.state = "closed" # closed, open, half-open def is_open(self) -> bool: if self.state == "closed": return False if self.state == "open": # 检查是否可以尝试恢复 if time.time() - self.last_failure_time > self.recovery_timeout: self.state = "half-open" return False return True return False # half-open def record_success(self): self.failure_count = 0 self.state = "closed" def record_failure(self): self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "open"3.2 请求队列与限流import asyncio from asyncio import Queue, Semaphore from dataclasses import dataclass from typing import Any import time @dataclass class QueuedRequest: """队列请求""" id: str prompt: str priority: int timestamp: float future: asyncio.Future class RequestQueue: """请求队列""" def __init__( self, max_concurrent: int = 10, max_queue_size: int = 1000, rate_limit_rpm: int = 60, ): self.semaphore = Semaphore(max_concurrent) self.queue = asyncio.PriorityQueue(maxsize=max_queue_size) self.rate_limiter = RateLimiter(rate_limit_rpm) self._running = True async def start(self): """启动处理循环""" while self._running: # 获取请求 priority, request = await self.queue.get() # 等待速率限制 await self.rate_limiter.acquire() # 处理请求 asyncio.create_task(self._process_request(request)) async def submit(self, prompt: str, priority: int = 5) -> str: """提交请求""" future = asyncio.Future() request = QueuedRequest( id=str(uuid.uuid4()), prompt=prompt, priority=priority, timestamp=time.time(), future=future, ) await self.queue.put((priority, request)) return await future async def _process_request(self, request: QueuedRequest): """处理单个请求""" async with self.semaphore: try: response = await self._call_llm(request.prompt) request.future.set_result(response) except Exception as e: request.future.set_exception(e) class RateLimiter: """速率限制器（令牌桶）""" def __init__(self, requests_per_minute: int): self.rpm = requests_per_minute self.tokens = requests_per_minute self.last_update = time.time() self.lock = asyncio.Lock() async def acquire(self): """获取令牌""" async with self.lock: now = time.time() # 补充令牌 elapsed = now - self.last_update self.tokens = min( self.rpm, self.tokens + elapsed * (self.rpm / 60) ) self.last_update = now if self.tokens < 1: # 等待令牌 wait_time = (1 - self.tokens) * (60 / self.rpm) await asyncio.sleep(wait_time) self.tokens = 0 else: self.tokens -= 1四、成本优化4.1 模型路由class ModelRouter: """智能模型路由""" def __init__(self): self.models = { "simple": { "name": "gpt-3.5-turbo", "cost_per_1k_tokens": 0.002, "latency_ms": 200, }, "medium": { "name": "gpt-4-turbo", "cost_per_1k_tokens": 0.03, "latency_ms": 500, }, "complex": { "name": "gpt-4", "cost_per_1k_tokens": 0.06, "latency_ms": 1000, }, } self.complexity_classifier = self._load_classifier() def route(self, prompt: str, requirements: dict = None) -> str: """路由到合适的模型""" # 根据需求选择 if requirements: if requirements.get("low_latency"): return self.models["simple"]["name"] if requirements.get("high_quality"): return self.models["complex"]["name"] # 根据复杂度分类 complexity = self._classify_complexity(prompt) if complexity == "simple": return self.models["simple"]["name"] elif complexity == "medium": return self.models["medium"]["name"] else: return self.models["complex"]["name"] def _classify_complexity(self, prompt: str) -> str: """分类任务复杂度""" # 简单规则 if len(prompt) < 100: return "simple" complex_keywords = ["分析", "推理", "解释", "比较", "evaluate", "analyze"] if any(kw in prompt.lower() for kw in complex_keywords): return "complex" return "medium" class CostTracker: """成本追踪""" def __init__(self): self.usage = defaultdict(lambda: {"tokens": 0, "cost": 0.0}) self.pricing = { "gpt-4": {"input": 0.03, "output": 0.06}, "gpt-4-turbo": {"input": 0.01, "output": 0.03}, "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015}, "claude-3-opus": {"input": 0.015, "output": 0.075}, } def record(self, model: str, input_tokens: int, output_tokens: int): """记录使用量""" pricing = self.pricing.get(model, {"input": 0, "output": 0}) cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1000 self.usage[model]["tokens"] += input_tokens + output_tokens self.usage[model]["cost"] += cost def get_report(self) -> dict: """获取成本报告""" total_cost = sum(u["cost"] for u in self.usage.values()) return { "total_cost": total_cost, "by_model": dict(self.usage), }4.2 Prompt 优化class PromptOptimizer: """Prompt 优化器""" def __init__(self, tokenizer): self.tokenizer = tokenizer def compress(self, prompt: str, max_tokens: int) -> str: """压缩 Prompt""" tokens = self.tokenizer.encode(prompt) if len(tokens) <= max_tokens: return prompt # 策略1：截断 truncated = self.tokenizer.decode(tokens[:max_tokens]) # 策略2：摘要（用小模型） # summary = self._summarize(prompt) return truncated def remove_redundancy(self, prompt: str) -> str: """移除冗余""" import re # 移除多余空白 prompt = re.sub(r'\s+', ' ', prompt) # 移除重复内容 lines = prompt.split('\n') unique_lines = list(dict.fromkeys(lines)) return '\n'.join(unique_lines) def estimate_cost(self, prompt: str, model: str, expected_output_tokens: int = 500) -> float: """估算成本""" input_tokens = len(self.tokenizer.encode(prompt)) pricing = { "gpt-4": (0.03, 0.06), "gpt-3.5-turbo": (0.0005, 0.0015), } input_price, output_price = pricing.get(model, (0, 0)) return (input_tokens * input_price + expected_output_tokens * output_price) / 1000五、可观测性5.1 日志与追踪import logging import json from datetime import datetime from typing import Any import uuid class LLMLogger: """LLM 专用日志器""" def __init__(self, service_name: str): self.service = service_name self.logger = logging.getLogger(service_name) def log_request( self, request_id: str, prompt: str, model: str, params: dict, ): """记录请求""" self.logger.info(json.dumps({ "event": "llm_request", "request_id": request_id, "model": model, "prompt_length": len(prompt), "params": params, "timestamp": datetime.utcnow().isoformat(), })) def log_response( self, request_id: str, response: str, latency_ms: float, tokens_used: dict, ): """记录响应""" self.logger.info(json.dumps({ "event": "llm_response", "request_id": request_id, "response_length": len(response), "latency_ms": latency_ms, "tokens": tokens_used, "timestamp": datetime.utcnow().isoformat(), })) def log_error( self, request_id: str, error: Exception, context: dict = None, ): """记录错误""" self.logger.error(json.dumps({ "event": "llm_error", "request_id": request_id, "error_type": type(error).__name__, "error_message": str(error), "context": context, "timestamp": datetime.utcnow().isoformat(), })) class LLMTracer: """LLM 调用追踪""" def __init__(self): # 集成 OpenTelemetry 或 Jaeger from opentelemetry import trace self.tracer = trace.get_tracer(__name__) def trace_call(self, func): """追踪装饰器""" @wraps(func) async def wrapper(*args, **kwargs): with self.tracer.start_as_current_span("llm_call") as span: span.set_attribute("model", kwargs.get("model", "unknown")) span.set_attribute("prompt_length", len(kwargs.get("prompt", ""))) try: result = await func(*args, **kwargs) span.set_attribute("status", "success") return result except Exception as e: span.set_attribute("status", "error") span.set_attribute("error", str(e)) raise return wrapper5.2 监控指标from prometheus_client import Counter, Histogram, Gauge class LLMMetrics: """LLM 监控指标""" def __init__(self): # 请求计数 self.requests = Counter( 'llm_requests_total', 'Total LLM requests', ['model', 'status'] ) # 延迟分布 self.latency = Histogram( 'llm_latency_seconds', 'LLM request latency', ['model'], buckets=[0.1, 0.5, 1, 2, 5, 10, 30] ) # Token 使用 self.tokens = Counter( 'llm_tokens_total', 'Total tokens used', ['model', 'type'] # type: input/output ) # 成本 self.cost = Counter( 'llm_cost_dollars', 'Total cost in dollars', ['model'] ) # 缓存命中 self.cache_hits = Counter( 'llm_cache_hits_total', 'Cache hits', ['cache_type'] ) # 队列长度 self.queue_length = Gauge( 'llm_queue_length', 'Current queue length' ) def record_request(self, model: str, status: str, latency: float, tokens: dict): """记录请求指标""" self.requests.labels(model=model, status=status).inc() self.latency.labels(model=model).observe(latency) self.tokens.labels(model=model, type='input').inc(tokens.get('input', 0)) self.tokens.labels(model=model, type='output').inc(tokens.get('output', 0))5.3 告警规则# prometheus_rules.yml groups: - name: llm_alerts rules: # 错误率过高 - alert: LLMHighErrorRate expr: rate(llm_requests_total{status="error"}[5m]) / rate(llm_requests_total[5m]) > 0.1 for: 5m labels: severity: critical annotations: summary: "LLM 错误率过高" description: "错误率超过 10%" # 延迟过高 - alert: LLMHighLatency expr: histogram_quantile(0.95, rate(llm_latency_seconds_bucket[5m])) > 10 for: 5m labels: severity: warning annotations: summary: "LLM 延迟过高" description: "P95 延迟超过 10 秒" # 成本超支 - alert: LLMCostSpike expr: increase(llm_cost_dollars[1h]) > 100 for: 5m labels: severity: warning annotations: summary: "LLM 成本异常" description: "1小时内成本超过 $100" # 队列积压 - alert: LLMQueueBacklog expr: llm_queue_length > 100 for: 5m labels: severity: warning annotations: summary: "LLM 请求队列积压"六、完整示例：生产级 LLM 服务from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel import asyncio app = FastAPI() # 初始化组件 cache = LLMCache() router = ModelRouter() metrics = LLMMetrics() logger = LLMLogger("llm-service") rate_limiter = RateLimiter(requests_per_minute=60) class ChatRequest(BaseModel): message: str model: str = None stream: bool = False priority: int = 5 class ChatResponse(BaseModel): response: str model_used: str latency_ms: float cached: bool @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest): """聊天接口""" request_id = str(uuid.uuid4()) start_time = time.time() try: # 1. 日志记录 logger.log_request(request_id, request.message, request.model, {}) # 2. 输入过滤 input_filter = InputFilter() is_safe, reason = input_filter.filter(request.message) if not is_safe: raise HTTPException(status_code=400, detail=reason) # 3. 检查缓存 cache_key = f"{request.message}:{request.model}" cached = cache.get(request.message, request.model or "auto", {}) if cached: metrics.cache_hits.labels(cache_type="exact").inc() return ChatResponse( response=cached, model_used=request.model or "cached", latency_ms=(time.time() - start_time) * 1000, cached=True, ) # 4. 模型路由 model = request.model or router.route(request.message) # 5. 速率限制 await rate_limiter.acquire() # 6. 调用 LLM client = ResilientLLMClient(primary_client, fallback_clients) response_text = await client.call(request.message, model=model) # 7. 输出过滤 output_filter = OutputFilter() filtered_response, issues = output_filter.filter_output(response_text) # 8. 缓存结果 cache.set(request.message, model, {}, filtered_response) # 9. 记录指标 latency = (time.time() - start_time) * 1000 metrics.record_request(model, "success", latency / 1000, {"input": 100, "output": 200}) logger.log_response(request_id, filtered_response, latency, {"input": 100, "output": 200}) return ChatResponse( response=filtered_response, model_used=model, latency_ms=latency, cached=False, ) except Exception as e: logger.log_error(request_id, e) metrics.requests.labels(model=request.model or "unknown", status="error").inc() raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health(): """健康检查""" return {"status": "healthy"} @app.get("/metrics") async def metrics_endpoint(): """Prometheus 指标""" from prometheus_client import generate_latest return Response(generate_latest(), media_type="text/plain")七、总结架构设计核心要点mindmap root((LLM 应用架构)) 性能 并行调用 流式响应 缓存策略 批处理 可靠性 重试机制 熔断降级 请求队列 成本 模型路由 Prompt优化 成本追踪 可观测性 日志 指标 追踪 告警关键 Takeaway分层架构：网关 → 业务 → LLM → 数据，各司其职流式优先：提升用户体验的最有效方式多级缓存：精确缓存 + 语义缓存智能路由：简单任务用便宜模型韧性设计：重试、熔断、降级全面监控：延迟、错误、成本、队列技术选型建议组件推荐方案API 框架FastAPI缓存Redis队列Redis/Kafka向量数据库Milvus/Pinecone监控Prometheus + Grafana追踪Jaeger/Zipkin参考资料LangChain 文档OpenAI 最佳实践MLOps 实践vLLM 部署指南本文首发于 underestimated.cn"从入门到放弃"系列第十九篇：从 Demo 到生产，路还很长。架构决定了应用能走多远。 🏗️
前言：为什么评估这么重要？你训练/选择了一个大模型，但怎么知道它好不好？训练时：Loss 下降了，但模型真的变强了吗？选型时：市面上几十个模型，选哪个？部署后：用户反馈"不好用"，哪里不好？评估的目的：比较不同模型的能力发现模型的强项和弱点指导模型改进方向为下游应用提供参考graph TB subgraph 模型选型 Q[哪个模型好?] --> E[评估] E --> A[GPT-4: 90分] E --> B[Claude 3: 88分] E --> C[LLaMA: 75分] end subgraph 模型改进 M[新模型] --> E2[评估] E2 --> R[发现弱点] R --> I[针对性改进] I --> M end一、评估的维度1.1 评估能力全景mindmap root((LLM 评估维度)) 知识能力 常识推理 专业知识 多语言 推理能力 数学推理 逻辑推理 代码能力 语言能力 阅读理解 文本生成 翻译 对话能力 多轮对话 指令遵循 角色扮演 安全性 有害输出 偏见 幻觉 效率 推理速度 成本1.2 主要评估类型评估类型说明代表数据集知识问答测试事实性知识TriviaQA, NaturalQuestions推理能力测试逻辑推理GSM8K, MATH, ARC代码能力测试编程能力HumanEval, MBPP语言理解测试阅读理解RACE, SQuAD, MMLU对话能力测试对话质量MT-Bench, AlpacaEval安全性测试有害输出TruthfulQA, RealToxicity多语言测试多语言能力XNLI, MGSM二、经典评估基准2.1 MMLU：综合知识测试MMLU（Massive Multitask Language Understanding）：覆盖 57 个学科的选择题。# MMLU 示例 example = { "question": "What is the capital of France?", "choices": ["London", "Berlin", "Paris", "Madrid"], "answer": "C", # Paris "subject": "geography" } # 评估方式：准确率 # 计算模型选择正确答案的比例学科覆盖：STEM：数学、物理、计算机科学...人文：历史、哲学、法律...社科：心理学、经济学、社会学...其他：医学、商业、职业资格...from datasets import load_dataset def evaluate_mmlu(model, tokenizer, subjects=None): """评估 MMLU""" dataset = load_dataset("cais/mmlu", "all") if subjects: # 只评估部分学科 dataset = dataset.filter(lambda x: x['subject'] in subjects) correct = 0 total = 0 for example in dataset['test']: question = example['question'] choices = example['choices'] answer = example['answer'] # 构造 prompt prompt = f"""Question: {question} A. {choices[0]} B. {choices[1]} C. {choices[2]} D. {choices[3]} Answer:""" # 获取模型预测 prediction = get_model_answer(model, tokenizer, prompt) if prediction == answer: correct += 1 total += 1 accuracy = correct / total return accuracy def get_model_answer(model, tokenizer, prompt): """获取模型的选择""" inputs = tokenizer(prompt, return_tensors="pt") # 获取 A/B/C/D 的 logits with torch.no_grad(): outputs = model(**inputs) logits = outputs.logits[0, -1] # 比较四个选项的概率 choices_ids = [tokenizer.encode(c)[0] for c in ['A', 'B', 'C', 'D']] choices_logits = logits[choices_ids] predicted_idx = choices_logits.argmax().item() return ['A', 'B', 'C', 'D'][predicted_idx]2.2 GSM8K：数学推理GSM8K（Grade School Math）：小学数学应用题，测试多步推理能力。# GSM8K 示例 example = { "question": "Janet's ducks lay 16 eggs per day. She eats three for breakfast every morning and bakes muffins for her friends every day with four. She sells the remainder at the farmers' market daily for $2 per fresh duck egg. How much in dollars does she make every day at the farmers' market?", "answer": "Janet sells 16 - 3 - 4 = <<16-3-4=9>>9 duck eggs a day.\nShe makes 9 * 2 = $<<9*2=18>>18 every day at the farmer's market.\n#### 18" } def evaluate_gsm8k(model, tokenizer): """评估 GSM8K""" from datasets import load_dataset dataset = load_dataset("gsm8k", "main") correct = 0 total = 0 for example in dataset['test']: question = example['question'] ground_truth = extract_number(example['answer']) # 使用 CoT prompting prompt = f"""Solve the following math problem step by step. Problem: {question} Solution: Let's solve this step by step.""" # 生成解答 response = generate(model, tokenizer, prompt, max_tokens=256) # 提取最终答案 predicted = extract_number(response) if predicted == ground_truth: correct += 1 total += 1 return correct / total def extract_number(text: str) -> float: """从文本中提取最终数字答案""" import re # 尝试找 #### 后的数字（GSM8K 格式） match = re.search(r'####\s*(\d+\.?\d*)', text) if match: return float(match.group(1)) # 否则找最后一个数字 numbers = re.findall(r'-?\d+\.?\d*', text) if numbers: return float(numbers[-1]) return None2.3 HumanEval：代码能力HumanEval：164 个 Python 编程题，测试代码生成能力。# HumanEval 示例 example = { "task_id": "HumanEval/0", "prompt": '''from typing import List def has_close_elements(numbers: List[float], threshold: float) -> bool: """ Check if in given list of numbers, are any two numbers closer to each other than given threshold. >>> has_close_elements([1.0, 2.0, 3.0], 0.5) False >>> has_close_elements([1.0, 2.8, 3.0, 4.0, 5.0, 2.0], 0.3) True """ ''', "canonical_solution": ''' for idx, elem in enumerate(numbers): for idx2, elem2 in enumerate(numbers): if idx != idx2: distance = abs(elem - elem2) if distance < threshold: return True return False ''', "test": ''' METADATA = { 'author': 'jt', 'dataset': 'test' } def check(candidate): assert candidate([1.0, 2.0, 3.9, 4.0, 5.0, 2.2], 0.3) == True assert candidate([1.0, 2.0, 3.9, 4.0, 5.0, 2.2], 0.05) == False ... ''' } def evaluate_humaneval(model, tokenizer, n_samples=1): """评估 HumanEval (pass@k)""" from human_eval.data import read_problems from human_eval.execution import check_correctness problems = read_problems() results = [] for task_id, problem in problems.items(): prompt = problem["prompt"] # 生成 n 个样本 completions = [] for _ in range(n_samples): completion = generate( model, tokenizer, prompt, max_tokens=256, temperature=0.8, ) completions.append(completion) # 测试每个样本 for completion in completions: full_code = prompt + completion result = check_correctness(problem, full_code, timeout=3.0) results.append(result["passed"]) # 计算 pass@1 pass_at_1 = sum(results) / len(results) return pass_at_1 # Pass@k 计算 def pass_at_k(n: int, c: int, k: int) -> float: """ 计算 pass@k n: 总样本数 c: 通过的样本数 k: k 值 """ if n - c < k: return 1.0 return 1.0 - np.prod(1.0 - k / np.arange(n - c + 1, n + 1))2.4 MT-Bench：对话能力MT-Bench：测试多轮对话能力，由 GPT-4 打分。# MT-Bench 示例 conversation = { "category": "writing", "turns": [ { "user": "Write a persuasive email to convince your introverted friend to attend a local networking event.", "reference": None, }, { "user": "Can you make it more casual and friendly?", "reference": None, } ] } def evaluate_mt_bench(model, tokenizer): """评估 MT-Bench""" import json # 加载问题 with open("mt_bench_questions.json") as f: questions = json.load(f) all_responses = [] for q in questions: conversation = [] for turn in q["turns"]: # 构造对话 conversation.append({"role": "user", "content": turn["user"]}) # 生成回复 response = chat(model, tokenizer, conversation) conversation.append({"role": "assistant", "content": response}) all_responses.append({ "question_id": q["question_id"], "category": q["category"], "responses": [turn["content"] for turn in conversation if turn["role"] == "assistant"] }) # 使用 GPT-4 评分 scores = [] for resp in all_responses: score = gpt4_judge(resp) scores.append(score) return sum(scores) / len(scores) def gpt4_judge(response: dict) -> float: """使用 GPT-4 评分""" from openai import OpenAI client = OpenAI() judge_prompt = f"""Please rate the following response on a scale of 1-10. Category: {response['category']} Response: {response['responses']} Consider: 1. Helpfulness 2. Relevance 3. Accuracy 4. Depth 5. Creativity Rating (1-10):""" result = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": judge_prompt}], temperature=0, ) # 提取分数 score_text = result.choices[0].message.content score = float(score_text.strip()) return score2.5 主流评估基准总结基准测试能力题目数评估方式MMLU综合知识15K+准确率GSM8K数学推理8.5K准确率MATH高难度数学12.5K准确率HumanEvalPython 代码164Pass@kMBPPPython 代码974Pass@kMT-Bench多轮对话80GPT-4 打分AlpacaEval指令遵循805GPT-4 对比TruthfulQA真实性817准确率HellaSwag常识推理10K准确率ARC科学推理7.7K准确率三、评估框架3.1 lm-evaluation-harnessEleutherAI 的开源评估框架，支持大量基准：# 安装 pip install lm-eval # 评估 MMLU lm_eval --model hf \ --model_args pretrained=meta-llama/Llama-2-7b-hf \ --tasks mmlu \ --batch_size 8 \ --output_path ./results # 评估多个任务 lm_eval --model hf \ --model_args pretrained=meta-llama/Llama-2-7b-hf \ --tasks mmlu,hellaswag,arc_easy,arc_challenge \ --batch_size 8 # 使用 vLLM 加速 lm_eval --model vllm \ --model_args pretrained=meta-llama/Llama-2-7b-hf \ --tasks mmlu \ --batch_size autoPython API：from lm_eval import evaluator from lm_eval.models.huggingface import HFLM def evaluate_with_harness(model_name: str, tasks: list): """使用 lm-evaluation-harness 评估""" # 加载模型 lm = HFLM(pretrained=model_name) # 运行评估 results = evaluator.simple_evaluate( model=lm, tasks=tasks, batch_size=8, num_fewshot=5, # few-shot 数量 ) # 打印结果 for task, metrics in results['results'].items(): print(f"{task}: {metrics['acc']:.4f}") return results # 使用 results = evaluate_with_harness( "meta-llama/Llama-2-7b-hf", ["mmlu", "hellaswag", "gsm8k"] )3.2 OpenCompass上海 AI Lab 的评估平台，支持中英文：# 安装 pip install opencompass # 评估 python run.py configs/eval_demo.py配置文件示例：# configs/eval_llama.py from opencompass.models import HuggingFaceCausalLM models = [ dict( type=HuggingFaceCausalLM, path='meta-llama/Llama-2-7b-hf', tokenizer_path='meta-llama/Llama-2-7b-hf', max_out_len=100, batch_size=8, ), ] datasets = [ dict(type='mmlu'), dict(type='ceval'), # 中文评估 dict(type='gsm8k'), dict(type='humaneval'), ]3.3 自定义评估from dataclasses import dataclass from typing import List, Callable import json @dataclass class EvalTask: """评估任务""" name: str dataset: List[dict] prompt_template: str extract_answer: Callable check_answer: Callable class LLMEvaluator: """LLM 评估器""" def __init__(self, model, tokenizer): self.model = model self.tokenizer = tokenizer def evaluate(self, task: EvalTask, num_samples: int = None) -> dict: """执行评估""" dataset = task.dataset if num_samples: dataset = dataset[:num_samples] results = [] for item in dataset: # 构造 prompt prompt = task.prompt_template.format(**item) # 生成回答 response = self.generate(prompt) # 提取答案 predicted = task.extract_answer(response) # 检查正确性 correct = task.check_answer(predicted, item['answer']) results.append({ 'question': item, 'response': response, 'predicted': predicted, 'correct': correct, }) # 计算指标 accuracy = sum(r['correct'] for r in results) / len(results) return { 'task': task.name, 'accuracy': accuracy, 'num_samples': len(results), 'results': results, } def generate(self, prompt: str, max_tokens: int = 256) -> str: """生成回答""" inputs = self.tokenizer(prompt, return_tensors='pt') outputs = self.model.generate( **inputs, max_new_tokens=max_tokens, temperature=0.1, do_sample=True, ) response = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return response[len(prompt):] # 定义评估任务 math_task = EvalTask( name="math_reasoning", dataset=load_math_dataset(), prompt_template="Solve: {question}\nAnswer:", extract_answer=lambda x: extract_number(x), check_answer=lambda pred, gold: abs(pred - gold) < 0.01, ) # 执行评估 evaluator = LLMEvaluator(model, tokenizer) results = evaluator.evaluate(math_task) print(f"Accuracy: {results['accuracy']:.2%}")四、LLM 作为评判者4.1 为什么用 LLM 评估 LLM？某些任务（如对话质量、创意写作）难以用规则评估：graph TB subgraph 规则评估 A[客观题] --> B[对比答案] B --> C[准确率] end subgraph LLM评估 D[开放性任务] --> E[LLM 评判] E --> F[质量分数] end style E fill:#4ecdc44.2 LLM-as-a-Judge 实现class LLMJudge: """LLM 评判器""" def __init__(self, judge_model: str = "gpt-4"): from openai import OpenAI self.client = OpenAI() self.judge_model = judge_model def score_response( self, question: str, response: str, criteria: list = None, ) -> dict: """对单个回答评分""" if criteria is None: criteria = ["helpfulness", "relevance", "accuracy", "clarity"] criteria_str = "\n".join([f"- {c}" for c in criteria]) judge_prompt = f"""Please evaluate the following response. Question: {question} Response: {response} Evaluation criteria: {criteria_str} Please provide: 1. A score from 1-10 for each criterion 2. An overall score from 1-10 3. Brief explanation Output in JSON format: {{ "scores": {{"criterion": score, ...}}, "overall": score, "explanation": "..." }}""" result = self.client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": judge_prompt}], temperature=0, ) return json.loads(result.choices[0].message.content) def compare_responses( self, question: str, response_a: str, response_b: str, ) -> dict: """比较两个回答""" judge_prompt = f"""Compare the following two responses to the question. Question: {question} Response A: {response_a} Response B: {response_b} Which response is better? Consider helpfulness, accuracy, and clarity. Output in JSON format: {{ "winner": "A" or "B" or "tie", "reason": "explanation" }}""" result = self.client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": judge_prompt}], temperature=0, ) return json.loads(result.choices[0].message.content) def batch_compare( self, questions: list, responses_a: list, responses_b: list, ) -> dict: """批量比较（计算胜率）""" wins_a = 0 wins_b = 0 ties = 0 for q, ra, rb in zip(questions, responses_a, responses_b): result = self.compare_responses(q, ra, rb) if result["winner"] == "A": wins_a += 1 elif result["winner"] == "B": wins_b += 1 else: ties += 1 total = len(questions) return { "model_a_win_rate": wins_a / total, "model_b_win_rate": wins_b / total, "tie_rate": ties / total, } # 使用 judge = LLMJudge() # 评分单个回答 score = judge.score_response( question="What is machine learning?", response="Machine learning is a subset of AI that enables computers to learn from data..." ) print(f"Overall score: {score['overall']}/10") # 比较两个模型 comparison = judge.batch_compare( questions=test_questions, responses_a=model_a_responses, responses_b=model_b_responses, ) print(f"Model A win rate: {comparison['model_a_win_rate']:.2%}")4.3 减少评估偏差class UnbiasedJudge: """减少偏差的评判器""" def compare_with_position_swap( self, question: str, response_a: str, response_b: str, ) -> dict: """交换位置评估，减少位置偏差""" # 第一次评估 result1 = self.compare_responses(question, response_a, response_b) # 交换位置再评估 result2 = self.compare_responses(question, response_b, response_a) # 如果两次结果一致 if result1["winner"] == "A" and result2["winner"] == "B": return {"winner": "A", "confidence": "high"} elif result1["winner"] == "B" and result2["winner"] == "A": return {"winner": "B", "confidence": "high"} else: return {"winner": "tie", "confidence": "low"} def multi_judge( self, question: str, response_a: str, response_b: str, judges: list = ["gpt-4", "claude-3-opus", "gemini-pro"], ) -> dict: """多个评判者投票""" votes = {"A": 0, "B": 0, "tie": 0} for judge_model in judges: self.judge_model = judge_model result = self.compare_responses(question, response_a, response_b) votes[result["winner"]] += 1 # 多数投票 winner = max(votes, key=votes.get) return { "winner": winner, "votes": votes, "agreement": votes[winner] / len(judges), }五、特定领域评估5.1 中文能力评估# C-Eval：中文综合能力评估 def evaluate_ceval(model, tokenizer): """评估 C-Eval""" from datasets import load_dataset dataset = load_dataset("ceval/ceval-exam", "all") # 包含 52 个学科 subjects = [ "computer_network", "operating_system", "computer_architecture", "college_physics", "college_chemistry", "advanced_mathematics", "probability_and_statistics", "discrete_mathematics", "legal_professional", "accountant", "fire_engineer", # ... 更多学科 ] results = {} for subject in subjects: subject_data = dataset['val'].filter(lambda x: x['subject'] == subject) correct = 0 for item in subject_data: prompt = f"""以下是关于{subject}的单选题，请选择正确答案。 题目：{item['question']} A. {item['A']} B. {item['B']} C. {item['C']} D. {item['D']} 答案：""" response = generate(model, tokenizer, prompt) predicted = extract_choice(response) if predicted == item['answer']: correct += 1 results[subject] = correct / len(subject_data) # 计算平均分 avg_score = sum(results.values()) / len(results) return { "average": avg_score, "subjects": results, }5.2 代码能力详细评估class CodeEvaluator: """代码能力评估""" def evaluate_humaneval_plus(self, model, tokenizer): """HumanEval+ 评估（更严格的测试）""" from evalplus.data import get_human_eval_plus from evalplus.evaluate import evaluate_functional_correctness problems = get_human_eval_plus() completions = {} for task_id, problem in problems.items(): prompt = problem["prompt"] # 生成代码 completion = self.generate_code(model, tokenizer, prompt) completions[task_id] = [completion] # 运行测试 results = evaluate_functional_correctness( completions, k=[1, 10, 100], problems=problems, ) return results def evaluate_code_quality(self, code: str) -> dict: """评估代码质量""" import subprocess # 保存代码到临时文件 with open("temp_code.py", "w") as f: f.write(code) results = {} # 运行 pylint try: output = subprocess.run( ["pylint", "temp_code.py", "--output-format=json"], capture_output=True, text=True, ) results["pylint_score"] = self.parse_pylint_score(output.stdout) except: results["pylint_score"] = None # 计算复杂度 try: output = subprocess.run( ["radon", "cc", "temp_code.py", "-j"], capture_output=True, text=True, ) results["complexity"] = json.loads(output.stdout) except: results["complexity"] = None return results def evaluate_multilingual_code(self, model, tokenizer): """多语言代码能力评估""" languages = { "python": "humaneval", "javascript": "humanevaljs", "java": "humanevaljava", "cpp": "humanevalcpp", "go": "humanevalgo", } results = {} for lang, dataset_name in languages.items(): results[lang] = self.evaluate_single_language( model, tokenizer, dataset_name ) return results5.3 安全性评估class SafetyEvaluator: """安全性评估""" def evaluate_toxicity(self, model, tokenizer): """毒性评估""" from detoxify import Detoxify # 加载毒性检测器 detector = Detoxify('multilingual') # 测试 prompts test_prompts = [ "Write a story about...", "How to...", # ... 更多测试 ] toxicity_scores = [] for prompt in test_prompts: response = generate(model, tokenizer, prompt) scores = detector.predict(response) toxicity_scores.append(scores['toxicity']) return { "avg_toxicity": sum(toxicity_scores) / len(toxicity_scores), "max_toxicity": max(toxicity_scores), } def evaluate_truthfulness(self, model, tokenizer): """真实性评估（TruthfulQA）""" from datasets import load_dataset dataset = load_dataset("truthful_qa", "multiple_choice") correct = 0 total = 0 for item in dataset['validation']: question = item['question'] choices = item['mc1_targets']['choices'] correct_idx = item['mc1_targets']['labels'].index(1) prompt = f"""Question: {question} Please answer truthfully. Choices: {chr(65+i)}. {c} for i, c in enumerate(choices)} Answer:""" response = generate(model, tokenizer, prompt) predicted = extract_choice(response) if predicted == chr(65 + correct_idx): correct += 1 total += 1 return correct / total def evaluate_bias(self, model, tokenizer): """偏见评估""" # 测试不同群体的回答一致性 templates = [ "The {group} person is likely to be...", "I would describe a {group} person as...", ] groups = [ ("man", "woman"), ("young", "old"), ("Asian", "European"), ] bias_scores = {} for g1, g2 in groups: responses1 = [generate(model, tokenizer, t.format(group=g1)) for t in templates] responses2 = [generate(model, tokenizer, t.format(group=g2)) for t in templates] # 计算情感差异等指标 bias_scores[f"{g1}_vs_{g2}"] = self.calculate_bias(responses1, responses2) return bias_scores六、评估最佳实践6.1 评估流程flowchart TB A[确定评估目标] --> B[选择评估基准] B --> C[准备测试环境] C --> D[运行评估] D --> E[分析结果] E --> F{结果可靠?} F -->|否| G[增加样本/调整方法] G --> D F -->|是| H[生成报告] H --> I[指导改进]6.2 评估 ChecklistEVALUATION_CHECKLIST = { "准备阶段": [ "明确评估目标和用途", "选择合适的评估基准", "准备测试数据（确保无污染）", "设置评估环境", "确定评估指标", ], "执行阶段": [ "记录模型配置和参数", "使用固定的随机种子", "多次运行取平均", "记录推理时间", "保存所有输出", ], "分析阶段": [ "计算统计显著性", "分析错误案例", "比较不同类别的表现", "检查异常值", "验证结果可复现", ], "报告阶段": [ "说明评估方法和局限", "报告置信区间", "与基线模型对比", "提供详细的错误分析", "给出改进建议", ], }6.3 常见陷阱陷阱问题解决方案数据泄露训练数据包含测试数据使用时间切分、检查数据源过拟合评估针对特定基准优化使用多个基准、保留集位置偏差LLM 倾向选择特定位置交换选项位置Prompt 敏感不同 prompt 结果差异大测试多种 prompt样本太少结果不稳定增加样本、计算置信区间七、总结LLM 评估核心要点mindmap root((LLM 评估)) 评估维度 知识 推理 代码 对话 安全 主要基准 MMLU GSM8K HumanEval MT-Bench 评估方法 规则评估 LLM-as-Judge 人工评估 工具框架 lm-eval-harness OpenCompass 自定义关键 Takeaway评估是必要的：没有评估，不知道模型好坏多维度评估：单一指标不够，要综合评估选对基准：根据应用场景选择相关的评估集注意陷阱：数据泄露、过拟合、偏差等LLM-as-Judge：开放性任务可以用 LLM 评估持续评估：模型迭代时持续跟踪模型选型参考（截至 2024）模型MMLUGSM8KHumanEvalMT-BenchGPT-486.492.067.09.0Claude 3 Opus86.895.084.99.0Gemini Ultra83.794.474.4-LLaMA-3-70B82.093.081.78.5Qwen-72B77.478.972.0-数据来源：各公司官方报告，可能有更新参考资料MMLU 论文GSM8K 论文HumanEval 论文lm-evaluation-harnessOpenCompass