LLM运维

AI系统可观测性的三个支柱 传统软件的可观测性关注延迟、吞吐、错误率。AI系统需要额外关注：token消耗、模型质量漂移、幻觉率、工具调用成功率等AI特有指标。 指标采集 核心指标定义 from prometheus_client import Counter, Histogram, Gauge # 请求指标 REQUEST_TOTAL = Counter('ai_requests_total', 'Total AI requests', ['model', 'status']) REQUEST_LATENCY = Histogram('ai_request_duration_seconds', 'Request duration', ['model']) ACTIVE_REQUESTS = Gauge('ai_active_requests', 'Active requests') # Token指标 TOKEN_INPUT = Counter('ai_tokens_input_total', 'Input tokens', ['model']) TOKEN_OUTPUT = Counter('ai_tokens_output_total', 'Output tokens', ['model']) TOKEN_COST = Counter('ai_token_cost_usd', 'Token cost in USD', ['model']) # 质量指标 HALLUCINATION_RATE = Gauge('ai_hallucination_rate', 'Hallucination rate', ['model']) TOOL_CALL_SUCCESS = Counter('ai_tool_calls_total', 'Tool calls', ['tool', 'status']) # 缓存指标 CACHE_HIT_RATE = Gauge('ai_cache_hit_rate', 'Cache hit rate') 中间件实现 class ObservabilityMiddleware: def __init__(self, app): self.app = app async def __call__(self, request): ACTIVE_REQUESTS.inc() start = time.time() model = request.json.get("model", "unknown") try: response = await self.app(request) duration = time.time() - start REQUEST_TOTAL.labels(model=model, status="success").inc() REQUEST_LATENCY.labels(model=model).observe(duration) if "usage" in response: TOKEN_INPUT.labels(model=model).inc(response["usage"]["prompt_tokens"]) TOKEN_OUTPUT.labels(model=model).inc(response["usage"]["completion_tokens"]) cost = self.calculate_cost(model, response["usage"]) TOKEN_COST.labels(model=model).inc(cost) return response except Exception as e: REQUEST_TOTAL.labels(model=model, status="error").inc() raise finally: ACTIVE_REQUESTS.dec() def calculate_cost(self, model, usage): pricing = {"gpt-4": 0.03, "qwen3-32b": 0.002, "claude-3": 0.015} rate = pricing.get(model, 0.01) return (usage["prompt_tokens"] + usage["completion_tokens"]) / 1000 * rate 链路追踪 from opentelemetry import trace tracer = trace.get_tracer(__name__) class TracedLLMCall: def __init__(self, llm_client): self.client = llm_client async def chat(self, messages, **kwargs): with tracer.start_as_current_span("llm_chat") as span: span.set_attribute("llm.model", kwargs.get("model", "unknown")) span.set_attribute("llm.messages_count", len(messages)) span.set_attribute("llm.temperature", kwargs.get("temperature", 0.7)) start = time.time() response = await self.client.chat(messages, **kwargs) duration = time.time() - start span.set_attribute("llm.duration_ms", duration * 1000) span.set_attribute("llm.prompt_tokens", response["usage"]["prompt_tokens"]) span.set_attribute("llm.completion_tokens", response["usage"]["completion_tokens"]) return response 质量监控 class QualityMonitor: def __init__(self, sample_rate=0.05): self.sample_rate = sample_rate # 采样5%的请求做质量评估 async def evaluate_response(self, query, response, context=None): """异步评估响应质量""" import random if random.random() > self.sample_rate: return None metrics = {} # 幻觉检测 metrics["hallucination"] = await self.detect_hallucination(response, context) # 相关性 metrics["relevance"] = await self.score_relevance(query, response) # 毒性检测 metrics["toxicity"] = await self.detect_toxicity(response) # 记录到Prometheus if metrics["hallucination"]: HALLUCINATION_RATE.inc() else: HALLUCINATION_RATE.dec(0.01) return metrics 告警规则 # Prometheus告警规则 groups: - name: ai_alerts rules: - alert: HighErrorRate expr: rate(ai_requests_total{status="error"}[5m]) / rate(ai_requests_total[5m]) > 0.05 for: 5m annotations: summary: "AI error rate > 5%" - alert: HighLatency expr: histogram_quantile(0.95, ai_request_duration_seconds_bucket) > 30 for: 10m annotations: summary: "P95 latency > 30s" - alert: HighCost expr: rate(ai_token_cost_usd[1h]) > 100 for: 30m annotations: summary: "Hourly cost > $100" - alert: ModelDegradation expr: ai_hallucination_rate > 0.15 for: 1h annotations: summary: "Hallucination rate > 15%" Grafana仪表板 关键面板： ...

为什么需要AI网关？ 当企业使用多个LLM提供商（OpenAI、Anthropic、本地模型等）时，直接对接各家API会面临：密钥管理分散、无法统一限流、缺乏请求日志、故障切换困难。AI网关统一管理所有LLM请求，提供路由、缓存、限流、监控等基础设施。 核心架构 客户端 → AI网关 → LLM提供商A → LLM提供商B → 本地vLLM 实现方案 统一API接口 from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): model: str messages: list temperature: float = 0.7 max_tokens: int = 2048 stream: bool = False # 提供商配置 PROVIDERS = { "openai": {"base_url": "https://api.openai.com/v1", "api_key": "..."}, "anthropic": {"base_url": "https://api.anthropic.com", "api_key": "..."}, "local": {"base_url": "http://localhost:8000/v1", "api_key": "..."}, } # 模型到提供商的路由 MODEL_ROUTING = { "gpt-4": "openai", "claude-3": "anthropic", "qwen3-32b": "local", } @app.post("/v1/chat/completions") async def chat_completions(request: ChatRequest): provider = MODEL_ROUTING.get(request.model, "local") config = PROVIDERS[provider] # 转发请求 async with httpx.AsyncClient() as client: response = await client.post( f"{config['base_url']}/chat/completions", json=request.dict(), headers={"Authorization": f"Bearer {config['api_key']}"}, timeout=120 ) return response.json() 故障切换 class FailoverRouter: def __init__(self, routing_config): self.routing = routing_config # {model: [provider1, provider2, ...]} self.health = {p: True for providers in routing_config.values() for p in providers} async def route(self, model, request): providers = self.routing.get(model, ["local"]) for provider in providers: if not self.health[provider]: continue try: result = await self.call_provider(provider, request) return result except Exception as e: logger.warning(f"Provider {provider} failed: {e}") self.health[provider] = False continue raise ServiceUnavailableError("All providers failed") 请求缓存 import hashlib import redis.asyncio as redis class ResponseCache: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) def cache_key(self, model, messages, temperature): content = json.dumps({"model": model, "messages": messages, "temp": temperature}) return hashlib.sha256(content.encode()).hexdigest() async def get(self, model, messages, temperature): key = self.cache_key(model, messages, temperature) cached = await self.redis.get(key) return json.loads(cached) if cached else None async def set(self, model, messages, temperature, response, ttl=3600): key = self.cache_key(model, messages, temperature) await self.redis.setex(key, ttl, json.dumps(response)) 限流 from datetime import datetime, timedelta class RateLimiter: def __init__(self, redis): self.redis = redis async def check(self, user_id, limit=60, window=60): key = f"rate:{user_id}:{datetime.now().strftime('%Y%m%d%H%M')}" current = await self.redis.incr(key) if current == 1: await self.redis.expire(key, window) if current > limit: return False return True 日志与监控 class RequestLogger: def __init__(self): self.logger = structlog.get_logger() async def log(self, request, response, user_id, duration): self.logger.info("llm_request", user_id=user_id, model=request.model, input_tokens=response.get("usage", {}).get("prompt_tokens", 0), output_tokens=response.get("usage", {}).get("completion_tokens", 0), duration_ms=duration * 1000, provider=response.get("provider", "unknown"), status="success" if response.get("choices") else "error" ) 部署配置 Docker Compose version: '3.8' services: gateway: build: . ports: - "8080:8080" environment: - REDIS_URL=redis://redis:6379 depends_on: - redis redis: image: redis:7-alpine ports: - "6379:6379" prometheus: image: prom/prometheus ports: - "9090:9090" grafana: image: grafana/grafana ports: - "3000:3000" 结语 AI网关是LLM生产基础设施的核心组件。统一API、故障切换、缓存、限流和监控这五大功能，让LLM服务具备企业级的可靠性和可观测性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Ollama：简化LLM本地部署 Ollama是2026年最受欢迎的本地LLM部署工具之一。它以极简的命令行界面和自动化的模型管理，让在本地运行大模型变得前所未有的简单。但从"能跑"到"生产可用"之间，还有大量工程细节需要处理。 安装与环境准备 系统要求 # Linux安装 curl -fsSL https://ollama.com/install.sh | sh # 验证GPU支持 nvidia-smi # 确认GPU可用 ollama --version GPU显存规划 不同模型的显存需求： 模型 参数量 FP16显存 INT4显存 推荐GPU Qwen-3-7B 7B 14GB 5GB RTX 4060 8GB+ Llama-3-8B 8B 16GB 6GB RTX 4070 12GB+ Qwen-3-32B 32B 64GB 20GB RTX 4090 24GB+ Llama-3-70B 70B 140GB 40GB 2×A100 80GB Ollama服务配置 # 自定义模型存储路径 export OLLAMA_MODELS=/data/ollama/models # 监听所有网络接口（生产环境配合防火墙） export OLLAMA_HOST=0.0.0.0:11434 # 并发请求数 export OLLAMA_NUM_PARALLEL=4 # 上下文长度 export OLLAMA_CONTEXT_LENGTH=8192 # GPU层数（-1为全部卸载到GPU） export OLLAMA_NUM_GPU=-1 # 启动服务 ollama serve 模型管理 Modelfile自定义 # 基于Qwen-3创建自定义模型 FROM qwen3:32b # 系统提示词 SYSTEM """ 你是一个专业的技术助手。请提供准确、简洁的回答。 如果不确定，请明确说明。 """ # 参数调优 PARAMETER temperature 0.7 PARAMETER top_p 0.9 PARAMETER top_k 40 PARAMETER num_ctx 8192 PARAMETER num_gpu 50 PARAMETER stop "<|im_end|>" PARAMETER stop "<|endoftext|>" # 模板 TEMPLATE """ {{ if .System }}<|im_start|>system {{ .System }}<|im_end|> {{ end }}{{ range .Messages }}{{ if eq .Role "user" }}<|im_start|>user {{ .Content }}<|im_end|> {{ end }}{{ if eq .Role "assistant" }}<|im_start|>assistant {{ .Content }}<|im_end|> {{ end }}{{ end }}<|im_start|>assistant """ # 构建自定义模型 ollama create my-qwen -f Modelfile # 运行 ollama run my-qwen 模型量化 # Ollama自动选择量化级别 ollama pull llama3:70b # 默认INT4量化 ollama pull llama3:70b-q8_0 # 指定INT8 ollama pull llama3:70b-fp16 # FP16精度 # 从GGUF文件导入 ollama create my-model --file ./model.gguf API服务 REST API import requests # 基础对话 response = requests.post( "http://localhost:11434/api/chat", json={ "model": "my-qwen", "messages": [ {"role": "user", "content": "解释Transformer的注意力机制"} ], "stream": False, "options": { "temperature": 0.7, "num_ctx": 8192, } } ) print(response.json()["message"]["content"]) # 流式响应 response = requests.post( "http://localhost:11434/api/chat", json={"model": "my-qwen", "messages": [...], "stream": True}, stream=True ) for line in response.iter_lines(): if line: chunk = json.loads(line) print(chunk["message"]["content"], end="", flush=True) OpenAI兼容API Ollama提供OpenAI兼容接口，方便迁移现有应用： ...