持久化

Agent记忆的三层架构 Agent需要像人类一样管理不同类型的记忆： 工作记忆：当前对话的上下文（短期） 情景记忆：过去交互的具体经历（中期） 语义记忆：从交互中提炼的知识（长期） 工作记忆：对话窗口管理 from collections import deque class WorkingMemory: def __init__(self, max_messages=20, max_tokens=4096): self.messages = deque(maxlen=max_messages) self.max_tokens = max_tokens self.token_counter = 0 def add(self, role, content): tokens = len(content) // 4 self.messages.append({"role": role, "content": content, "tokens": tokens}) self.token_counter += tokens # 超出token限制时移除最早的消息 while self.token_counter > self.max_tokens and len(self.messages) > 2: old = self.messages.popleft() self.token_counter -= old["tokens"] def get_context(self): return list(self.messages) def summarize_old_context(self): """当记忆过长时，摘要旧对话""" if len(self.messages) < 10: return old_messages = list(self.messages)[:len(self.messages)//2] summary = await self.summarize(old_messages) # 用摘要替换旧消息 for _ in range(len(old_messages)): self.messages.popleft() self.messages.appendleft({"role": "system", "content": f"之前的对话摘要：{summary}"}) 情景记忆：交互历史存储 class EpisodicMemory: def __init__(self, vector_store, llm): self.store = vector_store # 向量数据库 self.llm = llm async def save_interaction(self, user_id, user_message, assistant_response, metadata=None): """保存一次交互记录""" interaction = { "user_id": user_id, "user_message": user_message, "assistant_response": assistant_response, "timestamp": datetime.now().isoformat(), "metadata": metadata or {}, } # 生成摘要用于检索 summary = await self.llm.generate( f"用一句话概括这次交互：\n用户：{user_message}\n助手：{assistant_response}" ) # 向量化并存储 embedding = await self.llm.embed(summary) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps(interaction, ensure_ascii=False), metadata={"user_id": user_id, "summary": summary} ) async def recall(self, user_id, query, top_k=5): """检索相关的历史交互""" query_embedding = await self.llm.embed(query) results = await self.store.search( query_embedding, filter={"user_id": user_id}, top_k=top_k ) return [json.loads(r.document) for r in results] 语义记忆：知识图谱 class SemanticMemory: def __init__(self, graph_store): self.graph = graph_store # 图数据库（如Neo4j） async def learn(self, user_id, fact): """从交互中提取并存储知识""" # 使用LLM提取结构化知识 extracted = await self.extract_facts(fact) for fact_data in extracted: await self.graph.add_triple( subject=fact_data["subject"], predicate=fact_data["predicate"], object=fact_data["object"], metadata={"user_id": user_id, "source": "interaction"} ) async def query(self, user_id, entity): """查询关于某实体的知识""" triples = await self.graph.query( "MATCH (s)-[p]->(o) WHERE s.name = $entity RETURN s, p, o", {"entity": entity} ) knowledge = [] for triple in triples: knowledge.append(f"{triple['s']} {triple['p']} {triple['o']}") return knowledge async def extract_facts(self, text): """从文本中提取三元组""" prompt = f"""从以下文本中提取知识三元组（主语-谓语-宾语）： 文本：{text} 输出JSON数组： [{{"subject": "...", "predicate": "...", "object": "..."}}]""" result = await self.llm.generate(prompt) return json.loads(result) 记忆整合 class AgentMemorySystem: def __init__(self, working, episodic, semantic): self.working = working # 工作记忆 self.episodic = episodic # 情景记忆 self.semantic = semantic # 语义记忆 async def build_context(self, user_id, current_message): """构建完整的记忆上下文""" context_parts = [] # 1. 工作记忆（当前对话） working_ctx = self.working.get_context() context_parts.append({"type": "working", "messages": working_ctx}) # 2. 情景记忆（相关历史交互） episodic_results = await self.episodic.recall( user_id, current_message, top_k=3 ) if episodic_results: context_parts.append({ "type": "episodic", "memories": [r["summary"] for r in episodic_results] }) # 3. 语义记忆（相关知识） entities = await self.extract_entities(current_message) for entity in entities: knowledge = await self.semantic.query(user_id, entity) if knowledge: context_parts.append({ "type": "semantic", "entity": entity, "facts": knowledge }) return self.format_context(context_parts) def format_context(self, parts): """格式化记忆上下文""" context = "" for part in parts: if part["type"] == "working": context += "## 当前对话\n" for msg in part["messages"]: context += f"{msg['role']}: {msg['content']}\n" elif part["type"] == "episodic": context += "\n## 相关历史\n" for mem in part["memories"]: context += f"- {mem}\n" elif part["type"] == "semantic": context += f"\n## 关于{part['entity']}的知识\n" for fact in part["facts"]: context += f"- {fact}\n" return context 遗忘机制 class ForgettingMechanism: def __init__(self, decay_rate=0.01): self.decay_rate = decay_rate async def decay(self, memory_store): """时间衰减：降低旧记忆的重要性""" now = datetime.now() memories = await memory_store.get_all() for mem in memories: age_days = (now - mem["timestamp"]).days importance = mem.get("importance", 1.0) importance *= (1 - self.decay_rate) ** age_days if importance < 0.1: await memory_store.delete(mem["id"]) else: await memory_store.update(mem["id"], importance=importance) async def consolidate(self, memory_store): """记忆整合：将频繁出现的情景记忆转为语义记忆""" # 找出高频出现的事实 memories = await memory_store.get_all(min_importance=0.5) # 聚类相似记忆 clusters = self.cluster_memories(memories) for cluster in clusters: if len(cluster) >= 3: # 出现3次以上的模式转为知识 summary = await self.summarize_cluster(cluster) await self.semantic_memory.learn(summary) # 降低原始记忆的重要性 for mem in cluster: await memory_store.update(mem["id"], importance=mem["importance"] * 0.3) 存储选择 记忆类型 推荐存储 特点 工作记忆 内存（Redis） 快速读写，无需持久化 情景记忆 向量数据库 语义检索，按相似度召回 语义记忆 图数据库 关系查询，知识推理 结语 Agent记忆系统是构建长期智能助手的核心基础设施。工作记忆处理当前对话，情景记忆保存交互历史，语义记忆积累结构化知识。三层记忆的协同工作加上遗忘和整合机制，让Agent像人类一样"记得住该记的，忘得掉该忘的"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent的状态管理是一个容易被忽视但至关重要的架构问题。无状态的Agent就像一个没有记忆的人——每次对话都从零开始。而有状态的Agent则能记住过去的交互、维持任务进度、在故障后恢复执行。 2026年，随着Agent从简单对话走向复杂任务执行，状态管理已经成为Agent架构中最具挑战性的设计问题之一。 一、Agent状态的类型 1.1 对话状态 最基础的状态类型，包括： 当前对话的历史消息 用户的偏好和上下文 对话中的实体和指代关系 1.2 任务状态 Agent执行复杂任务时的进度状态： 当前任务的目标和约束 已完成的步骤和待完成的步骤 中间结果和待处理的数据 任务的时间线和依赖关系 1.3 环境状态 Agent所处环境的状态快照： 已连接的工具和服务 文件系统的当前状态 外部系统的状态（数据库、API等） 环境变量和配置 1.4 学习状态 Agent在学习过程中积累的状态： 已学到的策略和模式 反思记录和经验教训 性能基线和改进目标 二、状态管理架构 2.1 状态存储分层 ┌─────────────────────────────────┐ │ 内存状态（In-Memory） │ ← 最快，易失 ├─────────────────────────────────┤ │ 会话存储（Redis/Memcached） │ ← 快速，会话级 ├─────────────────────────────────┤ │ 持久存储（PostgreSQL/MongoDB） │ ← 可靠，长期 ├─────────────────────────────────┤ │ 归档存储（S3/OSS） │ ← 低成本，历史 └─────────────────────────────────┘ 2.2 状态序列化 Agent状态需要序列化后存储。2026年的主流方案： JSON序列化：可读性好，兼容性强。适合简单状态和小规模数据。 Protocol Buffers：高效紧凑，支持schema演进。适合大规模生产环境。 自定义二进制格式：极致性能，但维护成本高。适合特殊场景。 选择时需要考虑：序列化/反序列化速度、存储大小、schema演进支持、可读性需求。 2.3 状态版本控制 状态会随时间变化，需要版本控制来支持： 回滚：出问题时回退到之前的版本 审计：追踪状态变更历史 调试：复现问题时需要知道当时的状态 建议采用类似Git的版本控制策略：每次状态变更生成一个diff，按时间链存储。 ...

引言 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 的"状态"是什么？是对话历史、是工作流进度、是工具调用结果、是用户偏好、是 Agent 的"记忆"。无状态 Agent 简单但健忘；有状态 Agent 智能但复杂。2026年，随着 Agent 处理的任务越来越长（从分钟级到天级），状态管理成为架构设计的核心挑战。 一、Agent 状态的分类 ┌─────────────────────────────────────────────────────┐ │ Agent 状态全景图 │ ├──────────────┬──────────────────┬───────────────────┤ │ 状态类型 │ 生命周期 │ 存储介质 │ ├──────────────┼──────────────────┼───────────────────┤ │ 会话状态 │ 单次会话 │ 内存 / Redis │ │ (消息历史) │ 30分钟-24小时 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 工作流状态 │ 任务执行期间 │ Redis / 数据库 │ │ (执行进度) │ 分钟-天 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 用户状态 │ 用户生命周期 │ 数据库 │ │ (偏好/画像) │ 永久 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 检查点状态 │ 可恢复期间 │ 对象存储/数据库 │ │ (快照) │ 可配置 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 共享状态 │ 多Agent协作期间 │ Redis/共享存储 │ │ (黑板/消息) │ 会话级 │ │ └──────────────┴──────────────────┴───────────────────┘ 二、会话状态管理 2.1 会话状态模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum class MessageRole(Enum): SYSTEM = "system" USER = "user" ASSISTANT = "assistant" TOOL = "tool" @dataclass class Message: role: MessageRole content: str tool_calls: list | None = None tool_call_id: str | None = None timestamp: datetime = field(default_factory=datetime.now) metadata: dict = field(default_factory=dict) @dataclass class SessionState: """完整的会话状态""" session_id: str user_id: str agent_id: str messages: list[Message] = field(default_factory=list) context: dict = field(default_factory=dict) # 上下文变量 active_tools: list[str] = field(default_factory=list) pending_tool_calls: list[dict] = field(default_factory=list) created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) expires_at: datetime | None = None status: str = "active" # active / paused / completed / error 2.2 会话存储实现 class SessionStore: """会话状态存储——分层缓存策略""" def __init__(self, redis, postgres): self.redis = redis # 热数据 self.postgres = postgres # 冷数据/持久化 self.ttl = 86400 * 7 # 7天过期 async def get(self, session_id: str) -> SessionState | None: # L1: Redis 缓存 cached = await self.redis.get(f"session:{session_id}") if cached: return SessionState.from_json(cached) # L2: PostgreSQL row = await self.postgres.fetchrow( "SELECT * FROM agent_sessions WHERE session_id = $1", session_id ) if not row: return None state = SessionState.from_db(row) # 回填缓存 await self.redis.setex( f"session:{session_id}", 3600, # 缓存1小时 state.to_json() ) return state async def save(self, state: SessionState): state.updated_at = datetime.now() # 写入 Redis（热路径） await self.redis.setex( f"session:{state.session_id}", 3600, state.to_json() ) # 异步写入 PostgreSQL（冷路径） asyncio.create_task(self._persist_to_db(state)) async def _persist_to_db(self, state: SessionState): await self.postgres.execute(""" INSERT INTO agent_sessions (session_id, user_id, agent_id, state, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (session_id) DO UPDATE SET state = $4, updated_at = $5 """, state.session_id, state.user_id, state.agent_id, state.to_json(), state.updated_at) 2.3 上下文窗口管理 class ContextWindowManager: """管理 Agent 的上下文窗口""" def __init__(self, max_tokens: int = 128000): self.max_tokens = max_tokens self.reserved_for_output = 4096 self.available = max_tokens - self.reserved_for_output def prepare_context( self, system_prompt: str, messages: list[Message], tools_schema: list[dict] ) -> list[dict]: """在 Context Window 内准备消息""" # 计算各部分 Token system_tokens = self._count_tokens(system_prompt) tools_tokens = self._count_tokens(json.dumps(tools_schema)) remaining = self.available - system_tokens - tools_tokens # 从最新消息向前保留 prepared = [] total = 0 for msg in reversed(messages): msg_tokens = self._count_tokens(msg.content) if total + msg_tokens > remaining: break prepared.insert(0, msg) total += msg_tokens # 如果截断了，添加摘要提示 if len(prepared) < len(messages): summary = self._generate_summary(messages[:len(messages) - len(prepared)]) prepared.insert(0, Message( role=MessageRole.SYSTEM, content=f"[Earlier conversation summary: {summary}]" )) return prepared def _count_tokens(self, text: str) -> int: # 使用 tiktoken 精确计算 import tiktoken enc = tiktoken.encoding_for_model("gpt-5") return len(enc.encode(text)) 三、工作流状态与检查点 3.1 检查点机制 class CheckpointManager: """Agent 执行检查点管理""" def __init__(self, storage): self.storage = storage async def save_checkpoint( self, execution_id: str, state: WorkflowState, step_index: int, step_name: str ): """保存执行检查点""" checkpoint = Checkpoint( execution_id=execution_id, step_index=step_index, step_name=step_name, state=state, timestamp=datetime.now() ) await self.storage.save(checkpoint) # 保留最近 N 个检查点 await self._prune_old_checkpoints(execution_id, keep=10) async def restore(self, execution_id: str) -> tuple[WorkflowState, int]: """从最新检查点恢复""" latest = await self.storage.get_latest(execution_id) if not latest: raise NoCheckpointError(execution_id) logger.info( f"Restoring from checkpoint: step={latest.step_index}, " f"name={latest.step_name}" ) return latest.state, latest.step_index async def list_checkpoints(self, execution_id: str) -> list[Checkpoint]: """列出所有检查点（用于调试）""" return await self.storage.list(execution_id) @dataclass class WorkflowState: """工作流状态——可序列化""" execution_id: str current_step: str step_index: int results: dict # {step_name: result} variables: dict # 工作流变量 pending_actions: list # 待执行操作 error: str | None # 错误信息（如果有的话） iteration: int # 循环计数 def serialize(self) -> bytes: return pickle.dumps(self) # 或使用 JSON @classmethod def deserialize(cls, data: bytes) -> "WorkflowState": return pickle.loads(data) 3.2 可恢复的 Agent 扥行器 class ResumableAgentExecutor: """支持断点续传的 Agent 执行器""" def __init__(self, checkpoint_mgr: CheckpointManager): self.checkpoints = checkpoint_mgr async def execute( self, workflow: Workflow, initial_state: WorkflowState, execution_id: str | None = None ) -> WorkflowState: execution_id = execution_id or str(uuid.uuid4()) # 尝试从检查点恢复 try: state, start_step = await self.checkpoints.restore(execution_id) logger.info(f"Resuming from step {start_step}") except NoCheckpointError: state = initial_state start_step = 0 # 获取工作流步骤 steps = workflow.get_steps() for i, step in enumerate(steps[start_step:], start=start_step): try: # 执行前保存检查点 state.current_step = step.name state.step_index = i await self.checkpoints.save_checkpoint( execution_id, state, i, step.name ) # 执行步骤 result = await step.execute(state) # 更新状态 state.results[step.name] = result state.variables.update(result.get("variables", {})) # 条件分支 if step.condition: next_step = step.condition(result) if next_step: state.pending_actions = [next_step] except Exception as e: state.error = str(e) await self.checkpoints.save_checkpoint( execution_id, state, i, step.name ) # 重试逻辑 if step.retry_policy: retry_count = state.variables.get(f"retry_{step.name}", 0) if retry_count < step.retry_policy.max_attempts: state.variables[f"retry_{step.name}"] = retry_count + 1 await asyncio.sleep(step.retry_policy.backoff(retry_count)) # 重新执行当前步骤 continue raise return state 四、多 Agent 共享状态 4.1 黑板模式 class SharedBlackboard: """多 Agent 共享黑板""" def __init__(self, redis_client): self.redis = redis_client self.namespace = "blackboard" async def write( self, key: str, value: any, agent_id: str, ttl: int = 3600 ): """写入共享状态""" entry = { "value": value, "writer": agent_id, "timestamp": time.time(), } await self.redis.hset( f"{self.namespace}:{key}", mapping={k: json.dumps(v) for k, v in entry.items()} ) await self.redis.expire(f"{self.namespace}:{key}", ttl) # 通知订阅者 await self.redis.publish( f"{self.namespace}:updates", json.dumps({"key": key, "writer": agent_id}) ) async def read(self, key: str) -> any: """读取共享状态""" data = await self.redis.hgetall(f"{self.namespace}:{key}") if not data: return None return json.loads(data.get("value", "null")) async def subscribe( self, key_pattern: str, callback: callable ): """订阅状态变更""" pubsub = self.redis.pubsub() await pubsub.subscribe(f"{self.namespace}:updates") async for message in pubsub.listen(): if message["type"] == "message": data = json.loads(message["data"]) if fnmatch.fnmatch(data["key"], key_pattern): await callback(data) 4.2 Agent 间消息传递 class AgentMessageBus: """Agent 间异步消息总线""" def __init__(self, redis_client): self.redis = redis_client self.queues = {} # {agent_id: Queue} async def send( self, from_agent: str, to_agent: str, message_type: str, payload: dict, reply_to: str | None = None ): """发送消息给另一个 Agent""" msg = AgentMessage( id=str(uuid.uuid4()), from_agent=from_agent, to_agent=to_agent, type=message_type, payload=payload, reply_to=reply_to, timestamp=datetime.now() ) # 推入接收者的队列 await self.redis.lpush( f"agent:inbox:{to_agent}", msg.to_json() ) async def receive( self, agent_id: str, timeout: int = 30 ) -> AgentMessage | None: """接收消息""" result = await self.redis.brpop( f"agent:inbox:{agent_id}", timeout=timeout ) if result: return AgentMessage.from_json(result[1]) return None async def request_reply( self, from_agent: str, to_agent: str, message_type: str, payload: dict, timeout: int = 60 ) -> dict | None: """请求-回复模式""" reply_channel = f"reply:{uuid.uuid4()}" await self.send( from_agent, to_agent, message_type, payload, reply_to=reply_channel ) # 等待回复 result = await self.redis.brpop(reply_channel, timeout=timeout) if result: return json.loads(result[1]) return None 五、状态序列化与迁移 class StateSerializer: """状态序列化器""" SCHEMA_VERSION = "2.0" def serialize(self, state: any) -> str: """序列化状态为 JSON""" data = { "schema_version": self.SCHEMA_VERSION, "type": type(state).__name__, "data": self._to_dict(state), "timestamp": datetime.now().isoformat() } return json.dumps(data, ensure_ascii=False, default=str) def deserialize(self, raw: str) -> any: """反序列化""" data = json.loads(raw) # 版本迁移 if data["schema_version"] != self.SCHEMA_VERSION: data = self._migrate(data) return self._from_dict(data["type"], data["data"]) def _migrate(self, data: dict) -> dict: """状态版本迁移""" migrations = [ ("1.0", "1.1", self._migrate_1_0_to_1_1), ("1.1", "2.0", self._migrate_1_1_to_2_0), ] current = data["schema_version"] for from_v, to_v, migrator in migrations: if current == from_v: data = migrator(data) current = to_v return data 六、状态管理架构选型 场景 推荐方案 原因 短对话（< 30min） 内存 + Redis 低延迟、自动过期 长对话（> 1h） Redis + PostgreSQL 持久化 + 快速访问 长流程工作流 Redis + 检查点 + DB 断点续传 多 Agent 协作 Redis 黑板 + 消息总线 实时共享 用户画像 PostgreSQL + 向量DB 持久化 + 语义检索 跨设备同步 CRDT + Redis 冲突解决 七、状态管理 Checklist □ 会话状态分层存储（内存 → Redis → 数据库） □ 上下文窗口管理策略（滑动窗口 + 摘要） □ 工作流检查点定期保存 □ 检查点支持断点续传 □ 多 Agent 共享状态通过消息总线 □ 状态序列化支持版本迁移 □ 过期状态自动清理 □ 状态加密敏感字段 □ 状态变更审计日志 □ 状态一致性测试（并发读写） 结语 状态管理是 Agent 从"玩具"到"产品"的分水岭。无状态 Agent 是函数——输入即输出；有状态 Agent 是伙伴——它记得你、理解上下文、能从中断处继续。但状态也带来了复杂性：一致性、持久化、恢复、迁移。好的状态管理架构是透明的——开发者不需要关心状态的存储和恢复，Agent 始终如丝般顺滑地运行。这是工程的艺术。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...