架构设计

1. 从传统数据库到 AI 原生数据库 传统关系数据库为结构化数据设计，以行/列为基本单位，通过 B+ 树索引加速点查询和范围查询。AI 时代新增了一类核心需求：向量检索——在高维空间中寻找与查询向量最相似的 Top-K 条记录。 AI 原生数据库（AI-Native Database）将向量检索与结构化查询深度融合，支持： SELECT * FROM articles WHERE category = 'AI' ORDER BY embedding <-> query_vector LIMIT 10 语义相似性过滤、混合排序、元数据 + 向量联合查询 代表系统：Pinecone、Weaviate、Chroma、Qdrant、Milvus、pgvector（PostgreSQL 扩展）、StarRocks（向量化）、ClickHouse（向量化）。 2. 存储引擎架构 2.1 整体架构 ┌──────────────────────────────────────────────────┐ │ SQL / GraphQL / REST API │ ├──────────────────────────────────────────────────┤ │ Query Optimizer & Planner │ │ ┌─────────────┐ ┌─────────────┐ ┌────────┐ │ │ │Scalar Filter│ │ Vector Index│ │Sort/Max│ │ │ │ Planner │ │ Planner │ │ Planner│ │ │ └─────────────┘ └─────────────┘ └────────┘ │ ├──────────────────────────────────────────────────┤ │ Execution Engine (Vectorized) │ ├──────────────────────────────────────────────────┤ │ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │ │ Vector │ │ Row │ │ Column │ │ │ │ Index │ │ Store │ │ Store │ │ │ │(HNSW/IVF)│ │ (Pilosa) │ │(Arrow/Parquet)│ │ │ └──────────┘ └──────────┘ └──────────────┘ │ ├──────────────────────────────────────────────────┤ │ Distributed Storage Layer │ │ ┌────────────────────────────────────────────┐ │ │ │ S3 / HDFS / Local SSD / Remote Memory │ │ │ └────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────┘ 2.2 向量存储格式 from dataclasses import dataclass, field from typing import Optional import numpy as np @dataclass class VectorRecord: id: str vector: np.ndarray # 原始向量（float32/float16) metadata: dict = field(default_factory=dict) version: int = 0 # MVCC 版本号 deleted: bool = False def to_bytes(self) -> bytes: """序列化：4字节维度 + 原始向量 + 元数据JSON + 版本""" dim = len(self.vector) vec_bytes = self.vector.astype(np.float32).tobytes() meta_bytes = json.dumps(self.metadata).encode("utf-8") header = struct.pack("<i", dim) version_bytes = struct.pack("<i", self.version) deleted_byte = b"\x01" if self.deleted else b"\x00" return header + version_bytes + deleted_byte + vec_bytes + meta_bytes @classmethod def from_bytes(cls, data: bytes) -> "VectorRecord": dim, version = struct.unpack("<ii", data[:8]) deleted = data[8] == 1 vec_bytes = data[9:9 + dim * 4] meta_bytes = data[9 + dim * 4:].decode("utf-8") return cls( id="", # id 需要从索引层获取 vector=np.frombuffer(vec_bytes, dtype=np.float32), metadata=json.loads(meta_bytes), version=version, deleted=deleted ) class VectorPage: """向量分页存储单元""" def __init__(self, page_size: int = 4096): self.page_size = page_size self.records: dict[str, bytes] = {} # id -> serialized bytes self.current_size = 0 def add(self, record: VectorRecord) -> bool: record_bytes = record.to_bytes() if self.current_size + len(record_bytes) > self.page_size: return False self.records[record.id] = record_bytes self.current_size += len(record_bytes) return True 3. 向量索引算法 3.1 HNSW（Hierarchical Navigable Small World） HNSW 是当前最流行的向量索引算法，在召回率和延迟之间取得优秀平衡。 ...

1. 为什么 LLM 需要缓存 LLM 调用成本高、延迟高。以 GPT-4o 为例： 延迟：单次请求 500ms - 3000ms 成本：输入 $2.5/1M tokens，输出 $10/1M tokens 并发限制：RPM（每分钟请求数）和 TPM（每分钟 tokens）受限 缓存可以显著降低延迟和成本。但 LLM 缓存不同于传统缓存：用户问题"今天天气"和"天气怎么样"语义相近，应该命中同一缓存条目——这就是语义缓存的核心价值。 2. 语义缓存原理 2.1 相似度计算 from dataclasses import dataclass import numpy as np @dataclass class CacheEntry: query: str query_embedding: np.ndarray response: str model: str tokens_used: int timestamp: float hit_count: int = 0 ttl_seconds: int = 3600 class SemanticCache: """ 语义缓存：基于向量相似度的缓存命中 核心思想：如果新查询与已缓存查询的语义距离 < 阈值，则复用缓存结果 """ def __init__(self, embedding_model, similarity_threshold: float = 0.95): self.embedding_model = embedding_model self.threshold = similarity_threshold self.cache: list[CacheEntry] = [] self.index = None # 可选：FAISS/Annoy 加速 async def get(self, query: str) -> tuple[bool, str | None]: query_embedding = await self.embedding_model.embed(query) # 线性搜索（生产环境用向量索引） for entry in self.cache: similarity = self._cosine_similarity(query_embedding, entry.query_embedding) if similarity >= self.threshold: entry.hit_count += 1 return True, entry.response return False, None async def set(self, query: str, response: str, model: str, tokens: int): embedding = await self.embedding_model.embed(query) entry = CacheEntry( query=query, query_embedding=embedding, response=response, model=model, tokens_used=tokens, timestamp=time.time() ) self.cache.append(entry) def _cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float: return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))) 2.2 阈值调优 import matplotlib.pyplot as plt class ThresholdOptimizer: """语义缓存阈值优化器""" def __init__(self, cache: SemanticCache, test_data: list[tuple[str, str, bool]]): """ test_data: [(query1, cached_query, should_hit), ...] """ self.cache = cache self.test_data = test_data def evaluate(self, threshold: float) -> dict: self.cache.threshold = threshold true_positives = 0 false_positives = 0 false_negatives = 0 true_negatives = 0 for query, cached, should_hit in self.test_data: # 手动计算相似度 e1 = await self.cache.embedding_model.embed(query) e2 = await self.cache.embedding_model.embed(cached) sim = self.cache._cosine_similarity(e1, e2) is_hit = sim >= threshold if should_hit and is_hit: true_positives += 1 elif not should_hit and is_hit: false_positives += 1 elif should_hit and not is_hit: false_negatives += 1 else: true_negatives += 1 precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0 recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0 return {"threshold": threshold, "precision": precision, "recall": recall, "f1": f1} def find_optimal_threshold(self) -> float: thresholds = np.arange(0.80, 1.0, 0.01) results = [self.evaluate(t) for t in thresholds] best = max(results, key=lambda x: x["f1"]) return best["threshold"] 推荐阈值： ...

1. 为什么需要 LLM 网关 当企业接入多个 LLM 供应商（OpenAI、Anthropic、Google、本地模型等），直接调用会面临： 供应商锁定：切换模型需要修改所有调用方代码 成本失控：无法统一监控和控制 token 消耗 可靠性差：单供应商宕机即服务不可用 安全风险：API Key 分散在各处，难以统一管理 LLM 网关作为统一接入层，解决以上所有问题。 2. 整体架构 ┌─────────────────────────────────────────────────┐ │ LLM Gateway │ │ │ │ ┌─────────┐ ┌──────────┐ ┌───────────────┐ │ │ │ Router │ │ Load │ │ Rate │ │ │ │ Engine │→ │ Balancer │→ │ Limiter │ │ │ └─────────┘ └──────────┘ └───────────────┘ │ │ │ │ │ ┌────┴────────────────────────────────────┐ │ │ │ Model Adapter Pool │ │ │ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │ │ │OpenAI│ │Claude│ │Gemini│ │Local │ │ │ │ │ └──────┘ └──────┘ └──────┘ └──────┘ │ │ │ └─────────────────────────────────────────┘ │ │ │ │ ┌──────────┐ ┌──────────┐ ┌────────────┐ │ │ │ Cache │ │ Metrics │ │ Audit │ │ │ │ Layer │ │ Collector│ │ Logger │ │ │ └──────────┘ └──────────┘ └────────────┘ │ └─────────────────────────────────────────────────┘ 3. 多模型路由引擎 3.1 路由策略 from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Optional import hashlib import time @dataclass class LLMRequest: model: str messages: list[dict] temperature: float = 0.7 max_tokens: int = 2048 stream: bool = False metadata: dict = field(default_factory=dict) @dataclass class ModelEndpoint: provider: str # openai, anthropic, local model_id: str # gpt-4o, claude-3.5-sonnet endpoint_url: str api_key: str max_concurrent: int = 10 cost_per_1k_input: float = 0.0 cost_per_1k_output: float = 0.0 avg_latency_ms: float = 500 availability: float = 99.9 current_load: int = 0 class RoutingStrategy(ABC): @abstractmethod def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: pass class CostOptimizedRouting(RoutingStrategy): """最低成本优先路由""" def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") return min(available, key=lambda c: c.cost_per_1k_input) class LatencyOptimizedRouting(RoutingStrategy): """最低延迟优先路由""" def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") return min(available, key=lambda c: c.avg_latency_ms) class WeightedRoundRobinRouting(RoutingStrategy): """加权轮询路由""" def __init__(self): self._index = 0 self._weights: dict[str, int] = {} def set_weight(self, model_id: str, weight: int): self._weights[model_id] = weight def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") # 构建加权列表 weighted = [] for c in available: w = self._weights.get(c.model_id, 1) weighted.extend([c] * w) selected = weighted[self._index % len(weighted)] self._index += 1 return selected class CapabilityBasedRouting(RoutingStrategy): """基于任务能力的智能路由""" CAPABILITY_MAP = { "code_generation": ["gpt-4o", "claude-3.5-sonnet", "deepseek-coder"], "long_context": ["gemini-1.5-pro", "claude-3.5-sonnet"], "vision": ["gpt-4o", "gemini-1.5-pro"], "low_cost": ["gpt-4o-mini", "claude-3-haiku", "gemini-1.5-flash"], } def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: required_cap = request.metadata.get("capability") if not required_cap: return candidates[0] preferred_models = self.CAPABILITY_MAP.get(required_cap, []) for model_id in preferred_models: for c in candidates: if c.model_id == model_id and c.current_load < c.max_concurrent: return c # 降级：选任意可用的 return next((c for c in candidates if c.current_load < c.max_concurrent), candidates[0]) 3.2 路由策略对比 策略 优势 劣势 适用场景 成本优先 最低开销 可能高延迟 批量处理 延迟优先 响应快 成本高 实时对话 加权轮询 负载均衡 不考虑成本 通用场景 能力路由 任务匹配最优 需维护映射 混合任务 4. 限流与降级 4.1 多维度限流 from collections import defaultdict, deque import time class MultiDimensionRateLimiter: def __init__(self): # 按用户限流 self.user_limits: dict[str, tuple[int, float]] = {} # user -> (max_rpm, window_s) self.user_counters: dict[str, deque] = defaultdict(deque) # 按模型限流 self.model_limits: dict[str, tuple[int, float]] = {} self.model_counters: dict[str, deque] = defaultdict(deque) # 全局限流 self.global_limit = (1000, 60) # 1000 rpm self.global_counter: deque = deque() def check(self, user_id: str, model: str) -> bool: now = time.time() if not self._check_counter(self.global_counter, self.global_limit, now): return False if user_id in self.user_limits: if not self._check_counter(self.user_counters[user_id], self.user_limits[user_id], now): return False if model in self.model_limits: if not self._check_counter(self.model_counters[model], self.model_limits[model], now): return False return True def _check_counter(self, counter: deque, limit: tuple, now: float) -> bool: max_count, window = limit while counter and counter[0] < now - window: counter.popleft() if len(counter) >= max_count: return False counter.append(now) return True 4.2 熔断降级链 class FallbackChain: """模型降级链：主模型失败时自动切换到备选模型""" def __init__(self): self.chains: dict[str, list[str]] = { "gpt-4o": ["claude-3.5-sonnet", "gemini-1.5-pro", "gpt-4o-mini"], "claude-3.5-sonnet": ["gpt-4o", "gemini-1.5-pro"], } async def execute_with_fallback(self, gateway, request: LLMRequest) -> dict: primary = request.model chain = [primary] + self.chains.get(primary, []) last_error = None for model_id in chain: try: request.model = model_id result = await gateway.forward(request) return result except Exception as e: last_error = e continue raise last_error 5. 统一适配器层 class ModelAdapter(ABC): @abstractmethod async def chat(self, request: LLMRequest) -> dict: pass @abstractmethod async def stream_chat(self, request: LLMRequest): pass @abstractmethod def count_tokens(self, messages: list[dict]) -> int: pass class OpenAIAdapter(ModelAdapter): def __init__(self, endpoint: ModelEndpoint): self.endpoint = endpoint self.client = httpx.AsyncClient( base_url=endpoint.endpoint_url, headers={"Authorization": f"Bearer {endpoint.api_key}"}, timeout=120 ) async def chat(self, request: LLMRequest) -> dict: resp = await self.client.post("/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "temperature": request.temperature, "max_tokens": request.max_tokens, }) resp.raise_for_status() data = resp.json() return { "content": data["choices"][0]["message"]["content"], "model": data["model"], "usage": data["usage"], } async def stream_chat(self, request: LLMRequest): async with self.client.stream("POST", "/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "stream": True, }) as resp: async for line in resp.aiter_lines(): if line.startswith("data: "): yield line[6:] def count_tokens(self, messages: list[dict]) -> int: # tiktoken 计算 import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") total = sum(len(enc.encode(m["content"])) for m in messages) return total class AnthropicAdapter(ModelAdapter): async def chat(self, request: LLMRequest) -> dict: # 适配 Anthropic API 格式 system = next((m["content"] for m in request.messages if m["role"] == "system"), "") user_messages = [m for m in request.messages if m["role"] != "system"] resp = await self.client.post("/v1/messages", json={ "model": request.model, "system": system, "messages": user_messages, "max_tokens": request.max_tokens, }) # 转换为统一格式 ... class LocalModelAdapter(ModelAdapter): """本地 vLLM/Ollama 适配器""" async def chat(self, request: LLMRequest) -> dict: resp = await self.client.post("/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "temperature": request.temperature, }) ... 6. 成本控制与计量 @dataclass class UsageRecord: user_id: str model: str input_tokens: int output_tokens: int cost: float timestamp: float request_id: str class CostTracker: def __init__(self, redis_client): self.redis = redis_client async def record(self, record: UsageRecord): pipe = self.redis.pipeline() # 按用户累计 pipe.hincrby(f"cost:user:{record.user_id}:daily:{date.today()}", "total", int(record.cost * 10000)) pipe.hincrby(f"cost:user:{record.user_id}:monthly:{date.today().strftime('%Y-%m')}", "total", int(record.cost * 10000)) # 按模型累计 pipe.hincrby(f"cost:model:{record.model}:daily:{date.today()}", "total", int(record.cost * 10000)) # 预算检查 pipe.hget(f"budget:user:{record.user_id}:monthly", "limit") results = await pipe.execute() budget_limit = results[-1] if budget_limit and results[1] > int(budget_limit): raise BudgetExceededError(f"User {record.user_id} exceeded monthly budget") async def get_usage_report(self, user_id: str, period: str = "daily") -> dict: key = f"cost:user:{user_id}:{period}:{date.today()}" data = await self.redis.hgetall(key) return {"user": user_id, "period": period, "cost_cents": int(data.get("total", 0)) / 10000} 7. 可观测性 class GatewayMetrics: def __init__(self): self.request_count = Counter("gateway_requests_total", ["model", "status"]) self.latency = Histogram("gateway_latency_seconds", ["model"]) self.token_usage = Counter("gateway_tokens_total", ["model", "direction"]) self.cost = Counter("gateway_cost_total", ["model"]) self.active_connections = Gauge("gateway_active_connections") def record_request(self, model: str, status: str, duration: float, input_tokens: int, output_tokens: int, cost: float): self.request_count.labels(model=model, status=status).inc() self.latency.labels(model=model).observe(duration) self.token_usage.labels(model=model, direction="input").inc(input_tokens) self.token_usage.labels(model=model, direction="output").inc(output_tokens) self.cost.labels(model=model).inc(cost) 8. 部署架构 ┌─────────┐ │ CDN │ └────┬────┘ │ ┌──────────┼──────────┐ │ │ │ ┌────┴───┐ ┌───┴────┐ ┌──┴─────┐ │GW #1 │ │GW #2 │ │GW #3 │ │(active)│ │(active)│ │(active)│ └────┬───┘ └───┬────┘ └──┬─────┘ │ │ │ ┌────┴─────────┴─────────┴────┐ │ Redis Cluster │ │ (rate limits, cache, cost) │ └──────────────────────────────┘ │ │ ┌────┴────┐ ┌────┴────┐ │ Model A │ │ Model B │ │ Provider│ │ Provider│ └─────────┘ └─────────┘ 9. 总结 LLM 网关是企业级 AI 应用的必备基础设施。核心设计要点： ...

1. 引言：为什么需要多智能体编排 单个 LLM Agent 在复杂任务中面临上下文窗口限制、角色混淆、推理链断裂等问题。多智能体架构通过任务分解、角色专精和协作机制，将复杂问题分配给多个专业化 Agent 协同完成。然而，如何编排这些 Agent——谁来调度、如何通信、何时同步——是工程落地的核心挑战。 2. 三种核心编排模式 2.1 中心化编排（Orchestrator Pattern） 一个中央编排器（Orchestrator）负责任务分配、状态管理和结果汇总。所有 Agent 只与编排器通信，互不直接交互。 ┌──────────────────────────────────┐ │ Orchestrator │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │Worker│ │Worker│ │Worker│ │ │ │ A │ │ B │ │ C │ │ │ └──────┘ └──────┘ └──────┘ │ └──────────────────────────────────┘ 核心代码实现： from abc import ABC, abstractmethod from typing import Any, Optional import asyncio class Agent(ABC): def __init__(self, name: str, system_prompt: str): self.name = name self.system_prompt = system_prompt self.message_history: list[dict] = [] @abstractmethod async def execute(self, task: str, context: dict) -> str: pass class Orchestrator: def __init__(self): self.agents: dict[str, Agent] = {} self.task_queue: list[dict] = [] self.results: dict[str, Any] = {} def register(self, agent: Agent): self.agents[agent.name] = agent async def dispatch(self, agent_name: str, task: str, context: dict = None) -> str: agent = self.agents[agent_name] result = await agent.execute(task, context or {}) self.results[f"{agent_name}:{task[:20]}"] = result return result async def run_pipeline(self, plan: list[dict]) -> dict: """按计划顺序执行任务，支持依赖传递""" for step in plan: agent_name = step["agent"] task = step["task"] deps = step.get("depends_on", []) merged_context = {d: self.results.get(d) for d in deps} await self.dispatch(agent_name, task, merged_context) return self.results 适用场景： 工作流明确的任务（如代码审查流水线、文档生成管线） ...

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 生产推荐： ...