架构设计

引言 Agent系统的发布比传统应用复杂得多——一个Prompt的微调可能导致Agent行为完全改变，一个工具的版本升级可能影响所有依赖它的Agent。传统的"停机发布"在Agent系统中不可接受，而简单的"滚动更新"也无法满足Agent系统对质量保障的高要求。 2026年，金丝雀发布 + 自动回滚已成为Agent系统的标准发布实践，但Agent系统的灰度发布有其独特的挑战和解决方案。 Agent发布的特殊性 维度 传统应用 Agent系统 变更类型 代码逻辑 Prompt/模型/工具/代码 质量评估 单元测试+集成测试 需要LLM评估+人工审核 回滚速度 秒级 秒级（代码）/分钟级（模型） 影响范围 功能正确性 对话质量、安全性、成本 监控指标 错误率、延迟 +质量评分、Token消耗、用户满意度 灰度发布策略 策略一：金丝雀发布 class CanaryReleaseManager: """金丝雀发布管理器""" def __init__(self, traffic_router, metrics_collector): self.router = traffic_router self.metrics = metrics_collector async def canary_deploy( self, new_version: str, stages: list = None ) -> bool: """渐进式金丝雀发布""" if stages is None: stages = [ {"traffic_percent": 5, "duration_minutes": 10}, {"traffic_percent": 20, "duration_minutes": 15}, {"traffic_percent": 50, "duration_minutes": 20}, {"traffic_percent": 100, "duration_minutes": 30}, ] baseline = await self.metrics.get_baseline() for stage in stages: # 调整流量分配 await self.router.set_traffic_split({ "stable": 100 - stage["traffic_percent"], "canary": stage["traffic_percent"] }) logger.info( f"Canary stage: {stage['traffic_percent']}% traffic " f"for {stage['duration_minutes']}min" ) # 等待观察期 await asyncio.sleep(stage["duration_minutes"] * 60) # 评估金丝雀指标 canary_metrics = await self.metrics.collect("canary") evaluation = self._evaluate(baseline, canary_metrics) if evaluation["action"] == "rollback": logger.warning( f"Canary failed at {stage['traffic_percent']}%: " f"{evaluation['reason']}" ) await self._rollback() return False elif evaluation["action"] == "hold": logger.info(f"Pausing canary: {evaluation['reason']}") await self._notify_human(evaluation) await self._wait_for_approval() # 所有阶段通过，完成发布 await self.router.promote_canary() return True def _evaluate(self, baseline: dict, canary: dict) -> dict: """评估金丝雀健康度""" checks = [ self._check_error_rate(baseline, canary), self._check_latency(baseline, canary), self._check_quality_score(baseline, canary), self._check_cost(baseline, canary), self._check_safety(baseline, canary), ] for check in checks: if check["status"] == "fail": return {"action": "rollback", "reason": check["reason"]} if check["status"] == "warn": return {"action": "hold", "reason": check["reason"]} return {"action": "proceed", "reason": "All checks passed"} def _check_quality_score(self, baseline: dict, canary: dict) -> dict: """质量评分检查——Agent特有的评估维度""" quality_drop = baseline["quality_score"] - canary["quality_score"] if quality_drop > 0.1: # 质量下降超过10% return { "status": "fail", "reason": f"Quality dropped {quality_drop:.1%}" } elif quality_drop > 0.05: return { "status": "warn", "reason": f"Quality dropped {quality_drop:.1%}, review needed" } return {"status": "pass"} def _check_safety(self, baseline: dict, canary: dict) -> dict: """安全检查——检测有害输出""" safety_violation_rate = canary.get("safety_violation_rate", 0) if safety_violation_rate > 0.001: # 0.1%安全违规 return { "status": "fail", "reason": f"Safety violation rate: {safety_violation_rate:.3%}" } return {"status": "pass"} 策略二：蓝绿部署 # K8s蓝绿部署配置 apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: agent-service spec: replicas: 10 strategy: blueGreen: activeService: agent-service-active previewService: agent-service-preview autoPromotionEnabled: false # 手动确认 scaleDownDelaySeconds: 30 prePromotionAnalysis: templates: - templateName: agent-quality-check args: - name: service-name value: agent-service-preview selector: matchLabels: app: agent-service template: metadata: labels: app: agent-service spec: containers: - name: agent image: agent/service:{{ .Values.version }} env: - name: AGENT_VERSION value: "{{ .Values.version }}" - name: MODEL_ENDPOINT value: "http://llm-service:8080/v1" 策略三：流量镜像 流量镜像是Agent系统特别适合的灰度策略——将生产流量复制一份到新版本，不影响真实用户： ...

引言 Agent系统本质上是一个工作流编排系统——理解意图、检索知识、调用工具、评估结果、生成回复，每一步都是工作流中的一个节点。选择合适的工作流引擎，直接决定了Agent系统的可靠性、可观测性和开发效率。 2026年，工作流引擎领域已形成了清晰的格局。Temporal凭借其强大的状态管理和重试机制成为Agent系统的热门选择，Airflow在数据处理管道中依然占有一席之地，而自研引擎则在对性能和灵活性有极致要求的场景中仍有市场。 Agent工作流的特殊需求 与传统数据处理工作流不同，Agent工作流有其独特的需求特征： 需求维度 传统工作流 Agent工作流 执行时长 分钟到小时 秒到分钟 分支复杂度 低（线性DAG） 高（动态分支、循环） 人机交互 罕见 频繁（澄清、确认） 失败处理 重试或告警 重新规划、降级策略 状态管理 简单 复杂（对话历史、中间结果） 实时性 批处理 实时或近实时 动态性 静态DAG 运行时动态生成 三大方案深度对比 Temporal：Agent工作流的最佳搭档 from temporalio import workflow, activity from datetime import timedelta @activity.defn async def understand_intent(user_input: str) -> dict: """意图理解活动""" # 调用LLM进行意图分类 result = await llm_client.classify(user_input) return { "intent": result.intent, "confidence": result.confidence, "entities": result.entities } @activity.defn async def retrieve_memory(query: str, top_k: int = 5) -> list: """记忆检索活动""" memories = await vector_db.search(query, top_k=top_k) return [{"content": m.text, "score": m.score} for m in memories] @activity.defn async def execute_tool(tool_name: str, params: dict) -> dict: """工具执行活动""" tool = tool_registry.get(tool_name) result = await tool.run(**params) return result @activity.defn async def generate_response(prompt: str, context: dict) -> str: """响应生成活动""" response = await llm_client.generate(prompt, **context) return response @workflow.defn class AgentWorkflow: """Agent主工作流""" @workflow.run async def run(self, user_input: str) -> str: # Step 1: 意图理解 intent_result = await workflow.execute_activity( understand_intent, user_input, start_to_close_timeout=timedelta(seconds=10), retry_policy=RetryPolicy( initial_interval=timedelta(seconds=1), maximum_interval=timedelta(seconds=10), maximum_attempts=3 ) ) # 需要澄清时，等待用户输入 if intent_result["confidence"] < 0.6: clarification = await workflow.wait_for_signal( "user_clarification", timeout=timedelta(minutes=5) ) intent_result = await workflow.execute_activity( understand_intent, clarification, start_to_close_timeout=timedelta(seconds=10) ) # Step 2: 并行检索记忆和执行工具 memory_task = workflow.execute_activity( retrieve_memory, user_input, start_to_close_timeout=timedelta(seconds=5) ) tool_task = workflow.execute_activity( execute_tool, intent_result["intent"], intent_result["entities"], start_to_close_timeout=timedelta(seconds=30) ) memories, tool_results = await asyncio.gather(memory_task, tool_task) # Step 3: 生成响应 prompt = build_prompt(user_input, memories, tool_results) response = await workflow.execute_activity( generate_response, prompt, {"intent": intent_result["intent"]}, start_to_close_timeout=timedelta(seconds=15) ) return response Temporal的优势在Agent场景中极为突出： ...

引言 Agent系统的扩展性挑战与传统Web应用截然不同。LLM推理是GPU密集型操作，工具执行可能是CPU或IO密集型，而向量检索则是内存密集型。这意味着简单的"加机器"策略无法有效解决Agent系统的扩展问题。 2026年，K8s + GPU Operator已成为Agent系统部署的事实标准，但如何高效利用集群资源仍然是工程团队面临的核心挑战。 扩展维度分析 Agent系统需要在多个维度上独立扩展： ┌─────────────────────────────────────────────────────────┐ │ Agent系统扩展维度 │ ├─────────────┬──────────────┬──────────────┬────────────┤ │ 并发会话数 │ 推理吞吐量 │ 工具执行并发 │ 记忆检索延迟│ │ (CPU/Mem) │ (GPU) │ (CPU/IO) │ (RAM/SSD) │ ├─────────────┼──────────────┼──────────────┼────────────┤ │ 水平扩展 │ GPU水平扩展 │ 水平扩展 │ 分片+副本 │ │ +Stateless │ +模型并行 │ +无状态 │ +读副本 │ └─────────────┴──────────────┴──────────────┴────────────┘ 从单机到集群的演进路径 Phase 1：单机优化 在扩展之前，先榨干单机性能： import asyncio import uvicorn from concurrent.futures import ThreadPoolExecutor class SingleNodeAgent: """单机Agent——最大化单节点利用率""" def __init__(self): # CPU密集型任务（工具执行） self.cpu_pool = ThreadPoolExecutor( max_workers=8, thread_name_prefix="tool-exec" ) # IO密集型任务（网络请求） self.io_pool = ThreadPoolExecutor( max_workers=32, thread_name_prefix="io-op" ) # LLM推理使用GPU，通过信号量控制并发 self.llm_semaphore = asyncio.Semaphore(4) async def process_request(self, user_input: str) -> str: # 并行执行独立任务 memory_task = asyncio.create_task(self._retrieve_memory(user_input)) tool_task = asyncio.create_task(self._execute_tools(user_input)) memory = await memory_task tool_results = await tool_task # LLM推理（GPU受限） async with self.llm_semaphore: response = await self._llm_inference(user_input, memory, tool_results) return response Phase 2：水平拆分 将不同负载特征的服务拆分到不同节点： ...

引言 Agent的状态管理是系统设计中最容易被忽视却又最关键的环节。一个Agent在执行任务时，可能经历"理解意图→检索记忆→调用工具→评估结果→生成回复"等多个阶段，每个阶段都有不同的状态和转移条件。状态管理不当会导致上下文丢失、重复执行、死循环等严重问题。 2026年，随着Agent系统复杂度的指数级增长，系统化的状态管理架构已成为生产部署的必备条件。 Agent状态的三个层次 第一层：会话状态（Session State） 会话状态是最基础的状态层，管理单次用户交互的上下文： from dataclasses import dataclass, field from datetime import datetime from enum import Enum class SessionStatus(Enum): ACTIVE = "active" PAUSED = "paused" COMPLETED = "completed" FAILED = "failed" TIMEOUT = "timeout" @dataclass class SessionState: """会话状态——管理单次交互的完整生命周期""" session_id: str user_id: str status: SessionStatus created_at: datetime updated_at: datetime message_history: list = field(default_factory=list) active_tools: list = field(default_factory=list) pending_actions: list = field(default_factory=list) context_window: dict = field(default_factory=dict) metadata: dict = field(default_factory=dict) def add_message(self, role: str, content: str): self.message_history.append({ "role": role, "content": content, "timestamp": datetime.now().isoformat() }) self.updated_at = datetime.now() def is_expired(self, ttl_seconds: int = 3600) -> bool: elapsed = (datetime.now() - self.updated_at).total_seconds() return elapsed > ttl_seconds 第二层：工作流状态（Workflow State） 工作流状态管理Agent执行复杂多步骤任务时的进度： from enum import Enum class WorkflowStepStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" RETRYING = "retrying" @dataclass class WorkflowStep: step_id: str step_name: str status: WorkflowStepStatus dependencies: list # 前置步骤ID inputs: dict outputs: dict retry_count: int = 0 max_retries: int = 3 started_at: datetime = None completed_at: datetime = None @dataclass class WorkflowState: """工作流状态——管理多步骤任务的执行进度""" workflow_id: str session_id: str steps: dict # step_id -> WorkflowStep current_step: str context: dict # 跨步骤共享的上下文 def get_ready_steps(self) -> list: """获取可执行的步骤（依赖已完成）""" ready = [] for step_id, step in self.steps.items(): if step.status != WorkflowStepStatus.PENDING: continue deps_satisfied = all( self.steps[dep].status == WorkflowStepStatus.SUCCESS for dep in step.dependencies ) if deps_satisfied: ready.append(step) return ready def is_complete(self) -> bool: return all( s.status in [WorkflowStepStatus.SUCCESS, WorkflowStepStatus.SKIPPED] for s in self.steps.values() ) 第三层：持久状态（Persistent State） 持久状态跨越会话边界，包括用户偏好、长期记忆和已学习的模式： ...

引言：为什么Agent系统需要事件驱动？ 传统的同步HTTP调用在Agent系统中面临严峻挑战：当一个复杂任务涉及数十次工具调用、记忆检索和多轮对话时，同步调用会导致漫长的等待时间和服务间的强耦合。更关键的是，Agent系统天然需要处理异步、长时运行的任务——一次网页搜索可能耗时数秒，一次代码执行可能需要更长时间。 事件驱动架构通过解耦服务间的时间依赖，让Agent系统能够优雅地处理异步性和并发性。 事件驱动架构的核心模式 发布-订阅模式 ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Tool │───▶│ Message │───▶│ Memory │ │ Service │ │ Bus │ │ Service │ └─────────────┘ │ │ └─────────────┘ │ ┌─────────┐ │ ┌─────────────┐ ┌─────────────┐ │ │ Topic │ │───▶│ LLM │ │ Router │───▶│ │ Router │ │ │ Inference │ │ Service │ │ └─────────┘ │ └─────────────┘ └─────────────┘ └─────────────┘ ▲ ┌────┴────┐ │Event │ │Producer │ └─────────┘ 事件溯源模式 在Agent系统中，事件溯源（Event Sourcing）能够完整记录Agent的思考和决策过程： ...

引言：为什么Agent系统需要微服务化？ 2026年，随着Agent系统在生产环境中的大规模部署，单体架构的局限性日益凸显。一个典型的Agent系统包含意图理解、工具调用、记忆检索、上下文管理和响应生成等多个子系统，当这些功能耦合在一个进程中时，扩展性、可用性和迭代速度都会受到严重制约。 本文将从实际工程角度，系统阐述Agent系统从单体到微服务的完整演进路径。 单体Agent架构的痛点 在早期阶段，大多数Agent系统采用单体架构——所有功能模块运行在同一个进程中： ┌─────────────────────────────────────┐ │ Monolithic Agent │ │ ┌───────┐ ┌───────┐ ┌───────────┐ │ │ │ LLM │ │ Tools │ │ Memory │ │ │ │ Call │ │ Exec │ │ Store │ │ │ └───────┘ └───────┘ └───────────┘ │ │ ┌───────┐ ┌───────┐ ┌───────────┐ │ │ │ Router│ │ Prompt│ │ Response │ │ │ │ │ │ Builder│ │ Formatter │ │ │ └───────┘ └───────┘ └───────────┘ │ └─────────────────────────────────────┘ 核心痛点包括： ...

从Naive RAG到Agentic RAG的演进 RAG技术的发展经历了三个阶段： Naive RAG (2023) → Advanced RAG (2024) → Agentic RAG (2025-2026) 朴素检索 增强检索 智能体检索 单次查询 多次优化 自主决策 固定管道 模块化 动态规划 传统RAG是一个"固定管道"：查询→检索→拼接→生成，一锤子买卖。Agentic RAG将LLM作为"大脑"，让它自主决定是否需要检索、检索什么、检索几次、何时停止。这就像从"查字典"升级为"咨询一个会查资料的研究员"。 Agentic RAG的核心架构 ┌─────────────────────────────────────────────────────┐ │ Agentic RAG Architecture │ ├─────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────────────────────────┐ │ │ │ User │───▶│ Agent Controller │ │ │ │ Query │ │ (LLM + Planning + Memory) │ │ │ └──────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌─────────┼─────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌───────┐ ┌─────────┐ │ │ │ Router │ │ Query │ │ Evaluator│ │ │ │ (路由) │ │Rewrite│ │ (评估器) │ │ │ └────┬────┘ └───┬───┘ └────┬────┘ │ │ │ │ │ │ │ ┌─────────┼──────────┼──────────┘ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌────────┐ ┌──────────┐ ┌─────────┐ │ │ │ Vector │ │ Web │ │ SQL │ │ Graph │ │ │ │ Store │ │ Search │ │ DB │ │ Store │ │ │ └──────────┘ └────────┘ └──────────┘ └─────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 四大核心组件 组件 职责 关键能力 Agent Controller 任务规划与决策 分解问题、决定工具调用顺序 Router 工具路由 选择合适的数据源（向量库/搜索引擎/数据库） Query Rewriter 查询改写 将用户问题改写为更有效的检索查询 Evaluator 结果评估 判断检索结果是否充分，决定是否需要追加检索 核心模式解析 模式1：迭代检索（Iterative Retrieval） from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): question: str retrieved_docs: Annotated[List[str], operator.add] answer: str iteration: int sufficient: bool def retrieve_node(state: AgentState) -> dict: """基于当前问题+已有信息生成查询并检索""" query = query_rewriter(state["question"], state["retrieved_docs"]) docs = vector_store.similarity_search(query, k=5) return {"retrieved_docs": docs, "iteration": state["iteration"] + 1} def evaluate_node(state: AgentState) -> dict: """评估检索结果是否足以回答问题""" if state["iteration"] >= 3: return {"sufficient": True} prompt = f"""基于以下检索到的文档，能否完整回答用户问题？ 问题: {state['question']} 文档: {state['retrieved_docs']} 如果信息充分回答"SUFFICIENT"，否则回答"INSUFFICIENT"并说明缺什么。""" response = llm.invoke(prompt) sufficient = "SUFFICIENT" in response return {"sufficient": sufficient} def answer_node(state: AgentState) -> dict: """基于所有检索文档生成最终回答""" prompt = f"问题: {state['question']}\n\n参考资料:\n{state['retrieved_docs']}" return {"answer": llm.invoke(prompt)} # 构建工作流 workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve_node) workflow.add_node("evaluate", evaluate_node) workflow.add_node("answer", answer_node) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "evaluate") workflow.add_conditional_edges( "evaluate", lambda state: "answer" if state["sufficient"] else "retrieve" ) workflow.add_edge("answer", END) agent = workflow.compile() 模式2：多路检索（Multi-Route Retrieval） def route_query(state: AgentState) -> str: """根据问题类型选择检索路径""" routing_prompt = f""" 分析以下问题，选择最合适的数据源： 问题: {state['question']} 选项: - "vector": 适合从内部文档库检索事实性信息 - "web": 适合需要最新信息的时效性问题 - "sql": 适合结构化数据查询 - "graph": 适合关系推理和多方关联 返回JSON: {{"route": "...", "reason": "..."}} """ decision = llm.invoke(routing_prompt) return json.loads(decision)["route"] 模式3：自我纠正检索（Self-Corrective RAG） def grade_documents(state: AgentState) -> dict: """对检索文档打分，过滤低质量结果""" graded_docs = [] for doc in state["retrieved_docs"]: score_prompt = f"""评估文档与问题的相关性(0-10): 问题: {state['question']} 文档: {doc.page_content[:500]} 只返回数字。""" score = int(llm.invoke(score_prompt).strip()) if score >= 6: graded_docs.append(doc) # 如果所有文档都不相关，触发查询改写 if not graded_docs: new_query = llm.invoke(f"改写这个查询以获得更好的检索结果: {state['question']}") return {"retrieved_docs": [], "question": new_query} return {"retrieved_docs": graded_docs} 工程实现：基于LangGraph的完整Agentic RAG from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 定义工具 @tool def vector_search(query: str) -> str: """从内部知识库检索相关文档""" docs = vector_store.similarity_search(query, k=5) return "\n\n".join([d.page_content for d in docs]) @tool def web_search(query: str) -> str: """从网络搜索最新信息""" results = search_api.search(query, num_results=5) return "\n\n".join([r["content"] for r in results]) @tool def sql_query(question: str) -> str: """查询业务数据库""" sql = text2sql(question) return db.execute(sql) # Agent决策循环 tools = [vector_search, web_search, sql_query] llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_decision(state): messages = state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response]} def should_continue(state): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END # 构建Graph graph = StateGraph(dict) graph.add_node("agent", agent_decision) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue) graph.add_edge("tools", "agent") # 工具执行后回到agent做下一轮决策 app = graph.compile() # 执行 result = app.invoke({ "messages": [HumanMessage(content="对比2026年Q1三大云厂商的AI服务收入")] }) 性能基准对比 我们在三个数据集上对比了不同RAG架构的表现： ...

为什么需要多模态RAG？ 传统RAG只能处理文字，但现实世界的信息是多维的。技术文档里穿插架构图、产品手册里有演示截图、研究报告里有数据图表——这些信息，纯文本RAG完全丢失了。 用户问"系统的整体架构是什么样的？"，纯文本RAG只会返回一段文字描述，而无法返回架构图。这就是多模态RAG要解决的问题。 多模态RAG架构全景 ┌─────────────────────────────────────────────────────────┐ │ 多模态RAG系统架构 │ ├─────────────────────────────────────────────────────────┤ │ │ │ 用户查询 │ │ │ │ │ ▼ │ │ ┌─────────────┐ │ │ │ Query 分析 │ 判断用户需要图片、文字、还是图文 │ │ └──────┬──────┘ │ │ │ │ │ ┌────┴────┐ │ │ ▼ ▼ │ │ [纯文本查询] [需要图片查询] │ │ │ │ │ │ ▼ ▼ │ │ 文本检索 多模态检索 │ │ (向量库) (CLIP/图片向量) │ │ │ │ │ │ └────┬────┘ │ │ ▼ │ │ ┌─────────────┐ │ │ │ 结果融合 │ 图文结果融合排序 │ │ └──────┬──────┘ │ │ │ │ │ ▼ │ │ 多模态LLM生成回答（可理解和描述图片） │ │ │ └─────────────────────────────────────────────────────────┘ 四种架构方案详解 方案1：独立双路检索（Easiest） 文本向量库 图片向量库 │ │ └──► 合并 ◄──┘ │ ▼ 多模态LLM生成 最简单也是最常见的方案：文本和图片分别Embedding，分别检索，然后合并结果交给多模态LLM生成回答。 ...

Agent编排：从简单到复杂的演进之路 2026年，Agent编排模式已经从早期的"提示词+循环"发展为成熟的架构模式体系。本文将系统梳理从最简单到最复杂的编排模式，帮助你在实际项目中做出正确的架构选择。 模式全景 复杂度 ──────────────────────────────────────────→ 串行 ─→ 并行 ─→ 路由 ─→ 循环 ─→ 递归 ─→ 图式 ─→ 自治 │ │ │ │ │ │ │ │ │ │ │ │ │ └─ Agent自主决策 │ │ │ │ │ └─ DAG + 条件边 + 子图 │ │ │ │ └─ 分治递归 │ │ │ └─ ReAct循环 / 反思循环 │ │ └─ if-else路由 / 内容分发 │ └─ 扇出-扇入 / Map-Reduce └─ Pipeline直线流程 模式1：串行编排（Pipeline） 最基础的编排模式，Agent按固定顺序执行。 ...

Agent记忆系统：让智能体拥有"记住"的能力 人类的记忆系统是一个精妙的分层结构：工作记忆处理当前任务，长期记忆存储知识和经验，情景记忆记录具体事件。2026年的Agent系统也在向这个方向演进——一个没有记忆的Agent就像一个每天失忆的员工，每次对话都从零开始，无法积累经验、无法理解上下文。 记忆系统三层架构 ┌───────────────────────────────────────────────┐ │ 记忆系统架构 │ ├───────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────┐ │ │ │ Layer 1: 短期记忆 (Working Memory) │ │ │ │ - 当前对话上下文 │ │ │ │ - 当前任务状态 │ │ │ │ - 最近的工具调用结果 │ │ │ │ 容量: ~32K tokens | TTL: 会话结束 │ │ │ └─────────────────────────────────────────┘ │ │ ↓ 溢出压缩 │ │ ┌─────────────────────────────────────────┐ │ │ │ Layer 2: 情景记忆 (Episodic Memory) │ │ │ │ - 具体事件记录（时间、地点、人物） │ │ │ │ - 对话历史摘要 │ │ │ │ - 任务执行日志 │ │ │ │ 容量: 无限 | TTL: 90天衰减 │ │ │ └─────────────────────────────────────────┘ │ │ ↓ 知识提取 │ │ ┌─────────────────────────────────────────┐ │ │ │ Layer 3: 长期记忆 (Long-term Memory) │ │ │ │ - 用户偏好和习惯 │ │ │ │ - 学到的知识和规则 │ │ │ │ - 经验教训 │ │ │ │ - 语义关联 │ │ │ │ 容量: 无限 | TTL: 永久 │ │ │ └─────────────────────────────────────────┘ │ │ │ └───────────────────────────────────────────────┘ Layer 1: 短期记忆（Working Memory） 短期记忆是Agent在当前会话中保持的上下文信息，主要通过LLM的上下文窗口实现。 ...