Posts

引言 vLLM是2026年最流行的高性能LLM推理引擎。从PagedAttention到连续批处理，vLLM不断创新推理优化技术。本文将全面介绍2026年vLLM社区的最新进展。 vLLM 2026核心特性 PagedAttention 2.0 vLLM的招牌技术，2026年升级到2.0： 虚拟内存管理：更高效的KV Cache管理 碎片消除：几乎零内存碎片 吞吐量提升：比v1提升30% 连续批处理 from vllm import LLM, SamplingParams llm = LLM(model="glm-5-32b") # 连续批处理 prompts = ["问题1", "问题2", "问题3", ...] sampling_params = SamplingParams(temperature=0.7, max_tokens=500) outputs = llm.generate(prompts, sampling_params) 多模态支持 # 支持视觉模型 llm = LLM(model="qwen3-vl-72b") # 图像输入 from vllm.multimodal import ImageFeature outputs = llm.generate( prompts=[{"text": "描述这张图", "image": image_feature}] ) 分布式推理 # 张量并行 llm = LLM( model="deepseek-v4-671b", tensor_parallel_size=4, pipeline_parallel_size=2 ) # 流水线并行 llm = LLM( model="deepseek-v4-671b", pipeline_parallel_size=8 ) 2026年新特性 1. Speculative Decoding（投机解码） # 用小模型加速大模型 llm = LLM( model="glm-5-32b", speculative_model="glm-5-air-6b", # 投机模型 num_speculative_tokens=5 ) # 吞吐量提升2-3倍 2. 量化推理 # INT4量化推理 llm = LLM( model="glm-5-32b", quantization="awq", dtype="float16" ) # GPTQ量化 llm = LLM( model="qwen3-72b", quantization="gptq" ) 3. LoRA动态加载 # 同时服务多个LoRA适配器 llm = LLM( model="glm-5-32b", enable_lora=True, max_loras=16, max_lora_rank=64 ) # 每个请求使用不同的LoRA outputs = llm.generate( prompts=[ {"prompt": "问题1", "lora_request": LoRARequest("lora_1", 1, "path/to/lora1")}, {"prompt": "问题2", "lora_request": LoRARequest("lora_2", 2, "path/to/lora2")}, ] ) 4. 语法引导生成 # 约束输出为JSON from vllm.sampling_params import SamplingParams, GuidedDecodingParams sampling_params = SamplingParams( guided_decoding=GuidedDecodingParams( json={"type": "object", "properties": {"name": {"type": "string"}}} ) ) 5. 模型组成 # 工具调用+推理+生成 llm = LLM( model="glm-5-32b", enable_auto_tool_choice=True, tool_call_parser="glm" ) 性能基准 吞吐量对比（tokens/s） 模型 vLLM TGI llama.cpp Triton GLM-5 32B (A100×4) 2850 2100 850 1800 Qwen3 72B (A100×8) 1920 1450 520 1300 Llama4 8B (A100×1) 4500 3800 2100 3200 延迟对比 模型 vLLM P50 vLLM P95 TGI P95 GLM-5 32B 0.8s 2.1s 3.5s Qwen3 7B 0.2s 0.5s 0.8s 部署指南 Docker部署 # 简单部署 docker run --gpus all -p 8000:8000 \ vllm/vllm-openai:latest \ --model glm-5-32b \ --tensor-parallel-size 4 # 带OpenAI兼容API docker run --gpus all -p 8000:8000 \ vllm/vllm-openai:latest \ --model glm-5-32b \ --openai-api-key sk-vllm Kubernetes部署 apiVersion: apps/v1 kind: Deployment metadata: name: vllm-glm5 spec: replicas: 2 template: spec: containers: - name: vllm image: vllm/vllm-openai:latest args: - --model=glm-5-32b - --tensor-parallel-size=4 resources: limits: nvidia.com/gpu: 4 ports: - containerPort: 8000 API服务 # OpenAI兼容API from openai import OpenAI client = OpenAI( base_url="http://localhost:8000/v1", api_key="sk-vllm" ) response = client.chat.completions.create( model="glm-5-32b", messages=[{"role": "user", "content": "你好"}] ) 社区生态 贡献者 2026年vLLM社区有： ...

提示词也是代码 在LLM应用中，提示词是影响输出质量最关键的变量。但提示词往往散落在代码中、聊天记录里、文档中，缺乏系统的版本管理。将提示词纳入Git版本管理，是LLM工程化的基本要求。 提示词仓库结构 prompts/ ├── system/ │ ├── assistant.md # 通用助手系统提示 │ ├── coding_assistant.md # 编程助手系统提示 │ └── rag_assistant.md # RAG助手系统提示 ├── templates/ │ ├── chat.j2 # 对话模板 │ ├── summarize.j2 # 摘要模板 │ └── extract.j2 # 信息提取模板 ├── few_shot/ │ ├── classification.json # 分类示例 │ └── extraction.json # 提取示例 ├── versions/ │ ├── v1.0/ # 历史版本 │ └── v2.0/ └── config.yaml # 提示词配置 提示词模板管理 from jinja2 import Environment, FileSystemLoader import git class PromptManager: def __init__(self, prompts_dir="./prompts"): self.env = Environment(loader=FileSystemLoader(prompts_dir)) self.repo = git.Repo(prompts_dir) def get_prompt(self, template_name, **variables): """渲染提示词模板""" template = self.env.get_template(template_name) return template.render(**variables) def get_version(self, template_name, commit_hash): """获取指定版本的提示词""" blob = self.repo.commit(commit_hash).tree / template_name return blob.data_stream.read().decode() def diff_versions(self, template_name, v1, v2): """比较两个版本的差异""" diff = self.repo.git.diff(v1, v2, template_name) return diff def list_versions(self, template_name): """列出提示词的所有修改历史""" commits = list(self.repo.iter_commits(paths=template_name)) return [{"hash": c.hexsha[:8], "message": c.message, "date": c.committed_datetime} for c in commits] 提示词配置 # config.yaml prompts: assistant: template: "system/assistant.md" model: "qwen3-32b" temperature: 0.7 max_tokens: 2048 variables: - name: user_name required: true - name: context required: false default: "" summarize: template: "templates/summarize.j2" model: "qwen3-7b" # 摘要用小模型 temperature: 0.3 # 低温度保持一致性 max_tokens: 512 提示词A/B测试 class PromptABTest: def __init__(self, prompt_manager, variant_a, variant_b, split=0.5): self.pm = prompt_manager self.variant_a = variant_a # 版本A的commit hash self.variant_b = variant_b # 版本B的commit hash self.split = split self.results = {"a": [], "b": []} def get_prompt(self, template_name, user_id, **variables): """基于用户ID确定性分流""" hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) variant = "a" if (hash_val % 100) / 100 < self.split else "b" commit = self.variant_a if variant == "a" else self.variant_b template_str = self.pm.get_version(template_name, commit) from jinja2 import Template return Template(template_str).render(**variables), variant def record_result(self, variant, score): self.results[variant].append(score) def get_winner(self): avg_a = sum(self.results["a"]) / len(self.results["a"]) if self.results["a"] else 0 avg_b = sum(self.results["b"]) / len(self.results["b"]) if self.results["b"] else 0 return "a" if avg_a >= avg_b else "b" CI/CD集成 # .github/workflows/prompt-review.yml name: Prompt Review on: pull_request: paths: ['prompts/**'] jobs: validate: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Validate prompt templates run: | python scripts/validate_prompts.py --dir prompts/ - name: Run prompt tests run: | python scripts/test_prompts.py --model qwen3-7b --test-cases test_cases/ - name: Compare with previous version run: | python scripts/compare_prompts.py --base main --head ${{ github.head_ref }} - name: Quality regression check run: | python scripts/quality_check.py --threshold 0.85 提示词评估 class PromptEvaluator: def __init__(self, llm, test_cases): self.llm = llm self.test_cases = test_cases # 预标注的测试用例 async def evaluate(self, prompt_template, prompt_version): """评估提示词版本的质量""" results = [] for case in self.test_cases: # 渲染提示词 prompt = self.render(prompt_template, prompt_version, case["input"]) # 生成响应 response = await self.llm.generate(prompt) # 评估 score = self.score(response, case["expected"]) results.append({ "case_id": case["id"], "score": score, "response": response, }) avg_score = sum(r["score"] for r in results) / len(results) return {"version": prompt_version, "avg_score": avg_score, "details": results} 提示词回滚 # 回滚到上一个版本 git log --oneline prompts/system/assistant.md # a1b2c3d 优化系统提示措辞 # d4e5f6g 初始版本 # 查看差异 git diff d4e5f6g a1b2c3d prompts/system/assistant.md # 回滚 git checkout d4e5f6g -- prompts/system/assistant.md git commit -m "rollback: 回滚assistant提示词到初始版本" 实践建议 提示词与代码分离：提示词文件独立存放，不硬编码在代码中 模板化：使用Jinja2等模板引擎，支持变量注入 评审流程：提示词修改需要通过PR评审和自动化测试 版本标注：重要版本打tag，便于快速回滚 多语言管理：不同语言的提示词分目录管理 文档化：每个提示词文件包含描述、适用场景、注意事项 结语 提示词是LLM应用中投入产出比最高的优化点。将提示词纳入Git版本管理，配合模板化、A/B测试、自动化评估和CI/CD流程，可以让提示词迭代从"凭感觉改"变为"数据驱动改"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

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仪表板 关键面板： ...

引言 Ollama已经成为本地运行LLM最流行的工具。从2023年的简单命令行工具，到2026年的完整生态系统，Ollama让在本地运行大模型变得像安装App一样简单。本文将全面解析2026年的Ollama生态系统。 Ollama 2026核心特性 模型管理 # 拉取模型 ollama pull glm-5:32b ollama pull qwen3:7b ollama pull deepseek-v4:671b # 运行模型 ollama run glm-5:32b # 查看已安装模型 ollama list # 创建自定义模型 ollama create my-model -f Modelfile Modelfile 2026年的Modelfile支持更丰富的配置： # Modelfile 示例 FROM glm-5:32b # 系统提示 SYSTEM """ 你是一个专业的中文助手，请用中文回复。 """ # 参数 PARAMETER temperature 0.7 PARAMETER top_p 0.9 PARAMETER num_ctx 8192 PARAMETER stop "<|im_end|>" # 适配器（LoRA） ADAPTER ./my-lora.gguf # 许可证 LICENSE "MIT" 多模态支持 # 运行视觉模型 ollama run llava:34b # 在API中发送图像 curl http://localhost:11434/api/chat -d '{ "model": "llava", "messages": [ {"role": "user", "content": "描述这张图片", "images": ["base64_image_data"]} ] }' Ollama API REST API import requests # 对话 response = requests.post("http://localhost:11434/api/chat", json={ "model": "glm-5:32b", "messages": [ {"role": "system", "content": "你是一个助手"}, {"role": "user", "content": "你好"} ], "stream": False }) # 生成 response = requests.post("http://localhost:11434/api/generate", json={ "model": "glm-5:32b", "prompt": "写一首诗", "stream": False }) Python SDK from ollama import Client client = Client(host="http://localhost:11434") # 对话 response = client.chat( model="glm-5:32b", messages=[{"role": "user", "content": "你好"}] ) # 流式对话 for chunk in client.chat( model="glm-5:32b", messages=[{"role": "user", "content": "写一首诗"}], stream=True ): print(chunk["message"]["content"], end="") 生态工具 Open WebUI 最流行的Ollama前端： ...

LLM版本管理的挑战 传统的软件版本管理（Git）无法处理大模型文件（数十GB）。LLM的版本管理需要同时追踪代码、配置、数据、模型权重和评估结果。 版本管理工具链 MLflow模型注册 import mlflow # 记录模型版本 with mlflow.start_run(run_name="qwen3-32b-v2"): mlflow.log_params({ "base_model": "Qwen-3-32B", "fine_tune_method": "LoRA", "learning_rate": 2e-4, "epochs": 3, "dataset": "instruction-v3", }) mlflow.log_metrics({ "eval_loss": 0.45, "eval_accuracy": 0.89, "human_eval_score": 4.2, }) # 注册模型 mlflow.register_model( "runs:/abc123/model", "qwen3-32b-instruct", tags={ "version": "v2.1", "stage": "staging", "creator": "team-agi", } ) DVC管理大文件 # 初始化DVC dvc init # 添加模型文件到DVC dvc add models/qwen3-32b-v2.1/ # 推送到远程存储 dvc remote add -d storage s3://my-bucket/models dvc push # Git只追踪.dvc文件（指针），不追踪实际大文件 git add models/qwen3-32b-v2.1/.dvc git commit -m "Add qwen3-32b v2.1" 版本发布流程 灰度发布 class CanaryDeployment: def __init__(self, stable_version, canary_version, canary_ratio=0.1): self.stable = stable_version self.canary = canary_version self.ratio = canary_ratio self.metrics = {"stable": [], "canary": []} def route(self, request): """灰度路由""" import random if random.random() < self.ratio: self.metrics["canary"].append({"time": time.time()}) return self.canary else: self.metrics["stable"].append({"time": time.time()}) return self.stable def evaluate(self): """评估灰度结果""" canary_latency = self.get_avg_latency("canary") stable_latency = self.get_avg_latency("stable") canary_error = self.get_error_rate("canary") stable_error = self.get_error_rate("stable") # 灰度通过条件 if canary_latency > stable_latency * 1.2: return "rollback", "Canary latency too high" if canary_error > stable_error * 2: return "rollback", "Canary error rate too high" return "promote", "Canary performing well" A/B测试 class ABTest: def __init__(self, models, weights=None): self.models = models self.weights = weights or [1/len(models)] * len(models) self.results = {m: {"satisfied": 0, "total": 0} for m in models} def route(self, user_id): # 基于用户ID的确定性路由 hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) cumulative = 0 for model, weight in zip(self.models, self.weights): cumulative += weight if (hash_val % 1000) / 1000 < cumulative: return model def record_feedback(self, model, satisfied): self.results[model]["total"] += 1 if satisfied: self.results[model]["satisfied"] += 1 def get_winner(self): rates = {m: r["satisfied"]/r["total"] for m, r in self.results.items() if r["total"] > 0} return max(rates, key=rates.get) if rates else None CI/CD管线 # .github/workflows/model-deploy.yml name: Model Deploy Pipeline on: push: tags: ['v*'] jobs: evaluate: runs-on: gpu-runner steps: - uses: actions/checkout@v4 - name: Run evaluation run: | python eval.py --model checkpoints/latest --benchmark mmlu,gsm8k,humaneval - name: Check quality gates run: | python check_gates.py --min_accuracy 0.85 --max_regression 0.02 deploy_staging: needs: evaluate runs-on: deploy-runner steps: - name: Deploy to staging run: | ./deploy.sh --env staging --version ${{ github.ref_name }} - name: Run smoke tests run: | python smoke_test.py --env staging canary: needs: deploy_staging runs-on: deploy-runner steps: - name: Canary deployment (10%) run: | ./deploy.sh --env production --version ${{ github.ref_name }} --canary 0.1 - name: Monitor for 1 hour run: | python monitor.py --duration 3600 --check latency,error_rate full_deploy: needs: canary runs-on: deploy-runner steps: - name: Full deployment run: | ./deploy.sh --env production --version ${{ github.ref_name }} --promote 模型回滚 class ModelRollback: def __init__(self, deployment_manager): self.deployment = deployment_manager self.version_history = [] async def rollback(self, target_version=None): """回滚到指定版本或上一个稳定版本""" if target_version is None: target_version = self.get_previous_stable() logger.info(f"Rolling back to {target_version}") # 快速切换流量 await self.deployment.switch_traffic( from_version="current", to_version=target_version, ratio=1.0 # 100%切换 ) # 验证回滚 health = await self.deployment.health_check(target_version) if not health: logger.error("Rollback target also unhealthy!") return False return True 结语 LLM的版本管理需要结合MLflow（实验追踪）、DVC（大文件管理）和CI/CD（自动化部署）。灰度发布和快速回滚是降低部署风险的关键能力。建立完善的MLOps流程，可以让模型迭代从"手动谨慎"变为"自动自信"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 评估数据的质量直接决定了评估结果的可信度。“垃圾进，垃圾出"在LLM评估中同样适用。一个有偏见的评估数据集可能让你做出错误的模型选择。2026年，评估数据策划已经成为一门专门的学问。本文将系统介绍评估数据策划的方法论。 评估数据的挑战 挑战一：代表性 评估数据是否代表了真实使用场景？ 问题：评估数据全部来自新闻文本 真实场景：用户输入包含口语、错别字、混合语言 → 评估结果无法反映真实效果 挑战二：多样性 评估数据是否覆盖了各种输入类型？ 问题：情感分析评估集只有明显正面/负面 缺失：讽刺、中性、混合情感 → 模型在边界情况上的表现未知 挑战三：数据污染 评估数据是否泄露到了训练集中？ 挑战四：时效性 评估数据是否过时？ 挑战五：偏见 评估数据是否对某些群体或观点有偏见？ 评估数据策划流程 步骤一：需求分析 eval_data_requirements = { "task": "客服对话质量评估", "dimensions": { "准确性": "回复信息是否正确", "有用性": "是否解决了用户问题", "态度": "回复是否礼貌友好", "效率": "是否在3轮内解决" }, "coverage": { "query_types": ["咨询", "投诉", "退款", "技术支持"], "difficulty": ["简单", "中等", "困难"], "languages": ["中文"], "domains": ["电商", "金融", "教育"] }, "size": { "minimum": 500, "recommended": 2000, "ideal": 5000 } } 步骤二：数据收集 来源一：真实用户数据 def collect_real_user_data(production_logs, n=1000): """ 从生产日志中采样真实用户数据 """ # 随机采样 sampled = random.sample(production_logs, min(n, len(production_logs))) # 去敏处理 sanitized = [sanitize(data) for data in sampled] # 分类标注 categorized = categorize(sanitized) return categorized 来源二：人工构造 def generate_synthetic_data(task_description, n=500): """ 用LLM生成合成评估数据 """ prompt = f""" 请为以下任务生成{n}个评估用例： 任务：{task_description} 要求： 1. 覆盖不同难度（简单/中等/困难） 2. 包含边界情况 3. 包含对抗性输入 4. 输入多样化 以JSON格式输出。 """ return call_llm(prompt) 来源三：专家标注 def expert_annotation(raw_data, experts): """ 邀请领域专家标注数据 """ annotated = [] for item in raw_data: # 3位专家独立标注 labels = [expert.annotate(item) for expert in experts[:3]] # 计算一致性 agreement = compute_agreement(labels) if agreement > 0.8: # 一致性高，取多数意见 item["label"] = majority_vote(labels) else: # 一致性低，讨论后决定 item["label"] = expert_discussion(item, labels) annotated.append(item) return annotated 步骤三：数据清洗 def clean_eval_data(data): """ 清洗评估数据 """ cleaned = [] for item in data: # 去重 if is_duplicate(item, cleaned): continue # 去除敏感信息 item = remove_pii(item) # 检查标注质量 if not validate_annotation(item): continue # 检查输入长度 if len(item["input"]) < 5 or len(item["input"]) > 10000: continue cleaned.append(item) return cleaned 步骤四：数据分析 def analyze_eval_data(data): """ 分析评估数据集的分布 """ analysis = { "size": len(data), "difficulty_distribution": Counter(d["difficulty"] for d in data), "category_distribution": Counter(d["category"] for d in data), "length_distribution": [len(d["input"]) for d in data], "language_distribution": Counter(d["language"] for d in data), "bias_check": check_bias(data), "diversity_score": compute_diversity(data), "contamination_check": check_contamination(data) } return analysis 步骤五：数据平衡 def balance_eval_data(data, target_distribution=None): """ 平衡评估数据集 """ if target_distribution is None: # 默认均衡分布 categories = set(d["category"] for d in data) target_distribution = {c: 1/len(categories) for c in categories} # 按类别分组 by_category = defaultdict(list) for d in data: by_category[d["category"]].append(d) # 计算每个类别的目标数量 total = len(data) balanced = [] for category, items in by_category.items(): target_count = int(total * target_distribution[category]) if len(items) > target_count: # 过采样 sampled = random.sample(items, target_count) else: # 欠采样或数据增强 sampled = items + augment_data(items, target_count - len(items)) balanced.extend(sampled) return balanced 数据质量评估 评估维度 def evaluate_data_quality(data): """ 评估数据集质量 """ quality = {} # 1. 覆盖度：是否覆盖了所有需要评估的场景 quality["coverage"] = evaluate_coverage(data) # 2. 多样性：输入是否足够多样 quality["diversity"] = evaluate_diversity(data) # 3. 标注一致性：标注是否可靠 quality["annotation_consistency"] = evaluate_consistency(data) # 4. 偏见检测：是否存在偏见 quality["bias"] = detect_bias(data) # 5. 污染检测：是否与训练数据重叠 quality["contamination"] = detect_contamination(data) # 6. 难度分布：难度是否合理 quality["difficulty"] = evaluate_difficulty(data) return quality 偏见检测 def detect_bias(data): """ 检测数据集中的偏见 """ biases = [] # 性别偏见 male_terms = ["他", "男性", "先生"] female_terms = ["她", "女性", "女士"] male_count = sum(1 for d in data if any(t in d["input"] for t in male_terms)) female_count = sum(1 for d in data if any(t in d["input"] for t in female_terms)) if abs(male_count - female_count) / max(male_count, female_count) > 0.3: biases.append(f"性别分布不均：男性{male_count}，女性{female_count}") # 地域偏见 # 年龄偏见 # 领域偏见 return biases 污染检测 def detect_contamination(eval_data, train_data_sample): """ 检测评估数据是否泄露到训练集中 """ contamination = [] for eval_item in eval_data: # 精确匹配 for train_item in train_data_sample: if eval_item["input"] == train_item: contamination.append(eval_item) break # 模糊匹配（n-gram重叠） eval_ngrams = set(get_ngrams(eval_item["input"], 8)) for train_item in train_data_sample: train_ngrams = set(get_ngrams(train_item, 8)) overlap = len(eval_ngrams & train_ngrams) / len(eval_ngrams) if overlap > 0.8: contamination.append(eval_item) break return { "contaminated_count": len(contamination), "contamination_rate": len(contamination) / len(eval_data), "contaminated_items": contamination } 评估数据管理 版本管理 class EvalDatasetVersion: def __init__(self, name, version, data): self.name = name self.version = version self.data = data self.created_at = datetime.now() self.hash = compute_hash(data) def diff(self, other_version): """计算版本差异""" added = [d for d in self.data if d not in other_version.data] removed = [d for d in other_version.data if d not in self.data] return {"added": added, "removed": removed} 数据集文档 # eval_dataset_card.yaml name: "客服对话评估集 v2.0" version: "2.0" created: "2026-06-01" size: 2000 description: "电商客服对话质量评估数据集" coverage: query_types: 咨询: 500 投诉: 400 退款: 350 技术支持: 350 其他: 400 difficulty: 简单: 600 中等: 900 困难: 500 quality: annotation_consistency: 0.87 diversity_score: 0.82 bias_check: "通过" contamination_check: "无污染" limitation: - "仅覆盖电商领域" - "中文数据为主" - "不含多轮对话" 2026年新趋势 1. 动态评估数据 评估数据集定期更新，防止数据污染和过时。 ...

引言 AI应用的监控比传统软件复杂得多。除了常规的系统指标（CPU、内存、延迟），还需要监控AI特有的指标（输出质量、幻觉率、安全事件）。2026年，AI性能监控已经发展成为一个专门的领域。本文将系统介绍AI性能监控体系的构建。 AI监控的独特需求 传统软件监控 vs AI监控 维度 传统软件 AI应用 延迟 毫秒级 秒级（可接受） 错误类型 崩溃、超时 幻觉、不当内容 质量指标 功能正确性 输出准确性、相关性 成本 服务器成本 API调用成本（按token计） 变化来源 代码部署 代码+模型版本+提示 AI监控的核心指标 AI监控指标体系 ├── 性能指标 │ ├── 延迟（P50/P95/P99） │ ├── 吞吐量 │ └── 并发数 ├── 质量指标 │ ├── 输出准确率 │ ├── 幻觉率 │ ├── 拒绝率 │ └── 用户满意度 ├── 成本指标 │ ├── 每次请求成本 │ ├── 每日总成本 │ └── token效率 ├── 安全指标 │ ├── 有害内容率 │ ├── 注入攻击次数 │ └── 数据泄露事件 └── 可靠性指标 ├── 可用性 ├── 错误率 └── 降级率 监控架构 数据采集层 class MetricsCollector: def __init__(self): self.collectors = [ LatencyCollector(), QualityCollector(), CostCollector(), SafetyCollector(), ReliabilityCollector() ] def record_request(self, request_id, request, response, metadata): """记录每次请求""" for collector in self.collectors: collector.record(request_id, request, response, metadata) 指标计算层 class MetricsCalculator: def calculate(self, raw_metrics): return { "latency": { "p50": percentile(raw_metrics["latencies"], 50), "p95": percentile(raw_metrics["latencies"], 95), "p99": percentile(raw_metrics["latencies"], 99), }, "quality": { "accuracy": raw_metrics["correct"] / raw_metrics["total"], "hallucination_rate": raw_metrics["hallucinations"] / raw_metrics["total"], "refusal_rate": raw_metrics["refusals"] / raw_metrics["total"], }, "cost": { "per_request": raw_metrics["total_cost"] / raw_metrics["total"], "daily": raw_metrics["total_cost"], "token_efficiency": raw_metrics["output_tokens"] / raw_metrics["input_tokens"], }, "safety": { "harmful_rate": raw_metrics["harmful"] / raw_metrics["total"], "injection_attempts": raw_metrics["injections"], }, "reliability": { "availability": 1 - raw_metrics["downtime"] / raw_metrics["total_time"], "error_rate": raw_metrics["errors"] / raw_metrics["total"], } } 告警层 class AlertManager: def __init__(self): self.rules = [ AlertRule("high_latency", "p95_latency > 5000", severity="warning"), AlertRule("critical_latency", "p99_latency > 10000", severity="critical"), AlertRule("high_error", "error_rate > 0.05", severity="critical"), AlertRule("quality_drop", "accuracy < 0.85", severity="warning"), AlertRule("hallucination_spike", "hallucination_rate > 0.1", severity="critical"), AlertRule("cost_spike", "daily_cost > budget * 1.2", severity="warning"), AlertRule("safety_incident", "harmful_rate > 0.01", severity="critical"), ] def check(self, metrics): alerts = [] for rule in self.rules: if rule.evaluate(metrics): alerts.append(rule.create_alert(metrics)) if alerts: self.notify(alerts) return alerts 关键监控实现 延迟监控 class LatencyMonitor: def __init__(self): self.latencies = SlidingWindow(size=10000) def record(self, request_id, start_time, end_time): latency = end_time - start_time self.latencies.append(latency) # 实时检查 if latency > 10: # 超过10秒 self.alert(f"请求 {request_id} 延迟 {latency:.1f}s") def get_stats(self): return { "p50": self.latencies.percentile(50), "p95": self.latencies.percentile(95), "p99": self.latencies.percentile(99), "max": self.latencies.max(), "avg": self.latencies.mean() } 质量监控 class QualityMonitor: def __init__(self): self.sample_rate = 0.1 # 采样10%进行质量评估 self.evaluator = LLMJudge(model="gpt-5") # 用GPT-5评估 async def evaluate(self, request, response): """异步评估输出质量""" if random.random() > self.sample_rate: return # 采样 # 用LLM评估 score = await self.evaluator.evaluate( input=request, output=response, criteria=["accuracy", "relevance", "completeness"] ) if score["accuracy"] < 0.7: self.alert(f"低质量输出检测：{score}") return score 成本监控 class CostMonitor: def __init__(self, daily_budget=100): self.daily_budget = daily_budget self.today_cost = 0 self.costs = [] def record(self, input_tokens, output_tokens, model): cost = calculate_cost(input_tokens, output_tokens, model) self.today_cost += cost self.costs.append({"timestamp": datetime.now(), "cost": cost}) # 预算检查 if self.today_cost > self.daily_budget * 0.8: self.alert("日预算已用80%") if self.today_cost > self.daily_budget: self.alert("日预算超支！") return "stop" # 触发熔断 安全监控 class SafetyMonitor: def __init__(self): self.content_filter = ContentFilter() self.injection_detector = InjectionDetector() def check_input(self, user_input): """检查输入安全性""" if self.injection_detector.is_injection(user_input): self.log_incident("injection_attempt", user_input) return False if self.content_filter.is_harmful(user_input): self.log_incident("harmful_input", user_input) return False return True def check_output(self, output): """检查输出安全性""" if self.content_filter.is_harmful(output): self.log_incident("harmful_output", output) return False return True 可视化仪表板 class MonitoringDashboard: def render(self): return { "overview": { "status": "healthy", # healthy/warning/critical "uptime": "99.97%", "requests_today": 154289, "avg_latency": "1.2s", "cost_today": "$45.30" }, "latency_chart": self.render_latency_chart(), "quality_trend": self.render_quality_trend(), "cost_trend": self.render_cost_trend(), "alerts": self.get_active_alerts(), "top_errors": self.get_top_errors() } 告警策略 告警分级 级别 条件 响应时间 通知方式 P0 系统不可用 立即 电话+短信+邮件 P1 关键指标超标 15分钟 短信+邮件 P2 质量下降 1小时 邮件+IM P3 预警 4小时 IM 告警抑制 def should_suppress(alert, recent_alerts): """避免告警风暴""" # 同类告警5分钟内只发一次 for recent in recent_alerts: if (recent["type"] == alert["type"] and (datetime.now() - recent["timestamp"]).seconds < 300): return True return False 2026年新趋势 1. AI自监控 用AI监控AI：模型自己评估输出质量，自动发现异常。 ...

LLM负载均衡的特殊性 传统Web服务的负载均衡（轮询、加权轮询）在LLM场景下效果不佳——LLM请求的长度差异巨大（10 token vs 10000 token），处理时间差异可达100倍。简单的轮询会导致某些节点被长请求占满，而短请求也被迫排队。 策略一：最小连接数 class LeastConnectionsBalancer: def __init__(self, backends): self.backends = {b: 0 for b in backends} # backend -> active_connections self.lock = asyncio.Lock() async def get_backend(self): async with self.lock: backend = min(self.backends, key=self.backends.get) self.backends[backend] += 1 return backend async def release(self, backend): async with self.lock: self.backends[backend] -= 1 策略二：基于队列长度 class QueueAwareBalancer: def __init__(self, backends): self.queues = {b: asyncio.Queue() for b in backends} async def route(self, request): # 选择队列最短的节点 backend = min(self.queues, key=lambda b: self.queues[b].qsize()) await self.queues[backend].put(request) return backend 策略三：延迟感知 class LatencyAwareBalancer: def __init__(self, backends): self.backends = backends self.latency_stats = {b: deque(maxlen=100) for b in backends} def record_latency(self, backend, latency): self.latency_stats[backend].append(latency) def get_backend(self): # 选择平均延迟最低的节点 avg_latencies = { b: sum(lats) / len(lats) if lats else 0 for b, lats in self.latency_stats.items() } return min(avg_latencies, key=avg_latencies.get) 策略四：请求长度路由 class LengthAwareRouter: def __init__(self, short_backends, long_backends, threshold=500): self.short_backends = short_backends # 小模型，处理短请求 self.long_backends = long_backends # 大模型，处理长请求 self.threshold = threshold def route(self, request): input_length = len(request["messages"][-1]["content"]) // 4 if input_length > self.threshold: return self.select_least_loaded(self.long_backends) else: return self.select_least_loaded(self.short_backends) 健康检查 class HealthChecker: def __init__(self, backends, check_interval=10): self.backends = {b: {"healthy": True, "last_check": 0} for b in backends} self.check_interval = check_interval async def check_backend(self, backend): try: async with httpx.AsyncClient() as client: resp = await client.get(f"{backend}/health", timeout=5) return resp.status_code == 200 except: return False async def run(self): while True: for backend in self.backends: healthy = await self.check_backend(backend) self.backends[backend]["healthy"] = healthy if not healthy: logger.warning(f"Backend {backend} unhealthy") await asyncio.sleep(self.check_interval) 结语 LLM负载均衡需要考虑请求长度差异、节点异构性和KV Cache状态。最小连接数+延迟感知的组合策略在大多数场景下表现最佳。配合健康检查和自动故障转移，可以构建高可用的LLM推理服务。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么需要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论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 模型卡（Model Card）是AI透明度的核心工具。它像食品的营养标签一样，让用户清楚地了解AI模型的能力、局限和风险。2026年，随着AI法规的完善，模型卡已经成为AI系统的必备文档。本文将介绍2026年模型卡的最佳实践和模板。 模型卡的价值 对开发者 促进系统性评估 发现模型局限 建立用户信任 对用户 了解模型能力边界 评估适用性 知情决策 对监管机构 审查合规性 评估风险 追溯责任 2026年模型卡模板 # 模型卡：[模型名称] ## 1. 基本信息 | 字段 | 内容 | |------|------| | 模型名称 | GLM-5 | | 版本 | 5.0.2 | | 发布日期 | 2026-06-15 | | 开发者 | 智谱AI | | 许可证 | 商业许可 | | 联系方式 | support@zhipuai.cn | ## 2. 模型描述 ### 架构 - 类型：Transformer (MoE) - 参数量：未公开 - 上下文长度：256K tokens - 训练数据截止：2026-04 ### 预期用途 - 通用对话和问答 - 文本生成和摘要 - 代码生成和调试 - 中文理解和生成 ### 不建议用途 - 医疗诊断（未获医疗器械认证） - 法律建议（不构成法律意见） - 自动化决策（需人工审核） - 高风险场景（如自动驾驶） ## 3. 性能评估 ### 基准测试结果 | 基准 | 得分 | 评估条件 | |------|------|---------| | MMLU-Pro | 86.2% | 5-shot | | C-Eval 2026 | 92.1% | 5-shot | | GSM8K | 94.1% | 0-shot CoT | | HumanEval | 86.5% | 0-shot | | GPQA Diamond | 66.3% | 0-shot | ### 分群体性能 | 群体 | 准确率 | 样本量 | |------|--------|--------| | 中文用户 | 93.5% | 5000 | | 英文用户 | 87.2% | 5000 | | 低资源语言 | 72.3% | 1000 | ### 局限性 1. **知识截止**：训练数据截止2026年4月，之后的事件可能不知道 2. **幻觉**：在罕见话题上可能产生看似合理但错误的信息 3. **数学**：虽然GSM8K得分高，但在更高级数学上可能出错 4. **多模态**：不支持图像和视频输入 5. **实时性**：不支持实时信息检索 ## 4. 安全与伦理 ### 安全评估 | 测试 | 结果 | |------|------| | 有害内容拒绝率 | 98.7% | | 提示注入防御 | 通过 | | 隐私泄露测试 | 通过 | | 偏见测试 | 轻微性别偏见（已记录） | ### 已知风险 1. **社会偏见**：在职业描述上可能存在性别偏见 2. **信息幻觉**：在知识边界外可能编造信息 3. **提示注入**：虽然有多层防御，但不是100%安全 ### 缓解措施 1. RLHF对齐训练 2. 安全过滤器 3. 输出后处理 4. 持续红队测试 ## 5. 部署信息 ### 硬件要求 | 部署方式 | 最低配置 | |----------|---------| | API | 无需硬件 | | 自托管 | 8×A100 80GB | | 量化部署 | 4×A100 (INT4) | ### 性能指标 | 指标 | 数值 | |------|------| | 平均延迟 | 1.2s | | P95延迟 | 3.5s | | 吞吐量 | 245 t/s | | 可用性 | 99.9% | ### 成本 | 项目 | 价格 | |------|------| | 输入 | ¥2.0/百万token | | 输出 | ¥8.0/百万token | ## 6. 版本历史 | 版本 | 日期 | 变更 | |------|------|------| | 5.0.2 | 2026-06-15 | 修复安全漏洞 | | 5.0.1 | 2026-06-01 | 性能优化 | | 5.0.0 | 2026-05-20 | 初始发布 | ## 7. 引用 @misc{glm5_2026, title={GLM-5: …}, author={ZhipuAI}, year={2026} } ...