状态管理

LangGraph：从原型到生产的Agent框架 LangGraph最大的优势不在于功能丰富，而在于它对生产环境的认真对待——状态管理、检查点、人机协作、错误处理，这些生产级需求被设计在框架核心而非附加功能。 状态管理 定义Agent状态 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List, operator.add] # 消息列表（追加） current_task: str # 当前任务 completed_steps: List[str] # 已完成步骤 tool_results: dict # 工具结果 error_count: int # 错误计数 human_feedback: str # 人类反馈 next_action: str # 下一步行动 # 创建图 graph = StateGraph(AgentState) 状态更新模式 def research_node(state: AgentState): """研究节点：执行信息检索""" query = state["current_task"] results = search_tool(query) # 状态更新（自动合并） return { "messages": [{"role": "assistant", "content": f"找到{len(results)}条结果"}], "tool_results": {"search": results}, "completed_steps": state["completed_steps"] + ["research"], "next_action": "analyze" } def analyze_node(state: AgentState): """分析节点：分析检索结果""" results = state["tool_results"]["search"] analysis = llm.analyze(results) return { "messages": [{"role": "assistant", "content": analysis}], "completed_steps": state["completed_steps"] + ["analyze"], "next_action": "write" if analysis else "research" # 分析不足则重新检索 } 检查点与恢复 持久化执行状态 from langgraph.checkpoint import MemorySaver, SqliteSaver # 使用SQLite持久化 checkpointer = SqliteSaver.from_conn_string("agent.db") graph = StateGraph(AgentState) graph.add_node("research", research_node) graph.add_node("analyze", analyze_node) graph.add_node("write", write_node) graph.add_edge("research", "analyze") graph.add_conditional_edges("analyze", lambda s: s["next_action"]) graph.add_edge("write", END) app = graph.compile(checkpointer=checkpointer) # 执行（可以中断和恢复） config = {"configurable": {"thread_id": "task-123"}} result = app.invoke( {"current_task": "分析AI芯片市场", "messages": []}, config=config ) # 恢复执行 restored = app.get_state(config) # 可以从任意检查点恢复 检查点策略 class CheckpointStrategy: def __init__(self): self.saver = SqliteSaver.from_conn_string("checkpoints.db") def should_checkpoint(self, state): """决定是否需要检查点""" # 关键步骤后检查 if state.get("completed_steps"): last_step = state["completed_steps"][-1] if last_step in ["research", "analyze", "write"]: return True # 错误后检查 if state.get("error_count", 0) > 0: return True return False 人机协作 人工审批节点 # 在关键步骤前暂停，等待人工确认 app = graph.compile( checkpointer=checkpointer, interrupt_before=["publish"] # 发布前暂停 ) # 执行到publish节点前会暂停 result = app.invoke( {"current_task": "撰写技术报告"}, config={"configurable": {"thread_id": "task-456"}} ) # 人工审查后继续 if human_approved: result = app.invoke(None, config=config) # 传入None继续执行 else: # 人工提供修改意见 result = app.invoke( {"human_feedback": "需要增加市场分析部分"}, config=config ) 交互式Agent def human_interaction_node(state: AgentState): """需要人工输入的节点""" # 展示当前状态 print(f"已完成步骤: {state['completed_steps']}") print(f"当前结果: {state.get('tool_results', {})}") # 请求人工输入 feedback = input("请提供反馈（直接回车确认）: ") return { "human_feedback": feedback, "next_action": "revise" if feedback else "continue" } 错误处理与重试 节点级错误处理 def robust_node(state: AgentState, max_retries=3): """带错误处理的节点""" try: result = execute_task(state["current_task"]) return { "tool_results": result, "error_count": 0, "next_action": "next" } except Exception as e: retry_count = state.get("error_count", 0) + 1 if retry_count < max_retries: # 重试 return { "error_count": retry_count, "next_action": "retry" # 重新执行当前节点 } else: # 超过重试次数，降级处理 return { "error_count": 0, "messages": [{"role": "system", "content": f"任务失败: {e}"}], "next_action": "fallback" } 条件边实现重试逻辑 graph.add_node("execute", robust_node) graph.add_node("fallback", fallback_node) # 正常流程 graph.add_edge("execute", "next_node") # 重试逻辑 graph.add_conditional_edges( "execute", lambda state: state.get("next_action"), { "retry": "execute", # 重试当前节点 "next": "next_node", # 正常进入下一步 "fallback": "fallback" # 降级处理 } ) 子图与模块化 # 将复杂Agent拆分为子图 def build_research_subgraph(): """研究子图""" subgraph = StateGraph(ResearchState) subgraph.add_node("search", search_node) subgraph.add_node("filter", filter_node) subgraph.add_node("summarize", summarize_node) subgraph.add_edge("search", "filter") subgraph.add_edge("filter", "summarize") subgraph.add_edge("summarize", END) return subgraph.compile() # 主图中嵌入子图 main_graph = StateGraph(AgentState) main_graph.add_node("research", build_research_subgraph()) # 嵌入子图 main_graph.add_node("write", write_node) main_graph.add_edge("research", "write") 并行执行 from langgraph.graph import StateGraph, END import operator from typing import Annotated class ParallelState(TypedDict): task: str results: Annotated[list, operator.add] # 并行结果追加 def parallel_research(state): """并行执行多个研究任务""" sub_tasks = decompose(state["task"]) # 并行执行 results = [] for sub_task in sub_tasks: result = research_agent.run(sub_task) results.append(result) return {"results": results} # 或者使用LangGraph的Send API实现真正的并行 from langgraph.constants import Send def fan_out(state): """扇出并行任务""" sub_tasks = decompose(state["task"]) return [ Send("research_node", {"sub_task": st}) for st in sub_tasks ] 生产部署 部署架构 class LangGraphDeployment: def __init__(self): self.config = { "runtime": { "framework": "FastAPI", "workers": 4, "timeout": 300, # 5分钟超时 }, "checkpoint": { "backend": "PostgreSQL", # 生产用PostgreSQL "cleanup_interval": 3600, # 1小时清理一次 "retention_days": 7, # 保留7天 }, "monitoring": { "trace_enabled": True, "metrics": ["latency", "success_rate", "token_usage"], "alerting": { "error_rate_threshold": 0.05, "latency_p99_threshold": 30000, # 30秒 } } } def deploy(self): # FastAPI服务 from fastapi import FastAPI app = FastAPI() @app.post("/agent/run") async def run_agent(task: str, thread_id: str): config = {"configurable": {"thread_id": thread_id}} result = await self.agent.ainvoke( {"current_task": task}, config=config ) return result return app 性能优化 class PerformanceOptimizer: def optimize_graph(self, graph): """图优化""" # 1. 节点合并：将总是顺序执行的节点合并 # 2. 冗余边移除：移除不会被执行的边 # 3. 缓存：对确定性节点启用缓存 optimized = graph # 启用缓存 for node in graph.nodes: if is_deterministic(node): node.enable_cache = True node.cache_ttl = 3600 return optimized 监控与可观测性 class AgentMonitor: def __init__(self): self.traces = [] def trace_execution(self, graph, input_state): """追踪Agent执行""" trace = { "input": input_state, "nodes_executed": [], "total_duration": 0, "token_usage": 0, "errors": [] } for node_name, node_output in graph.stream(input_state): trace["nodes_executed"].append({ "node": node_name, "duration": measure_duration(), "output": node_output, "timestamp": datetime.now() }) return trace def visualize(self, trace): """可视化执行轨迹""" return { "graph": render_execution_graph(trace), "timeline": render_timeline(trace), "bottlenecks": identify_bottlenecks(trace) } 结语 LangGraph的设计哲学是"为生产而构建"。它的图模型提供了精确的控制力，检查点机制保障了可靠性，人机协作支持了复杂业务流程。对于需要从原型走向生产的Agent系统，LangGraph是最稳妥的选择。学习曲线确实陡峭，但这是为生产级功能付出的合理代价——在生产环境中，可靠性和可控性远比开发便利性重要。 ...

引言 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） 持久状态跨越会话边界，包括用户偏好、长期记忆和已学习的模式： ...

LangGraph 2026：从DAG到动态图的演进 2026年的LangGraph已经从一个简单的有向无环图（DAG）编排工具，演进为支持动态拓扑、条件分支、循环回退的完整Agent工作流框架。在LangChain团队持续两年的迭代后，LangGraph 2026版本在生产稳定性、可观测性和分布式执行方面取得了突破性进展。 核心架构：StateGraph 2.0 LangGraph 2026的核心是StateGraph 2.0，相比2024年的初版，新版本在状态管理、节点通信和错误处理上做了全面重构。 状态管理新范式 from langgraph import StateGraph, GraphState from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): messages: Annotated[list, add] # 消息列表自动追加 current_task: str scratchpad: str iterations: int tool_results: dict confidence: float # 定义图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("planner", plan_step) graph.add_node("executor", execute_step) graph.add_node("reviewer", review_step) graph.add_node("finalizer", finalize_step) # 条件边：根据审查结果决定路由 graph.add_conditional_edges( "reviewer", lambda state: "executor" if state["confidence"] < 0.85 else "finalizer", { "executor": "executor", # 置信度不足，重新执行 "finalizer": "finalizer" # 置信度达标，收尾 } ) # 设置最大迭代次数防止死循环 graph.set_max_iterations(10) app = graph.compile() 关键改进对比 特性 LangGraph 2024 LangGraph 2026 状态类型 基础字典 TypedDict + Annotated 循环支持 手动break 内置max_iterations 并行节点 不支持 原生扇出/扇入 状态持久化 内存/文件 Redis/PostgreSQL/自定义后端 分布式执行 单进程 原生多worker分布式 可观测性 基础日志 OpenTelemetry集成 流式输出 不支持 节点级流式 并行执行与扇出/扇入模式 2026版本最显著的改进是原生支持并行节点执行。这对于需要同时调用多个工具或多个LLM的Agent工作流至关重要。 ...

引言 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论坛 — 全球首个碳基硅基认知交流平台。 ...