架构设计

1. 从轮询到事件驱动 传统 Agent 采用轮询模式：定期检查是否有新任务或状态变化。这种模式简单但低效——要么延迟高（轮询间隔长），要么浪费资源（轮询间隔短）。 事件驱动架构（EDA）将 Agent 从"主动检查"转变为"被动响应"：当有意义的事件发生时，系统主动通知 Agent 处理。核心优势： 低延迟：事件发生即触发，无需等待轮询周期 低资源消耗：无事件时 Agent 处于休眠状态 天然解耦：事件生产者和消费者完全分离 弹性扩展：事件积压时可动态增加消费者 2. 事件驱动 Agent 整体架构 ┌──────────┐ ┌──────────┐ ┌──────────┐ │ GitHub │ │ Slack │ │ Sensor │ 事件源 │ Webhook │ │ Events │ │ Data │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ ▼ ▼ ▼ ┌────────────────────────────────────────┐ │ Event Ingestion Layer │ │ (API Gateway / Message Queue) │ └───────────────────┬────────────────────┘ │ ▼ ┌────────────────────────────────────────┐ │ Event Router / Filter │ └───────────────────┬────────────────────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Agent A │ │ Agent B │ │ Agent C │ 事件消费者 │(Coder) │ │(Reviewer)│ │(Notifier)│ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────┐ │ Action Executor │ │ (API calls, DB writes, etc.) │ └─────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────┐ │ Event Publisher (新事件) │ └─────────────────────────────────────┘ 3. 事件模型设计 3.1 事件 Schema from pydantic import BaseModel, Field from datetime import datetime from enum import Enum from typing import Any, Optional import uuid class EventSeverity(str, Enum): INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" class AgentEvent(BaseModel): """统一事件格式 - CloudEvents 兼容""" event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) event_type: str # "github.pr.opened", "agent.task.completed" source: str # 事件来源标识 subject: str = "" # 事件主题（如 PR #123） data: dict[str, Any] # 事件负载 time: datetime = Field(default_factory=datetime.now) severity: EventSeverity = EventSeverity.INFO trace_id: str = "" # 分布式追踪 ID correlation_id: str = "" # 关联 ID（同一会话） # 事件版本控制 spec_version: str = "1.0" data_content_type: str = "application/json" class TaskCompletedEvent(AgentEvent): event_type: str = "agent.task.completed" data: dict[str, Any] = Field(description="包含 task_id, result, duration 等") class ErrorEvent(AgentEvent): event_type: str = "agent.error" severity: EventSeverity = EventSeverity.ERROR 3.2 事件注册表 class EventRegistry: """事件类型注册表 - 管理 Schema 和路由规则""" def __init__(self): self._schemas: dict[str, type[BaseModel]] = {} self._handlers: dict[str, list[callable]] = {} def register(self, event_type: str, schema: type[BaseModel]): self._schemas[event_type] = schema def subscribe(self, event_type: str, handler: callable): self._handlers.setdefault(event_type, []).append(handler) def get_handlers(self, event_type: str) -> list[callable]: # 支持通配符匹配: "github.pr.*" 匹配 "github.pr.opened" handlers = self._handlers.get(event_type, []) for pattern, h_list in self._handlers.items(): if "*" in pattern: prefix = pattern.replace("*", "") if event_type.startswith(prefix): handlers.extend(h_list) return handlers def validate(self, event: AgentEvent) -> bool: schema = self._schemas.get(event.event_type) if schema: schema(**event.model_dump()) return True 4. 事件总线实现 4.1 内存事件总线（单机） import asyncio from collections import defaultdict from concurrent.futures import ThreadPoolExecutor class InMemoryEventBus: def __init__(self, max_workers: int = 10): self._subscribers: dict[str, list] = defaultdict(list) self._executor = ThreadPoolExecutor(max_workers=max_workers) self._dead_letter_queue: list[tuple[AgentEvent, Exception]] = [] self._middleware: list[callable] = [] def use(self, middleware: callable): """注册中间件：日志、认证、限流等""" self._middleware.append(middleware) async def subscribe(self, event_type: str, handler: callable): self._subscribers[event_type].append(handler) async def publish(self, event: AgentEvent): # 执行中间件链 for mw in self._middleware: event = await mw(event) if event is None: return # 被中间件拦截 handlers = self._subscribers.get(event.event_type, []) # 匹配通配符订阅 for pattern, h_list in self._subscribers.items(): if "*" in pattern: prefix = pattern.replace("*", "") if event.event_type.startswith(prefix): handlers.extend(h_list) # 并行执行所有处理器 tasks = [self._safe_execute(h, event) for h in handlers] await asyncio.gather(*tasks, return_exceptions=True) async def _safe_execute(self, handler: callable, event: AgentEvent): try: if asyncio.iscoroutinefunction(handler): await handler(event) else: await asyncio.get_event_loop().run_in_executor( self._executor, handler, event ) except Exception as e: self._dead_letter_queue.append((event, e)) # 发布错误事件 error_event = ErrorEvent( source="event_bus", data={"original_event": event.event_id, "error": str(e)}, trace_id=event.trace_id, ) await self.publish(error_event) 4.2 Kafka 事件总线（分布式） from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json class KafkaEventBus: def __init__(self, bootstrap_servers: str = "localhost:9092"): self.bootstrap_servers = bootstrap_servers self._producer: AIOKafkaProducer = None self._consumers: list[AIOKafkaConsumer] = [] self._handlers: dict[str, callable] = {} async def start(self): self._producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v.model_dump(mode="json")).encode(), key_serializer=lambda k: k.encode() if k else None, acks="all", # 等待所有副本确认 enable_idempotence=True, # 幂等生产者 compression_type="lz4", # 压缩 max_in_flight_requests_per_connection=5, ) await self._producer.start() async def publish(self, event: AgentEvent, topic: str = None): topic = topic or event.event_type.split(".")[0] # 按事件类型分 Topic partition_key = event.correlation_id or event.source await self._producer.send_and_wait( topic, event, key=partition_key ) async def subscribe(self, event_type: str, handler: callable, group_id: str = "agent-group"): topic = event_type.split(".")[0] consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, group_id=group_id, value_deserializer=lambda v: AgentEvent(**json.loads(v.decode())), auto_offset_reset="latest", enable_auto_commit=False, ) await consumer.start() self._consumers.append(consumer) self._handlers[event_type] = handler async for msg in consumer: event = msg.value if event.event_type == event_type or event_type.endswith("*"): try: await handler(event) await consumer.commit() except Exception as e: # 重试或进入死信队列 await self._handle_error(event, e) async def stop(self): await self._producer.stop() for c in self._consumers: await c.stop() 5. Agent 状态机 事件驱动 Agent 的核心是一个状态机：不同事件触发不同状态转换。 ...

成本问题的本质 LLM 计费公式：成本 = Token 数 × 单价。优化方向只有两个：减少 Token 数、降低单价。 ┌──────────────────────────────────────────────┐ │ 总 API 成本 │ │ ┌──────────────────────────────────────┐ │ │ │ Token 优化 (减少输入) │ │ │ │ - Prompt 压缩 │ │ │ │ - 历史截断 │ │ │ │ - Few-shot 精简 │ │ │ └──────────────────────────────────────┘ │ │ ┌──────────────────────────────────────┐ │ │ │ 模型路由 (降低单价) │ │ │ │ - 简单 → 小模型 ($0.15/1M) │ │ │ │ - 复杂 → 大模型 ($10/1M) │ │ │ └──────────────────────────────────────┘ │ │ ┌──────────────────────────────────────┐ │ │ │ 架构优化 (减少调用) │ │ │ │ - 缓存 (命中率 60%) │ │ │ │ - 批量推理 │ │ │ │ - 流式输出提前终止 │ │ │ └──────────────────────────────────────┘ │ └──────────────────────────────────────────────┘ 一、Token 优化 1.1 Prompt 压缩 class PromptCompressor: """多层 Prompt 压缩""" def compress(self, system_prompt: str, history: list, query: str) -> str: # 1. 系统提示词精简 system = self._trim_system(system_prompt) # 2. 历史对话压缩 history = self._compress_history(history) # 3. Few-shot 示例选择（只保留最相关的） examples = self._select_relevant_examples(query, k=2) # 4. 组装 return f"{system}\n\n{examples}\n\n历史：{history}\n\n问题：{query}" def _trim_system(self, prompt: str) -> str: """去除冗余描述，保留核心指令""" # 去除重复空格和换行 import re prompt = re.sub(r'\n{3,}', '\n\n', prompt) prompt = re.sub(r' {2,}', ' ', prompt) # 去除注释性文字 prompt = re.sub(r'#.*\n', '', prompt) return prompt.strip() def _compress_history(self, history: list, max_turns: int = 5) -> str: """压缩对话历史：保留最近 N 轮 + 摘要""" if len(history) <= max_turns: return "\n".join([f"用户: {h['user']}\n助手: {h['assistant']}" for h in history]) # 早期对话生成摘要 old = history[:-max_turns] recent = history[-max_turns:] summary = self._summarize(old) recent_text = "\n".join([f"用户: {h['user']}\n助手: {h['assistant']}" for h in recent]) return f"[之前对话摘要：{summary}]\n\n{recent_text}" 1.2 Token 计算与监控 import tiktoken class TokenMonitor: def __init__(self, model="gpt-4"): self.encoder = tiktoken.encoding_for_model(model) def count(self, text: str) -> int: return len(self.encoder.encode(text)) def estimate_cost(self, input_tokens: int, output_tokens: int, model: str) -> float: PRICING = { "gpt-4": {"input": 0.03, "output": 0.06}, "gpt-4o": {"input": 0.005, "output": 0.015}, "gpt-4o-mini": {"input": 0.00015, "output": 0.0006}, "claude-3.5-sonnet": {"input": 0.003, "output": 0.015}, } p = PRICING.get(model, PRICING["gpt-4"]) return (input_tokens * p["input"] + output_tokens * p["output"]) / 1000 1.3 压缩效果对比 优化手段 Token 节省 质量影响 去除冗余空白 5-10% 无 历史摘要 40-60% 轻微 Few-shot 精简 20-30% 可控 输出 max_tokens 限制 15-25% 无 函数调用替代文本解析 10-15% 正面 二、模型路由 核心思路：80% 的请求用小模型，20% 的请求用大模型。 ...

为什么传统负载均衡不够？ 传统 LB（Nginx round-robin）对 LLM 不适用。原因： 请求耗时差异巨大：生成 10 字 vs 1000 字，耗时差 100 倍 模型异构：不同实例可能运行不同模型（GPT-4 / Llama / Mistral） 显存敏感：并发请求过多导致 OOM 流式响应：SSE 长连接需要特殊处理 架构总览 ┌──────────────────────────┐ │ API Gateway / LB │ │ (Nginx / Traefik / ENI) │ └────────────┬─────────────┘ ▼ ┌──────────────────────────┐ │ Router (模型感知路由) │ │ - 模型映射 │ │ - 负载策略 │ │ - 熔断/降级 │ └────────────┬─────────────┘ ┌────────────┼────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Model A │ │ Model B │ │ Model C │ │ x3 实例 │ │ x2 实例 │ │ x1 实例 │ │ (GPT-4) │ │(Llama-3)│ │(Mistral)│ └─────────┘ └─────────┘ └─────────┘ 负载均衡策略 1. 加权轮询（Weighted Round Robin） 根据实例的 GPU 数量和模型大小分配权重。 ...

为什么 LLM 需要专项可观测性？ 传统 APM 不够：LLM 有 Token 计费、Prompt 变体、模型路由、工具调用链等特有维度。一个请求可能涉及 3 个模型 + 5 个工具调用 + 2 次检索，没有 Tracing 根本无法定位问题。 三位一体架构 ┌──────────────────────────────────────────────────┐ │ 用户请求 │ │ trace_id = xxx │ └──────────────────────┬───────────────────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Logs │ │ Traces │ │ Metrics │ │ 结构化 │ │ 链路 │ │ 聚合 │ │ 日志 │ │ 追踪 │ │ 指标 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ ELK / │ │ Jaeger /│ │Prometheus│ │ Loki │ │ Langfuse│ │ + Grafana│ └─────────┘ └─────────┘ └─────────┘ │ │ │ └──────────────┼──────────────┘ ▼ ┌─────────────────┐ │ AlertManager │ │ 告警 + 通知 │ └─────────────────┘ 一、结构化日志 import structlog import json # 配置 structlog structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], ) logger = structlog.get_logger() class LLMLogger: """LLM 专用结构化日志""" def log_request(self, trace_id: str, user_id: str, model: str, prompt: str, **kwargs): logger.info("llm_request", trace_id=trace_id, user_id=user_id, model=model, prompt_length=len(prompt), prompt_tokens=kwargs.get("input_tokens"), max_tokens=kwargs.get("max_tokens"), temperature=kwargs.get("temperature", 1.0), tools=kwargs.get("tools"), timestamp=datetime.utcnow().isoformat(), ) def log_response(self, trace_id: str, response: str, input_tokens: int, output_tokens: int, latency_ms: float, model: str, **kwargs): logger.info("llm_response", trace_id=trace_id, model=model, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, latency_ms=latency_ms, finish_reason=kwargs.get("finish_reason"), cost_usd=self._calc_cost(model, input_tokens, output_tokens), ) def log_tool_call(self, trace_id: str, tool_name: str, params: dict, result: dict, latency_ms: float): logger.info("tool_call", trace_id=trace_id, tool=tool_name, params_keys=list(params.keys()), result_status="success" if result.get("success") else "failed", latency_ms=latency_ms, ) def log_error(self, trace_id: str, error: Exception, context: dict): logger.error("llm_error", trace_id=trace_id, error_type=type(error).__name__, error_message=str(error), context=context, ) 日志查询示例 # ELK / Loki 查询：查找高延迟请求 # Kibana KQL: # llm_response AND latency_ms > 5000 AND model: "gpt-4" # Grafana Loki LogQL: # {app="llm-service"} |= "llm_response" | json | latency_ms > 5000 二、分布式链路追踪 LLM 请求的典型链路：API → Router → Cache → Model → Tool → Model → Response ...

微调流水线全景 ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据准备 │───▶│ 训练配置 │───▶│ 训练执行 │───▶│ 评估 │───▶│ 部署发布 │ │ 清洗/标注 │ │ LoRA/QLoRA│ │ GPU 集群 │ │ 自动化 │ │ 灰度/AB │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据版本 │ │ 模型注册 │ │ 监控告警 │ │ DVC/MLflow│ │ MLflow │ │ 回滚机制 │ └──────────┘ └──────────┘ └──────────┘ 一、数据准备 1.1 数据清洗 import re import json from datasets import Dataset class DataCleaner: def __init__(self, min_length=10, max_length=8192): self.min_length = min_length self.max_length = max_length def clean(self, samples: list[dict]) -> list[dict]: cleaned = [] for s in samples: text = s.get("text", "") # 去除 HTML 标签 text = re.sub(r'<[^>]+>', '', text) # 去除多余空白 text = re.sub(r'\s+', ' ', text).strip() # 长度过滤 if self.min_length <= len(text) <= self.max_length: # 去重（基于内容哈希） cleaned.append({**s, "text": text}) # 去重 seen = set() unique = [] for s in cleaned: h = hash(s["text"][:200]) if h not in seen: seen.add(h) unique.append(s) return unique def to_chat_format(self, samples: list[dict]) -> list[dict]: """转换为 chatml 格式""" formatted = [] for s in samples: formatted.append({ "messages": [ {"role": "system", "content": s.get("system", "你是一个有用的助手")}, {"role": "user", "content": s["input"]}, {"role": "assistant", "content": s["output"]} ] }) return formatted 1.2 数据增强 class DataAugmenter: """使用大模型生成训练数据变体""" AUGMENT_PROMPT = """基于以下示例，生成 3 个语义相同但表达不同的变体： 原文：{original} 要求： 1. 保持意图一致 2. 变化表达方式（句式/用词） 3. 不要改变关键信息 输出 JSON 数组格式。""" async def augment(self, sample: dict, llm_client) -> list[dict]: prompt = self.AUGMENT_PROMPT.format(original=sample["input"]) resp = await llm_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) variants = json.loads(resp.choices[0].message.content) return [ {"input": v["input"], "output": sample["output"]} for v in variants.get("variants", []) ] 1.3 数据集分割 from sklearn.model_selection import train_test_split def split_dataset(data: list[dict], train=0.8, val=0.1, test=0.1): train_data, temp = train_test_split(data, test_size=1-train, random_state=42) val_data, test_data = train_test_split(temp, test_size=test/(test+val), random_state=42) return {"train": train_data, "val": val_data, "test": test_data} 二、训练配置 2.1 训练方法对比 方法 显存需求 训练速度 效果 适用场景 Full Fine-tune 极高（全部参数） 慢 最好 数据充足、预算充足 LoRA 低（0.1-1% 参数） 快 接近全量 通用首选 QLoRA 极低（4bit 量化） 中 略低于 LoRA 显存受限 P-Tuning v2 低 快 中等 特定任务 2.2 LoRA 训练配置 from peft import LoraConfig, get_peft_model, TaskType from transformers import AutoModelForCausalLM, TrainingArguments from trl import SFTTrainer def setup_lora_training(model_name="Qwen/Qwen2.5-7B"): model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype="auto", device_map="auto", ) lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # LoRA 秩，越大效果越好但显存越多 lora_alpha=128, # 通常为 r 的 2 倍 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none", ) model = get_peft_model(model, lora_config) return model, lora_config training_args = TrainingArguments( output_dir="./output/qwen-lora", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, warmup_ratio=0.1, learning_rate=2e-4, lr_scheduler_type="cosine", logging_steps=10, save_strategy="epoch", eval_strategy="epoch", bf16=True, gradient_checkpointing=True, optim="adamw_torch", max_grad_norm=1.0, ) trainer = SFTTrainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=eval_dataset, packing=True, # 序列打包提升效率 max_seq_length=2048, ) 2.3 QLoRA（显存优化） from transformers import BitsAndBytesConfig bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype="bfloat16", bnb_4bit_use_double_quant=True, ) model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=bnb_config, device_map="auto", ) 三、评估流程 class ModelEvaluator: def __init__(self, model_path, test_data): self.model_path = model_path self.test_data = test_data def evaluate(self) -> dict: results = { "loss": self._eval_loss(), "bleu": self._eval_bleu(), "rouge": self._eval_rouge(), "human_like": self._eval_human_like(), "safety": self._eval_safety(), "latency_p50": self._eval_latency(), } return results def _eval_safety(self) -> float: """安全评估：检测有害输出比例""" harmful_count = 0 for sample in self.test_data: output = self._generate(sample["input"]) if self._is_harmful(output): harmful_count += 1 return 1.0 - harmful_count / len(self.test_data) def _is_harmful(self, text: str) -> bool: harmful_patterns = [ r"如何(制造|获取).*(武器|毒品)", r"(自杀|自残)的方法", r"歧视.*(种族|性别|宗教)", ] return any(re.search(p, text) for p in harmful_patterns) 四、版本管理 import mlflow class ModelRegistry: def __init__(self, tracking_uri="http://mlflow:5000"): mlflow.set_tracking_uri(tracking_uri) def register_model(self, model_path, name, metrics, tags=None): with mlflow.start_run(): mlflow.log_metrics(metrics) mlflow.log_artifacts(model_path) mlflow.register_model( f"runs:/{mlflow.active_run().info.run_id}/model", name, tags=tags or {} ) def get_version(self, name, stage="Production"): client = mlflow.tracking.MlflowClient() versions = client.get_latest_versions(name, stages=[stage]) return versions[0] if versions else None 五、灰度发布与 A/B 测试 class CanaryDeployer: """灰度发布：逐步增加新模型流量比例""" def __init__(self, old_model: str, new_model: str): self.old_model = old_model self.new_model = new_model self.traffic_split = 0.0 # 新模型流量比例 self.metrics = {"old": [], "new": []} def should_use_new(self) -> bool: import random return random.random() < self.traffic_split def canary_stages(self): """分阶段灰度""" stages = [ {"split": 0.05, "duration": "1h", "check": "error_rate < 1%"}, {"split": 0.20, "duration": "6h", "check": "error_rate < 1%, latency_p99 < 10s"}, {"split": 0.50, "duration": "24h", "check": "all_metrics_stable"}, {"split": 1.00, "duration": "∞", "check": "promoted"}, ] return stages def evaluate_and_promote(self): """评估指标决定是否推进""" new_error_rate = self._calc_error_rate("new") old_error_rate = self._calc_error_rate("old") new_latency = self._calc_p99("new") old_latency = self._calc_p99("old") if new_error_rate > old_error_rate * 1.5: self._rollback() return "ROLLBACK: error rate too high" if new_latency > old_latency * 1.3: self._rollback() return "ROLLBACK: latency regression" return "PROMOTE: metrics OK" 六、回滚机制 class RollbackManager: def __init__(self, registry: ModelRegistry): self.registry = registry def rollback(self, model_name: str, reason: str): """回滚到上一个 Production 版本""" client = mlflow.tracking.MlflowClient() versions = client.search_model_versions( f"name='{model_name}'", order_by=["version_number DESC"] ) prod_versions = [v for v in versions if v.current_stage == "Production"] archived = [v for v in versions if v.current_stage == "Archived"] if len(prod_versions) >= 1 and archived: # 当前 prod 版本归档，上一个 archived 版本恢复 client.transition_model_version_stage( name=model_name, version=prod_versions[0].version, stage="Archived", ) client.transition_model_version_stage( name=model_name, version=archived[0].version, stage="Production", ) logger.info(f"Rolled back {model_name}: {reason}") return True return False 总结 LLM 微调 MLOps 流水线的核心环节：数据质量决定上限，LoRA/QLoRA 平衡效果与成本，评估必须覆盖质量+安全+性能三维度，灰度发布配合自动回滚是生产安全的最后防线。建议使用 MLflow 统一管理模型版本，从训练到部署全链路可追溯。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

从 POC 到生产的鸿沟 POC 阶段一个 LangChain + Chroma 就能跑通，但百万级查询的生产系统需要解决：检索精度、延迟、扩展性、成本、更新策略。 生产架构总览 ┌──────────────────────────────────────────────────────────────┐ │ 用户查询 │ └──────────────────────────┬───────────────────────────────────┘ ▼ ┌──────────────────────────────────────────────────────────────┐ │ 1. Query 处理层 │ │ - 查询改写 / HyDE / 多查询扩展 │ │ - 意图识别 / 路由 │ └──────────────────────────┬───────────────────────────────────┘ ▼ ┌──────────────────────────────────────────────────────────────┐ │ 2. 检索层 │ │ ┌────────────────┐ ┌────────────────┐ │ │ │ 向量检索 (ANN) │ │ 关键词检索 (BM25)│ │ │ │ top_k=20 │ │ top_k=20 │ │ │ └───────┬────────┘ └───────┬────────┘ │ │ └────────┬──────────┘ │ │ ▼ │ │ ┌────────────────┐ │ │ │ 融合 (RRF) │ │ │ │ top_k=50 │ │ │ └───────┬────────┘ │ └──────────────────┼─────────────────────────────────────────┘ ▼ ┌──────────────────────────────────────────────────────────────┐ │ 3. 重排层 (Reranker) │ │ Cross-Encoder 重排 → top_k=5 │ └──────────────────────────┬───────────────────────────────────┘ ▼ ┌──────────────────────────────────────────────────────────────┐ │ 4. 生成层 │ │ Context + Query → LLM → Response + Citations │ └──────────────────────────┬───────────────────────────────────┘ ▼ ┌──────────────────────────────────────────────────────────────┐ │ 5. 缓存层 │ │ 语义缓存 → 响应缓存 │ └──────────────────────────────────────────────────────────────┘ 向量数据库选型 数据库 索引算法 百万级 QPS 扩展性 生态 许可证 Milvus HNSW/IVF/DiskANN 高 分布式原生 丰富 Apache 2.0 Pinecone 专有 高 全托管 SaaS only 商业 Weaviate HNSW 中高 分布式 良好 BSD-3 Qdrant HNSW 高 Rust 单机/分布式 新锐 Apache 2.0 pgvector IVFFlat/HNSW 中 PG 生态 最广 PostgreSQL Redis HNSW/FLAT 高 Redis 生态 良好 Redis SSPL 生产推荐： ...

为什么需要多区域部署？ 单区域部署的延迟：美国用户访问亚洲 LLM 服务，RTT 200-300ms，加上推理时间，总延迟 5-10 秒。多区域部署将延迟降到 1-2 秒。 同时，数据合规（GDPR/CCPA/PIPL）要求数据不跨境，多区域是必选项。 架构总览 ┌───────────────────────┐ │ Global DNS (GeoDNS) │ │ Anycast / Route53 │ └───────────┬───────────┘ ┌───────────────────┼───────────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ US-West │ │ EU-West │ │ AP-East │ │ (Oregon) │ │ (Ireland) │ │ (Singapore) │ │ │ │ │ │ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ │ LLM GPU │ │ │ │ LLM GPU │ │ │ │ LLM GPU │ │ │ │ Cluster │ │ │ │ Cluster │ │ │ │ Cluster │ │ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ │ Redis │ │ │ │ Redis │ │ │ │ Redis │ │ │ │ + Vector│ │ │ │ + Vector│ │ │ │ + Vector│ │ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └───────────────────┼───────────────────┘ ┌───────────┴───────────┐ │ Global Model Registry │ │ (Central Storage) │ │ S3 + CDN 分发 │ └───────────────────────┘ 流量路由 GeoDNS 配置 # AWS Route53 Geo DNS 配置 Type: A Name: api.llm.example.com TTL: 60 Records: - Region: us-west-2 Value: 54.x.x.x # US-West LLM endpoint HealthCheck: llm-us-health - Region: eu-west-1 Value: 52.x.x.x # EU-West LLM endpoint HealthCheck: llm-eu-health - Region: ap-southeast-1 Value: 13.x.x.x # AP-East LLM endpoint HealthCheck: llm-ap-health - Region: default Value: 54.x.x.x # 默认路由到 US # 健康检查失败时自动故障转移 Failover: primary: us-west-2 → ap-southeast-1 secondary: eu-west-1 → us-west-2 智能路由器 import geoip2.database from datetime import datetime class GeoRouter: def __init__(self, regions: list[dict]): self.regions = regions # {"name": "us-west", "endpoint": "...", "latencies": {}} self.health = {r["name"]: {"healthy": True, "p99": 0} for r in regions} self.reader = geoip2.database.Reader('GeoLite2-City.mmdb') def route(self, request_ip: str, user_region: str = None) -> str: # 1. 用户指定区域优先 if user_region and self._is_healthy(user_region): return self._get_endpoint(user_region) # 2. GeoIP 自动判断 try: resp = self.reader.city(request_ip) user_continent = resp.continent.code region = self._match_region(user_continent) if region and self._is_healthy(region): return self._get_endpoint(region) except Exception: pass # 3. 延迟优先选择 return self._lowest_latency_region() def _match_region(self, continent: str) -> str: mapping = {"NA": "us-west", "SA": "us-west", "EU": "eu-west", "AS": "ap-east", "OC": "ap-east", "AF": "eu-west"} return mapping.get(continent) def _lowest_latency_region(self) -> str: healthy = [r for r in self.regions if self._is_healthy(r["name"])] if not healthy: raise Exception("No healthy regions") return min(healthy, key=lambda r: self.health[r["name"]]["p99"])["endpoint"] 数据合规 区域数据隔离 class CompliantDataRouter: """确保用户数据不跨境""" REGION_POLICIES = { "eu-west": {"regulation": "GDPR", "data_residency": "EU"}, "us-west": {"regulation": "CCPA", "data_residency": "US"}, "ap-east": {"regulation": "PIPL", "data_residency": "CN/SG"}, } def __init__(self, region: str): self.region = region self.policy = self.REGION_POLICIES[region] def validate_request(self, user_data: dict) -> bool: """验证请求数据是否符合区域合规要求""" # 检查用户是否同意数据在该区域处理 user_region = user_data.get("region", "") if not self._is_compliant(user_region, self.policy["data_residency"]): raise ComplianceError( f"User from {user_region} cannot be processed in {self.region}" ) return True def _is_compliant(self, user_region: str, data_residency: str) -> bool: rules = { "EU": ["EU", "UK"], # EU 数据不离开欧洲 "US": ["US", "CA", "MX"], # US 数据在北美 "CN/SG": ["CN", "SG", "JP", "KR"], # 亚太数据在亚太 } allowed = rules.get(data_residency, []) return user_region in allowed 合规要求对比 法规 适用区域 核心要求 违规罚款 GDPR EU 数据不离开欧盟，用户有权删除 €20M 或 4% 营收 CCPA California 用户有权知道数据用途，可要求删除 $7,500/违规 PIPL China 数据本地化，跨境需审批 营收 5% 或 5000 万元 LGPD Brazil 类似 GDPR 营收 2% 模型同步 多区域部署最大的挑战：如何保证各区域模型版本一致。 ...

为什么需要 AI 网关 当你只用一个 LLM 提供商时，直接调 API 就行。但现实是： GPT-4 写代码好，Claude 写文章好，Llama 3 性价比高 OpenAI 偶尔宕机，需要自动故障转移 不同业务线有不同的成本预算和延迟要求 需要灰度切换模型而不改业务代码 AI 网关就是解决这些问题的统一入口。它不是简单的反向代理，而是包含路由决策、流量控制、缓存优化、成本管理的完整中间层。 网关核心职责 职责 描述 关键指标 智能路由 根据任务类型/成本/延迟选择模型 路由延迟 <5ms 限流保护 租户级 + 全局级速率限制 误杀率 <0.1% 语义缓存 缓存相似查询的结果 缓存命中率 15-30% 成本控制 实时 Token 计费和预算预警 计费误差 <0.01% 故障转移 自动切换到备用模型/提供商 切换时间 <2s 可观测性 全链路追踪和监控 日志延迟 <100ms 灰度发布 逐步迁移流量到新模型 回滚时间 <10s 架构总览 ┌──────────────────────────────────────────────────┐ │ Client SDK │ │ (统一接口，隐藏后端模型差异) │ └──────────────────┬───────────────────────────────┘ │ ┌──────────────────▼───────────────────────────────┐ │ AI Gateway │ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │ │ Router │ │ RateLimit │ │ Cache │ │ │ │ (模型选择) │ │ (限流) │ │ (语义缓存) │ │ │ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │ │ ┌─────▼──────────────▼───────────────▼──────┐ │ │ │ Request Pipeline │ │ │ │ Auth → Sanitize → Enrich → Route → Proxy │ │ │ └────────────────────────────────────────────┘ │ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │ │ Metrics │ │ Billing │ │ Failover │ │ │ │ (监控) │ │ (计费) │ │ (故障转移) │ │ │ └────────────┘ └────────────┘ └────────────┘ │ └──────────────────┬───────────────────────────────┘ │ ┌──────────────┼──────────────┬───────────────┐ ▼ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌──────────┐ ┌──────────┐ │ OpenAI │ │Claude │ │ Llama │ │ 自部署 │ │ API │ │ API │ │ (vLLM) │ │ (TGI) │ └────────┘ └────────┘ └──────────┘ └──────────┘ 智能路由引擎 from dataclasses import dataclass from typing import Optional import asyncio @dataclass class ModelCandidate: provider: str # "openai" | "anthropic" | "self-hosted" model: str # "gpt-4-turbo" | "claude-3-opus" | "llama-3-70b" priority: int # 优先级（数字越小越优先） cost_per_1k: float # 每千 token 成本（美元） avg_latency_ms: int # 平均延迟 health_score: float # 健康分（0-1） rate_limit_remaining: int # 剩余配额 class IntelligentRouter: def __init__(self, redis, metrics_collector): self.redis = redis self.metrics = metrics_collector self.routing_rules = self._load_routing_rules() async def select_model( self, request: dict, constraints: dict = None ) -> ModelCandidate: """根据请求内容和约束选择最佳模型""" constraints = constraints or {} # 1. 获取所有候选模型 candidates = await self._get_candidates(request, constraints) if not candidates: raise NoModelAvailableError("No healthy model found") # 2. 应用硬约束过滤 filtered = self._apply_constraints(candidates, constraints) if not filtered: raise ConstraintViolationError("No model meets constraints") # 3. 按规则打分排序 scored = [] for c in filtered: score = await self._score_model(c, request, constraints) scored.append((c, score)) scored.sort(key=lambda x: x[1], reverse=True) # 4. 返回最优模型 best = scored[0][0] logger.info(f"Routed to {best.provider}/{best.model} (score={scored[0][1]:.3f})") return best async def _score_model(self, candidate: ModelCandidate, request, constraints) -> float: score = 1.0 # 健康分（权重 0.3） score *= candidate.health_score * 0.3 # 成本分（权重 0.25）：越便宜分越高 max_cost = max(c.cost_per_1k for c in [candidate]) cost_score = 1 - (candidate.cost_per_1k / max(max_cost, 0.001)) score += cost_score * 0.25 # 延迟分（权重 0.2） latency_score = 1 - (candidate.avg_latency_ms / 5000) score += max(latency_score, 0) * 0.2 # 剩余配额分（权重 0.15） quota_score = min(candidate.rate_limit_remaining / 10000, 1.0) score += quota_score * 0.15 # 规则匹配加分（权重 0.1） rule_bonus = self._match_routing_rules(candidate, request) score += rule_bonus * 0.1 return score def _match_routing_rules(self, candidate, request) -> float: bonus = 0.0 task_type = request.get("metadata", {}).get("task_type", "general") # 代码任务优先路由到 GPT-4 if task_type == "code" and "gpt-4" in candidate.model: bonus += 0.5 # 长文写作优先路由到 Claude if task_type == "writing" and "claude" in candidate.model: bonus += 0.5 # 简单问答优先路由到 Llama（成本低） if task_type == "simple_qa" and "llama" in candidate.model: bonus += 0.5 return bonus 路由规则配置 # routing_rules.yaml rules: - name: "code_generation" condition: task_type: "code" min_context_length: 0 preferred_models: - { provider: "openai", model: "gpt-4-turbo", weight: 1.0 } - { provider: "anthropic", model: "claude-3-opus", weight: 0.8 } fallback: - { provider: "self-hosted", model: "llama-3-70b" } - name: "long_context" condition: min_context_length: 50000 preferred_models: - { provider: "anthropic", model: "claude-3-sonnet", weight: 1.0 } # 200K context fallback: - { provider: "openai", model: "gpt-4-turbo", weight: 0.9 } # 128K context - name: "cost_optimized" condition: max_cost_per_request: 0.01 preferred_models: - { provider: "self-hosted", model: "llama-3-8b", weight: 1.0 } fallback: - { provider: "openai", model: "gpt-3.5-turbo", weight: 0.7 } - name: "low_latency" condition: max_latency_ms: 500 preferred_models: - { provider: "self-hosted", model: "llama-3-8b", weight: 1.0 } - { provider: "openai", model: "gpt-3.5-turbo", weight: 0.8 } 语义缓存 传统缓存用 hash key 精确匹配，LLM 场景下用户问 “Python 怎么读文件” 和 “如何在 Python 中读取文件” 是同一个问题，应该命中缓存。 ...

为什么要把 LLM 放到边缘设备上 2026 年，端侧运行 LLM 不再是学术论文里的概念，而是产品需求。原因很直接： 延迟：云上 LLM 往返延迟 200-800ms，端侧可以做到 20-50ms 隐私：医疗、金融场景的数据不能离开设备 离线：网络不稳定场景（工业、户外）需要离线推理 成本：云端推理 $0.01/千 token，端侧摊到设备折旧里几乎免费 但挑战也真实：手机内存 8-16GB，要跑 7B 模型需要 14GB（FP16），得压缩。 模型压缩三板斧 量化：用更少的 bit 表示权重 量化方案 位宽 7B 模型大小 精度损失 硬件支持 FP16 (基线) 16-bit 13.5 GB 0% 全部 INT8 8-bit 6.8 GB <1% CPU/GPU/NPU INT4 (GPTQ) 4-bit 3.5 GB 2-3% GPU/NPU INT4 (AWQ) 4-bit 3.5 GB 1-2% GPU/NPU INT3 3-bit 2.6 GB 5-8% 部分 NPU 1.58-bit ~2-bit 1.8 GB 8-12% 实验阶段 # 使用 llama.cpp 的量化工具链 # 1. 转换为 GGUF 格式 # python convert.py models/llama-3-8b --outtype f16 --outfile llama-3-8b-f16.gguf # 2. 量化为 INT4 # ./quantize llama-3-8b-f16.gguf llama-3-8b-q4_k_m.gguf Q4_K_M # 在 Python 中使用 llama-cpp-python 加载 from llama_cpp import Llama llm = Llama( model_path="llama-3-8b-q4_k_m.gguf", n_ctx=2048, # 上下文窗口（移动端建议 2048-4096） n_threads=4, # CPU 线程数 n_gpu_layers=0, # 移动端通常 0（纯 CPU） n_batch=512, # 批处理大小 mmap=True, # 内存映射加载，减少内存占用 mmap_timeout=30000, ) response = llm( "解释量子计算的基本原理", max_tokens=256, temperature=0.7, top_p=0.9, ) 蒸馏：小模型学大模型 # 使用 Hugging Face 进行知识蒸馏 from transformers import AutoModelForCausalLM, AutoTokenizer import torch import torch.nn.functional as F teacher = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3-70B") student = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3-1B") teacher.eval() student.train() optimizer = torch.optim.AdamW(student.parameters(), lr=5e-5) for batch in dataloader: with torch.no_grad(): teacher_logits = teacher(batch["input_ids"]).logits teacher_probs = F.softmax(teacher_logits / 2.0, dim=-1) # 温度=2 student_logits = student(batch["input_ids"]).logits student_log_probs = F.log_softmax(student_logits / 2.0, dim=-1) # KL 散度损失 loss = F.kl_div(student_log_probs, teacher_probs, reduction="batchmean") # 加上 hard label loss loss += 0.3 * F.cross_entropy( student_logits.view(-1, student_logits.size(-1)), batch["labels"].view(-1) ) loss.backward() optimizer.step() optimizer.zero_grad() 剪枝：去掉不重要的权重 # 结构化剪枝：移除整个注意力头或层 import torch.nn.utils.prune as prune def structured_prune(model, amount=0.3): """移除 30% 最不重要的注意力头""" for name, module in model.named_modules(): if isinstance(module, torch.nn.Linear): # 按 L1 范数剪枝 prune.l1_unstructured(module, name="weight", amount=amount) # 固化剪枝 prune.remove(module, "weight") return model # 评估剪枝后的精度 def evaluate_pruned(model, eval_dataset): pruned = structured_prune(model, amount=0.3) # 重新微调恢复精度 trainer = Trainer( model=pruned, args=TrainingArguments( output_dir="./pruned-finetune", num_train_epochs=2, learning_rate=2e-5, ), train_dataset=train_dataset, eval_dataset=eval_dataset, ) trainer.train() return pruned 端侧推理引擎对比 llama.cpp Mobile // Android NDK 集成示例 // JNI 接口：从 Java/Kotlin 调用 llama.cpp #include "llama.h" #include <jni.h> #include <android/log.h> extern "C" JNIEXPORT jlong JNICALL Java_com_example_llm_LlamaEngine_nativeInit(JNIEnv* env, jobject thiz, jstring model_path) { const char* path = env->GetStringUTFChars(model_path, nullptr); llama_backend_init(false); llama_model_params model_params = llama_model_default_params(); model_params.n_gpu_layers = 0; // 移动端纯 CPU model_params.use_mmap = true; llama_model* model = llama_load_model_from_file(path, model_params); env->ReleaseStringUTFChars(model_path, path); return (jlong)model; } extern "C" JNIEXPORT jstring JNICALL Java_com_example_llm_LlamaEngine_nativeGenerate( JNIEnv* env, jobject thiz, jlong model_ptr, jstring prompt, jint max_tokens ) { llama_model* model = (llama_model*)model_ptr; const char* prompt_str = env->GetStringUTFChars(prompt, nullptr); llama_context_params ctx_params = llama_context_default_params(); ctx_params.n_ctx = 2048; ctx_params.n_threads = 4; ctx_params.n_batch = 512; llama_context* ctx = llama_new_context_with_model(model, ctx_params); // Tokenize std::string text(prompt_str); std::vector<llama_token> tokens = llama_tokenize(ctx, text, true); // 生成循环 std::string result; for (int i = 0; i < max_tokens; i++) { if (llama_decode(ctx, llama_batch_get_one(&tokens.back(), 1, i, 0))) break; llama_token new_token = llama_sample_token(ctx, nullptr, 0.8f, 0.9f, 32); if (new_token == llama_token_eos(ctx)) break; tokens.push_back(new_token); result += llama_token_to_piece(ctx, new_token); } llama_free(ctx); env->ReleaseStringUTFChars(prompt, prompt_str); return env->NewStringUTF(result.c_str()); } MLC-LLM # MLC-LLM：使用 TVM 编译器优化，支持 Android/iOS # 编译模型到移动端 """ # 1. 编译模型 python -m mlc_llm compile \ --model llama-3-8b \ --quantization q4f16_1 \ --target android \ --output llama-3-8b-q4f16_1-android.tar # 2. 在 Android 上加载 """ # Swift (iOS) 示例 """ import MLCEngine let engine = MLCEngine() try await engine.load( model: "Llama-3-8B-q4f16_1", modelPath: Bundle.main.path(forResource: "llama-3-8b", ofType: "mlc")! ) let response = try await engine.generate( prompt: "写一首关于秋天的诗", maxTokens: 256, temperature: 0.7 ) """ 引擎性能对比（Llama-3-8B INT4） 引擎 平台 首 token 延迟 吞吐 (tok/s) 内存峰值 包大小 llama.cpp Android 800ms 12-18 4.2 GB 3.5 GB MLC-LLM Android 500ms 18-25 4.0 GB 3.6 GB MLC-LLM iOS (Metal) 200ms 30-40 3.8 GB 3.5 GB ExecuTorch Android 600ms 15-22 4.1 GB 3.5 GB ONNX Runtime Android 900ms 10-15 4.5 GB 4.0 GB NPU 加速 现代手机都有 NPU（神经网络处理器），但 LLM 在 NPU 上的支持还很不成熟。 ...

为什么多租户 LLM 架构不同于传统 SaaS 传统 SaaS 多租户解决的是数据隔离问题——数据库加个 tenant_id 列基本就搞定了。LLM SaaS 不一样，它涉及计算资源隔离、模型权重隔离、Prompt 安全、Token 计费四层问题，每一层都有坑。 核心矛盾：LLM 推理极其昂贵，你不可能给每个租户分配独立 GPU，但又不能让大租户的请求挤占小租户的资源。 租户隔离的四个层次 层次一：数据隔离 数据隔离是最基础的要求。用户输入的 Prompt、对话历史、上传的文档，必须严格按租户隔离。 # 租户上下文注入中间件 class TenantContextMiddleware: def __init__(self, db_pool, redis): self.db_pool = db_pool self.redis = redis async def __call__(self, request): tenant_id = request.headers.get("X-Tenant-Id") if not tenant_id: raise HTTPException(401, "Missing tenant context") # 从 Redis 加载租户配置（带缓存） tenant_config = await self.redis.get(f"tenant:{tenant_id}:config") if not tenant_config: tenant_config = await self.load_tenant_config(tenant_id) await self.redis.setex(f"tenant:{tenant_id}:config", 300, tenant_config) request.state.tenant = json.loads(tenant_config) request.state.tenant_id = tenant_id return request async def load_tenant_config(self, tenant_id): async with self.db_pool.acquire() as conn: row = await conn.fetchrow( "SELECT * FROM tenants WHERE id = $1 AND status = 'active'", tenant_id ) if not row: raise HTTPException(403, "Tenant not found or inactive") return json.dumps(dict(row)) 关键原则： 所有数据查询必须强制带 tenant_id 条件，最好在 ORM 层面做硬约束，而不是靠开发者自觉。 ...