架构设计

Agent可观测性：你无法优化你看不见的东西 传统软件的可观测性已经相当成熟——我们有Prometheus做指标、ELK做日志、Jaeger做追踪。但Agent系统引入了新的可观测性挑战：非确定性的执行路径、不可预测的token消耗、LLM输出的质量评估、多Agent协作的链路追踪。2026年，Agent可观测性已经形成了一套独立的最佳实践。 Agent可观测性的三大支柱 ┌───────────────────────────────────────────────────┐ │ Agent 可观测性三大支柱 │ ├─────────────┬─────────────┬───────────────────────┤ │ │ │ │ │ 追踪 │ 日志 │ 指标 │ │ (Traces) │ (Logs) │ (Metrics) │ │ │ │ │ │ Agent执行 │ 结构化事件 │ 性能计数器 │ │ 链路追踪 │ 决策记录 │ 资源消耗 │ │ 跨Agent关联 │ 上下文快照 │ 质量评估 │ │ │ │ │ └─────────────┴─────────────┴───────────────────────┘ 1. 分布式追踪 Agent Trace模型 Agent系统的追踪比传统微服务更复杂，因为一个"请求"可能涉及多轮LLM调用、多次工具调用、跨多个Agent。 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional import uuid @dataclass class AgentSpan: """Agent追踪的基本单元""" span_id: str = field(default_factory=lambda: str(uuid.uuid4())) parent_id: Optional[str] = None trace_id: str = "" # 基本信息 name: str = "" # span名称 span_type: str = "" # llm_call / tool_call / agent_step / sub_agent agent_name: str = "" # 哪个Agent start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None # Agent特有信息 input_data: Any = None output_data: Any = None model: str = "" # 使用的LLM模型 prompt_tokens: int = 0 completion_tokens: int = 0 cost: float = 0.0 # 状态 status: str = "ok" # ok / error / timeout error_message: Optional[str] = None # 元数据 attributes: dict = field(default_factory=dict) def finish(self, output=None, status="ok", error=None): self.end_time = datetime.now() self.output_data = output self.status = status self.error_message = error @property def duration_ms(self) -> float: if self.end_time: return (self.end_time - self.start_time).total_seconds() * 1000 return 0 追踪实现 class AgentTracer: """Agent追踪器""" def __init__(self, service_name="agent-service"): self.service_name = service_name self.spans: list[AgentSpan] = [] self.current_span_stack: list[AgentSpan] = [] def start_trace(self, name: str, agent_name: str) -> AgentSpan: """开始一个新的追踪（根span）""" trace_id = str(uuid.uuid4()) span = AgentSpan( trace_id=trace_id, name=name, span_type="agent_step", agent_name=agent_name ) self.spans.append(span) self.current_span_stack.append(span) return span def start_span( self, name: str, span_type: str, agent_name: str = "", input_data: Any = None ) -> AgentSpan: """开始一个子span""" parent = self.current_span_stack[-1] if self.current_span_stack else None span = AgentSpan( trace_id=parent.trace_id if parent else str(uuid.uuid4()), parent_id=parent.span_id if parent else None, name=name, span_type=span_type, agent_name=agent_name, input_data=input_data ) self.spans.append(span) self.current_span_stack.append(span) return span def end_span(self, span: AgentSpan, output=None, status="ok", error=None): """结束一个span""" span.finish(output, status, error) if self.current_span_stack and self.current_span_stack[-1].span_id == span.span_id: self.current_span_stack.pop() def get_trace_tree(self, trace_id: str) -> dict: """获取追踪树""" trace_spans = [s for s in self.spans if s.trace_id == trace_id] return self._build_tree(trace_spans, parent_id=None) def _build_tree(self, spans: list[AgentSpan], parent_id: str | None) -> dict: children = [s for s in spans if s.parent_id == parent_id] return [ { "span_id": s.span_id, "name": s.name, "type": s.span_type, "agent": s.agent_name, "duration_ms": s.duration_ms, "tokens": s.prompt_tokens + s.completion_tokens, "cost": s.cost, "status": s.status, "children": self._build_tree(spans, s.span_id) } for s in children ] 使用示例 tracer = AgentTracer() # 开始追踪 root = tracer.start_trace("用户咨询", "router_agent") # Agent执行LLM调用 llm_span = tracer.start_span("LLM调用", "llm_call", "router_agent", input_data="用户问题") response = await llm.complete("...") tracer.end_span(llm_span, output=response.text, status="ok") llm_span.prompt_tokens = response.usage.prompt_tokens llm_span.completion_tokens = response.usage.completion_tokens llm_span.cost = calculate_cost(response.usage, "gpt-4o") # Agent调用工具 tool_span = tracer.start_span("搜索知识库", "tool_call", "router_agent") results = await knowledge_base.search("query") tracer.end_span(tool_span, output=results) # 路由到子Agent sub_agent_span = tracer.start_span("专家Agent处理", "sub_agent", "expert_agent") # ... 子Agent内部会有自己的span ... tracer.end_span(sub_agent_span, output="最终回答") # 结束追踪 tracer.end_span(root, output="最终回答") # 查看追踪树 trace_tree = tracer.get_trace_tree(root.trace_id) 追踪可视化 追踪树可以渲染为瀑布图： ...

多Agent系统：从单体到协同的架构挑战 2026年，随着Agent能力的提升，单个Agent已经难以处理复杂的企业任务。多Agent系统（Multi-Agent System, MAS）成为解决复杂问题的主流架构。但多Agent也带来了新的架构挑战：Agent之间如何通信？如何协调行动？当多个Agent得出矛盾结论时如何解决？ 核心挑战 ┌─────────────────────────────────────────────────┐ │ 多Agent系统的三大核心挑战 │ ├─────────────────────────────────────────────────┤ │ │ │ 1. 通信 (Communication) │ │ Agent A 如何告诉 Agent B 它需要什么？ │ │ - 消息格式 │ │ - 通信协议 │ │ - 异步 vs 同步 │ │ │ │ 2. 协调 (Coordination) │ │ 多个Agent如何协同完成一个任务？ │ │ - 任务分配 │ │ - 资源竞争 │ │ - 死锁避免 │ │ │ │ 3. 冲突解决 (Conflict Resolution) │ │ 当Agent意见不一致时怎么办？ │ │ - 投票机制 │ │ - 仲裁机制 │ │ - 置信度加权 │ │ │ └─────────────────────────────────────────────────┘ 1. 通信协议设计 消息格式标准化 在多Agent系统中，统一的消息格式是互操作的基础： ...

引言 混合专家（Mixture of Experts, MoE）架构通过将模型参数划分为多个"专家"子网络，每次推理只激活其中一部分，实现了参数规模与计算量的解耦。这一设计使得MoE模型在保持稠密模型性能的同时，大幅降低了推理成本。2026年，MoE已成为主流大模型（Mixtral、DeepSeek-V3、Qwen3-MoE）的核心架构。本文将深入解析MoE的路由机制与负载均衡策略。 MoE基础架构 标准MoE Layer MoE层由 $N$ 个专家网络和1个门控网络（Router/Gating Network）组成： class MoELayer(nn.Module): def __init__(self, d_model, n_experts, n_active, expert_dim): super().__init__() self.n_experts = n_experts self.n_active = n_active # Top-K激活专家数 self.gate = nn.Linear(d_model, n_experts, bias=False) self.experts = nn.ModuleList([ ExpertBlock(d_model, expert_dim) for _ in range(n_experts) ]) def forward(self, x): """ x: [batch_size, seq_len, d_model] """ B, T, D = x.shape # 门控分数 gate_logits = self.gate(x) # [B, T, n_experts] gate_probs = F.softmax(gate_logits, dim=-1) # Top-K选择 topk_probs, topk_indices = gate_probs.topk(self.n_active, dim=-1) # 重新归一化 topk_probs = topk_probs / topk_probs.sum(dim=-1, keepdim=True) # 专家计算（分散式） output = torch.zeros_like(x) for i in range(self.n_active): expert_idx = topk_indices[..., i] # [B, T] expert_weight = topk_probs[..., i].unsqueeze(-1) # [B, T, 1] # 对每个专家ID，批量计算 for eid in range(self.n_experts): mask = (expert_idx == eid) if mask.any(): expert_input = x[mask] expert_output = self.experts[eid](expert_input) output[mask] += expert_weight[mask] * expert_output return output 主流MoE模型对比 模型 总参数 激活参数 专家数 Top-K 特点 Mixtral 8x7B 47B 13B 8 2 首批开源MoE DeepSeek-V2 236B 21B 160 6 细粒度专家 DeepSeek-V3 671B 37B 256 8 共享专家+细粒度 Qwen3-235B 235B 35B 128 8 GQA+MoE Mixtral 8x22B 141B 39B 8 2 大规模MoE 路由机制详解 Top-K路由 标准Top-K路由： ...

MoE（Mixture of Experts）已成为2026年大模型架构的事实标准。从DeepSeek-V3到GPT-5，MoE让模型在保持推理效率的同时实现万亿级参数规模。本文将深入MoE的技术内核。 1. MoE架构基础 1.1 从稠密到稀疏 标准Transformer的FFN层对所有输入执行相同计算。MoE将FFN替换为多个"专家"网络，每个token仅激活少数专家： $$\text{FFN}{MoE}(x) = \sum{i \in \mathcal{I}(x)} g_i(x) \cdot E_i(x)$$ 其中 $\mathcal{I}(x)$ 为路由到token $x$ 的专家索引集合，$g_i(x)$ 为门控权重，$E_i$ 为第 $i$ 个专家。 1.2 参数与计算量的解耦 MoE的关键优势：参数总量与计算量解耦。 模型 总参数 激活参数 FLOPs/token 等效稠密模型 Mixtral 8x7B 47B 13B ~13B ~13B DeepSeek-V3 671B 37B ~37B ~37B GPT-5 (估) 3T 300B ~300B ~300B MoE用3T参数获得了300B稠密模型的效果，但推理仅需300B的计算量。 2. 路由机制详解 2.1 Top-K路由 最基础也最常用的路由策略： class TopKRouter(nn.Module): def __init__(self, d_model, num_experts, top_k=2): super().__init__() self.gate = nn.Linear(d_model, num_experts, bias=False) self.top_k = top_k self.num_experts = num_experts def forward(self, x): # x: [batch * seq_len, d_model] logits = self.gate(x) # [batch * seq_len, num_experts] # Top-K选择 topk_logits, topk_indices = torch.topk(logits, self.top_k, dim=-1) # Softmax归一化（仅对选中的专家） weights = F.softmax(topk_logits, dim=-1) return weights, topk_indices 2.2 Expert Choice路由 反转路由方向：不是token选专家，而是专家选token。 ...

2026年是AI Agent元年。如果说2023-2024年大家在探索"Agent能做什么"，2025年大家在尝试"怎么构建Agent"，那么2026年的核心问题已经变成"用哪个框架构建Agent"。LangChain（含LangGraph）、Microsoft AutoGen、CrewAI三足鼎立，各有千秋。本文将通过架构分析、功能对比和真实项目测试，给出最全面的选型指南。 一、三大框架概览 LangChain + LangGraph 1.0 LangChain在2025年底完成了架构重组，将核心库精简为langchain-core，同时LangGraph从实验项目升级为正式的Agent编排框架。 定位：全功能AI应用开发框架 核心组件： LangChain：链式调用、工具集成、文档处理 LangGraph：基于状态图的Agent编排，支持循环、分支、并行 LangSmith：调试、监控、评估平台 LangServe：API部署 2026年关键更新： LangGraph 1.0正式发布，引入"持久化状态"和"人在环路"原生支持 预构建Agent模板：ReAct、Plan-and-Execute、Reflexion等 原生支持Claude 5的Computer Use能力 Microsoft AutoGen 0.4 AutoGen在2026年发布了0.4版本（经过重大重写），从"对话式多Agent"进化为"事件驱动的异步Agent框架"。 定位：多Agent对话与协作框架 核心组件： Agent Runtime：异步消息传递的Agent运行时 GroupChat：多Agent群聊协作 Code Executor：安全的代码执行沙箱 AgentChat：高级API，简化多Agent对话 2026年关键更新： 完全异步架构，支持高并发 跨语言支持：Python + .NET 与Microsoft Azure AI深度集成 新增"Agent Workflow"可视化编排界面 CrewAI 0.5 CrewAI在2026年获得了$2000万A轮融资，成为增长最快的Agent框架。它专注于"角色扮演式多Agent协作"——定义不同的Agent角色，分配任务，让他们像团队一样工作。 定位：简洁直观的多Agent协作框架 核心组件： Crew：Agent团队定义 Agent：角色定义（角色、目标、背景故事） Task：任务定义（描述、期望输出、分配的Agent） Process：任务执行模式（顺序/层级/共识） Tools：工具集成 2026年关键更新： 新增"共识决策"模式——Agent通过投票/讨论达成共识 支持Agent记忆持久化 内置多种行业模板（研究、营销、开发等） CrewAI Cloud：托管部署平台 二、架构设计对比 核心架构差异 LangGraph：状态机模型 LangGraph将Agent工作流建模为状态图（StateGraph）： 每个节点是一个处理步骤（LLM调用、工具调用、条件判断） 边定义了状态转移规则 支持循环（用于ReAct等迭代推理模式） 状态在节点间传递，可持久化 优势：精确控制流程，支持复杂工作流，可调试性强 劣势：学习曲线陡，简单任务显得过度工程化 ...

引言 Agent 的执行过程是黑盒——你看到输入和输出，但中间发生了什么？调用了什么工具？为什么选择这个路径？Token 花在哪里？结构化日志是打开这个黑盒的钥匙。2026年，随着 Agent 系统复杂度增长，日志不再是"给人看的文本"，而是"给系统查询的数据"。 一、Agent 日志设计原则 传统日志 vs Agent 日志 维度 传统日志 Agent 日志 格式 半结构化文本 全结构化 JSON 粒度 请求级 步骤级（每轮迭代） 关联 request_id trace_id + session_id + step_id 内容 状态和错误 决策推理、工具调用、Token消耗 用途 故障排查 故障排查 + 质量分析 + 成本归因 查询 grep/正则 结构化查询 + 聚合分析 设计原则 一切皆结构化：每条日志都是可查询的 JSON 因果链完整：从输入到输出的每一步都可追溯 上下文丰富：每条日志携带足够的上下文独立理解 成本感知：Token 和费用信息嵌入每条日志 隐私安全：PII 自动脱敏 二、日志数据模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum import uuid class LogLevel(Enum): DEBUG = "debug" INFO = "info" WARN = "warn" ERROR = "error" CRITICAL = "critical" class EventType(Enum): # Agent 生命周期 AGENT_START = "agent.start" AGENT_END = "agent.end" AGENT_INTERRUPT = "agent.interrupt" # LLM 交互 LLM_REQUEST = "llm.request" LLM_RESPONSE = "llm.response" LLM_ERROR = "llm.error" LLM_RETRY = "llm.retry" # 工具调用 TOOL_DECISION = "tool.decision" TOOL_CALL_START = "tool.call.start" TOOL_CALL_END = "tool.call.end" TOOL_ERROR = "tool.error" # 决策推理 REASONING = "reasoning" PLANNING = "planning" REFLECTION = "reflection" # 状态变更 STATE_UPDATE = "state.update" CONTEXT_PRUNED = "context.pruned" # 错误恢复 ERROR_RECOVERY = "error.recovery" FALLBACK_TRIGGERED = "fallback.triggered" @dataclass class AgentLogEntry: """Agent 结构化日志条目""" # 标识 log_id: str = field(default_factory=lambda: str(uuid.uuid4())) timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) # 关联 trace_id: str = "" # 贯穿整个请求 session_id: str = "" # 会话ID step_id: str = "" # 当前步骤ID parent_step_id: str = "" # 父步骤（用于嵌套） # 事件 event_type: EventType = EventType.AGENT_START level: LogLevel = LogLevel.INFO # Agent 信息 agent_name: str = "" agent_version: str = "" # 内容 message: str = "" data: dict = field(default_factory=dict) # 性能 duration_ms: float = 0 # 成本 tokens_in: int = 0 tokens_out: int = 0 cost_usd: float = 0 # 上下文 user_id: str = "" tenant_id: str = "" environment: str = "production" 三、结构化日志实现 3.1 日志记录器 import structlog from structlog.contextvars import bind_contextvars, clear_contextvars class AgentLogger: """Agent 结构化日志记录器""" def __init__(self): self.logger = structlog.get_logger("agent") def bind_request_context( self, trace_id: str, session_id: str, user_id: str, agent_name: str, agent_version: str ): """绑定请求级上下文""" bind_contextvars( trace_id=trace_id, session_id=session_id, user_id=user_id, agent_name=agent_name, agent_version=agent_version, ) def log_agent_start( self, query: str, available_tools: list[str], max_iterations: int ): """记录 Agent 启动""" self.logger.info( "agent.start", event_type=EventType.AGENT_START.value, query_preview=query[:200], query_length=len(query), available_tools=available_tools, max_iterations=max_iterations, ) def log_llm_call( self, model: str, messages_count: int, input_tokens: int, temperature: float, tools_provided: bool ): """记录 LLM 调用""" self.logger.info( "llm.request", event_type=EventType.LLM_REQUEST.value, model=model, messages_count=messages_count, input_tokens=input_tokens, temperature=temperature, tools_provided=tools_provided, ) def log_llm_response( self, model: str, output_tokens: int, duration_ms: float, cost_usd: float, tool_calls: list[dict] | None, finish_reason: str ): """记录 LLM 响应""" self.logger.info( "llm.response", event_type=EventType.LLM_RESPONSE.value, model=model, output_tokens=output_tokens, duration_ms=round(duration_ms, 2), cost_usd=round(cost_usd, 6), tool_calls_count=len(tool_calls) if tool_calls else 0, tool_calls=[ {"tool": tc["function"]["name"], "args_preview": str(tc["function"]["arguments"])[:100]} for tc in (tool_calls or []) ], finish_reason=finish_reason, ) def log_tool_decision( self, selected_tool: str, available_tools: list[str], reasoning: str, confidence: float | None = None ): """记录工具选择决策""" self.logger.debug( "tool.decision", event_type=EventType.TOOL_DECISION.value, selected_tool=selected_tool, available_tools=available_tools, reasoning=reasoning, confidence=confidence, ) def log_tool_execution( self, tool_name: str, args: dict, result: any, duration_ms: float, success: bool, error: str | None = None ): """记录工具执行""" log_data = { "event_type": EventType.TOOL_CALL_END.value, "tool": tool_name, "args_preview": self._truncate_args(args), "duration_ms": round(duration_ms, 2), "success": success, } if success: log_data["result_preview"] = str(result)[:500] log_data["result_size"] = len(str(result)) else: log_data["error"] = error self.logger.info("tool.call.end", **log_data) def log_reasoning( self, step: int, thought: str, action: str, observation: str ): """记录 ReAct 推理过程""" self.logger.debug( "reasoning", event_type=EventType.REASONING.value, step=step, thought=thought[:500], action=action, observation=observation[:500], ) def log_agent_end( self, total_iterations: int, total_tokens_in: int, total_tokens_out: int, total_cost: float, total_duration_ms: float, tools_used: list[str], status: str ): """记录 Agent 结束""" self.logger.info( "agent.end", event_type=EventType.AGENT_END.value, total_iterations=total_iterations, total_tokens_in=total_tokens_in, total_tokens_out=total_tokens_out, total_cost_usd=round(total_cost, 6), total_duration_ms=round(total_duration_ms, 2), tools_used=tools_used, status=status, ) def _truncate_args(self, args: dict, max_len: int = 200) -> dict: """截断过长的参数""" result = {} for k, v in args.items(): s = str(v) result[k] = s[:max_len] + "..." if len(s) > max_len else v return result def clear_context(self): """清理上下文""" clear_contextvars() 3.2 日志中间件 class AgentLoggingMiddleware: """Agent 日志中间件——自动记录""" def __init__(self, logger: AgentLogger): self.logger = logger async def wrap_agent( self, agent: Agent, request: Request ) -> Response: """包装 Agent 执行，自动记录日志""" trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4())) # 绑定上下文 self.logger.bind_request_context( trace_id=trace_id, session_id=request.session_id, user_id=request.user_id, agent_name=agent.name, agent_version=agent.version ) start_time = time.time() try: # 记录启动 self.logger.log_agent_start( query=request.query, available_tools=agent.get_tool_names(), max_iterations=agent.max_iterations ) # 执行 Agent（内部会通过回调记录各步骤） response = await agent.run(request.query) # 记录结束 self.logger.log_agent_end( total_iterations=agent.iteration_count, total_tokens_in=agent.total_input_tokens, total_tokens_out=agent.total_output_tokens, total_cost=agent.total_cost, total_duration_ms=(time.time() - start_time) * 1000, tools_used=agent.tools_used, status="success" ) return response except Exception as e: self.logger.logger.error( "agent.error", event_type="agent.error", error_type=type(e).__name__, error_message=str(e), duration_ms=(time.time() - start_time) * 1000, ) raise finally: self.logger.clear_context() 四、日志输出配置 import structlog import logging import sys def configure_logging(environment: str = "production"): """配置结构化日志""" if environment == "production": # 生产环境：JSON 输出到 stdout structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), _add_server_info(), _pii_redactor(), # PII 脱敏 structlog.processors.JSONRenderer(ensure_ascii=False), ], wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), logger_factory=structlog.PrintLoggerFactory(), ) elif environment == "development": # 开发环境：彩色控制台输出 structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.dev.ConsoleRenderer(colors=True), ], ) # 同时写入文件（轮转） file_handler = logging.handlers.RotatingFileHandler( "/var/log/agent/agent.log", maxBytes=100_000_000, # 100MB backupCount=10, ) file_handler.setFormatter( logging.Formatter('{"message": "%(message)s"}') ) def _add_server_info(): """添加服务器信息处理器""" def processor(logger, method_name, event_dict): event_dict["hostname"] = socket.gethostname() event_dict["pid"] = os.getpid() return event_dict return processor def _pii_redactor(): """PII 脱敏处理器""" patterns = { "email": (r'[\w.-]+@[\w.-]+\.\w+', '[REDACTED_EMAIL]'), "phone": (r'\b1[3-9]\d{9}\b', '[REDACTED_PHONE]'), "id_card": (r'\b\d{17}[\dXx]\b', '[REDACTED_ID]'), } def redact(text: str) -> str: for pattern, replacement in patterns.values(): text = re.sub(pattern, replacement, text) return text def processor(logger, method_name, event_dict): for key, value in event_dict.items(): if isinstance(value, str): event_dict[key] = redact(value) elif isinstance(value, dict): event_dict[key] = { k: redact(v) if isinstance(v, str) else v for k, v in value.items() } return event_dict return processor 五、日志查询与分析 5.1 日志查询接口 class AgentLogQuery: """Agent 日志查询接口""" async def get_trace(self, trace_id: str) -> list[dict]: """获取完整执行链路""" return await self.elasticsearch.search( index="agent-logs-*", body={ "query": {"term": {"trace_id": trace_id}}, "sort": [{"timestamp": "asc"}] } ) async def find_slow_agents( self, threshold_ms: float = 30000, time_range: str = "1h" ) -> list[dict]: """查找慢 Agent 执行""" return await self.elasticsearch.search( index="agent-logs-*", body={ "query": { "bool": { "filter": [ {"term": {"event_type": "agent.end"}}, {"range": { "total_duration_ms": {"gte": threshold_ms} }}, {"range": { "timestamp": {"gte": f"now-{time_range}"} }} ] } }, "sort": [{"total_duration_ms": "desc"}], "size": 50 } ) async def find_expensive_sessions( self, min_cost: float = 0.10, time_range: str = "24h" ) -> list[dict]: """查找高成本会话""" return await self.elasticsearch.search( index="agent-logs-*", body={ "query": { "bool": { "filter": [ {"term": {"event_type": "agent.end"}}, {"range": {"total_cost_usd": {"gte": min_cost}}}, {"range": {"timestamp": {"gte": f"now-{time_range}"}}} ] } }, "sort": [{"total_cost_usd": "desc"}] } ) async def get_tool_failure_rate( self, time_range: str = "1h" ) -> dict: """工具失败率统计""" result = await self.elasticsearch.search( index="agent-logs-*", body={ "size": 0, "query": { "bool": { "filter": [ {"term": {"event_type": "tool.call.end"}}, {"range": {"timestamp": {"gte": f"now-{time_range}"}}} ] } }, "aggs": { "by_tool": { "terms": {"field": "tool"}, "aggs": { "success_count": { "filter": {"term": {"success": True}} }, "failure_count": { "filter": {"term": {"success": False}} } } } } } ) return { bucket["key"]: { "total": bucket["doc_count"], "success": bucket["success_count"]["doc_count"], "failure": bucket["failure_count"]["doc_count"], "failure_rate": bucket["failure_count"]["doc_count"] / bucket["doc_count"] } for bucket in result["aggregations"]["by_tool"]["buckets"] } 5.2 执行链路回放 class TraceReplay: """Agent 执行链路回放""" async def replay(self, trace_id: str) -> str: """生成可读的执行链路报告""" events = await self.query.get_trace(trace_id) if not events: return f"No trace found for {trace_id}" report = [] report.append(f"=== Agent Trace Replay: {trace_id} ===\n") total_tokens = 0 total_cost = 0 total_duration = 0 for event in events: ts = event["timestamp"] event_type = event["event_type"] data = event.get("data", {}) if event_type == "agent.start": report.append(f"[{ts}] 🚀 Agent started") report.append(f" Query: {data.get('query_preview', '')[:100]}") report.append(f" Tools: {data.get('available_tools', [])}") elif event_type == "llm.request": report.append(f"[{ts}] 📤 LLM call → {data.get('model')}") report.append(f" Input tokens: {data.get('input_tokens', 0)}") total_tokens += data.get('input_tokens', 0) elif event_type == "llm.response": report.append(f"[{ts}] 📥 LLM response ← {data.get('model')}") report.append(f" Output tokens: {data.get('output_tokens', 0)}") report.append(f" Duration: {data.get('duration_ms', 0):.0f}ms") report.append(f" Cost: ${data.get('cost_usd', 0):.6f}") if data.get('tool_calls_count', 0) > 0: report.append(f" Tool calls: {data['tool_calls']}") total_tokens += data.get('output_tokens', 0) total_cost += data.get('cost_usd', 0) total_duration += data.get('duration_ms', 0) elif event_type == "tool.call.end": status = "✅" if data.get('success') else "❌" report.append(f"[{ts}] 🔧 {status} {data.get('tool')}") report.append(f" Duration: {data.get('duration_ms', 0):.0f}ms") if not data.get('success'): report.append(f" Error: {data.get('error', '')}") total_duration += data.get('duration_ms', 0) elif event_type == "agent.end": report.append(f"\n[{ts}] 🏁 Agent finished") report.append(f" Status: {data.get('status')}") report.append(f" Total iterations: {data.get('total_iterations')}") report.append(f"\n=== Summary ===") report.append(f"Total tokens: {total_tokens}") report.append(f"Total cost: ${total_cost:.6f}") report.append(f"Total duration: {total_duration:.0f}ms") return "\n".join(report) 输出示例： ...

引言 Agent 不是孤立的——多个用户同时请求、多个工具并行执行、多个 Agent 协作完成任务。并发控制确保这些并行活动不互相干扰、不超出资源限制、不产生数据竞争。2026年，随着 Agent 集群规模扩大到数百实例，并发控制从单机问题升级为分布式问题。 一、Agent 并发的四个层次 ┌──────────────────────────────────────────────────┐ │ Agent 并发控制层次 │ ├──────────────────┬───────────────────────────────┤ │ 请求级 │ 多用户并发请求 │ │ (Request) │ → 限流 + 队列 + 优先级 │ ├──────────────────┼───────────────────────────────┤ │ Agent级 │ 单用户多Agent实例 │ │ (Agent) │ → 信号量 + 状态隔离 │ ├──────────────────┼───────────────────────────────┤ │ 工具级 │ 多工具并行调用 │ │ (Tool) │ → 并发限制 + 超时控制 │ ├──────────────────┼───────────────────────────────┤ │ 资源级 │ 共享资源访问 │ │ (Resource) │ → 分布式锁 + 乐观锁 │ └──────────────────┴───────────────────────────────┘ 二、请求级并发控制 2.1 多维限流器 import asyncio import time from collections import defaultdict class MultiDimensionRateLimiter: """多维度限流器""" def __init__(self, redis_client): self.redis = redis_client async def check( self, user_id: str, agent_name: str, limits: dict ) -> bool: """多维度限流检查 limits = { "qps": 10, # 每秒请求数 "concurrent": 5, # 最大并发 "daily_tokens": 500000, # 日Token上限 "monthly_cost": 100, # 月成本上限 } """ checks = [] # QPS 限流（滑动窗口） if "qps" in limits: checks.append(self._check_sliding_window( f"qps:{user_id}:{agent_name}", limits["qps"], window=1 )) # 并发限流 if "concurrent" in limits: checks.append(self._check_concurrent( f"conc:{user_id}:{agent_name}", limits["concurrent"] )) # Token 配额 if "daily_tokens" in limits: checks.append(self._check_quota( f"tokens:{user_id}:{date.today()}", limits["daily_tokens"] )) results = await asyncio.gather(*checks) return all(results) async def _check_sliding_window( self, key: str, limit: int, window: int ) -> bool: """滑动窗口限流""" now = time.time() pipe = self.redis.pipeline() # 移除窗口外的记录 pipe.zremrangebyscore(key, 0, now - window) # 添加当前请求 pipe.zadd(key, {str(uuid.uuid4()): now}) # 统计窗口内请求数 pipe.zcard(key) # 设置过期时间 pipe.expire(key, window * 2) results = await pipe.execute() count = results[2] if count > limit: # 移除刚添加的记录 pipe.zrem(key, str(uuid.uuid4())) return False return True async def _check_concurrent(self, key: str, limit: int) -> bool: """并发数限流""" current = await self.redis.incr(key) if current == 1: await self.redis.expire(key, 300) # 5分钟自动过期 if current > limit: await self.redis.decr(key) return False return True async def release_concurrent(self, key: str): """释放并发槽""" await self.redis.decr(key) class PriorityRequestQueue: """优先级请求队列""" def __init__(self, max_concurrent: int = 100): self.semaphore = asyncio.Semaphore(max_concurrent) self.queues = { "high": asyncio.Queue(maxsize=50), "normal": asyncio.Queue(maxsize=500), "low": asyncio.Queue(maxsize=1000), } self._running = True async def submit( self, request: Request, priority: str = "normal" ) -> Response: """提交请求""" queue = self.queues.get(priority, self.queues["normal"]) if queue.full(): raise QueueFullError(f"{priority} queue is full") future = asyncio.Future() await queue.put((request, future)) # 启动消费者（如果未启动） asyncio.create_task(self._consume()) return await future async def _consume(self): """消费请求：按优先级调度""" async with self.semaphore: # 按优先级获取请求 for priority in ["high", "normal", "low"]: queue = self.queues[priority] if not queue.empty(): request, future = await queue.get() try: result = await self._process(request) future.set_result(result) except Exception as e: future.set_exception(e) return # 所有队列为空 await asyncio.sleep(0.01) 三、工具级并发控制 3.1 并行工具调用管理 class ParallelToolExecutor: """并行工具调用管理器""" def __init__(self): self.tool_limits = {} # {tool_name: Semaphore} self.global_limit = asyncio.Semaphore(10) def register_tool( self, name: str, max_concurrent: int = 5, timeout: float = 30.0 ): """注册工具并发限制""" self.tool_limits[name] = { "semaphore": asyncio.Semaphore(max_concurrent), "timeout": timeout } async def execute_parallel( self, tool_calls: list[ToolCall] ) -> list[ToolResult]: """并行执行多个工具调用""" tasks = [] for call in tool_calls: task = asyncio.create_task( self._execute_single(call) ) tasks.append(task) # 等待所有完成，设置全局超时 results = await asyncio.gather(*tasks, return_exceptions=True) return [ result if not isinstance(result, Exception) else ToolResult(success=False, error=str(result)) for result in results ] async def _execute_single(self, call: ToolCall) -> ToolResult: """执行单个工具调用（带并发限制和超时）""" tool_limit = self.tool_limits.get(call.tool) if not tool_limit: raise UnknownToolError(call.tool) # 双重信号量：全局 + 工具级 async with self.global_limit: async with tool_limit["semaphore"]: try: result = await asyncio.wait_for( self._call_tool(call), timeout=tool_limit["timeout"] ) return ToolResult(success=True, data=result) except asyncio.TimeoutError: return ToolResult( success=False, error=f"Tool {call.tool} timed out" ) # 注册工具并发限制 executor = ParallelToolExecutor() executor.register_tool("web_search", max_concurrent=5, timeout=10) executor.register_tool("database_query", max_concurrent=3, timeout=15) executor.register_tool("file_read", max_concurrent=20, timeout=5) executor.register_tool("file_write", max_concurrent=2, timeout=10) executor.register_tool("send_email", max_concurrent=1, timeout=30) 四、资源级并发控制 4.1 分布式锁 class DistributedLock: """基于 Redis 的分布式锁""" def __init__(self, redis_client): self.redis = redis_client self._local_locks = {} # 本地锁缓存 async def acquire( self, resource: str, holder: str, ttl: int = 30, timeout: float = 10.0 ) -> bool: """获取分布式锁 Args: resource: 锁定的资源名 holder: 持有者标识（如 agent_id + session_id） ttl: 锁的生存时间（秒） timeout: 等待获取锁的超时时间 """ lock_key = f"lock:{resource}" start = time.time() while time.time() - start < timeout: # 尝试获取锁（原子操作） acquired = await self.redis.set( lock_key, holder, nx=True, ex=ttl ) if acquired: # 启动看门狗续期 asyncio.create_task(self._watchdog(resource, holder, ttl)) return True # 检查持有者是否是自己（重入） current = await self.redis.get(lock_key) if current == holder: await self.redis.expire(lock_key, ttl) return True # 等待重试 await asyncio.sleep(0.1) return False async def release(self, resource: str, holder: str) -> bool: """释放锁（使用 Lua 脚本保证原子性）""" lua_script = """ if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end """ result = await self.redis.eval( lua_script, 1, f"lock:{resource}", holder ) return bool(result) async def _watchdog(self, resource: str, holder: str, ttl: int): """看门狗：自动续期锁""" lock_key = f"lock:{resource}" renew_interval = ttl * 0.3 # 在30% TTL时续期 while True: await asyncio.sleep(renew_interval) # 检查是否仍持有锁 current = await self.redis.get(lock_key) if current != holder: break # 锁已被释放或被抢 # 续期 await self.redis.expire(lock_key, ttl) class AgentResourceLock: """Agent 资源锁管理""" def __init__(self, lock_manager: DistributedLock): self.locks = lock_manager async def with_resource_lock( self, resource: str, agent_id: str, action: callable, ttl: int = 30 ): """在锁保护下执行操作""" acquired = await self.locks.acquire( resource, agent_id, ttl=ttl ) if not acquired: raise LockAcquisitionError( f"Could not acquire lock for {resource}" ) try: return await action() finally: await self.locks.release(resource, agent_id) 4.2 乐观锁 class OptimisticLock: """乐观锁：适用于低冲突场景""" async def update_with_version( self, resource_id: str, update_fn: callable, max_retries: int = 3 ) -> any: """带版本控制的更新""" for attempt in range(max_retries): # 1. 读取当前版本 current = await self.db.get(resource_id) if not current: raise NotFoundError(resource_id) version = current.version # 2. 计算更新 updated = await update_fn(current.data) # 3. 乐观写入（带版本检查） result = await self.db.update_with_version( resource_id, data=updated, expected_version=version ) if result.success: return updated # 版本冲突，重试 logger.warning( f"Optimistic lock conflict on {resource_id}, " f"attempt {attempt+1}" ) await asyncio.sleep(0.05 * attempt) # 短暂退避 raise ConcurrencyError( f"Failed after {max_retries} retries due to conflicts" ) 五、Agent 级并发控制 class AgentInstanceManager: """Agent 实例管理——控制单用户的 Agent 并发""" def __init__(self, redis_client): self.redis = redis_client self.limits = { "free": 1, # 免费用户：1个并发Agent "pro": 5, # Pro用户：5个 "enterprise": 20, # 企业用户：20个 } async def acquire_slot( self, user_id: str, tier: str, agent_type: str ) -> AgentSlot: """获取 Agent 执行槽""" limit = self.limits.get(tier, 1) key = f"agent_slots:{user_id}" # 检查当前活跃数 active = await self.redis.scard(key) if active >= limit: raise ConcurrencyLimitError( f"User {user_id} has {active}/{limit} active agents" ) # 分配槽位 slot_id = str(uuid.uuid4()) await self.redis.sadd(key, slot_id) await self.redis.expire(key, 3600) # 1小时过期 return AgentSlot( slot_id=slot_id, user_id=user_id, agent_type=agent_type, acquired_at=time.time() ) async def release_slot(self, slot: AgentSlot): """释放 Agent 执行槽""" key = f"agent_slots:{slot.user_id}" await self.redis.srem(key, slot.slot_id) class AgentCoordinator: """Agent 协调器——管理多Agent并发执行""" async def run_agents_concurrent( self, agents: list[Agent], max_parallel: int = 5 ) -> list[AgentResult]: """并发运行多个 Agent""" semaphore = asyncio.Semaphore(max_parallel) async def run_one(agent: Agent) -> AgentResult: async with semaphore: try: result = await agent.run() return AgentResult( agent_id=agent.id, success=True, result=result ) except Exception as e: return AgentResult( agent_id=agent.id, success=False, error=str(e) ) return await asyncio.gather(*[run_one(a) for a in agents]) async def run_agents_pipeline( self, agents: list[Agent], dependencies: dict # {agent_id: [depends_on_ids]} ) -> dict[str, AgentResult]: """按依赖关系运行 Agent（拓扑排序）""" results = {} completed = set() while len(completed) < len(agents): # 找到可执行的 Agent（依赖已完成） ready = [ agent for agent in agents if agent.id not in completed and all( dep in completed for dep in dependencies.get(agent.id, []) ) ] if not ready: # 检测死锁 raise DeadlockError("Dependency cycle detected") # 并发执行就绪的 Agent batch_results = await self.run_agents_concurrent(ready) for result in batch_results: results[result.agent_id] = result completed.add(result.agent_id) return results 六、并发控制监控 class ConcurrencyMonitor: """并发控制监控""" METRICS = { "active_requests": "当前活跃请求数", "queue_depth": "各优先级队列深度", "concurrent_agents": "各用户活跃Agent数", "tool_concurrent_usage": "各工具并发使用数", "lock_wait_time": "锁等待时间分布", "lock_contention_rate": "锁竞争率", "rate_limit_hits": "限流触发次数", } async def health_check(self) -> dict: return { "healthy": await self._check_health(), "active_connections": await self._count_active(), "queue_health": await self._check_queues(), "lock_health": await self._check_locks(), "alerts": await self._get_alerts(), } async def detect_issues(self) -> list[Issue]: issues = [] # 检测队列堆积 for priority, queue in self.queues.items(): if queue.qsize() > queue.maxsize * 0.8: issues.append(Issue( severity="warning", type="queue_backlog", detail=f"{priority} queue at {queue.qsize()}/{queue.maxsize}" )) # 检测锁饥饿 lock_waits = await self._get_lock_wait_times() for resource, wait_time in lock_waits.items(): if wait_time > 5.0: issues.append(Issue( severity="high", type="lock_starvation", detail=f"{resource} lock wait: {wait_time:.1f}s" )) # 检测限流频繁触发 rate_limit_hits = await self._get_rate_limit_hits() for user_id, hits in rate_limit_hits.items(): if hits > 100: # 每分钟超过100次 issues.append(Issue( severity="medium", type="excessive_rate_limiting", detail=f"User {user_id} hit rate limit {hits} times" )) return issues 七、并发控制 Checklist □ 多维限流（QPS + 并发数 + Token + 成本） □ 优先级队列（高/中/低优先级请求隔离） □ 工具级并发限制（不同工具有不同并发上限） □ 全局并发限制（总并发不超过系统容量） □ 分布式锁保护共享资源（带看门狗续期） □ 乐观锁用于低冲突场景 □ Agent 实例数按用户等级限制 □ 依赖感知的 Agent 调度（拓扑排序） □ 死锁检测和预防 □ 并发监控和告警 □ 队列堆积自动扩容 □ 压测验证并发上限 结语 并发控制是 Agent 系统从"能跑"到"能扛"的关键。单机时代，一个信号量就够了；分布式时代，你需要考虑锁的公平性、可重入性、死锁预防和故障恢复。最好的并发控制是"恰到好处"——限制太松会导致雪崩，限制太紧会浪费资源。通过压测找到你的系统边界，然后在那里设置护栏。记住：并发控制不是阻碍性能，而是保护性能。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent 在执行任务时可能遇到各种错误：LLM 超时、工具返回异常、网络中断、格式解析失败。传统软件的错误处理是确定性的——写好 if-else 分支即可。Agent 的错误处理需要应对非确定性的模型输出和动态环境。2026年，“自修复"Agent 已从概念走向实践。 一、Agent 错误分类体系 ┌──────────────────────────────────────────────────────────┐ │ Agent 错误分类 │ ├──────────────┬────────────┬──────────────────────────────┤ │ 错误类型 │ 可重试性 │ 恢复策略 │ ├──────────────┼────────────┼──────────────────────────────┤ │ LLM 超时 │ 可重试 │ 指数退避 + 模型降级 │ │ LLM 限流 │ 可重试 │ 令牌桶等待 + 队列排队 │ │ LLM 格式错误 │ 可重试 │ 格式修正 + 结构化输出重试 │ │ 工具超时 │ 看情况 │ 重试 + 备用工具 + 降级 │ │ 工具异常 │ 看情况 │ 参数修正 + 自修复 │ │ 网络中断 │ 可重试 │ 自动重连 + 状态恢复 │ │ 上下文溢出 │ 不可重试 │ 上下文压缩 + 分段处理 │ │ 幻觉/错误输出│ 需判断 │ 自我验证 + 重新推理 │ │ 死循环 │ 不可重试 │ 迭代上限 + 强制终止 │ │ 权限拒绝 │ 不可重试 │ 降级 + 人工介入 │ └──────────────┴────────────┴──────────────────────────────┘ 二、基础层：智能重试 2.1 分级重试策略 from enum import Enum import asyncio import random class RetryStrategy(Enum): FIXED = "fixed" EXPONENTIAL = "exponential" LINEAR = "linear" JITTERED = "jittered" class SmartRetryPolicy: """智能重试策略""" STRATEGIES = { "timeout": RetryPolicy( max_attempts=3, strategy=RetryStrategy.EXPONENTIAL, base_delay=1.0, max_delay=30.0, jitter=True ), "rate_limit": RetryPolicy( max_attempts=5, strategy=RetryStrategy.LINEAR, base_delay=5.0, max_delay=60.0, jitter=False # 限流重试不需要抖动 ), "connection": RetryPolicy( max_attempts=5, strategy=RetryStrategy.EXPONENTIAL, base_delay=0.5, max_delay=10.0, jitter=True ), "format_error": RetryPolicy( max_attempts=2, # 格式错误重试意义有限 strategy=RetryStrategy.FIXED, base_delay=0.0, max_delay=0.0, pre_action="repair_prompt" # 重试前修复 Prompt ), } def get_policy(self, error: Exception) -> RetryPolicy: if isinstance(error, TimeoutError): return self.STRATEGIES["timeout"] elif isinstance(error, RateLimitError): return self.STRATEGIES["rate_limit"] elif isinstance(error, ConnectionError): return self.STRATEGIES["connection"] elif isinstance(error, FormatError): return self.STRATEGIES["format_error"] else: return RetryPolicy(max_attempts=1) # 不重试 def compute_delay(self, attempt: int, policy: RetryPolicy) -> float: if policy.strategy == RetryStrategy.FIXED: delay = policy.base_delay elif policy.strategy == RetryStrategy.LINEAR: delay = policy.base_delay * attempt elif policy.strategy == RetryStrategy.EXPONENTIAL: delay = policy.base_delay * (2 ** attempt) delay = min(delay, policy.max_delay) if policy.jitter: delay += random.uniform(0, delay * 0.1) return delay async def with_retry( func: callable, error_handler: callable = None, policies: SmartRetryPolicy = None ): """带智能重试的函数调用""" policies = policies or SmartRetryPolicy() last_error = None for attempt in range(10): # 安全上限 try: return await func() except Exception as e: last_error = e policy = policies.get_policy(e) if attempt >= policy.max_attempts: raise # 重试前操作（如修复 Prompt） if policy.pre_action == "repair_prompt": if error_handler: await error_handler(e, attempt) delay = policies.compute_delay(attempt, policy) logger.warning( f"Attempt {attempt+1} failed: {e}. " f"Retrying in {delay:.1f}s..." ) await asyncio.sleep(delay) raise last_error 2.2 格式错误自修复 class FormatRepairAgent: """格式错误自动修复""" REPAIR_STRATEGIES = [ "json_extract", # 从文本中提取 JSON "json_repair", # 修复常见 JSON 错误 "re_prompt", # 重新请求 LLM "structured_output", # 切换到结构化输出 ] async def repair(self, raw_output: str, expected_schema: dict) -> dict: """尝试修复格式错误""" for strategy in self.REPAIR_STRATEGIES: try: result = await getattr(self, f"_strategy_{strategy}")( raw_output, expected_schema ) if result is not None: logger.info(f"Format repaired using: {strategy}") return result except Exception: continue raise FormatRepairError( f"Could not repair output after all strategies" ) async def _strategy_json_extract(self, raw: str, schema: dict) -> dict | None: """从文本中提取 JSON""" # 尝试找到 JSON 块 import re patterns = [ r'```json\s*(.*?)\s*```', # Markdown JSON 块 r'```\s*(.*?)\s*```', # 通用代码块 r'\{[^{}]*\}', # 裸 JSON 对象 r'\[.*\]', # 裸 JSON 数组 ] for pattern in patterns: match = re.search(pattern, raw, re.DOTALL) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: continue return None async def _strategy_json_repair(self, raw: str, schema: dict) -> dict | None: """修复常见 JSON 错误""" repaired = raw # 修复尾随逗号 repaired = re.sub(r',\s*}', '}', repaired) repaired = re.sub(r',\s*]', ']', repaired) # 修复单引号 repaired = repaired.replace("'", '"') # 修复未引用的键 repaired = re.sub(r'(\w+):', r'"\1":', repaired) # 修复省略号 repaired = repaired.replace('...', 'null') try: return json.loads(repaired) except json.JSONDecodeError: return None async def _strategy_re_prompt(self, raw: str, schema: dict) -> dict | None: """使用 LLM 修复格式""" repair_prompt = f"""The following text was supposed to be valid JSON matching this schema: {json.dumps(schema, indent=2)} But it had format errors. Fix it and return ONLY valid JSON: Original text: {raw[:2000]} Return the corrected JSON:""" response = await llm.invoke( repair_prompt, response_format={"type": "json_object"}, temperature=0.0 ) try: return json.loads(response.content) except: return None async def _strategy_structured_output(self, raw: str, schema: dict) -> dict | None: """使用结构化输出 API 重新请求""" response = await llm.invoke( f"Convert this to structured data:\n{raw[:2000]}", response_format={"type": "json_schema", "json_schema": schema}, temperature=0.0 ) return json.loads(response.content) if response.content else None 三、中间层：工具错误恢复 3.1 工具执行框架 class ResilientToolExecutor: """带错误恢复的工具执行器""" def __init__(self): self.fallback_chains = {} # {tool_name: [backup_tool1, backup_tool2]} self.error_handlers = {} # {tool_name: handler} async def execute( self, tool_call: ToolCall, context: ExecutionContext ) -> ToolResult: tool_name = tool_call.tool primary = self._get_tool(tool_name) try: result = await with_retry( lambda: primary.execute(tool_call.args, context), policies=self._get_retry_policy(tool_name) ) return ToolResult(success=True, data=result) except Exception as e: # 尝试错误处理器 handler = self.error_handlers.get(tool_name) if handler: repaired_call = await handler(e, tool_call, context) if repaired_call: return await self.execute(repaired_call, context) # 尝试备用工具链 for backup_name in self.fallback_chains.get(tool_name, []): backup = self._get_tool(backup_name) try: adapted_args = await self._adapt_args( tool_call.args, primary, backup ) result = await backup.execute(adapted_args, context) return ToolResult( success=True, data=result, degraded=True, used_fallback=backup_name ) except Exception: continue return ToolResult( success=False, error=str(e), tool=tool_name ) # 注册备用工具链 executor = ResilientToolExecutor() executor.fallback_chains = { "web_search": ["bing_search", "duckduckgo_search"], "database_query": ["cache_lookup", "default_response"], "send_email": ["queue_email", "save_to_retry"], } # 注册错误处理器 async def search_error_handler( error: Exception, tool_call: ToolCall, context: ExecutionContext ) -> ToolCall | None: """搜索工具错误处理器""" if isinstance(error, TimeoutError): # 减少结果数量重试 modified_args = tool_call.args.copy() modified_args["max_results"] = 3 # 减少结果数量 return ToolCall(tool=tool_call.tool, args=modified_args) elif isinstance(error, RateLimitError): return None # 不重试，直接走备用 return None executor.error_handlers["web_search"] = search_error_handler 3.2 LLM 输出验证与自修复 class OutputValidator: """LLM 输出验证与自修复""" async def validate_and_repair( self, output: str, task: str, constraints: list[str] ) -> ValidatedOutput: # 1. 结构验证 structural = self._check_structure(output, task) if not structural.valid: repaired = await self._repair_structure(output, structural.errors) if repaired: output = repaired # 2. 约束验证 constraint_results = await self._check_constraints(output, constraints) violated = [r for r in constraint_results if not r.passed] if not violated: return ValidatedOutput(valid=True, output=output) # 3. 自修复：让 LLM 自己修正 if self._is_repairable(violated): repaired_output = await self._self_repair( output, task, violated ) if repaired_output: # 重新验证 recheck = await self._check_constraints(repaired_output, constraints) if all(r.passed for r in recheck): return ValidatedOutput( valid=True, output=repaired_output, repaired=True ) return ValidatedOutput( valid=False, output=output, violations=[r.constraint for r in violated] ) async def _self_repair( self, original_output: str, task: str, violations: list[ConstraintViolation] ) -> str | None: """让 LLM 自我修复输出""" violation_desc = "\n".join( f"- {v.constraint}: {v.detail}" for v in violations ) repair_prompt = f"""Your previous response has issues that need to be fixed. Original task: {task} Your response: {original_output[:3000]} Issues found: {violation_desc} Please fix these issues and provide the corrected response. Only output the corrected response, no explanations.""" try: response = await llm.invoke(repair_prompt, temperature=0.0) return response.content except: return None 四、高级层：自修复 Agent class SelfHealingAgent: """自修复 Agent：能识别错误并自主修复""" async def run_with_healing( self, task: str, max_heal_attempts: int = 3 ) -> str: for attempt in range(max_heal_attempts): try: result = await self._execute(task) # 自我验证 validation = await self._self_validate(task, result) if validation.confidence > 0.8: return result # 识别问题 diagnosis = await self._diagnose( task, result, validation ) # 生成修复方案 fix_plan = await self._generate_fix_plan(diagnosis) # 执行修复 task = await self._apply_fix(task, fix_plan) logger.info( f"Self-healing attempt {attempt+1}: " f"diagnosis={diagnosis.issue}, " f"fix={fix_plan.action}" ) except CircularError as e: # 检测到循环，换策略 task = await self._reframe_task(task, e) except MaxIterationsError: # 达到迭代上限，简化任务 task = await self._simplify_task(task) # 自修复失败，返回最佳结果 return result or "I was unable to complete this task." async def _diagnose( self, task: str, result: str, validation: Validation ) -> Diagnosis: """诊断输出问题""" diagnosis_prompt = f"""Analyze why the following result may be incorrect: Task: {task} Result: {result[:2000]} Validation concerns: {validation.concerns} Identify the most likely issue: - wrong_approach: Used wrong method to solve the problem - incomplete: Missing important information or steps - hallucination: Contains fabricated information - format_error: Output format doesn't match requirements - logical_error: Reasoning contains logical flaws - other: Specify Respond in JSON: {{"issue": "...", "detail": "...", "confidence": 0.0-1.0}}""" response = await llm.invoke(diagnosis_prompt, temperature=0.0) return Diagnosis(**json.loads(response.content)) async def _generate_fix_plan(self, diagnosis: Diagnosis) -> FixPlan: """生成修复计划""" FIX_STRATEGIES = { "wrong_approach": "Rethink the approach and try a different method", "incomplete": "Identify missing information and supplement it", "hallucination": "Verify all claims using tools and remove unverified ones", "format_error": "Reformat the output to match requirements", "logical_error": "Review the reasoning chain step by step", } action = FIX_STRATEGIES.get(diagnosis.issue, "Retry with more careful approach") return FixPlan( action=action, issue=diagnosis.issue, detail=diagnosis.detail, strategy="adjust_and_retry" ) 五、错误恢复编排 class ErrorRecoveryOrchestrator: """错误恢复编排器""" async def execute_with_recovery( self, task: str, agent: Agent ) -> RecoveryResult: recovery_layers = [ ("retry", RetryLayer()), ("format_repair", FormatRepairLayer()), ("tool_fallback", ToolFallbackLayer()), ("output_validation", OutputValidationLayer()), ("self_healing", SelfHealingLayer()), ("human_escalation", HumanEscalationLayer()), ] current_task = task current_output = None for layer_name, layer in recovery_layers: try: result = await layer.process( current_task, current_output, agent ) if result.success: return RecoveryResult( output=result.output, recovery_layers_used=[layer_name], success=True ) # 更新任务或输出，传给下一层 if result.modified_task: current_task = result.modified_task if result.partial_output: current_output = result.partial_output except Exception as e: logger.error(f"Recovery layer {layer_name} failed: {e}") continue return RecoveryResult( output=current_output or "Unable to complete task", success=False, recovery_layers_used=["all_failed"] ) 六、错误恢复 Checklist □ 错误分类体系明确（可重试/不可重试/需判断） □ 重试策略匹配错误类型（指数退避/线性/固定） □ 格式错误自动修复（JSON提取/修复/重新请求） □ 工具备用链配置（主工具失败→备用工具） □ LLM 输出验证+自我修复 □ 自修复 Agent 能诊断并修复自身错误 □ 循环检测器防止自修复死循环 □ 人工升级机制（自修复失败时触发） □ 错误恢复全链路追踪 □ 定期演练错误恢复流程 结语 错误恢复是 Agent 可靠性的最后一道防线。从简单的重试到复杂的自修复，每一层恢复策略都在增加系统的韧性。但要注意：自修复不是万能的——它消耗额外 Token、增加延迟、可能引入新错误。最佳策略是分层防御：基础错误用重试解决，格式错误用修复解决，逻辑错误用自修复解决，复杂错误交给人工。让 Agent 像人一样——犯错后能自我纠正，但也知道什么时候该求助。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

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

引言 Agent 的能力边界由工具决定。2024年，工具是硬编码在 Agent 中的；2025年，工具变成可配置的插件；2026年，Anthropic 的 MCP（Model Context Protocol）让工具发现变成动态的、标准化的。本文梳理工具发现机制的完整演进路径。 一、工具发现的三代演进 第一代：静态注册 (2023-2024) ┌─────────────────────────────┐ │ Agent 代码 │ │ ┌──────┐ ┌──────┐ ┌──────┐│ │ │search│ │calc │ │write ││ ← 硬编码在代码中 │ └──────┘ └──────┘ └──────┘│ └─────────────────────────────┘ 问题：添加工具需要改代码、重新部署 第二代：配置化注册 (2024-2025) ┌─────────────────────────────┐ │ Agent 运行时 │ │ ┌─────────────────────┐ │ │ │ Tool Registry │ │ ← 从配置文件/YAML加载 │ │ ┌──┐┌──┐┌──┐┌──┐ │ │ │ │ │S ││C ││W ││E │ │ │ │ │ └──┘└──┘└──┘└──┘ │ │ │ └─────────────────────┘ │ └─────────────────────────────┘ 问题：工具集固定，无法按需扩展 第三代：动态发现 (2025-2026) ┌─────────────────────────────┐ │ Agent 运行时 │ │ ┌─────────────────────┐ │ │ │ Tool Discovery │ │ │ │ ┌────────────────┐ │ │ │ │ │ MCP Registry │ │ │ ← 标准化协议 │ │ │ ┌─┐┌─┐┌─┐┌─┐ │ │ │ │ │ │ │S││C││W││?│ │ │ │ ← 动态发现 │ │ │ └─┘└─┘└─┘└─┘ │ │ │ │ │ └────────────────┘ │ │ │ └─────────────────────┘ │ └─────────────────────────────┘ 优势：即插即用、运行时扩展、生态共享 二、第一代：静态注册 # 硬编码工具——最简单但最不灵活 class SimpleAgent: def __init__(self, llm): self.llm = llm self.tools = { "search": self._search, "calculate": self._calculate, "write_file": self._write_file, } async def run(self, query: str) -> str: # 将工具定义注入 Prompt tools_desc = "\n".join( f"- {name}: {func.__doc__}" for name, func in self.tools.items() ) prompt = f"""Tools available: {tools_desc} To use a tool, respond with JSON: {{"tool": "name", "args": {{...}}}} User: {query}""" response = await self.llm.invoke(prompt) # 解析并执行工具 ... 问题：添加工具需要修改代码、测试、部署，周期以天计。 ...