架构设计

为什么需要 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） 三个环节优化：检索前 + 检索中 + 检索后。 ...

从单兵作战到团队协作 单个 Agent 能力再强，也有上限：上下文有限、单线程思考、无法并行处理子任务。多 Agent 系统像是从"个人英雄"升级到"特种部队"——每个成员专精一项，协作完成复杂任务。 五种架构模式 模式一：中心化编排（Orchestrator） ┌─────────────┐ │ Orchestrator │ ← 规划、分配、汇总 └──┬──┬──┬──┬─┘ │ │ │ │ ┌──────┘ │ │ └──────┐ ↓ ↓ ↓ ↓ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │Agent│ │Agent│ │Agent│ │Agent│ │ A │ │ B │ │ C │ │ D │ └─────┘ └─────┘ └─────┘ └─────┘ class Orchestrator: def __init__(self): self.agents = { "researcher": ResearchAgent(), "writer": WriterAgent(), "reviewer": ReviewerAgent(), "fact_checker": FactCheckAgent(), } async def run(self, task): # 1. 分解任务 subtasks = self.decompose(task) # 2. 分配给专业 Agent results = {} for subtask in subtasks: agent = self.assign(subtask) results[subtask.id] = await agent.run(subtask) # 3. 汇总结果 return self.synthesize(results) # 适用场景：报告撰写、项目管理 # 优点：控制清晰，易于调试 # 缺点：Orchestrator 是瓶颈和单点故障 模式二：流水线（Pipeline） Input → [Agent A] → [Agent B] → [Agent C] → Output 搜索 写作 审校 class Pipeline: def __init__(self): self.stages = [ ResearchAgent(), # 搜索资料 OutlineAgent(), # 生成大纲 DraftAgent(), # 写初稿 ReviewAgent(), # 审校修改 FactCheckAgent(), # 事实核查 ] async def run(self, input): data = input for stage in self.stages: data = await stage.process(data) if data.rejected: # 回退到上一阶段 data = await self.stages[stage.id - 1].rework(data) return data # 适用场景：内容生产、数据处理 # 优点：简单直观，每个阶段可独立优化 # 缺点：串行执行，延迟高 模式三：辩论（Debate） ┌───────────┐ │ Judge │ └──┬────┬───┘ │ │ ┌──────┘ └──────┐ ↓ ↓ ┌─────────┐ ┌─────────┐ │ Agent A │ ←→ │ Agent B │ │ (正方) │ 辩论 │ (反方) │ └─────────┘ └─────────┘ class DebateSystem: async def run(self, question): pro_agent = Agent(role="supporter") con_agent = Agent(role="opponent") judge = Agent(role="judge") rounds = [] for round_num in range(3): # 3 轮辩论 pro_arg = await pro_agent.argue(question, rounds, side="pro") con_arg = await con_agent.argue(question, rounds, side="con") rounds.append({"pro": pro_arg, "con": con_arg}) verdict = await judge.evaluate(question, rounds) return verdict # 适用场景：决策支持、风险评估 # 优点：减少偏见，多角度分析 # 缺点：Token 消耗是单 Agent 的 3-5 倍 模式四：层级委托（Hierarchical） ┌──────────┐ │ Manager │ └────┬─────┘ ┌────────┼────────┐ ↓ ↓ ↓ ┌──────┐ ┌──────┐ ┌──────┐ │Team A │ │Team B │ │Team C │ │Lead │ │Lead │ │Lead │ └──┬───┘ └──┬───┘ └──┬───┘ ↓ ↓ ↓ Workers Workers Workers class Manager: async def handle(self, task): if self.can_handle(task): return self.do(task) # 委托给子团队 team = self.select_team(task) result = await team.lead.handle(task) return result class TeamLead: async def handle(self, task): subtasks = self.split(task) results = await asyncio.gather(*[ worker.run(t) for t, worker in zip(subtasks, self.workers) ]) return self.merge(results) # 适用场景：复杂项目、大规模任务 # 优点：可扩展性好，模拟人类组织 # 缺点：通信开销大，调试困难 模式五：自由协作（Swarm） ┌────────┐ │ Agent A │←──→┌────────┐ └────────┘ │ Agent B │ ↑ ↓ └────────┘ │ ↑ ↓ ↑ ┌──┴───┐ ┌───┴──┐ │Agent C│←──→│Agent D│ └────────┘ └──────┘ class SwarmMessageBus: """Agent 之间通过消息总线自由通信""" def __init__(self): self.agents = {} self.messages = asyncio.Queue() async def broadcast(self, sender, content): for agent_id, agent in self.agents.items(): if agent_id != sender: await agent.receive({ "from": sender, "content": content, }) async def run(self, task): # 所有 Agent 同时启动，自由协作 await asyncio.gather(*[ agent.start(task, self) for agent in self.agents.values() ]) # 适用场景：开放式探索、创意协作 # 优点：涌现行为，灵活性极高 # 缺点：不可预测，难以控制 通信机制 消息格式 @dataclass class AgentMessage: sender: str # 发送者 ID receiver: str # 接收者 ID 或 "broadcast" type: str # request / response / notify content: str # 消息内容 context: dict # 上下文（任务ID、对话历史等） timestamp: float # 时间戳 priority: int = 0 # 优先级 通信协议 class AgentProtocol: """简化的 Agent 间通信协议""" async def request(self, target, action, params): """请求-响应模式""" msg = AgentMessage( sender=self.id, receiver=target, type="request", content=json.dumps({"action": action, "params": params}), ) response = await self.bus.send_and_wait(msg, timeout=30) return response async def notify(self, target, event): """通知模式（不需要响应）""" msg = AgentMessage( sender=self.id, receiver=target, type="notify", content=json.dumps({"event": event}), ) await self.bus.send(msg) async def broadcast(self, event): """广播模式""" msg = AgentMessage( sender=self.id, receiver="broadcast", type="notify", content=json.dumps({"event": event}), ) await self.bus.send(msg) 冲突处理 当多个 Agent 给出矛盾结果： ...

Agent 记忆系统 2026：让 AI 真正「记住」 2026 年，Agent 记忆系统已经从「概念验证」进化到「生产级架构」。本文系统梳理当前最成熟的 Agent 记忆方案，从架构设计到落地实践。 记忆系统的三层架构 现代 Agent 记忆系统通常分为三层： 第一层：工作记忆（Working Memory） 当前会话的上下文，通常存在 LLM 的上下文窗口中 容量：4K-2M tokens（取决于模型） 特点：高速、容量有限、会话结束即丢失 第二层：情景记忆（Episodic Memory） 历史会话的关键信息，通过向量检索召回 存储：向量数据库（Chroma/Pinecone/Milvus） 容量：无限（受存储成本限制） 特点：语义检索、持久化、跨会话 第三层：长期记忆（Long-term Memory） 结构化的知识和用户偏好，通常以知识图谱形式存储 存储：Neo4j/TigerGraph + 向量数据库 特点：精确检索、关系推理、持续更新 2026 年主流方案对比 方案 记忆类型 检索方式 适用场景 代表项目 MemGPT 架构 分层记忆 函数调用调度 长文档处理 MemGPT/OpenClaw Vector-Only 向量检索 语义相似度 简单问答 Dify/RAGFlow Hybrid Memory 向量+图谱 向量+图查询 复杂推理 Hermes Agent Database Memory 结构化存储 SQL+向量 企业应用 Microsoft Copilot File-based Memory 文件系统 全文搜索 个人 Agent OpenClaw 工程实践：构建一个生产级记忆系统 Step 1：记忆提取 ...

单个 Agent 能力有限，多个 Agent 协作能解决更复杂的问题。本文探讨多智能体协作的核心模式、通信协议和工程挑战。