架构设计

为什么Agent需要记忆？ 传统大语言模型的上下文窗口本质上是"短期工作记忆"，一旦对话超出窗口长度，早期信息就会丢失。而真正的智能体需要在跨会话、跨任务的长周期运作中保持上下文连贯性——这就要求我们构建结构化的记忆系统。 记忆系统的三层架构 1. 工作记忆（Working Memory） 工作记忆对应Agent当前正在处理的上下文，通常映射为LLM的context window。工程实现上需要注意以下几点： 动态裁剪策略：不是简单截断，而是基于注意力权重或摘要压缩来保留关键信息 滑动窗口+摘要：将溢出的历史对话通过小模型生成摘要，注入到新窗口头部 多模态工作区：除文本外，缓存当前任务相关的图片、表格、代码片段 一个典型实现是使用Redis作为热数据存储，配合embedding检索，将相关历史片段动态注入context。 2. 情景记忆（Episodic Memory） 情景记忆记录Agent过去的交互经历，包括时间、场景、行为和结果。实现方式通常是向量数据库存储： 记忆条目结构: { "id": "ep_001", "timestamp": "2026-07-16T10:00:00Z", "task": "数据分析任务", "actions": ["读取CSV", "生成图表", "撰写报告"], "outcome": "成功生成季度销售分析报告", "embedding": [0.12, -0.34, ...], "metadata": {"user_id": "u123", "session_id": "s456"} } 检索时，将当前任务描述编码为向量，在数据库中做kNN查询，召回最相关的N条历史经历。 3. 语义记忆（Semantic Memory） 语义记忆是Agent的"知识库"，存储事实、概念和规则。与RAG系统类似，但更强调知识的结构化组织和增量更新： 本体图谱：使用知识图谱存储实体关系 文档库：向量索引+全文检索双路召回 技能库：Agent掌握的工具调用模式和Prompt模板 记忆的写入与遗忘 记忆系统最关键的不是存储，而是遗忘机制。人脑会自动遗忘无关信息以保持认知效率，Agent同样需要： 衰减权重：每条记忆有一个decay score，随时间递减，长期未被检索的记忆降权 合并压缩：相似记忆条目通过聚类合并，提取共性模式 主动遗忘：低质量记忆（如失败且无参考价值的尝试）直接删除 实践中的策略是设置三层保留周期： 7天内：完整保留所有细节 7-30天：保留摘要和关键决策点 30天以上：仅保留高价值模式和教训 工程实现要点 存储选型 记忆层 推荐存储 理由 工作记忆 Redis / 本地内存 低延迟，频繁读写 情景记忆 Milvus / Pinecone 向量检索为主 语义记忆 Neo4j + 向量库 图查询+向量混合 检索融合 最终注入prompt的记忆来自三层的融合结果。推荐的重排策略： ...

从单兵作战到团队协作 单个Agent的能力受限于单一模型的上下文窗口和推理能力。当任务复杂到需要多种专业能力时，多Agent协作成为必然选择。但如何编排多个Agent高效协作，是一个充满设计权衡的工程问题。 编排模式分类 1. 中心化编排（Orchestrator模式） 一个中心编排器分发任务给多个专业Agent： class Orchestrator: def __init__(self): self.agents = { "researcher": ResearchAgent(), "writer": WriterAgent(), "reviewer": ReviewerAgent(), "fact_checker": FactCheckAgent(), } self.task_decomposer = TaskDecomposer() def execute(self, task): # 1. 任务分解 subtasks = self.task_decomposer.decompose(task) # 2. 分配给合适的Agent results = {} for subtask in subtasks: agent = self._select_agent(subtask) result = agent.execute(subtask, context=results) results[subtask.id] = result # 3. 综合结果 final = self._synthesize(results) return final def _select_agent(self, subtask): """根据子任务类型选择Agent""" if subtask.type == "research": return self.agents["researcher"] elif subtask.type == "writing": return self.agents["writer"] elif subtask.type == "fact_check": return self.agents["fact_checker"] # ... 优势： 控制流清晰，易于调试 可以精确控制执行顺序 中心节点维护全局状态 劣势： 中心节点是性能瓶颈 所有通信经过中心，延迟高 中心节点故障则全系统故障 2. 去中心化编排（P2P模式） Agent之间直接通信，无中心节点： class P2PAgent: def __init__(self, name, capabilities): self.name = name self.capabilities = capabilities self.peers = {} # 已知的其他Agent self.message_queue = asyncio.Queue() async def run(self): """Agent主循环""" while True: message = await self.message_queue.get() if message.type == "task": # 处理任务 if self._can_handle(message.task): result = await self._handle(message.task) await self._send(message.sender, "result", result) else: # 转发给合适的peer peer = self._find_capable_peer(message.task) await self._send(peer, "task", message.task) elif message.type == "result": self._process_result(message) async def _send(self, peer_name, msg_type, content): """直接发送消息给peer""" peer = self.peers[peer_name] await peer.message_queue.put({ "type": msg_type, "content": content, "sender": self.name }) 优势： ...

从工具到平台：AI的操作系统化 传统操作系统管理硬件资源、提供系统调用、运行应用程序。AI操作系统（AI-OS）将这一范式升级：用大模型作为"内核"，用自然语言作为"系统调用"，用Agent作为"应用程序"。这不是科幻——多个团队已在构建雏形。 AI-OS的架构愿景 分层架构 ┌──────────────────────────────────────────┐ │ 用户交互层 │ │ (自然语言 / 多模态 / 脑机接口) │ ├──────────────────────────────────────────┤ │ Agent运行时层 │ │ (Agent调度 / 隔离 / 权限 / 通信) │ ├──────────────────────────────────────────┤ │ LLM内核层 │ │ (推理 / 规划 / 记忆 / 对齐) │ ├──────────────────────────────────────────┤ │ 工具与API层 │ │ (MCP Server / 函数调用 / 外部服务) │ ├──────────────────────────────────────────┤ │ 数据与知识层 │ │ (向量DB / 知识图谱 / 文件系统) │ ├──────────────────────────────────────────┤ │ 硬件抽象层 │ │ (GPU/NPU调度 / 存储 / 网络) │ └──────────────────────────────────────────┘ LLM内核 传统OS内核负责进程调度、内存管理、文件系统。AI-OS的LLM内核负责： ...

规划：Agent的核心能力 AI Agent与聊天机器人的本质区别在于规划能力——将复杂目标分解为可执行步骤并动态调整的能力。从简单的ReAct循环到复杂的树搜索规划，Agent规划算法经历了快速演进。 ReAct：推理与行动的交错 基本循环 ReAct（Reasoning + Acting）是Agent规划的基石，核心思想是交错推理和行动： Thought: 用户要查北京天气，我需要调用天气API Action: search_weather(city="北京") Observation: 北京今天晴，25°C Thought: 获取到天气信息，可以回答用户了 Answer: 北京今天天气晴朗，气温25°C，适合外出活动。 实现细节 class ReActAgent: def __init__(self, llm, tools, max_steps=10): self.llm = llm self.tools = tools self.max_steps = max_steps def run(self, task): trajectory = [] for step in range(self.max_steps): # 生成下一步思考和行为 prompt = self._build_prompt(task, trajectory) response = self.llm.generate(prompt) thought, action = self._parse(response) trajectory.append({"thought": thought, "action": action}) if action["type"] == "finish": return action["answer"] # 执行工具 if action["type"] == "tool": result = self.tools[action["name"]](**action["params"]) trajectory.append({"observation": result}) return "达到最大步数限制" def _build_prompt(self, task, trajectory): prompt = f"""任务：{task} 可用工具：{list(self.tools.keys())} 历史： """ for item in trajectory: if "thought" in item: prompt += f"Thought: {item['thought']}\n" prompt += f"Action: {item['action']}\n" elif "observation" in item: prompt += f"Observation: {item['observation']}\n" prompt += "Thought:" return prompt ReAct的局限 无记忆反馈：失败的经验不会影响后续尝试 线性思维：无法回溯到之前的选择点尝试其他路径 容易陷入循环：在复杂任务中反复尝试相同的失败方案 Reflexion：从失败中学习 核心改进 Reflexion在ReAct基础上增加了自我反思机制。当任务执行失败时，Agent会生成反思并存储到长期记忆中： ...

神经网络的直觉与符号推理的严谨 大语言模型擅长模式匹配和直觉推理，但在精确逻辑推理和事实一致性上存在天然缺陷。知识图谱作为结构化的符号知识表示，恰好互补了LLM的短板。两者的融合——神经符号AI——正在成为构建可靠AI系统的重要方向。 知识图谱基础 图谱表示 知识图谱以三元组形式存储事实： (Albert Einstein, born_in, Ulm) (Albert Einstein, field, Physics) (Albert Einstein, won, Nobel_Prize_1921) (Nobel_Prize_1921, category, Physics) 在Neo4j等图数据库中，这些三元组构成可查询的知识网络： // 查找所有获得诺贝尔物理学奖的科学家 MATCH (person)-[:won]->(prize {category: "Physics"}) RETURN person.name, prize.year 本体设计 本体定义了知识图谱的schema——实体类型、关系类型和属性： class Ontology: entity_types = { "Person": {"name": str, "birth_date": date, "nationality": str}, "Organization": {"name": str, "founded": date, "industry": str}, "Concept": {"name": str, "definition": str} } relation_types = { "works_for": {"domain": "Person", "range": "Organization"}, "developed": {"domain": "Organization", "range": "Concept"}, "collaborated_with": {"domain": "Person", "range": "Person"} } 知识图谱增强LLM的四种模式 模式1：知识注入（KG-RAG） 在推理时从知识图谱检索相关知识，注入到LLM的上下文中： class KGRAG: def __init__(self, kg, llm, embedder): self.kg = kg # 知识图谱 self.llm = llm self.embedder = embedder def query(self, question): # 1. 实体链接 entities = self._extract_entities(question) # 2. 子图检索 subgraph = self._retrieve_subgraph(entities, hops=2) # 3. 路径排序 paths = self._rank_paths(question, subgraph) # 4. 文本化 context = self._serialize_paths(paths) # 5. LLM生成 prompt = f"""基于以下知识图谱信息回答问题： 知识： {context} 问题：{question} """ return self.llm.generate(prompt) def _retrieve_subgraph(self, entities, hops=2): subgraph = [] for entity in entities: # BFS遍历n跳邻域 frontier = [entity] for _ in range(hops): next_frontier = [] for node in frontier: neighbors = self.kg.get_neighbors(node) for neighbor, relation in neighbors: subgraph.append((node, relation, neighbor)) next_frontier.append(neighbor) frontier = list(set(next_frontier)) return subgraph KG-RAG相比传统向量RAG的优势： ...

MoE：用稀疏激活突破密集模型的效率瓶颈 混合专家模型（Mixture of Experts）通过稀疏激活机制，让模型总参数量巨大但每次推理只激活一小部分。DeepSeek-V3的671B参数中每次只激活37B，Llama 4的500B+参数只激活30B——这就是MoE的效率魔法。 MoE的数学基础 标准MoE层 给定输入 $x$，MoE层的计算过程： $$y = \sum_{i=1}^{N} g_i(x) \cdot f_i(x)$$ 其中 $f_i$ 是第 $i$ 个专家（一个FFN），$g_i$ 是路由函数： $$g_i(x) = \text{softmax}(W_g \cdot x)_i$$ Top-K路由只保留 $K$ 个最大的 $g_i$，其余置零： $$g_i(x) = \begin{cases} \frac{\exp(W_g \cdot x)i}{\sum{j \in \text{TopK}} \exp(W_g \cdot x)_j} & i \in \text{TopK} \ 0 & \text{otherwise} \end{cases}$$ 专家容量因子 为防止所有token路由到同一专家（路由崩塌），每个专家有一个容量上限： $$\text{capacity} = \frac{T \cdot K}{N} \times \text{capacity_factor}$$ 其中 $T$ 是token数，$K$ 是Top-K，$N$ 是专家数。超出容量的token被丢弃或传递到下一层。 路由机制设计 标准Top-K路由 最简单的路由方式。每个token独立选择Top-K个专家： class TopKRouter(nn.Module): def __init__(self, dim, n_experts, top_k=2): self.gate = nn.Linear(dim, n_experts) self.top_k = top_k def forward(self, x): # x: [batch * seq_len, dim] logits = self.gate(x) scores = F.softmax(logits, dim=-1) # Top-K选择 topk_scores, topk_indices = scores.topk(self.top_k, dim=-1) topk_scores = topk_scores / topk_scores.sum(dim=-1, keepdim=True) return topk_scores, topk_indices 专家选择路由 DeepSeek-V3使用了一种"专家选择"的反向路由：每个专家选择Top-K个token，而非token选择专家。这平衡了专家负载： ...

Agent框架的核心问题 构建AI Agent需要解决三个核心问题：任务规划与分解、工具调用与执行、多Agent协作。不同的Agent框架对这些问题的回答各不相同，形成了截然不同的编程范式。 LangGraph：图驱动的精确控制 设计哲学 LangGraph将Agent工作流建模为有向图：节点是处理步骤，边是控制流。这种设计赋予了开发者对Agent行为的精确控制。 核心概念 from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list tool_results: list next_step: str # 定义图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("planner", plan_task) graph.add_node("executor", execute_tools) graph.add_node("reviewer", review_results) # 添加边（控制流） graph.add_edge("planner", "executor") graph.add_conditional_edges( "executor", lambda state: "reviewer" if state["tool_results"] else END, {True: "reviewer", False: END} ) graph.add_conditional_edges( "reviewer", lambda state: "planner" if not state["success"] else END ) 优势 精确控制：图的拓扑结构让Agent的执行路径完全可见和可预测。这在需要审计和合规的场景中至关重要。 状态管理：LangGraph内置了检查点（checkpoint）机制，可以在任何节点暂停和恢复执行。这对于长时间运行的Agent任务非常有用。 人机协作：interrupt_before和interrupt_after参数允许在特定节点暂停，等待人类反馈： graph.compile( checkpointer=MemorySaver(), interrupt_before=["executor"] # 执行前等待人类确认 ) 局限 学习曲线陡峭，图思维需要适应 简单任务显得过度工程化 调试复杂图流程需要可视化工具辅助 AutoGen：多Agent对话的原生范式 设计哲学 AutoGen（微软研究院）将多Agent协作建模为对话。每个Agent有角色定义，通过消息传递完成协作。 核心模式 from autogen import AssistantAgent, UserProxyAgent, GroupChat # 创建Agent planner = AssistantAgent( name="Planner", system_message="你负责分析用户需求并制定执行计划", llm_config=llm_config ) coder = AssistantAgent( name="Coder", system_message="你负责根据计划编写代码", llm_config=llm_config ) reviewer = AssistantAgent( name="Reviewer", system_message="你负责审查代码质量", llm_config=llm_config ) user_proxy = UserProxyAgent( name="User", human_input_mode="NEVER", code_execution_config={"work_dir": "workspace"} ) # 创建群聊 group_chat = GroupChat( agents=[user_proxy, planner, coder, reviewer], messages=[], max_round=15 ) manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config) user_proxy.initiate_chat(manager, message="实现一个用户注册API") 优势 对话式协作自然：多Agent之间的协作就像团队会议，每个Agent专注于自己的职责。 ...

边缘AI的必要性 随着AI应用深入日常生活，将推理能力从云端下沉到边缘设备成为刚需。边缘AI具有三大核心优势：低延迟、隐私保护、离线可用。但在手机、IoT设备上运行数十亿参数的大模型是一项极具挑战性的工程任务。 硬件资源约束分析 手机端资源 2026年主流旗舰手机的AI相关硬件参数： 设备 芯片 NPU算力(TOPS) 可用内存(GB) iPhone 16 Pro A18 Pro 35 8 Galaxy S25 SD 8 Gen4 45 12 Pixel 9 Tense G4 40 12 在8GB内存的手机上运行7B模型（INT4量化后约3.5GB），留给系统和应用的内存仅约4GB，紧张但可行。 边缘服务器资源 以树莓派5（8GB RAM）为例： CPU：4核Cortex-A76 @ 2.4GHz 无NPU，纯CPU推理 可运行3B模型（INT4量化后约1.5GB） 模型压缩技术栈 量化：最有效的压缩手段 移动端部署首选INT4量化。以Qwen3-1.5B为例： FP16模型大小: 3.0GB INT8模型大小: 1.5GB INT4模型大小: 0.9GB ← 手机可运行 GGUF格式是llama.cpp生态的标准量化格式： # 使用llama.cpp量化模型 # Q4_K_M：4bit量化，平衡精度和大小 ./quantize model.gguf model-q4_k_m.gguf Q4_K_M 不同量化级别的精度损失： Q8_0：<0.5%精度损失，但文件大 Q4_K_M：1-2%精度损失，推荐移动端使用 Q3_K_S：3-5%精度损失，极限压缩场景 蒸馏：减小模型尺寸 知识蒸馏将大模型的能力迁移到小模型： class DistillationTrainer: def __init__(self, teacher, student, temperature=2.0): self.teacher = teacher # 冻结的大模型 self.student = student # 待训练的小模型 self.temperature = temperature def compute_loss(self, inputs): with torch.no_grad(): teacher_logits = self.teacher(inputs).logits / self.temperature student_logits = self.student(inputs).logits / self.temperature # KL散度损失 kl_loss = F.kl_div( F.log_softmax(student_logits, dim=-1), F.softmax(teacher_logits, dim=-1), reduction="batchmean" ) * (self.temperature ** 2) # 标准交叉熵损失 ce_loss = F.cross_entropy(student_logits, inputs["labels"]) return 0.7 * kl_loss + 0.3 * ce_loss 剪枝：移除冗余参数 结构化剪枝移除整个注意力头或FFN中间维度，不破坏模型结构： ...

MCP协议：AI工具调用的"USB-C"接口 2024年Anthropic推出的Model Context Protocol（MCP）正在改变AI Agent与外部工具的交互方式。如果说Function Calling是每个厂商各自的充电接口，那么MCP就是统一的USB-C标准。 MCP核心架构 MCP采用客户端-服务器架构，定义了三个核心原语： Resources（资源） 只读数据源，类似REST中的GET端点。Agent可以读取文件、数据库记录、API响应等： { "uri": "file:///project/src/main.py", "mimeType": "text/x-python", "content": "..." } Tools（工具） 可执行的操作，类似POST端点。每个工具定义包含名称、描述和JSON Schema参数： interface Tool { name: string; description: string; inputSchema: { type: "object"; properties: { ... }; required: string[]; }; } Prompts（提示模板） 预定义的提示词模板，支持参数化注入，方便复用。 与Function Calling的对比 传统Function Calling存在三个痛点：工具定义与模型耦合、跨厂商不兼容、缺乏状态管理。MCP通过标准化协议层解决了这些问题： 维度 Function Calling MCP 工具定义 内嵌在prompt中 独立服务，动态发现 跨厂商 各厂商格式不同 统一协议 状态管理 无状态 支持有状态会话 传输方式 API请求内嵌 stdio/SSE/HTTP 工程实现示例 以下是一个MCP Server的Python实现： from mcp.server import Server from mcp.types import Tool, TextContent server = Server("code-analyzer") @server.list_tools() async def list_tools(): return [ Tool( name="analyze_code", description="分析Python代码的复杂度和潜在问题", inputSchema={ "type": "object", "properties": { "code": {"type": "string", "description": "待分析的代码"}, "strictness": {"type": "string", "enum": ["low", "medium", "high"]} }, "required": ["code"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): if name == "analyze_code": result = await analyze(arguments["code"], arguments.get("strictness", "medium")) return [TextContent(type="text", text=result)] MCP在Agent编排中的价值 MCP最大的价值在于Agent编排场景。当多个Agent需要共享同一组工具时，MCP Server作为统一的工具提供者： ...

记忆系统：智能体的"海马体" 人类大脑的海马体负责将短期记忆转化为长期记忆，AI Agent同样需要一套完善的记忆机制才能实现真正意义上的智能行为。当前主流的Agent框架中，记忆系统设计往往是最被低估却又最关键的组件。 三层记忆架构模型 1. 工作记忆（Working Memory） 工作记忆对应Agent的上下文窗口，是当前正在处理的信息。以GPT-4 Turbo为例，128K token的上下文窗口就是其工作记忆容量。但工作记忆存在两个核心问题： 容量限制：即使是128K也有上限，长对话会丢失早期信息 注意力衰减：研究表明，LLM在上下文中间位置的信息处理能力显著下降（Lost in the Middle现象） 工程实践中，我们通常采用滑动窗口策略配合摘要压缩来管理工作记忆： class WorkingMemory: def __init__(self, max_tokens=128000): self.max_tokens = max_tokens self.messages = [] self.current_tokens = 0 def add(self, message): self.messages.append(message) self.current_tokens += count_tokens(message) if self.current_tokens > self.max_tokens * 0.8: self._compress() def _compress(self): # 保留最近N条消息，将早期消息摘要化 recent = self.messages[-10:] old = self.messages[:-10] summary = llm_summarize(old) self.messages = [{"role": "system", "content": summary}] + recent 2. 情景记忆（Episodic Memory） 情景记忆记录Agent过去的交互经历，包括具体对话、决策过程和结果反馈。这一层通常通过向量数据库实现，如Chroma、Pinecone或Weaviate。 关键设计点在于索引策略。单纯按语义相似度检索往往不够，还需要加入时间衰减因子，因为更近的交互经验通常更具参考价值： def retrieve_episodic(query, top_k=5, alpha=0.7): # 语义相似度 semantic_scores = vector_db.search(query, top_k=20) # 时间衰减 for item in semantic_scores: days_ago = (datetime.now() - item.timestamp).days time_score = math.exp(-0.05 * days_ago) item.final_score = alpha * item.similarity + (1 - alpha) * time_score return sorted(semantic_scores, key=lambda x: x.final_score, reverse=True)[:top_k] 3. 语义记忆（Semantic Memory） 语义记忆是Agent的长期知识库，存储经过验证的事实、规则和概念。与情景记忆的区别在于：情景记忆是"经历"，语义记忆是"知识"。 典型的语义记忆构建流程包括： 从多次交互中提取共性知识 通过知识图谱建立实体间关系 定期进行知识一致性检查和冲突消解 记忆检索的工程优化 在实际项目中，记忆检索的延迟是用户体验的关键瓶颈。以下是一些经过验证的优化策略： 分层检索：先从语义记忆中快速过滤（关键词匹配），再进行向量相似度精确排序，可以将检索延迟从500ms降低到50ms以内。 预计算缓存：对于高频查询，预先计算并缓存检索结果。利用用户意图分类器判断是否命中缓存。 异步写入：记忆写入不应阻塞主流程。采用写前日志（WAL）模式，先快速持久化到本地，再异步同步到向量数据库。 记忆遗忘机制 好的记忆系统不仅需要记住，还需要遗忘。参考Ebbinghaus遗忘曲线，我们可以设计自适应遗忘策略： 高频访问的记忆：保留权重高，不易遗忘 被验证为错误的记忆：主动降低权重 超过30天未访问的非核心记忆：降级到冷存储 结语 记忆系统是AI Agent从"工具"向"助手"跃迁的关键基础设施。随着Agent应用场景的复杂化，记忆架构的设计将越来越接近人类认知模型。下一篇文章我们将探讨基于MCP协议的记忆共享机制，实现多Agent间的知识传递。 ...