Posts

NVIDIA 的推理加速王牌 TensorRT-LLM 是 NVIDIA 在大模型推理领域的旗舰产品。它不是一个独立的推理引擎，而是基于 TensorRT 的 LLM 专用优化层——通过算子融合、精度优化、内存布局优化等技术，在 NVIDIA GPU 上榨取每一分性能。 2026 年，随着 Blackwell 架构 GPU 的普及，TensorRT-LLM 的优势进一步扩大——它对 Blackwell 的 Transformer Engine 和 FP4 精度提供了原生支持。 2026 核心特性 性能优势 特性 TensorRT-LLM 2026 vLLM 0.8 SGLang 0.3 峰值吞吐量 8,500 tok/s 4,200 tok/s 4,800 tok/s 首 Token 延迟 0.15s 0.5s 0.35s FP4 支持 ✅ (Blackwell) ❌ ❌ FP8 支持 ✅ (Hopper) ✅ ✅ 算子融合 深度 基础 基础 模型编译 AOT 编译 JIT JIT 多 GPU TP + PP + EP TP + PP TP + PP Blackwell 架构优化 # Blackwell B200 上的 FP4 推理 import tensorrt_llm as trtllm # 编译模型为 FP4 精度 builder = trtllm.Builder() config = builder.create_builder_config( precision="fp4", # FP4 量化 plugin_config=trtllm.PluginConfig( attention_plugin=True, nccl_plugin=True, gemm_plugin=True, rmsnorm_plugin=True, # Blackwell 专属 transformer_engine=True, moe_plugin=True, ), max_batch_size=256, max_input_len=32768, max_output_len=4096, max_num_tokens=8192, use_paged_context_fmha=True, # Paged Attention use_context_fmha=True, # Flash Attention multiple_profiles=True, # 多优化 Profile tensor_parallel=8, # 8 路张量并行 pipeline_parallel=1, ) # 编译（AOT，提前编译为优化引擎） engine = builder.build( model_dir="Qwen/Qwen3-72B-Instruct", config=config, output_dir="engines/qwen3-72b-fp4" ) # 编译后的引擎不可移植，绑定特定 GPU 架构 # 但性能比 JIT 方案高 30-60% 部署流程 1. 模型编译 # 步骤 1：从 HuggingFace 模型编译 TensorRT 引擎 import tensorrt_llm as trtllm from tensorrt_llm.models import QWenForCausalLM # 加载模型配置 model_config = trtllm.ModelConfig.from_pretrained( "Qwen/Qwen3-72B-Instruct" ) # 编译配置 build_config = trtllm.BuildConfig( max_input_len=32768, max_output_len=4096, max_batch_size=128, max_num_tokens=8192, opt_batch_size=32, opt_input_len=4096, # 精度配置 precision="fp8", # fp4/fp8/fp16 int8_kv_cache=True, # KV Cache 量化 # 插件配置 plugin_config=trtllm.PluginConfig( paged_kv_cache=True, attention_plugin=True, gemm_plugin=True, nccl_plugin=True, rmsnorm_plugin=True, rotary_plugin=True, remove_input_padding=True, # 移除 padding 优化 ), # 并行配置 tensor_parallel=4, pipeline_parallel=1, # 高级优化 use_fused_mlp=True, # MLP 算子融合 use_fused_qkv=True, # QKV 融合 use_dynamic_shape=True, # 动态形状 weight_sparsity=True, # 权重稀疏化 ) # 编译引擎 builder = trtllm.Builder() engine = builder.build_model( model_config=model_config, build_config=build_config, output_dir="engines/qwen3-72b-fp8-tp4" ) 2. 启动服务 # 使用 Triton Inference Server 部署 # 模型仓库结构 models/ └── qwen3-72b/ ├── config.pbtxt ├── 1/ │ └── model.py └── engines/ └── qwen3-72b-fp8-tp4/ ├── rank0.engine ├── rank1.engine ├── rank2.engine └── rank3.engine # config.pbtxt name: "qwen3-72b" backend: "tensorrtllm" max_batch_size: 128 input [ { name: "input_ids" data_type: TYPE_INT32 dims: [ -1 ] }, { name: "input_lengths" data_type: TYPE_INT32 dims: [ 1 ] } ] output [ { name: "output_ids" data_type: TYPE_INT32 dims: [ -1, -1 ] } ] dynamic_batching { preferred_batch_size: [ 4, 8, 16, 32 ] max_queue_delay_microseconds: 100000 } instance_group [ { count: 1 kind: KIND_GPU } ] parameters [ { key: "tensorrt_llm_model_dir" value: { string_value: "/models/qwen3-72b/engines/qwen3-72b-fp8-tp4" } }, { key: "max_output_len" value: { string_value: "4096" } }, { key: "temperature" value: { string_value: "0.7" } }, { key: "top_p" value: { string_value: "0.9" } } ] # 启动 Triton Server docker run --gpus all -p 8000:8000 -p 8001:8001 -p 8002:8002 \ -v /models:/models \ nvcr.io/nvidia/tritonserver:25.06-py3 \ tritonserver --model-repository=/models \ --backend-directory=/opt/tritonserver/backends \ --log-verbose=1 3. 客户端调用 import tritonclient.grpc as grpcclient import numpy as np class TensorRTLLMClient: def __init__(self, url="localhost:8001"): self.client = grpcclient.InferenceServerClient(url) def generate(self, prompt: str, max_tokens: int = 512, temperature: float = 0.7, stream: bool = True): # Tokenize input_ids = self.tokenizer.encode(prompt) inputs = [ grpcclient.InferInput("input_ids", [1, len(input_ids)], "INT32"), grpcclient.InferInput("input_lengths", [1, 1], "INT32"), ] inputs[0].set_data_from_numpy(np.array([input_ids], dtype=np.int32)) inputs[1].set_data_from_numpy(np.array([[len(input_ids)]], dtype=np.int32)) outputs = [grpcclient.InferRequestedOutput("output_ids")] # 推理 result = self.client.infer( model_name="qwen3-72b", inputs=inputs, outputs=outputs ) output_ids = result.as_numpy("output_ids") # Detokenize return self.tokenizer.decode(output_ids[0]) 性能优化 精度对比 精度 显存 (72B) 吞吐量 质量损失 推荐 GPU FP16 145 GB 3,200 tok/s 0% A100 80GB×2 FP8 75 GB 5,800 tok/s <1% H100/H200 INT4 AWQ 42 GB 4,500 tok/s ~3% 任意 FP4 38 GB 8,500 tok/s ~5% B200 算子融合效果 优化 吞吐量提升 延迟降低 基础（无融合） 基准 基准 QKV 融合 +15% -8% + MLP 融合 +25% -15% + RMSNorm 融合 +30% -20% + Rotary 融合 +35% -22% + 全部融合 +42% -28% 多 GPU 扩展 # 张量并行 + 流水线并行配置 config = trtllm.BuildConfig( tensor_parallel=4, # 4 路张量并行 pipeline_parallel=2, # 2 路流水线并行 # 总共 8 GPU # 专家并行（MoE 模型） moe_config=trtllm.MoEConfig( num_experts=256, expert_parallel_size=8, moe_plugin=True, ), ) 并行策略 GPU 数量 吞吐量 扩展效率 TP=1 1 3,200 100% TP=2 2 5,800 91% TP=4 4 9,500 74% TP=8 8 15,200 59% TP=4+PP=2 8 14,800 58% 与 vLLM 对比 维度 TensorRT-LLM vLLM 峰值性能 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ 部署难度 高 低 模型支持 跟随 NVIDIA 跟随社区 编译时间 10-60 分钟 即时 灵活性 低（AOT） 高（JIT） 跨平台 仅 NVIDIA NVIDIA + AMD 生态 NVIDIA 生态 开源生态 成本 需要 NVIDIA GPU 任意 GPU 适用场景 最适合 极致性能需求：延迟和吞吐量最优 NVIDIA 纯净环境：充分利用 GPU 特性 固定模型部署：AOT 编译换取性能 大规模生产：Triton Server 集群部署 Blackwell 用户：FP4 独家支持 不太适合 快速迭代：每次模型变更需要重新编译 多 GPU 品牌：仅支持 NVIDIA 小团队：部署和调优门槛高 实验性模型：新模型架构支持滞后于 vLLM 成本敏感：需要 NVIDIA GPU 许可 总结 TensorRT-LLM 在 2026 年仍然是"NVIDIA GPU 上最快的推理引擎"。它的 AOT 编译、深度算子融合、FP4/FP8 支持，让它在峰值性能上领先 vLLM 60-100%。这个优势在 Blackwell 架构上更加明显。 ...

TGI 2026：HuggingFace 的推理引擎 Text Generation Inference（TGI）是 HuggingFace 推出的大模型推理服务框架。与 vLLM 并列为开源推理引擎双雄。2026 年，TGI 在企业级特性方面持续强化，成为 HuggingFace 生态（Hub + Inference Endpoints + TGI）的核心组件。 2026 架构概览 ┌──────────────────────────────────────────────────┐ │ Client Layer │ │ REST API │ gRPC │ WebSocket │ Python/JS SDK │ ├──────────────────────────────────────────────────┤ │ Router Layer │ │ ┌────────────┐ ┌──────────┐ ┌────────────────┐ │ │ │ Load │ │ Queue │ │ Response │ │ │ │ Balancer │ │ Manager │ │ Aggregator │ │ │ └────────────┘ └──────────┘ └────────────────┘ │ ├──────────────────────────────────────────────────┤ │ Inference Layer │ │ ┌────────────┐ ┌──────────┐ ┌────────────────┐ │ │ │ Continuous │ │ Flash │ │ Speculative │ │ │ │ Batching │ │ Attention│ │ Decoding │ │ │ └────────────┘ └──────────┘ └────────────────┘ │ │ ┌────────────┐ ┌──────────┐ ┌────────────────┐ │ │ │ Tensor │ │ Pipeline │ │ Quantization │ │ │ │ Parallel │ │ Parallel │ │ (AWQ/GPTQ/FP8) │ │ │ └────────────┘ └──────────┘ └────────────────┘ │ ├──────────────────────────────────────────────────┤ │ Model Layer │ │ Safetensors │ GGUF │ Tokenizer │ Config │ └──────────────────────────────────────────────────┘ TGI vs vLLM 定位差异 维度 TGI vLLM 核心优势 HF 生态集成 极致吞吐量 部署方式 Docker 优先 灵活部署 模型格式 Safetensors 优先 GGUF/多种 企业特性 完善 基础 社区 HF 社区 独立社区 推理速度 快 最快 模型支持 跟随 HF 跟随社区 部署指南 Docker 部署 # 基础部署 docker run --gpus all -p 8080:80 \ -v /data/models:/models \ ghcr.io/huggingface/text-generation-inference:3.0 \ --model Qwen/Qwen2.5-32B-Instruct \ --quantization awq \ --max-total-tokens 32768 \ --max-batch-size 256 \ --max-concurrent-requests 512 高级配置 # 完整生产配置 docker run --gpus all -p 8080:80 \ -v /data/models:/models \ -v /data/cache:/data \ -e HUGGING_FACE_HUB_TOKEN=$HF_TOKEN \ ghcr.io/huggingface/text-generation-inference:3.0 \ --model Qwen/Qwen2.5-72B-Instruct \ --model-auto-config \ --revision main \ --quantization awq \ --dtype float16 \ --max-total-tokens 65536 \ --max-batch-size 128 \ --max-concurrent-requests 256 \ --max-batch-prefill-tokens 8192 \ --max-waiting-tokens 20 \ --max-waiting-batches 4 \ --waiting-served-ratio 1.2 \ --cuda-memory-fraction 0.90 \ --tensor-parallel-size 2 \ --num-shard 2 \ --sharded true \ --enable-flash-attention \ --enable-prefix-caching \ --enable-chunked-prefill \ --disable-custom-kernels false \ --json-output \ --google-service-account /data/gcp.json Python SDK 使用 from text_generation import Client, AsyncClient # 同步客户端 client = Client("http://localhost:8080") # 简单生成 response = client.generate( prompt="解释量子纠缠", max_new_tokens=512, temperature=0.7, top_p=0.9, repetition_penalty=1.1, stop=["<|im_end|>"], ) print(response.generated_text) # 流式生成 for token in client.generate_stream( prompt="写一首诗", max_new_tokens=200, ): print(token.token.text, end="", flush=True) # 批量生成 responses = client.generate_batch( prompts=["你好", "Hello", "Bonjour"], max_new_tokens=50 ) # 异步客户端 async_client = AsyncClient("http://localhost:8080") response = await async_client.generate("Hello", max_new_tokens=100) OpenAI 兼容 API from openai import OpenAI # TGI 兼容 OpenAI API client = OpenAI( base_url="http://localhost:8080/v1", api_key="tgi" ) response = client.chat.completions.create( model="Qwen/Qwen2.5-32B-Instruct", messages=[ {"role": "system", "content": "你是助手"}, {"role": "user", "content": "Hello"} ], stream=True, tools=[{ "type": "function", "function": { "name": "get_time", "parameters": {"type": "object", "properties": {}} } }] ) 核心优化 1. 连续批处理 TGI 的连续批处理（Continuous Batching）是其高吞吐量的核心： ...

引言：Transformer 的统治与挑战 自 2017 年 Google 提出 Transformer 架构以来，它已经统治了自然语言处理乃至整个深度学习领域长达九年。然而到了 2026 年，随着模型规模扩展到万亿参数、上下文窗口增长到百万 Token，原始 Transformer 架构的局限性日益凸显：注意力机制的 $O(n^2)$ 复杂度、推理时 KV Cache 的巨大内存开销、以及训练算力墙的逼近，都在倒逼架构创新。 本文将系统梳理 2026 年 Transformer 架构的三大演进方向：注意力机制优化、混合专家架构（MoE）的成熟、以及以 Mamba 为代表的状态空间模型（SSM）的崛起。 一、注意力机制的进化谱系 1.1 标准 Self-Attention 回顾 标准多头自注意力机制的核心计算为： $$\text{Attention}(Q, K, V) = \text{softmax}\left(\frac{QK^T}{\sqrt{d_k}}\right)V$$ 其中 $Q \in \mathbb{R}^{n \times d_k}$, $K \in \mathbb{R}^{n \times d_k}$, $V \in \mathbb{R}^{n \times d_v}$。其计算复杂度为 $O(n^2 \cdot d)$，空间复杂度同样为 $O(n^2)$，这在长序列场景下成为瓶颈。 1.2 2026 年的注意力新范式 ┌─────────────────────────────────────────────────┐ │ 注意力机制演进谱系 (2026) │ ├─────────────────────────────────────────────────┤ │ │ │ Standard MHA ──► Multi-Query (MQA) │ │ │ │ │ │ │ ▼ │ │ │ Grouped-Query (GQA) │ │ │ │ │ │ │ ▼ │ │ ▼ Latent Attention │ │ Linear Attention │ │ │ │ │ │ ▼ ▼ │ │ Flash Attention 3 Ring Attention │ │ (GPU优化) (分布式) │ └─────────────────────────────────────────────────┘ Latent Attention（潜注意力） 是 2025-2026 年最重要的架构创新之一，由 DeepSeek V3 首先大规模验证。其核心思想是将 Key 和 Value 压缩到低秩潜在空间： ...

vLLM 2026：推理引擎的事实标准 vLLM 在 2026 年已经成为大模型推理部署的事实标准。根据社区统计，全球超过 70% 的开源大模型生产部署使用 vLLM 作为推理引擎。它的核心优势在于 PagedAttention 技术带来的高吞吐量和低延迟，以及对各类开源模型的广泛支持。 2026 核心特性 版本演进 特性 vLLM 0.3 (2024) vLLM 0.8 (2026) PagedAttention v1 v3（内存效率+40%） 连续批处理 支持 支持 + 动态批大小 张量并行 支持 支持 + 专家并行 量化 AWQ/GPTQ AWQ/GPTQ/FP8/INT4 多模态 实验性 原生支持 Speculative Decoding 不支持 支持 长上下文 32k 1M+ 分离式推理 不支持 Prefill/Decode 分离 安装与环境准备 基础安装 # 创建虚拟环境 python -m venv vllm-env source vllm-env/bin/activate # Linux # vllm-env\Scripts\activate # Windows # 安装 vLLM（CUDA 12.1+） pip install vllm==0.8.5 # 验证安装 python -c "import vllm; print(vllm.__version__)" GPU 环境检查 # 检查 CUDA 版本 nvidia-smi # 需要 CUDA 12.1+，驱动 535+ # 检查 GPU 内存 nvidia-smi --query-gpu=name,memory.total,memory.free --format=csv 模型内存需求参考 模型 参数量 FP16 显存 INT8 显存 INT4 显存 推荐 GPU Qwen2.5-7B 7B 14 GB 8 GB 5 GB RTX 4090 Llama-4-8B 8B 16 GB 9 GB 5 GB RTX 4090 Qwen2.5-32B 32B 64 GB 34 GB 20 GB A100 80GB Llama-4-70B 70B 140 GB 75 GB 42 GB 2×A100 80GB Qwen3-72B 72B 145 GB 78 GB 44 GB 2×A100 80GB DeepSeek-V3 671B (MoE) 1.3 TB 700 GB 400 GB 8×H100 80GB 基础部署 单 GPU 部署 from vllm import LLM, SamplingParams # 加载模型 llm = LLM( model="Qwen/Qwen2.5-32B-Instruct", quantization="awq", # 使用 AWQ 量化 max_model_len=32768, # 最大上下文长度 gpu_memory_utilization=0.90, # GPU 内存利用率 tensor_parallel_size=1, # 张量并行度 dtype="float16", # 数据类型 trust_remote_code=True, enforce_eager=False, # 使用 CUDA Graph 优化 swap_space=4, # CPU 交换空间 (GB) max_num_seqs=256, # 最大并发序列数 ) # 配置采样参数 sampling_params = SamplingParams( temperature=0.7, top_p=0.9, top_k=50, max_tokens=2048, repetition_penalty=1.05, ) # 批量推理 prompts = [ "解释量子计算的基本原理", "写一首关于春天的诗", "用 Python 实现快速排序算法", ] outputs = llm.generate(prompts, sampling_params) for output in outputs: print(output.outputs[0].text) 多 GPU 张量并行 # 2×A100 80GB 部署 70B 模型 llm = LLM( model="meta-llama/Llama-4-70B-Instruct", tensor_parallel_size=2, # 2 路张量并行 pipeline_parallel_size=1, # 流水线并行 gpu_memory_utilization=0.92, max_model_len=65536, dtype="float16", enable_prefix_caching=True, # 前缀缓存 enable_chunked_prefill=True, # 分块预填充 ) OpenAI 兼容 API 服务 # 启动 API 服务器 python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-32B-Instruct \ --quantization awq \ --tensor-parallel-size 2 \ --gpu-memory-utilization 0.90 \ --max-model-len 32768 \ --port 8000 \ --host 0.0.0.0 \ --api-key sk-your-api-key \ --served-model-name qwen-32b \ --enable-prefix-caching \ --enable-chunked-prefill \ --max-num-seqs 256 \ --uvicorn-log-level info # 客户端调用（兼容 OpenAI SDK） from openai import OpenAI client = OpenAI( api_key="sk-your-api-key", base_url="http://localhost:8000/v1" ) response = client.chat.completions.create( model="qwen-32b", messages=[ {"role": "system", "content": "你是专业翻译"}, {"role": "user", "content": "Translate: Hello World"} ], temperature=0.3, max_tokens=100, stream=True # 支持流式输出 ) for chunk in response: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="") 性能优化 1. 量化策略 # FP8 量化（H100 专用，吞吐量最高） llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", quantization="fp8", dtype="bfloat16", gpu_memory_utilization=0.92, ) # INT4 AWQ 量化（最省显存） llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", quantization="awq", quantization_config={ "bits": 4, "group_size": 128, "zero_point": True, }, ) # GPTQ 量化 llm = LLM( model="TheBloke/Llama-4-70B-GPTQ", quantization="gptq", dtype="float16", ) 2. Speculative Decoding（投机解码） # 使用小模型加速大模型推理 llm = LLM( model="Qwen/Qwen2.5-72B-Instruct", speculative_model="Qwen/Qwen2.5-1.5B-Instruct", # 草稿模型 num_speculative_tokens=5, # 每次投机 5 个 token speculative_draft_tensor_parallel_size=1, gpu_memory_utilization=0.92, ) # 效果：吞吐量提升 2-3x，延迟降低 40-60% # 代价：草稿模型需要共享词表 3. 分离式推理（Prefill-Decode 分离） # 2026 新特性：将 Prefill 和 Decode 分离到不同 GPU # 适合高并发场景 # Prefill 节点（计算密集） python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-72B-Instruct \ --disaggregation-mode prefill \ --disaggregation-port 5001 \ --port 8000 # Decode 节点（内存密集） python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-72B-Instruct \ --disaggregation-mode decode \ --disaggregation-port 5001 \ --port 8001 # 路由层（自动分发请求） python -m vllm.entrypoints.disagg_router \ --prefill-endpoint http://gpu1:8000 \ --decode-endpoint http://gpu2:8001 \ --port 8080 4. 长上下文优化 # 1M 上下文部署 llm = LLM( model="Qwen/Qwen2.5-32B-Instruct-1M", max_model_len=1048576, # 1M tokens gpu_memory_utilization=0.95, # 长上下文优化 enable_chunked_prefill=True, # 分块预填充 max_num_batched_tokens=8192, # 每批最大 token 数 max_num_seqs=32, # 降低并发数以容纳长序列 # KV Cache 优化 block_size=16, # PagedAttention 块大小 swap_space=16, # CPU 交换空间 # 滑动窗口注意力（适用于超长上下文） sliding_window=131072, # 128K 滑动窗口 ) 生产部署架构 Kubernetes 部署 apiVersion: apps/v1 kind: Deployment metadata: name: vllm-qwen-32b spec: replicas: 2 selector: matchLabels: app: vllm-qwen-32b template: metadata: labels: app: vllm-qwen-32b spec: containers: - name: vllm image: vllm/vllm-openai:v0.8.5 args: - --model=Qwen/Qwen2.5-32B-Instruct - --quantization=awq - --tensor-parallel-size=2 - --gpu-memory-utilization=0.90 - --max-model-len=32768 - --port=8000 resources: limits: nvidia.com/gpu: 2 memory: 128Gi requests: nvidia.com/gpu: 2 memory: 64Gi ports: - containerPort: 8000 readinessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 120 periodSeconds: 10 livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 300 periodSeconds: 30 --- apiVersion: v1 kind: Service metadata: name: vllm-service spec: selector: app: vllm-qwen-32b ports: - port: 8000 targetPort: 8000 type: LoadBalancer 负载均衡配置 # Nginx 负载均衡 upstream vllm_backend { least_conn; # 最少连接策略 server gpu-node-1:8000 weight=1 max_fails=3 fail_timeout=30s; server gpu-node-2:8000 weight=1 max_fails=3 fail_timeout=30s; server gpu-node-3:8000 weight=1 max_fails=3 fail_timeout=30s; keepalive 32; keepalive_timeout 60s; } server { listen 80; location /v1/ { proxy_pass http://vllm_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; # 流式响应支持 proxy_buffering off; proxy_cache off; chunked_transfer_encoding on; # 超时设置 proxy_connect_timeout 10s; proxy_read_timeout 300s; proxy_send_timeout 60s; } location /health { proxy_pass http://vllm_backend/health; } } 性能基准 吞吐量对比（Qwen2.5-32B AWQ，2×A100 80GB） 配置 吞吐量 (tok/s) P50 延迟 P99 延迟 并发数 基础配置 2,800 0.8s 3.2s 64 + 前缀缓存 3,500 0.6s 2.5s 64 + 分块预填充 4,200 0.5s 2.1s 128 + Speculative 6,800 0.3s 1.2s 128 + FP8 (H100) 8,500 0.25s 0.9s 256 与其他推理引擎对比 引擎 吞吐量 延迟 显存效率 易用性 vLLM 0.8 4,200 tok/s 0.5s 92% ⭐⭐⭐⭐⭐ TGI 3.0 3,100 tok/s 0.7s 85% ⭐⭐⭐⭐ SGLang 0.3 4,800 tok/s 0.4s 90% ⭐⭐⭐⭐ TensorRT-LLM 5,200 tok/s 0.3s 95% ⭐⭐⭐ Ollama 1,800 tok/s 1.2s 70% ⭐⭐⭐⭐⭐ 监控与运维 Prometheus 指标 # vLLM 内置 Prometheus 指标 # 访问 http://localhost:8000/metrics # 关键指标： # vllm:num_requests_running - 正在运行的请求数 # vllm:num_requests_waiting - 等待队列长度 # vllm:gpu_cache_usage_perc - GPU 缓存使用率 # vllm:time_to_first_token - TTFT（首 Token 延迟） # vllm:time_per_output_token - TPOT（每 Token 延迟） # vllm:e2e_request_latency - 端到端延迟 # Prometheus 采集配置 scrape_configs: - job_name: 'vllm' static_configs: - targets: ['gpu-node-1:8000', 'gpu-node-2:8000'] metrics_path: /metrics scrape_interval: 10s 常见问题排查 问题 原因 解决方案 OOM 显存不足 降低 gpu_memory_utilization 或使用量化 首Token延迟高 Prefill 慢 启用 chunked_prefill 吞吐量低 批处理不足 增加 max_num_seqs 请求排队 并发过高 增加副本数或降低 max_model_len 模型加载慢 磁盘 I/O 使用本地 SSD 缓存模型 总结 vLLM 在 2026 年仍然是开源 LLM 推理部署的最佳选择。它的 PagedAttention v3、Speculative Decoding、分离式推理等特性让它在性能上保持领先，同时 OpenAI 兼容 API 降低了使用门槛。 ...

引言 Agent 的推理过程往往是漫长的等待——用户盯着加载动画，不知道 Agent 在做什么。流式响应把"等待结果"变成"实时观察思考"，是 Agent 用户体验的关键升级。2026年，三种流式协议各有优劣，选型不当会导致体验降级或工程复杂度爆炸。 一、三种协议对比 核心特性矩阵 特性 SSE WebSocket gRPC Stream 通信方向 服务器→客户端（单向） 双向 双向 底层协议 HTTP/1.1 或 HTTP/2 HTTP 升级 HTTP/2 数据格式 文本（text/event-stream） 文本/二进制 Protobuf（二进制） 自动重连 内置 需手动实现 需手动实现 浏览器支持 原生 EventSource 原生 WebSocket 需 gRPC-Web 代理/CDN兼容 优秀 良好 较差 连接数限制 浏览器6个/域名 无限制 无限制 序列化效率 低（文本） 中 高（Protobuf） 移动端友好 高 中 低 Agent 场景适配分析 Agent 流式需求频谱： 单向输出流 双向交互流 （LLM→用户） （用户↔Agent） │ │ │ ┌───────────┐ │ │ │ SSE 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ WebSocket │ │ │ │ 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ gRPC 最佳 │ │ │ └───────────┘ │ 简单聊天 ←─────────────→ 复杂多Agent 低延迟 ←─────────────→ 高吞吐 Web前端 ←─────────────→ 微服务后端 二、SSE 实现方案 2.1 服务端 from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() class AgentStreamEvent: """Agent 流式事件类型""" THINKING = "thinking" # Agent 思考中 TOOL_CALL = "tool_call" # 工具调用 TOOL_RESULT = "tool_result" # 工具结果 CONTENT = "content" # 内容输出 ERROR = "error" # 错误 DONE = "done" # 完成 async def agent_stream_generator( query: str, session_id: str ): """Agent SSE 流式生成器""" try: # 1. 发送思考状态 yield _format_sse(AgentStreamEvent.THINKING, { "message": "正在分析您的请求...", "session_id": session_id }) # 2. Agent 推理（流式 LLM 输出） async for chunk in agent.think_stream(query): if chunk.type == "tool_call": yield _format_sse(AgentStreamEvent.TOOL_CALL, { "tool": chunk.tool_name, "args": chunk.tool_args, "thinking": chunk.reasoning }) # 3. 工具执行 result = await agent.execute_tool(chunk.tool_call) yield _format_sse(AgentStreamEvent.TOOL_RESULT, { "tool": chunk.tool_name, "result": result.summary, "duration_ms": result.duration_ms }) elif chunk.type == "content": yield _format_sse(AgentStreamEvent.CONTENT, { "text": chunk.text, "tokens_so_far": chunk.token_count }) # 4. 完成 yield _format_sse(AgentStreamEvent.DONE, { "session_id": session_id, "total_tokens": agent.total_tokens, "duration_ms": agent.total_duration_ms }) except Exception as e: yield _format_sse(AgentStreamEvent.ERROR, { "message": str(e), "session_id": session_id }) def _format_sse(event_type: str, data: dict) -> str: return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @app.post("/api/agent/chat") async def chat(request: ChatRequest): return StreamingResponse( agent_stream_generator(request.query, request.session_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx 禁用缓冲 } ) 2.2 客户端 class AgentSSEClient { private eventSource: EventSource | null = null; private reconnectAttempts = 0; private maxReconnects = 3; connect(query: string, sessionId: string) { // 使用 fetch POST + ReadableStream（EventSource 仅支持 GET） fetch('/api/agent/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ query, sessionId }), }).then(response => { const reader = response.body!.getReader(); const decoder = new TextDecoder(); let buffer = ''; const read = () => { reader.read().then(({ done, value }) => { if (done) return; buffer += decoder.decode(value, { stream: true }); const events = buffer.split('\n\n'); buffer = events.pop() || ''; events.forEach(raw => this.handleEvent(raw)); read(); }); }; read(); }); } private handleEvent(raw: string) { const lines = raw.split('\n'); let event = 'message'; let data = ''; lines.forEach(line => { if (line.startsWith('event: ')) event = line.slice(7); if (line.startsWith('data: ')) data = line.slice(6); }); const parsed = JSON.parse(data); switch (event) { case 'thinking': this.onThinking?.(parsed); break; case 'tool_call': this.onToolCall?.(parsed); break; case 'tool_result': this.onToolResult?.(parsed); break; case 'content': this.onContent?.(parsed.text); break; case 'error': this.onError?.(parsed); break; case 'done': this.onDone?.(parsed); break; } } } 三、WebSocket 实现方案 3.1 服务端 from fastapi import WebSocket, WebSocketDisconnect class AgentConnectionManager: """Agent WebSocket 连接管理""" def __init__(self): self.active: dict[str, WebSocket] = {} # session_id → WebSocket self.agent_tasks: dict[str, asyncio.Task] = {} async def connect(self, ws: WebSocket, session_id: str): await ws.accept() self.active[session_id] = ws logger.info(f"WebSocket connected: {session_id}") async def disconnect(self, session_id: str): if session_id in self.active: del self.active[session_id] if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() del self.agent_tasks[session_id] async def handle_session(self, ws: WebSocket, session_id: str): """处理 WebSocket 会话""" await self.connect(ws, session_id) try: while True: # 接收客户端消息 message = await ws.receive_json() if message["type"] == "chat": # 启动 Agent 任务 task = asyncio.create_task( self._run_agent(ws, session_id, message["content"]) ) self.agent_tasks[session_id] = task elif message["type"] == "interrupt": # 用户中断当前 Agent 执行 if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() await ws.send_json({ "type": "interrupted", "session_id": session_id }) elif message["type"] == "feedback": # 用户实时反馈（Human-in-the-loop） await self._handle_feedback(session_id, message) except WebSocketDisconnect: await self.disconnect(session_id) async def _run_agent(self, ws: WebSocket, session_id: str, query: str): """运行 Agent 并通过 WebSocket 推送更新""" try: async for event in agent.run_stream(query): await ws.send_json({ "type": event.type, "data": event.data, "timestamp": time.time() }) except asyncio.CancelledError: logger.info(f"Agent task cancelled: {session_id}") except Exception as e: await ws.send_json({ "type": "error", "data": {"message": str(e)} }) manager = AgentConnectionManager() @app.websocket("/ws/agent/{session_id}") async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.handle_session(ws, session_id) 3.2 客户端 class AgentWSClient { private ws: WebSocket | null = null; private messageQueue: string[] = []; connect(sessionId: string) { this.ws = new WebSocket(`wss://api.example.com/ws/agent/${sessionId}`); this.ws.onopen = () => { // 发送排队消息 this.messageQueue.forEach(msg => this.ws!.send(msg)); this.messageQueue = []; }; this.ws.onmessage = (event) => { const data = JSON.parse(event.data); this.handleMessage(data); }; this.ws.onclose = () => { // 自动重连 setTimeout(() => this.connect(sessionId), 3000); }; } sendChat(content: string) { const msg = JSON.stringify({ type: 'chat', content }); if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(msg); } else { this.messageQueue.push(msg); } } interrupt() { this.ws?.send(JSON.stringify({ type: 'interrupt' })); } } 四、gRPC 流式方案 4.1 Proto 定义 // agent.proto syntax = "proto3"; service AgentService { // 服务端流式：Agent → 客户端 rpc ChatStream(ChatRequest) returns (stream ChatResponse); // 双向流式：支持实时交互 rpc ChatBidirectional(stream ChatMessage) returns (stream ChatResponse); } message ChatRequest { string session_id = 1; string query = 2; map<string, string> metadata = 3; } message ChatMessage { string session_id = 1; MessageType type = 2; // CHAT, INTERRUPT, FEEDBACK string content = 3; } message ChatResponse { ResponseType type = 1; // THINKING, TOOL_CALL, CONTENT, DONE, ERROR string session_id = 2; bytes data = 3; // JSON 编码的事件数据 int64 timestamp = 4; } enum MessageType { CHAT = 0; INTERRUPT = 1; FEEDBACK = 2; } enum ResponseType { THINKING = 0; TOOL_CALL = 1; TOOL_RESULT = 2; CONTENT = 3; DONE = 4; ERROR = 5; } 4.2 服务端实现 import grpc from concurrent import futures class AgentServicer(agent_pb2_grpc.AgentServiceServicer): def ChatStream(self, request, context): """服务端流式：逐条推送 Agent 事件""" try: for event in agent.run(request.query): response = agent_pb2.ChatResponse( type=self._map_event_type(event.type), session_id=request.session_id, data=json.dumps(event.data).encode(), timestamp=int(time.time()) ) yield response except Exception as e: yield agent_pb2.ChatResponse( type=agent_pb2.ERROR, data=json.dumps({"error": str(e)}).encode() ) def ChatBidirectional(self, request_iterator, context): """双向流式：支持中断和实时反馈""" session = None for message in request_iterator: if message.type == agent_pb2.CHAT: # 启动 Agent 执行 for event in agent.run(message.content): if not context.is_active(): break yield agent_pb2.ChatResponse( type=self._map_event_type(event.type), data=json.dumps(event.data).encode() ) elif message.type == agent_pb2.INTERRUPT: agent.interrupt() yield agent_pb2.ChatResponse( type=agent_pb2.DONE, data=json.dumps({"reason": "interrupted"}).encode() ) 五、性能基准测试 测试环境 服务端：4 vCPU / 16GB RAM / Python 3.12 / FastAPI 客户端：1000 并发连接 消息大小：平均 200 bytes / 消息 持续时间：5 分钟 结果对比 指标 SSE WebSocket gRPC Stream 最大并发连接 10,000 50,000 30,000 消息延迟 P50 12ms 8ms 5ms 消息延迟 P95 45ms 25ms 12ms 消息延迟 P99 120ms 60ms 30ms 吞吐量 (msg/s) 50,000 200,000 150,000 内存/连接 32KB 48KB 24KB CPU 利用率 (1k连接) 35% 28% 22% 带宽效率 基准 +15% +40% 六、选型决策树 ┌─────────────────┐ │ 是否需要双向通信？│ └────────┬────────┘ │ ┌──────────────┴──────────────┐ │ 否 │ 是 ▼ ▼ ┌────────────────┐ ┌──────────────────┐ │ 是否是微服务 │ │ 是否是浏览器前端？│ │ 内部通信？ │ └────────┬─────────┘ └───────┬────────┘ │ │ ┌───────┴───────┐ ┌───────┴───────┐ │ 否 │ 是 │ 是 │ 否 ▼ ▼ ▼ ▼ ┌──────────┐ ┌────────────┐ ┌──────┐ ┌────────┐ │ gRPC │ │ WebSocket │ │gRPC │ │ SSE │ │ Stream │ │ │ └──────┘ └────────┘ └──────────┘ └────────────┘ 场景推荐 场景 推荐协议 原因 Web 聊天界面 SSE 原生支持、简单、自动重连 实时协作编辑 WebSocket 双向低延迟 移动 App SSE 移动网络友好、自动重连 微服务间 Agent 通信 gRPC 高效序列化、强类型 多 Agent 系统 gRPC 流式 RPC 适配 Agent 通信 简单 LLM 问答 SSE 单向流足够、实现简单 Human-in-the-loop WebSocket 需要双向交互 大规模推送 SSE CDN 兼容、连接效率高 七、生产环境关键配置 Nginx SSE 配置 location /api/agent/chat { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; # 关键：禁用缓冲 proxy_cache off; proxy_read_timeout 300s; # 长连接超时 chunked_transfer_encoding on; } WebSocket 配置 location /ws/agent/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 86400; # 24小时 proxy_send_timeout 86400; } 八、流式架构 Checklist □ 协议选型基于通信方向和客户端类型 □ SSE 禁用代理缓冲（proxy_buffering off） □ WebSocket 实现心跳和重连机制 □ gRPC 配置 keepalive 和流控 □ 消息序列化使用高效格式（JSON/Protobuf） □ 背压机制防止慢客户端拖垮服务端 □ 连接超时和最大连接数限制 □ 流式错误不中断连接，通过事件传递 □ 客户端实现优雅降级（流式不可用时回退轮询） □ 监控流式连接的延迟和消息丢失率 结语 流式响应是 Agent 从"工具"到"伙伴"的关键体验升级。协议选型没有银弹：SSE 简单可靠适合 Web 场景，WebSocket 灵活双向适合交互场景，gRPC 高效强类型适合微服务场景。理解你的通信模式——是单向输出还是双向交互——然后选择最适合的工具。在 Agent 时代，好的流式架构让用户感觉 Agent 在"思考"，而不是在"卡住"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

安全护栏：LLM 应用的安全带和气囊 2026 年，大模型已经深度嵌入各行各业的关键业务流程。但 LLM 天生的不确定性——幻觉、越狱、有害输出——使其在生产环境中的风险不容忽视。安全护栏（Guardrails）就像汽车的安全带和气囊：正常行驶时无感，事故发生时救命。 一、安全护栏架构 1.1 三层护栏体系 用户请求 → [输入护栏] → LLM 推理 → [输出护栏] → 用户响应 │ │ ├─ 内容过滤 ├─ 有害内容检测 ├─ 意图检查 ├─ 幻觉检测 ├─ 注入检测 ├─ PII 脱敏 ├─ 速率限制 ├─ 格式验证 └─ 权限验证 └─ 合规审查 1.2 护栏设计原则 原则 说明 重要性 纵深防御 多层护栏，层层兜底 核心原则 最小延迟 护栏延迟不超过总延迟的20% 用户体验 可解释 拦截原因必须可追溯 合规要求 可配置 不同场景不同规则集 灵活性 可观测 所有拦截记录可审计 安全运营 低误报 正常请求拦截率<2% 可用性 二、输入护栏实现 2.1 内容过滤护栏 from dataclasses import dataclass from typing import List, Optional import re @dataclass class GuardrailResult: passed: bool category: str # 护栏类别 severity: str # low | medium | high | critical reason: str # 拦截原因 original_input: str sanitized_input: Optional[str] = None # 净化后的输入 class ContentFilterGuardrail: """内容过滤护栏""" def __init__(self): self.categories = { 'violence': { 'patterns': [ r'(?i)(如何|怎么|怎样).{0,10}(制造|制作|获取).{0,10}(武器|炸弹|枪)', r'(?i)(how\s+to|make|create).{0,15}(bomb|weapon|explosive)', r'(?i)伤害.{0,5}(他人|别人|人类)', ], 'severity': 'critical', 'action': 'block' }, 'self_harm': { 'patterns': [ r'(?i)(自杀|自残|自伤).{0,5}(方法|方式|怎么)', r'(?i)(suicide|self.harm|kill.myself)', r'(?i)(不想活|了结|结束生命)', ], 'severity': 'critical', 'action': 'block_and_resource' # 拦截并提供帮助资源 }, 'illegal': { 'patterns': [ r'(?i)(毒品|大麻|海洛因).{0,5}(购买|出售|制作|提炼)', r'(?i)(假证|假身份证).{0,5}(办理|制作|购买)', r'(?i)(洗钱|逃税).{0,5}(方法|操作|教程)', ], 'severity': 'high', 'action': 'block' }, 'hate_speech': { 'patterns': [ r'(?i)(仇恨|歧视|侮辱).{0,5}(种族|民族|宗教|性别)', r'(?i)(racial|ethnic|religious)\s+(slur|insult|attack)', ], 'severity': 'high', 'action': 'block' }, 'sexual': { 'patterns': [ r'(?i)(色情|淫秽|成人).{0,5}(内容|图片|视频)', r'(?i)(未成年|儿童|teenager).{0,5}(sexual|色情)', ], 'severity': 'critical', 'action': 'block' }, } def check(self, user_input: str) -> GuardrailResult: for category, config in self.categories.items(): for pattern in config['patterns']: if re.search(pattern, user_input): return GuardrailResult( passed=False, category=category, severity=config['severity'], reason=f"匹配到{category}类内容规则", original_input=user_input ) return GuardrailResult( passed=True, category='none', severity='low', reason='', original_input=user_input ) 2.2 意图分类护栏 class IntentClassificationGuardrail: """意图分类护栏——确保请求在允许范围内""" ALLOWED_INTENTS = [ 'product_inquiry', # 产品咨询 'technical_support', # 技术支持 'general_qa', # 通用问答 'content_creation', # 内容创作 'code_assistance', # 代码辅助 'translation', # 翻译 'summarization', # 总结 ] DISALLOWED_INTENTS = [ 'medical_diagnosis', # 医疗诊断 'legal_advice', # 法律建议 'financial_advice', # 金融建议 'harmful_request', # 有害请求 'jailbreak_attempt', # 越狱尝试 ] def __init__(self, llm_client): self.llm = llm_client def check(self, user_input: str, allowed_intents: list = None) -> GuardrailResult: allowed = allowed_intents or self.ALLOWED_INTENTS intent = self._classify_intent(user_input) if intent in self.DISALLOWED_INTENTS: return GuardrailResult( passed=False, category='disallowed_intent', severity='high', reason=f'检测到不允许的意图：{intent}', original_input=user_input ) if intent not in allowed: return GuardrailResult( passed=False, category='out_of_scope', severity='medium', reason=f'请求超出服务范围（意图：{intent}）', original_input=user_input ) return GuardrailResult( passed=True, category='intent', severity='low', reason='', original_input=user_input ) def _classify_intent(self, text: str) -> str: prompt = f"""对以下输入进行意图分类，从以下选项中选择最匹配的： {', '.join(self.ALLOWED_INTENTS + self.DISALLOWED_INTENTS)} 输入：{text} 意图：""" return self.llm.generate(prompt).strip().lower() 2.3 PII 检测与脱敏护栏 class PIIGuardrail: """PII（个人身份信息）检测与脱敏护栏""" PII_PATTERNS = { 'phone': { 'pattern': r'1[3-9]\d{9}', 'mask': 'PHONE***', }, 'id_card': { 'pattern': r'\d{17}[\dXx]', 'mask': 'IDCARD***', }, 'email': { 'pattern': r'[\w.-]+@[\w.-]+\.\w+', 'mask': 'EMAIL***', }, 'bank_card': { 'pattern': r'\d{16,19}', 'mask': 'BANKCARD***', }, 'address': { 'pattern': r'[\u4e00-\u9fa5]{2,}(省|市|区|县|镇|村|路|街|号)', 'mask': 'ADDRESS***', }, } def check(self, user_input: str, mode: str = 'mask') -> GuardrailResult: """检测并处理PII""" import re detected = {} sanitized = user_input for pii_type, config in self.PII_PATTERNS.items(): matches = re.findall(config['pattern'], user_input) if matches: detected[pii_type] = matches if mode == 'mask': sanitized = re.sub( config['pattern'], config['mask'], sanitized ) elif mode == 'block': return GuardrailResult( passed=False, category='pii_detected', severity='high', reason=f'检测到{pii_type}信息', original_input=user_input ) return GuardrailResult( passed=True, category='pii', severity='low' if not detected else 'medium', reason=f'检测到PII: {list(detected.keys())}' if detected else '', original_input=user_input, sanitized_input=sanitized ) 2.4 速率限制护栏 import time from collections import defaultdict class RateLimitGuardrail: """速率限制护栏""" def __init__(self): self.limits = { 'per_user': {'window': 60, 'max': 30}, # 每用户每分钟30次 'per_ip': {'window': 60, 'max': 100}, # 每IP每分钟100次 'global': {'window': 1, 'max': 50}, # 全局每秒50次 } self.counters = defaultdict(list) def check(self, user_id: str, ip: str) -> GuardrailResult: now = time.time() # 检查各维度限制 for dimension, limit in self.limits.items(): key = f"{dimension}:{user_id if dimension == 'per_user' else ip if dimension == 'per_ip' else 'global'}" # 清理过期记录 self.counters[key] = [ t for t in self.counters[key] if now - t < limit['window'] ] # 检查限制 if len(self.counters[key]) >= limit['max']: return GuardrailResult( passed=False, category='rate_limit', severity='medium', reason=f'{dimension}速率超限：{limit["max"]}/{limit["window"]}s', original_input='' ) self.counters[key].append(now) return GuardrailResult( passed=True, category='rate_limit', severity='low', reason='', original_input='' ) 三、输出护栏实现 3.1 有害内容检测护栏 class OutputSafetyGuardrail: """输出安全护栏""" def __init__(self, llm_client): self.llm = llm_client def check(self, output: str, context: dict = None) -> GuardrailResult: """检查输出是否安全""" checks = { 'harmful_content': self._check_harmful(output), 'pii_leak': self._check_pii_leak(output, context), 'instruction_leak': self._check_instruction_leak(output, context), 'hallucination': self._check_hallucination(output, context), } for check_name, result in checks.items(): if not result['safe']: return GuardrailResult( passed=False, category=check_name, severity=result['severity'], reason=result['reason'], original_input=output ) return GuardrailResult( passed=True, category='output_safety', severity='low', reason='', original_input=output ) def _check_harmful(self, output: str) -> dict: """检查有害内容""" prompt = f"""评估以下AI输出是否包含有害内容： {output[:500]} 检查维度： 1. 暴力或伤害建议 2. 违法活动指导 3. 仇恨言论 4. 性暗示内容 5. 自残引导 如果安全返回"SAFE"，否则返回"UNSAFE: [具体原因]"。""" result = self.llm.generate(prompt) if 'SAFE' in result: return {'safe': True} return { 'safe': False, 'severity': 'critical', 'reason': result } def _check_pii_leak(self, output: str, context: dict) -> dict: """检查是否泄露PII""" pii_guard = PIIGuardrail() result = pii_guard.check(output, mode='detect') if not result.passed: return { 'safe': False, 'severity': 'critical', 'reason': f'输出中包含PII: {result.reason}' } return {'safe': True} def _check_instruction_leak(self, output: str, context: dict) -> dict: """检查是否泄露系统指令""" system_prompt = context.get('system_prompt', '') if not system_prompt: return {'safe': True} # 计算输出与系统提示词的相似度 overlap = self._text_overlap(output, system_prompt) if overlap > 0.3: return { 'safe': False, 'severity': 'high', 'reason': f'输出可能与系统提示词重叠 ({overlap:.0%})' } return {'safe': True} def _check_hallucination(self, output: str, context: dict) -> dict: """检查幻觉""" sources = context.get('sources', []) if not sources: return {'safe': True} # 检查输出中的关键事实是否可被来源支持 prompt = f"""判断以下输出中的事实性陈述是否有来源支持。 输出：{output[:500]} 来源：{' '.join(sources)[:1000]} 对于每个事实性陈述，判断是否可被来源支持。 如果有不可支持的陈述，返回"HALLUCINATION: [具体内容]" 如果都可支持，返回"SUPPORTED".""" result = self.llm.generate(prompt) if 'SUPPORTED' in result: return {'safe': True} return { 'safe': False, 'severity': 'medium', 'reason': result } def _text_overlap(self, text1: str, text2: str) -> float: words1 = set(text1.lower().split()) words2 = set(text2.lower().split()) if not words2: return 0 overlap = words1 & words2 return len(overlap) / min(len(words1), len(words2)) 3.2 格式验证护栏 class OutputFormatGuardrail: """输出格式验证护栏""" def __init__(self, expected_format: dict): self.expected = expected_format def check(self, output: str) -> GuardrailResult: if self.expected['type'] == 'json': return self._check_json(output) elif self.expected['type'] == 'markdown': return self._check_markdown(output) elif self.expected['type'] == 'code': return self._check_code(output) return GuardrailResult(passed=True, ...) def _check_json(self, output: str) -> GuardrailResult: import json try: data = json.loads(output) except json.JSONDecodeError as e: # 尝试修复 try: # 提取JSON部分 start = output.index('{') end = output.rindex('}') + 1 data = json.loads(output[start:end]) except: return GuardrailResult( passed=False, category='format', severity='high', reason=f'JSON解析失败: {e}', original_input=output ) # Schema验证 if 'schema' in self.expected: from jsonschema import validate, ValidationError try: validate(instance=data, schema=self.expected['schema']) except ValidationError as e: return GuardrailResult( passed=False, category='format', severity='medium', reason=f'Schema验证失败: {e.message}', original_input=output ) return GuardrailResult(passed=True, ...) 四、护栏编排引擎 class GuardrailOrchestrator: """护栏编排引擎——组合多个护栏""" def __init__(self): self.input_guardrails = [] self.output_guardrails = [] def add_input_guardrail(self, guardrail, priority: int = 0): self.input_guardrails.append((priority, guardrail)) self.input_guardrails.sort(key=lambda x: x[0]) def add_output_guardrail(self, guardrail, priority: int = 0): self.output_guardrails.append((priority, guardrail)) self.output_guardrails.sort(key=lambda x: x[0]) def process_input(self, user_input: str, context: dict = None) -> dict: """处理输入——依次通过所有输入护栏""" context = context or {} current_input = user_input for priority, guardrail in self.input_guardrails: result = guardrail.check(current_input, **context) if not result.passed: # 记录拦截 self._log_block('input', guardrail.__class__.__name__, result) # 返回安全响应 return { 'allowed': False, 'reason': result.reason, 'category': result.category, 'severity': result.severity, 'safe_response': self._safe_response(result) } # 使用净化后的输入 if result.sanitized_input: current_input = result.sanitized_input return {'allowed': True, 'input': current_input} def process_output(self, output: str, context: dict = None) -> dict: """处理输出——依次通过所有输出护栏""" context = context or {} current_output = output for priority, guardrail in self.output_guardrails: result = guardrail.check(current_output, context) if not result.passed: self._log_block('output', guardrail.__class__.__name__, result) return { 'allowed': False, 'reason': result.reason, 'category': result.category, 'severity': result.severity, 'safe_response': self._safe_response(result) } return {'allowed': True, 'output': current_output} def _safe_response(self, result: GuardrailResult) -> str: """生成安全替代响应""" if result.category == 'self_harm': return "我注意到你可能在经历困难时期。请拨打心理援助热线：400-161-9995。" elif result.category == 'pii_detected': return "您的输入包含敏感个人信息，请去除后再提交。" elif result.category == 'rate_limit': return "请求过于频繁，请稍后再试。" else: return "抱歉，我无法处理这个请求。" 五、护栏配置示例 def create_production_guardrails(): """创建生产环境护栏配置""" orchestrator = GuardrailOrchestrator() # 输入护栏（按优先级排序） orchestrator.add_input_guardrail(RateLimitGuardrail(), priority=0) orchestrator.add_input_guardrail(ContentFilterGuardrail(), priority=1) orchestrator.add_input_guardrail(PIIGuardrail(mode='mask'), priority=2) orchestrator.add_input_guardrail( IntentClassificationGuardrail(llm), priority=3 ) # 输出护栏 orchestrator.add_output_guardrail( OutputSafetyGuardrail(llm), priority=0 ) orchestrator.add_output_guardrail( OutputFormatGuardrail({'type': 'json'}), priority=1 ) return orchestrator 六、护栏效果监控 class GuardrailMonitor: """护栏效果监控""" def __init__(self): self.stats = defaultdict(lambda: { 'total': 0, 'blocked': 0, 'false_positives': 0 }) def record(self, guardrail_name: str, result: GuardrailResult): self.stats[guardrail_name]['total'] += 1 if not result.passed: self.stats[guardrail_name]['blocked'] += 1 def report(self) -> dict: report = {} for name, stats in self.stats.items(): report[name] = { 'total': stats['total'], 'blocked': stats['blocked'], 'block_rate': stats['blocked'] / stats['total'], 'health': 'healthy' if stats['blocked'] / stats['total'] < 0.05 else 'warning' if stats['blocked'] / stats['total'] < 0.15 else 'critical' } return report 结语 安全护栏不是可选项，而是 LLM 生产应用的必需品。好的护栏体系应该是透明的（用户几乎感知不到）、智能的（低误报、高召回）、可演进的（随威胁变化而更新）。 ...

引言 Agent 的非确定性使得"感觉更好"不能作为决策依据。一个 Prompt 的微调可能提升某类任务的表现，却悄悄损害了另一类。2026年，A/B 测试已成为 Agent 优化的科学方法——用数据说话，用统计检验做决策。 一、Agent A/B 测试的特殊性 与传统 Web A/B 测试不同，Agent A/B 测试面临独特挑战： 维度 传统 A/B 测试 Agent A/B 测试 指标 点击率、转化率 输出质量、任务完成率、用户满意度 测量 确定性（点击=1/不点击=0） 非确定性（同一输入可能不同输出） 变量 UI 元素 Prompt、模型、工具、温度 噪声 低 高（LLM 输出方差大） 样本量 百万级 千级（成本限制） 指标延迟 即时 分钟级（需要完整执行） 二、实验设计框架 2.1 假设构建 @dataclass class ExperimentHypothesis: """实验假设""" name: str description: str independent_variable: str # 自变量（如 temperature） control_value: any # 对照组值（如 0.3） treatment_value: any # 实验组值（如 0.5） dependent_variables: list[str] # 因变量（如 task_completion_rate） expected_effect: str # 预期效果 min_detectable_effect: float # 最小可检测效应 (MDE) statistical_power: float # 统计功效 (通常 0.8) significance_level: float # 显著性水平 (通常 0.05) def required_sample_size(self) -> int: """计算所需样本量""" # 基于双比例检验的样本量计算 p1 = self.baseline_rate # 基线成功率 p2 = p1 + self.min_detectable_effect # 预期成功率 z_alpha = 1.96 # α=0.05 z_beta = 0.84 # power=0.8 n = ( (z_alpha * (2*p1*(1-p1))**0.5 + z_beta * (p1*(1-p1) + p2*(1-p2))**0.5) ** 2 ) / (p2 - p1) ** 2 return int(n) + 1 # 示例 hypothesis = ExperimentHypothesis( name="temperature_optimization", description="将 temperature 从 0.3 调至 0.5，预期能提升创意写作任务的用户满意度", independent_variable="temperature", control_value=0.3, treatment_value=0.5, dependent_variables=["user_satisfaction", "task_completion_rate"], expected_effect="满意度提升 5%", min_detectable_effect=0.05, statistical_power=0.8, significance_level=0.05, baseline_rate=0.75 # 当前满意度 75% ) # 所需样本量 ≈ 2,435 per group 2.2 实验配置 @dataclass class ExperimentConfig: experiment_id: str name: str hypothesis: ExperimentHypothesis traffic_allocation: float # 实验占总流量比例 (0-1) control_split: float # 对照组在实验流量中的比例 (通常 0.5) targeting_rules: list[Rule] # 目标用户筛选 duration_days: int # 预计运行天数 metrics: list[Metric] # 追踪指标 guardrail_metrics: list[Metric] # 护栏指标（不可恶化） early_stop_rules: list[Rule] # 提前停止规则 cost_budget: float # 实验成本预算 # 护栏指标示例 GUARDRAIL_METRICS = [ Metric(name="error_rate", type="counter", max_threshold=0.05), Metric(name="p95_latency", type="histogram", max_threshold_ms=10000), Metric(name="cost_per_request", type="histogram", max_threshold=0.15), Metric(name="toxic_output_rate", type="counter", max_threshold=0.01), ] 三、流量分配系统 class ExperimentRouter: """实验流量路由""" def __init__(self, redis_client): self.redis = redis_client async def assign( self, user_id: str, agent_name: str ) -> VariantAssignment: """为用户分配实验变体""" # 1. 获取活跃实验 experiments = await self._get_active_experiments(agent_name) for exp in experiments: # 2. 检查目标规则 if not self._matches_targeting(user_id, exp.targeting_rules): continue # 3. 检查是否已分配 existing = await self._get_assignment(user_id, exp.experiment_id) if existing: return existing # 保持一致性 # 4. 一致性哈希分配 bucket = self._hash_bucket(user_id, exp.experiment_id) # 5. 决定是否进入实验 if bucket < exp.traffic_allocation: # 在实验内部分配对照组/实验组 inner_bucket = self._hash_bucket( f"{user_id}:{exp.experiment_id}", "inner" ) if inner_bucket < exp.control_split: variant = "control" else: variant = "treatment" else: variant = "excluded" # 不参与实验 assignment = VariantAssignment( experiment_id=exp.experiment_id, user_id=user_id, variant=variant, config=exp.get_variant_config(variant), assigned_at=datetime.now() ) await self._save_assignment(assignment) return assignment # 没有匹配的实验 return VariantAssignment(variant="default", config={}) def _hash_bucket(self, key: str, salt: str = "") -> float: """一致性哈希，返回 0-1 之间的值""" h = hashlib.sha256(f"{key}:{salt}".encode()).hexdigest() return int(h[:8], 16) / 0xFFFFFFFF 四、指标收集与统计检验 4.1 指标收集器 class ExperimentMetricsCollector: """实验指标收集器""" async def record( self, experiment_id: str, user_id: str, variant: str, metrics: dict ): """记录单次实验观测""" event = { "experiment_id": experiment_id, "user_id": user_id, "variant": variant, "timestamp": time.time(), **metrics # task_completed, satisfaction_score, latency_ms, tokens_used, cost } # 写入时序数据库 await self.influxdb.write( measurement="experiment_events", tags={"experiment_id": experiment_id, "variant": variant}, fields=metrics, timestamp=event["timestamp"] ) async def aggregate( self, experiment_id: str, metric_name: str ) -> dict: """聚合实验指标""" return { "control": await self._compute_stats(experiment_id, "control", metric_name), "treatment": await self._compute_stats(experiment_id, "treatment", metric_name) } async def _compute_stats( self, exp_id: str, variant: str, metric: str ) -> MetricStats: values = await self.influxdb.query( f'SELECT "{metric}" FROM "experiment_events" ' f'WHERE "experiment_id" = \'{exp_id}\' ' f'AND "variant" = \'{variant}\'' ) return MetricStats( n=len(values), mean=statistics.mean(values), std=statistics.stdev(values) if len(values) > 1 else 0, median=statistics.median(values), p25=np.percentile(values, 25), p75=np.percentile(values, 75), p95=np.percentile(values, 95), ) 4.2 统计检验 from scipy import stats import numpy as np class StatisticalTester: """统计显著性检验""" def test_proportion( self, control_successes: int, control_total: int, treatment_successes: int, treatment_total: int, alpha: float = 0.05 ) -> TestResult: """比例检验（用于完成率等二值指标）""" # 卡方检验 contingency = [ [control_successes, control_total - control_successes], [treatment_successes, treatment_total - treatment_successes] ] chi2, p_value, _, _ = stats.chi2_contingency(contingency) # 效应量 p_control = control_successes / control_total p_treatment = treatment_successes / treatment_total effect_size = p_treatment - p_control # 置信区间 se = np.sqrt(p_control*(1-p_control)/control_total + p_treatment*(1-p_treatment)/treatment_total) ci_lower = effect_size - 1.96 * se ci_upper = effect_size + 1.96 * se return TestResult( test="chi_square", p_value=p_value, significant=p_value < alpha, effect_size=effect_size, confidence_interval=(ci_lower, ci_upper), control_rate=p_control, treatment_rate=p_treatment, interpretation=self._interpret( p_value, alpha, effect_size, p_control, p_treatment ) ) def test_continuous( self, control_values: list[float], treatment_values: list[float], alpha: float = 0.05 ) -> TestResult: """连续值检验（用于满意度分数、延迟等）""" # 正态性检验 _, p_normal_ctrl = stats.shapiro(control_values) _, p_normal_treat = stats.shapiro(treatment_values) if p_normal_ctrl > 0.05 and p_normal_treat > 0.05: # 正态分布：使用 t 检验 statistic, p_value = stats.ttest_ind( control_values, treatment_values, equal_var=False # Welch's t-test ) test_name = "welch_t_test" else: # 非正态：使用 Mann-Whitney U 检验 statistic, p_value = stats.mannwhitneyu( control_values, treatment_values, alternative='two-sided' ) test_name = "mann_whitney_u" # 效应量 (Cohen's d) pooled_std = np.sqrt( ((len(control_values)-1) * np.var(control_values, ddof=1) + (len(treatment_values)-1) * np.var(treatment_values, ddof=1)) / (len(control_values) + len(treatment_values) - 2) ) cohens_d = (np.mean(treatment_values) - np.mean(control_values)) / pooled_std return TestResult( test=test_name, p_value=p_value, significant=p_value < alpha, effect_size=cohens_d, control_mean=np.mean(control_values), treatment_mean=np.mean(treatment_values), interpretation=self._interpret_continuous( p_value, alpha, cohens_d, np.mean(control_values), np.mean(treatment_values) ) ) def _interpret(self, p_value, alpha, effect, p_ctrl, p_treat): if p_value >= alpha: return f"无统计显著差异 (p={p_value:.4f} ≥ {alpha})。建议继续收集数据或增大样本量。" direction = "提升" if effect > 0 else "下降" return ( f"统计显著 (p={p_value:.4f} < {alpha})。" f"实验组{direction}了{abs(effect)*100:.1f}个百分点" f"（{p_ctrl:.1%} → {p_treat:.1%}）。" ) 4.3 序贯检验（支持提前停止） class SequentialTester: """序贯检验：允许在实验过程中提前判断""" def __init__(self, alpha: float = 0.05, power: float = 0.8, num_looks: int = 5): # Bonferroni 校正 self.adjusted_alpha = alpha / num_looks self.looks = num_looks self.current_look = 0 def should_stop_early( self, control_data: list, treatment_data: list, sample_size_ratio: float # 当前样本量 / 计划样本量 ) -> EarlyStopDecision: """检查是否可以提前停止""" self.current_look = int(sample_size_ratio * self.looks) result = StatisticalTester().test_continuous( control_data, treatment_data, self.adjusted_alpha ) # 护栏指标检查 guardrail_ok = self._check_guardrails(control_data, treatment_data) if not guardrail_ok: return EarlyStopDecision( should_stop=True, reason="护栏指标恶化，建议立即停止实验", winner="control" ) if result.significant: if result.effect_size > 0: return EarlyStopDecision( should_stop=True, reason=f"实验组显著优于对照组 (p={result.p_value:.4f})", winner="treatment" ) else: return EarlyStopDecision( should_stop=True, reason=f"实验组显著劣于对照组 (p={result.p_value:.4f})", winner="control" ) # 计算当前功效 current_power = self._compute_power( len(control_data), result.effect_size ) if current_power > 0.8 and not result.significant: return EarlyStopDecision( should_stop=True, reason=f"功效充足({current_power:.1%})但无显著差异，停止实验", winner="tie" ) return EarlyStopDecision(should_stop=False) 五、LLM 特有的 A/B 测试方法 5.1 LLM-as-Judge A/B 测试 class LLMJudgeABTest: """使用 LLM 作为评判者的 A/B 测试""" async def judge_pair( self, prompt: str, response_a: str, response_b: str, criteria: list[str] ) -> JudgmentResult: """让 LLM 判断哪个回答更好""" judge_prompt = f"""You are an impartial judge. Compare two responses to the same prompt. Prompt: {prompt} Response A: {response_a} Response B: {response_b} Criteria: {', '.join(criteria)} Evaluate which response is better. Consider: 1. Accuracy and correctness 2. Completeness 3. Clarity and structure 4. Adherence to instructions Respond in JSON: {{ "winner": "A" | "B" | "tie", "confidence": 0.0-1.0, "reasoning": "explanation", "scores": {{"A": float, "B": float}} }}""" response = await self.judge_llm.invoke(judge_prompt, temperature=0.0) return JudgmentResult(**json.loads(response.content)) async def run_experiment( self, test_cases: list[TestCase], control_agent: Agent, treatment_agent: Agent, num_judges: int = 3 # 多评判者取平均 ) -> ExperimentResult: results = [] for case in test_cases: # 生成两组回答 response_ctrl = await control_agent.run(case.input) response_treat = await treatment_agent.run(case.input) # 多评判者投票 judgments = [] for i in range(num_judges): judge = self.judges[i] judgment = await judge.judge_pair( case.input, response_ctrl, response_treat, case.criteria ) judgments.append(judgment) # 多数投票 winner = self._majority_vote(judgments) results.append({ "test_id": case.id, "winner": winner, "confidence": np.mean([j.confidence for j in judgments]), }) # 统计分析 wins_treatment = sum(1 for r in results if r["winner"] == "treatment") wins_control = sum(1 for r in results if r["winner"] == "control") ties = sum(1 for r in results if r["winner"] == "tie") # Bradley-Terry 模型检验 bt_stat = self._bradley_terry_test(wins_treatment, wins_control, ties) return ExperimentResult( wins_treatment=wins_treatment, wins_control=wins_control, ties=ties, p_value=bt_stat.p_value, significant=bt_stat.p_value < 0.05, avg_confidence=np.mean([r["confidence"] for r in results]) ) 六、实验报告自动化 class ExperimentReporter: """自动化实验报告生成""" async def generate_report( self, experiment_id: str ) -> ExperimentReport: exp = await self.repo.get(experiment_id) metrics = await self.collector.aggregate_all(experiment_id) test_results = {} for metric_name, data in metrics.items(): if metric_name in ["task_completed", "user_thumbs_up"]: # 比例检验 result = self.tester.test_proportion( data["control"].successes, data["control"].total, data["treatment"].successes, data["treatment"].total ) else: # 连续值检验 result = self.tester.test_continuous( data["control"].values, data["treatment"].values ) test_results[metric_name] = result # 护栏指标检查 guardrail_status = self._check_guardrails(metrics, exp.guardrail_metrics) # 生成决策建议 recommendation = self._generate_recommendation( test_results, guardrail_status, exp.hypothesis ) return ExperimentReport( experiment=exp, sample_sizes={ "control": metrics["task_completed"]["control"].total, "treatment": metrics["task_completed"]["treatment"].total, }, results=test_results, guardrail_status=guardrail_status, recommendation=recommendation, summary=self._generate_summary(test_results, recommendation), generated_at=datetime.now() ) def _generate_recommendation(self, results, guardrails, hypothesis): primary = results.get(hypothesis.dependent_variables[0]) if not primary.significant: return Recommendation( action="continue_or_stop", reason=f"主指标无显著差异 (p={primary.p_value:.4f})。" f"建议：若已达到计划样本量则停止；否则继续收集数据。" ) if primary.effect_size > 0 and guardrails.all_passed: return Recommendation( action="ship", reason=f"主指标显著提升 (p={primary.p_value:.4f}, " f"效应量={primary.effect_size:.3f})。" f"护栏指标全部通过。建议全量发布。" ) if primary.effect_size < 0: return Recommendation( action="do_not_ship", reason=f"主指标显著下降 (p={primary.p_value:.4f})。不建议发布。" ) if not guardrails.all_passed: return Recommendation( action="do_not_ship", reason=f"主指标虽提升但护栏指标恶化：{guardrails.violated}。不建议发布。" ) 七、A/B 测试 Checklist □ 实验假设明确（自变量、因变量、预期效果） □ 样本量计算完成（MDE、power、alpha） □ 流量分配使用一致性哈希（同一用户体验一致） □ 护栏指标已定义并监控 □ 统计检验方法匹配指标类型（比例/连续） □ 序贯检验支持提前停止 □ LLM-as-Judge 评判使用多评判者 □ 实验报告自动生成 □ 决策建议基于数据而非直觉 □ 实验结果归档可追溯 结语 A/B 测试是 Agent 优化的科学基石。在 LLM 的非确定性世界里，直觉是不可靠的，只有统计检验才能区分真实效果和随机噪声。投资 A/B 测试平台不是开销，而是回报率最高的基础设施投资。让每一次 Prompt 修改、每一次模型升级都有数据支撑，这就是 Agent 工程的成熟标志。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

AI 对齐：让模型做正确的事 AI 对齐（AI Alignment）是确保 AI 系统的行为与人类意图、价值观和利益保持一致的研究领域。2026 年，随着模型能力逼近 AGI 水平，对齐问题从学术讨论变成了紧迫的工程挑战。从 RLHF 到 Constitutional AI 再到最新的自我对齐方法，对齐技术正在经历快速迭代。 一、对齐技术演进路线 2022 2023 2024 2025 2026 │ │ │ │ │ RLHF → DPO → Constitutional → Self-Play → Multi-Agent RLAIF AI (CAI) Alignment Constitutional KTO (SPA) Alignment (MACA) 对齐方法对比总览 方法 核心思想 人类标注成本 效果 计算成本 RLHF 人类反馈强化学习 极高 好 高 DPO 直接偏好优化 高 较好 中 Constitutional AI 宪法约束自我改进 中 好 中 RLAIF AI 反馈强化学习 低 中 高 SPA 自我对弈对齐 低 较好 高 MACA 多 Agent 宪法对齐 低 最好 极高 二、RLHF 回顾与 2026 新发展 2.1 经典 RLHF 流程 ┌──────────┐ ┌──────────────┐ ┌─────────────┐ ┌──────────┐ │ SFT模型 │ → │ 奖励模型训练 │ → │ PPO强化学习 │ → │ 对齐模型 │ │ (预训练) │ │ (人类偏好数据)│ │ (RL优化) │ │ (输出) │ └──────────┘ └──────────────┘ └─────────────┘ └──────────┘ 2.2 RLHF 2026 新改进 class RLHF2026: """2026年的RLHF改进版本""" def __init__(self): self.improvements = { 'reward_model': { 'traditional': '单一奖励模型', '2026': '多维度奖励模型集成（有用性、安全性、诚实性）', }, 'preference_data': { 'traditional': '人工标注偏好对', '2026': '人工+AI混合标注，AI标注后人工审核', }, 'optimization': { 'traditional': 'PPO', '2026': 'GRPO (Group Relative Policy Optimization) + 自适应KL惩罚', }, 'evaluation': { 'traditional': '人工评估', '2026': '多维度自动评估 + 人工抽样', } } 2.3 多维度奖励模型 import torch import torch.nn as nn from transformers import AutoModel class MultiDimensionalRewardModel(nn.Module): """多维度奖励模型——2026 RLHF 标配""" def __init__(self, base_model_name: str, dimensions: list = None): super().__init__() self.base = AutoModel.from_pretrained(base_model_name) self.dimensions = dimensions or ['helpfulness', 'safety', 'honesty', 'factuality'] # 每个维度一个奖励头 self.reward_heads = nn.ModuleDict({ dim: nn.Linear(self.base.config.hidden_size, 1) for dim in self.dimensions }) def forward(self, input_ids, attention_mask): outputs = self.base(input_ids, attention_mask=attention_mask) pooled = outputs.last_hidden_state[:, 0] # CLS token rewards = {} for dim, head in self.reward_heads.items(): rewards[dim] = head(pooled).squeeze(-1) return rewards def get_combined_reward(self, rewards: dict, weights: dict = None) -> float: """加权组合各维度奖励""" weights = weights or {d: 1.0/len(self.dimensions) for d in self.dimensions} return sum(weights[d] * rewards[d] for d in self.dimensions) 三、Constitutional AI (CAI) 3.1 核心思想 Constitutionual AI 由 Anthropic 提出，核心是用一组"宪法"原则来指导模型自我改进，减少对人类标注的依赖： ...

引言 Agent 系统的三个核心维度——Prompt、工具、模型——任何一个的变更都可能引发连锁反应。传统软件的版本管理主要针对代码，而 Agent 还需要管理自然语言"代码"（Prompt）、动态加载的工具和外部模型版本。本文将构建完整的 Agent 版本管理体系。 一、Agent 版本的复杂性 变更类型与风险 ┌──────────────────────────────────────────────────────┐ │ Agent 变更类型与风险矩阵 │ ├──────────────┬──────────┬──────────┬─────────────────┤ │ 变更类型 │ 频率 │ 风险等级 │ 影响范围 │ ├──────────────┼──────────┼──────────┼─────────────────┤ │ Prompt 修改 │ 每周 │ 中-高 │ 输出质量/行为 │ │ 工具更新 │ 每月 │ 中 │ 工具调用/结果 │ │ 模型升级 │ 每季度 │ 高 │ 全局行为变化 │ │ System配置 │ 每周 │ 低-中 │ 性能/限制 │ │ Few-shot示例 │ 每月 │ 中 │ 输出风格/格式 │ │ 工作流变更 │ 每月 │ 高 │ 执行路径/延迟 │ └──────────────┴──────────┴──────────┴─────────────────┘ 版本组合的笛卡尔积问题 Agent 行为是 Prompt版本 × 工具版本 × 模型版本 的组合。如果三者各自有 3 个版本，理论上存在 27 种组合。版本管理的目标就是确保任意组合的行为可预测、可回滚。 ...

上下文窗口的"房价"问题 2026 年，虽然主流模型的上下文窗口已达到 128K-1M tokens，但"窗口越大越不够用"——RAG 检索结果、工具调用返回、对话历史、知识库内容，每个环节都在争抢窗口空间。Prompt 压缩技术就像是在有限的土地上建造高层建筑，让每一个 token 都发挥最大价值。 一、Prompt 压缩的价值 1.1 成本与性能双优化 优化维度 压缩前 压缩后 改善 输入 Token 数 8000 4000 -50% API 成本（/千次） $24 $12 -50% 响应延迟 3.2s 1.8s -44% 上下文利用率 40% 75% +87.5% 信息保留率 100% 92-97% -3~8% 1.2 压缩策略分类 Prompt 压缩 ├── 无损压缩 │ ├── 符号化压缩（缩写、代码化） │ ├── 结构化压缩（JSON→紧凑格式） │ └── 去冗余压缩（删除重复信息） ├── 有损压缩 │ ├── 语义压缩（LLM 总结） │ ├── 选择性保留（截断低重要性内容） │ └── 信息蒸馏（提取关键信息） └── 混合压缩 └── 分层压缩策略 二、无损压缩技术 2.1 符号化压缩 class SymbolicCompressor: """符号化压缩——用短符号替代长文本""" SYMBOL_MAP = { # 常见指令缩写 "请分析以下内容并给出": "分析:", "请根据以上信息回答": "回答:", "以下是相关的背景信息": "背景:", "请注意以下重要事项": "注意:", # 角色缩写 "你是一个专业的": "角色:", "你的核心职责是": "职责:", # 格式缩写 "请用Markdown表格格式输出": "→MD表格", "请用JSON格式输出": "→JSON", "请用列表格式输出": "→列表", # 常见短语 "需要注意的是": "⚠", "重要提醒": "‼", "例如": "如", "也就是说": "即", } def compress(self, prompt: str) -> str: for full, symbol in self.SYMBOL_MAP.items(): prompt = prompt.replace(full, symbol) return prompt def decompress_guide(self) -> str: """生成符号说明（添加到System Prompt）""" guide = "符号说明: " for full, symbol in self.SYMBOL_MAP.items(): guide += f"{symbol}={full[:4]}.. " return guide 2.2 结构化压缩 class StructuralCompressor: """结构化压缩——压缩冗余的格式""" def compress_table(self, markdown_table: str) -> str: """压缩 Markdown 表格""" lines = markdown_table.strip().split('\n') if len(lines) < 3: return markdown_table # 提取表头和数据 headers = [h.strip() for h in lines[0].split('|')[1:-1]] data_rows = [] for line in lines[2:]: # 跳过分隔行 cells = [c.strip() for c in line.split('|')[1:-1]] data_rows.append(cells) # 紧凑格式：用 | 分隔，不用对齐 compact = '|'.join(headers) + '\n' for row in data_rows: compact += '|'.join(row) + '\n' return compact def compress_json(self, json_str: str) -> str: """压缩 JSON""" import json data = json.loads(json_str) return json.dumps(data, ensure_ascii=False, separators=(',', ':')) def compress_list(self, markdown_list: str) -> str: """压缩列表""" lines = markdown_list.strip().split('\n') items = [l.lstrip('- *').strip() for l in lines if l.strip()] return '; '.join(items) 2.3 去冗余压缩 class RedundancyRemover: """去冗余压缩""" def compress(self, prompt: str) -> str: # 1. 移除重复段落 prompt = self._remove_duplicate_paragraphs(prompt) # 2. 移除重复句子 prompt = self._remove_duplicate_sentences(prompt) # 3. 移除空白行 prompt = self._remove_blank_lines(prompt) # 4. 合并连续空格 import re prompt = re.sub(r' {2,}', ' ', prompt) return prompt def _remove_duplicate_paragraphs(self, text: str) -> str: paragraphs = text.split('\n\n') seen = set() unique = [] for p in paragraphs: normalized = p.strip().lower() if normalized and normalized not in seen: seen.add(normalized) unique.append(p) return '\n\n'.join(unique) def _remove_duplicate_sentences(self, text: str) -> str: import re sentences = re.split(r'(?<=[。.!?！？])\s+', text) seen = set() unique = [] for s in sentences: if s.strip() and s.strip() not in seen: seen.add(s.strip()) unique.append(s) return ' '.join(unique) 三、有损压缩技术 3.1 LLM 语义压缩 class SemanticCompressor: """使用 LLM 进行语义压缩""" COMPRESSION_PROMPT = """请压缩以下文本，要求： 1. 保留所有关键信息和数据 2. 保留逻辑结构和因果关系 3. 移除冗余描述和过渡语句 4. 用更简洁的表达替代冗长表达 5. 保持事实准确性 原始文本（{original_tokens} tokens）： {text} 输出压缩后的文本，目标：{target_tokens} tokens以内。""" def compress(self, text: str, target_ratio: float = 0.5, llm_client=None) -> str: original_tokens = self._estimate_tokens(text) target_tokens = int(original_tokens * target_ratio) prompt = self.COMPRESSION_PROMPT.format( original_tokens=original_tokens, text=text, target_tokens=target_tokens ) compressed = llm_client.generate(prompt) return compressed def _estimate_tokens(self, text: str) -> int: # 粗略估算 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars return chinese_chars * 2 + other_chars // 4 3.2 选择性保留压缩 class SelectiveCompressor: """选择性保留——基于重要性的压缩""" def compress(self, text: str, target_ratio: float = 0.5) -> str: # 1. 分割为句子 sentences = self._split_sentences(text) # 2. 计算每个句子的重要性分数 scored = self._score_sentences(sentences) # 3. 保留高重要性句子 target_count = int(len(sentences) * target_ratio) top_sentences = sorted(scored, key=lambda x: -x[1])[:target_count] # 4. 按原顺序排列 top_sentences.sort(key=lambda x: x[2]) # 按原始位置排序 return ' '.join(s[0] for s in top_sentences) def _score_sentences(self, sentences: list) -> list: """使用 TextRank 思想计算句子重要性""" # 基于句子间相似度构建图 n = len(sentences) scores = [1.0] * n for iteration in range(10): # 迭代计算 new_scores = [] for i, sent in enumerate(sentences): score = 0.15 # 基础分 for j, other in enumerate(sentences): if i != j: sim = self._sentence_similarity(sent, other) score += 0.85 * sim * scores[j] / max( sum(self._sentence_similarity(other, s2) for k, s2 in enumerate(sentences) if k != j), 1e-8 ) new_scores.append(score) scores = new_scores return [(sentences[i], scores[i], i) for i in range(n)] def _sentence_similarity(self, s1: str, s2: str) -> float: """计算两个句子的相似度""" words1 = set(s1.split()) words2 = set(s2.split()) intersection = words1 & words2 union = words1 | words2 return len(intersection) / max(len(union), 1) 3.3 信息蒸馏 class InformationDistiller: """信息蒸馏——提取关键信息，丢弃细节""" DISTILL_PROMPT = """从以下文本中提取关键信息，使用紧凑格式输出。 输出格式： - 主题：[1-2句话] - 关键事实：[每条1行，最多5条] - 数据：[数值和单位] - 结论：[1句话] 文本： {text}""" def distill(self, text: str, llm_client) -> str: prompt = self.DISTILL_PROMPT.format(text=text) return llm_client.generate(prompt) 四、分层压缩策略 class LayeredCompressor: """分层压缩策略——不同内容用不同压缩方法""" def __init__(self, llm_client): self.llm = llm_client self.symbolic = SymbolicCompressor() self.structural = StructuralCompressor() self.redundancy = RedundancyRemover() self.semantic = SemanticCompressor() self.selective = SelectiveCompressor() self.distiller = InformationDistiller(llm_client) def compress(self, prompt: str, target_ratio: float = 0.5) -> str: """分层压缩""" original_tokens = self._estimate_tokens(prompt) target_tokens = int(original_tokens * target_ratio) # Layer 1: 无损压缩（总是执行） prompt = self.symbolic.compress(prompt) prompt = self.structural.compress_json(prompt) prompt = self.redundancy.compress(prompt) current_tokens = self._estimate_tokens(prompt) if current_tokens <= target_tokens: return prompt # 无损压缩已达标 # Layer 2: 内容分类 sections = self._classify_sections(prompt) # Layer 3: 按类别压缩 compressed_sections = [] for section_type, content in sections: if section_type == 'rules': # 规则类：仅做符号压缩 compressed_sections.append(self.symbolic.compress(content)) elif section_type == 'knowledge': # 知识类：语义压缩 compressed_sections.append( self.semantic.compress(content, 0.4, self.llm) ) elif section_type == 'examples': # 示例类：选择性保留 compressed_sections.append( self.selective.compress(content, 0.6) ) elif section_type == 'context': # 上下文类：信息蒸馏 compressed_sections.append( self.distiller.distill(content, self.llm) ) else: compressed_sections.append(content) result = '\n\n'.join(compressed_sections) # Layer 4: 如果仍超标，全局压缩 if self._estimate_tokens(result) > target_tokens: result = self.semantic.compress(result, target_tokens / self._estimate_tokens(result), self.llm) return result def _classify_sections(self, prompt: str) -> list: """将 Prompt 分为不同类型的段落""" sections = [] current_section = "" current_type = "other" for line in prompt.split('\n'): if line.startswith('规则') or line.startswith('约束'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'rules' elif line.startswith('知识') or line.startswith('背景'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'knowledge' elif line.startswith('示例') or line.startswith('例子'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'examples' elif line.startswith('上下文') or line.startswith('历史'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'context' else: current_section += line + '\n' if current_section: sections.append((current_type, current_section)) return sections 五、对话历史压缩 class ConversationHistoryCompressor: """对话历史压缩——长对话的上下文管理""" def __init__(self, llm_client, max_history_tokens: int = 4000): self.llm = llm_client self.max_tokens = max_history_tokens def compress_history(self, messages: list) -> list: """压缩对话历史""" total_tokens = sum(self._estimate_tokens(m['content']) for m in messages) if total_tokens <= self.max_tokens: return messages # 不需要压缩 # 策略：保留最近 N 轮 + 压缩早期对话 recent_count = min(6, len(messages)) # 保留最近3轮 recent = messages[-recent_count:] old = messages[:-recent_count] # 压缩早期对话 summary = self._summarize_conversation(old) # 构建压缩后的历史 compressed = [ {"role": "system", "content": f"对话摘要：{summary}"}, *recent ] return compressed def _summarize_conversation(self, messages: list) -> str: """总结早期对话""" conversation_text = '\n'.join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请总结以下对话的关键信息，保留： 1. 用户的核心需求和偏好 2. 已达成的结论和决定 3. 未解决的问题 4. 重要的事实和数据 对话内容： {conversation_text} 总结（不超过200字）：""" return self.llm.generate(prompt) 六、压缩效果评估 class CompressionEvaluator: """压缩效果评估器""" def evaluate(self, original: str, compressed: str, test_cases: list, llm_client) -> dict: results = { 'compression_ratio': len(compressed) / len(original), 'token_reduction': 1 - self._tokens(compressed) / self._tokens(original), } # 信息保留率评估 info_retention = self._evaluate_info_retention(original, compressed, llm_client) results['info_retention'] = info_retention # 任务效果评估 original_scores = [] compressed_scores = [] for case in test_cases: # 使用原始 Prompt original_response = llm_client.generate(original + case['input']) original_scores.append(self._score(original_response, case['expected'])) # 使用压缩 Prompt compressed_response = llm_client.generate(compressed + case['input']) compressed_scores.append(self._score(compressed_response, case['expected'])) results['original_accuracy'] = sum(original_scores) / len(original_scores) results['compressed_accuracy'] = sum(compressed_scores) / len(compressed_scores) results['accuracy_drop'] = results['original_accuracy'] - results['compressed_accuracy'] # 成本节省 results['cost_saving'] = results['token_reduction'] return results 压缩效果实测数据 压缩方法 压缩率 信息保留 准确率变化 延迟改善 符号压缩 15% 100% 0% +10% 去冗余 20% 100% 0% +15% 语义压缩 45% 94% -3% +40% 选择性保留 50% 90% -5% +45% 信息蒸馏 65% 85% -8% +55% 分层压缩 50% 96% -2% +44% 七、最佳实践 先无损后有损：先尝试无损压缩，不够再考虑有损 分层压缩最优：不同内容用不同策略，综合效果最好 规则不可压缩：System Prompt 中的规则和约束不应被有损压缩 压缩vs精简：很多时候重新设计比压缩更有效 监控压缩质量：定期评估压缩后的任务效果 缓存压缩结果：相同输入的压缩结果可以缓存 结语 Prompt 压缩是在信息密度和效果之间寻找最优平衡点的艺术。2026 年的工具链已经足够成熟，可以实现接近无损的 50% 压缩——这意味着同样的上下文窗口可以容纳两倍的信息，同样的预算可以处理两倍的请求。 ...