架构设计

为什么需要多区域部署？ 单区域部署的延迟：美国用户访问亚洲 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 层面做硬约束，而不是靠开发者自觉。 ...

为什么需要 Agent 编排 单个 Agent 能力有限。复杂任务需要多个 Agent 协作：一个负责检索、一个负责分析、一个负责生成报告。如何编排这些 Agent 是生产环境的核心问题。 编排模式分类 模式 结构 适用场景 复杂度 串行 A -> B -> C 流水线任务 低 并行 A,B,C -> 合并 独立子任务 低 循环 A -> B -> 判断 -> (A 或 结束) 迭代优化 中 路由 Input -> Router -> [A/B/C] 分类分发 中 监督者 Supervisor -> [A,B,C] 中心调度 中 分层 Top Supervisor -> Sub-Supervisors -> Workers 大规模团队 高 图式 DAG/状态机 复杂工作流 高 1. Router Pattern 一个路由 Agent 根据输入决定调用哪个专家 Agent： ...

超越朴素 RAG 朴素 RAG 的流程是：用户提问 -> 检索 top-K 文档 -> 拼接到 Prompt -> LLM 生成。这对简单问题够用，但面对复杂问题力不从心。本文介绍 6 种高级 RAG 架构模式。 模式一览 模式 核心思想 适用场景 复杂度 多跳检索 分步检索，每步基于前步结果 多跳推理问题 中 递归检索 先粗后细，分层检索 层级结构文档 中 自适应检索 LLM 决定是否检索/检索什么 不确定是否需要检索 高 Self-RAG LLM 自我反思检索质量 高质量要求 高 CRAG 检索质量评估+Web兜底 知识库覆盖不全 高 模块化 RAG 可插拔模块组合 生产环境灵活部署 高 1. 多跳检索（Multi-Hop Retrieval） 面对"张三所在公司的CEO毕业于哪所大学？“这种问题，单次检索无法回答。需要： Hop 1: 检索"张三" -> 得到所在公司"ABC公司" Hop 2: 检索"ABC公司 CEO" -> 得到"李四" Hop 3: 检索"李四 教育背景" -> 得到"清华大学" class MultiHopRetriever: def __init__(self, llm_client, vector_store): self.llm = llm_client self.store = vector_store self.max_hops = 4 async def retrieve(self, query: str) -> list[dict]: all_docs = [] current_query = query for hop in range(self.max_hops): docs = await self.store.search(current_query, top_k=3) all_docs.extend(docs) prompt = f"""问题: {query} 已知信息: {self._format_docs(all_docs)} 如果已有足够信息回答问题，回复"SUFFICIENT"。 如果需要更多信息，生成下一个检索查询（只输出查询）。""" resp = await self.llm.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0 ) next_action = resp.choices[0].message.content.strip() if next_action == "SUFFICIENT": break current_query = next_action return all_docs def _format_docs(self, docs): return "\n".join(f"[{i}] {d['content']}" for i, d in enumerate(docs)) 2. 递归检索（Recursive Retrieval） 适用于层级结构文档。先检索章节级别（粗粒度），再在选中章节内检索段落级别（细粒度）： ...

扩展性挑战：为什么单机 Agent 走不远 单个 Agent 进程能处理的并发请求受限于 CPU、内存和网络连接数。一个典型的 LLM Agent 单次推理消耗 2-8GB 显存，单卡 GPU 同时只能服务少数几个请求。当 QPS 从个位数涨到百级、千级时，单机方案必然崩溃。 扩展 Agent 系统的核心挑战不是简单的"加机器"，而是： 挑战 描述 影响 状态管理 Agent 会话上下文在哪台机器？ 直接影响会话连续性 GPU 资源 LLM 推理是计算密集型 扩展瓶颈在 GPU 而非 CPU 长连接 流式输出需要持久连接 连接亲和性限制负载均衡 一致性 多 Agent 实例的内存状态同步 分布式共识开销 尾延迟 少数慢请求拖垮整体 P99 需要超时和降级机制 水平扩展策略 1. 无状态 Agent：将状态外部化 扩展的第一原则：Agent 进程本身不持有任何会话状态。所有状态存储在外部系统。 from dataclasses import dataclass, field from typing import Optional, Any import json # ❌ 错误做法：状态在内存中 class StatefulAgent: def __init__(self): self.conversations: dict[str, list] = {} # 内存中的会话状态 def chat(self, session_id: str, message: str) -> str: history = self.conversations[session_id] # 只在本机有效 history.append({"role": "user", "content": message}) # ...处理... return "response" # ✅ 正确做法：状态外部化 class StatelessAgent: """无状态 Agent，所有会话状态存入 Redis""" def __init__(self, state_store: 'SessionStore'): self.store = state_store def chat(self, session_id: str, message: str) -> str: # 每次从外部存储加载会话状态 history = self.store.get_history(session_id) history.append({"role": "user", "content": message}) # 处理逻辑 response = self._generate(history) history.append({"role": "assistant", "content": response}) # 持久化回外部存储 self.store.save_history(session_id, history) return response class SessionStore: """基于 Redis 的会话存储""" def __init__(self, redis_client): self.redis = redis_client self.ttl = 3600 # 会话过期时间 1 小时 def get_history(self, session_id: str) -> list[dict]: data = self.redis.get(f"session:{session_id}") return json.loads(data) if data else [] def save_history(self, session_id: str, history: list[dict]): self.redis.setex( f"session:{session_id}", self.ttl, json.dumps(history, ensure_ascii=False) ) def extend_ttl(self, session_id: str): self.redis.expire(f"session:{session_id}", self.ttl) 2. 分层架构：接入层 + 推理层 + 知识层 """ 三层架构设计： - 接入层：处理 HTTP/WebSocket 连接，会话路由 - 推理层：LLM 推理，可水平扩展 - 知识层：向量数据库、知识图谱 """ from abc import ABC, abstractmethod class GatewayLayer: """接入层：路由请求到推理节点""" def __init__(self, inference_pool: 'InferencePool'): self.pool = inference_pool async def handle_request(self, session_id: str, message: str): # 根据会话亲和性选择节点 node = self.pool.select_node(session_id) return await node.process(session_id, message) class InferenceNode: """推理层：单个推理节点""" def __init__(self, node_id: str, model_path: str): self.node_id = node_id self.model = self._load_model(model_path) self.active_sessions = 0 self.max_sessions = 10 async def process(self, session_id: str, message: str) -> str: self.active_sessions += 1 try: # 推理逻辑 result = await self._infer(message) return result finally: self.active_sessions -= 1 @property def available(self) -> bool: return self.active_sessions < self.max_sessions class InferencePool: """推理节点池""" def __init__(self): self.nodes: list[InferenceNode] = [] def add_node(self, node: InferenceNode): self.nodes.append(node) def select_node(self, session_id: str = None) -> InferenceNode: # 优先选择可用且负载最低的节点 available = [n for n in self.nodes if n.available] if not available: raise RuntimeError("No available inference nodes") return min(available, key=lambda n: n.active_sessions) 3. 推理与工具调用分离 LLM 推理和工具执行有不同的资源特征：推理需要 GPU，工具调用主要消耗 CPU 和网络 I/O。将它们分离到不同服务。 ...

实时需求场景 用户对 AI Agent 的期望正在从"能用"变成"好用"。好用的核心指标之一是响应速度——用户发送消息后多久能看到第一个字。 延迟感知分级 场景 可接受首字延迟 可接受完整响应时间 技术要求 闲聊对话 <500ms <3s 流式输出 + 缓存 知识问答 <1s <5s RAG + 流式推理 代码助手 <1s <10s 流式 + 增量渲染 语音助手 <200ms <2s 流式 ASR/TTS + 边缘部署 实时翻译 <300ms <1s 超低延迟管道 金融监控告警 <100ms <500ms 事件触发 + 预计算 传统请求-响应模式下，用户需要等待 Agent 完成全部推理后才能看到结果。一个需要 5 秒推理的回复，用户体验就是"等 5 秒然后突然出一大段文字"。流式输出改变了这个体验——首字延迟降到几百毫秒，后续内容逐步呈现。 WebSocket/SSE 流式输出 SSE vs WebSocket 选型 特性 SSE (Server-Sent Events) WebSocket 通信方向 服务器 → 客户端（单向） 双向 协议 HTTP 独立协议 自动重连 浏览器原生支持 需手动实现 代理兼容性 好（基于 HTTP） 部分代理不支持 适用场景 流式文本输出 需要客户端实时输入（语音） 对于大多数 Agent 场景，SSE 足够且更简单。语音助手等双向实时场景用 WebSocket。 ...

为什么 Agent 需要记忆 LLM 是无状态的——每次调用都是全新的。但真正的 Agent 需要记住： 5 分钟前用户说了什么（工作记忆） 昨天的对话中用户提到了什么偏好（情景记忆） 用户公司的 API 文档（语义记忆） 上周犯的错误，不要再犯（反思记忆） 记忆类型与人类类比 记忆类型 人类类比 Agent 实现 持续时间 工作记忆 短期记忆 上下文窗口 单次对话 情景记忆 个人经历 对话历史存储 天-月 语义记忆 知识事实 向量知识库 永久 程序记忆 技能习惯 工具使用模式 永久 反思记忆 经验教训 总结+规则 永久 工作记忆管理 上下文窗口策略 class WorkingMemory: def __init__(self, max_tokens=8000): self.max_tokens = max_tokens self.messages = [] def add(self, message): self.messages.append(message) self._compress_if_needed() def _compress_if_needed(self): while self.token_count() > self.max_tokens: if len(self.messages) <= 2: break # 取出最早的几条消息做摘要 old_messages = self.messages[:3] summary = self.summarize(old_messages) self.messages = [ {"role": "system", "content": f"之前的对话摘要：{summary}"} ] + self.messages[3:] 注意力机制 class AttentionManager: """管理上下文中各部分的"重要性"""" def __init__(self): self.attention_scores = {} def update(self, message_id, content, user_query): """计算每条消息对当前问题的相关性""" self.attention_scores[message_id] = { "recency": self.recency_score(message_id), "relevance": self.relevance_score(content, user_query), "importance": self.importance_score(content), } def get_context(self, max_tokens=4000): """按注意力分数排序，选择最重要的上下文""" ranked = sorted( self.attention_scores.items(), key=lambda x: x[1]["recency"] * 0.2 + x[1]["relevance"] * 0.5 + x[1]["importance"] * 0.3, reverse=True ) context = [] token_count = 0 for msg_id, _ in ranked: msg = self.get_message(msg_id) token_count += count_tokens(msg) if token_count > max_tokens: break context.append(msg) return context 情景记忆：对话历史 存储结构 from dataclasses import dataclass from datetime import datetime @dataclass class EpisodicMemory: id: str session_id: str timestamp: datetime user_input: str agent_response: str tools_used: list[str] outcome: str # success / failure / partial embedding: list[float] # 用于检索 metadata: dict # 用户信息、环境等 检索策略 class EpisodicMemoryStore: async def recall(self, query, user_id, limit=5): """回忆与当前情况相关的历史对话""" # 时间衰减权重：越近的记忆权重越高 recency_weight = 0.3 # 语义相似度 query_emb = self.embedder.embed(query) results = await self.db.search( user_id=user_id, embedding=query_emb, limit=limit * 3, # 多取一些做重排 ) # 综合排序 for r in results: days_ago = (datetime.now() - r.timestamp).days r.score = r.similarity * (1 - recency_weight) + \ math.exp(-days_ago / 30) * recency_weight results.sort(key=lambda x: x.score, reverse=True) return results[:limit] 记忆巩固 class MemoryConsolidation: """将情景记忆转化为语义记忆（类似人类睡眠时的记忆巩固）""" async def consolidate(self, user_id): # 1. 提取最近7天的对话 recent = await self.episodic_store.get_recent(user_id, days=7) # 2. 用 LLM 提取可复用的知识 knowledge = await self.llm.extract( f"从以下对话中提取可复用的知识和模式：\n{recent}" ) # 3. 存入语义记忆 for item in knowledge: await self.semantic_store.add( content=item.content, embedding=self.embedder.embed(item.content), source="consolidated", confidence=item.confidence, ) # 4. 标记已巩固的情景记忆 await self.episodic_store.mark_consolidated(recent) 语义记忆：知识库 分层知识存储 class SemanticMemory: """三层知识存储""" def __init__(self): # 第一层：快速事实（Key-Value） self.facts = RedisStore() # "用户公司" → "ABC科技" # 第二层：文档知识（向量检索） self.documents = VectorStore() # API文档、产品手册 # 第三层：深层知识（知识图谱） self.graph = GraphStore() # 实体关系 知识更新 class KnowledgeUpdater: async def learn(self, new_info, source="conversation"): """学习新知识""" # 1. 检查是否与已有知识冲突 conflicts = await self.find_conflicts(new_info) if conflicts: # 2. 冲突解决：用更新的信息覆盖 for conflict in conflicts: if conflict.confidence < new_info.confidence: await self.update(conflict.id, new_info) else: # 不确定，存为"待确认" await self.flag_for_review(new_info, conflict) else: # 3. 新知识，直接存入 await self.semantic_store.add(new_info) 反思记忆：经验学习 class ReflectionSystem: """Agent 从失败中学习""" async def reflect(self, task, attempt, outcome): if outcome.success: return # 成功不需要反思 # 分析失败原因 analysis = await self.llm.analyze( task=task, actions=attempt.actions, error=outcome.error, prompt="分析为什么这次任务失败了，如何避免" ) # 提取规则 rules = self.extract_rules(analysis) # 存入反思记忆 for rule in rules: await self.reflection_store.add({ "situation": rule.situation, "mistake": rule.mistake, "better_action": rule.better_action, "confidence": rule.confidence, }) async def consult(self, task): """执行任务前，查阅反思记忆""" relevant_mistakes = await self.reflection_store.search(task) if relevant_mistakes: return f""" 注意：根据历史经验，类似任务有以下陷阱： {format_mistakes(relevant_mistakes)} 请避免这些错误。 """ return None 记忆架构实战 class AgentMemory: """完整的 Agent 记忆系统""" def __init__(self): self.working = WorkingMemory(max_tokens=8000) self.episodic = EpisodicMemoryStore() self.semantic = SemanticMemory() self.reflection = ReflectionSystem() async def prepare_context(self, query, user_id): """为 LLM 调用准备完整上下文""" context_parts = [] # 1. 工作记忆（当前对话） context_parts.append(self.working.get_recent(n=6)) # 2. 相关情景记忆 episodes = await self.episodic.recall(query, user_id, limit=3) if episodes: context_parts.append(f"历史相关对话：{format_episodes(episodes)}") # 3. 语义知识 knowledge = await self.semantic.search(query, limit=5) if knowledge: context_parts.append(f"相关知识：{format_knowledge(knowledge)}") # 4. 反思提醒 warning = await self.reflection.consult(query) if warning: context_parts.append(warning) return "\n\n".join(context_parts) 记忆数据库选型 记忆类型 推荐数据库 理由 工作记忆 Redis 低延迟，自动过期 情景记忆 PostgreSQL + pgvector 事务+向量检索 语义记忆 Qdrant / Milvus 高性能向量检索 反思记忆 MongoDB 灵活 Schema 知识图谱 Neo4j / NebulaGraph 图查询 记忆隐私 class MemoryPrivacy: """记忆系统中的隐私保护""" PII_PATTERNS = [ (r'\d{11}', 'PHONE'), # 手机号 (r'\d{18}', 'ID_CARD'), # 身份证 (r'[\w.]+@[\w.]+', 'EMAIL'), # 邮箱 ] def sanitize(self, memory): """存储前脱敏""" for pattern, pii_type in self.PII_PATTERNS: memory = re.sub(pattern, f'[{pii_type}]', memory) return memory def user_forget(self, user_id): """用户要求删除所有记忆""" await self.episodic.delete_by_user(user_id) await self.semantic.delete_by_user(user_id) await self.reflection.delete_by_user(user_id) # 生成删除确认报告 return {"deleted": True, "user_id": user_id} 结论 记忆是 Agent 从"工具"变成"伙伴"的关键。没有记忆的 Agent 每次都是陌生人，有了记忆它才能： ...

RAG 的三年代演进 RAG（Retrieval-Augmented Generation）从 2023 年的朴素方案到 2026 年的模块化架构，经历了三个阶段。每个阶段解决上一阶段的核心瓶颈。 第一阶段：朴素 RAG（Naive RAG） 用户问题 → 嵌入 → 向量搜索 → Top-K 文档 → LLM 生成答案 class NaiveRAG: def __init__(self): self.embedder = OpenAIEmbeddings() self.vectorstore = Chroma() self.llm = ChatOpenAI(model="gpt-4o") def query(self, question): # 1. 嵌入问题 q_emb = self.embedder.embed(question) # 2. 向量搜索 docs = self.vectorstore.similarity_search(q_emb, k=4) # 3. 拼接 Prompt context = "\n".join(d.page_content for d in docs) prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{question}" # 4. LLM 生成 return self.llm.invoke(prompt) 朴素 RAG 的问题 问题 原因 影响 检索不相关 语义相似 ≠ 相关 答案跑偏 丢失上下文 Chunk 切割破坏段落 信息残缺 无法回答多跳问题 单次检索不够 复杂问题失败 幻觉仍存在 LLM 无视检索结果 答案不可信 第二阶段：进阶 RAG（Advanced RAG） 三个环节优化：检索前 + 检索中 + 检索后。 ...