架构设计

引言 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) # 解析并执行工具 ... 问题：添加工具需要修改代码、测试、部署，周期以天计。 ...

Hermes Agent：面向企业的智能体架构 Hermes Agent（爱马仕智能体）是 2026 年企业级 Agent 领域的一匹黑马。它由前 Google Brain 和 Meta AI 团队成员联合创建，定位为"企业级通用智能体操作系统"。与消费级 Agent 不同，Hermes 从第一天起就为生产环境设计，在可靠性、安全性和可扩展性方面树立了新的标杆。 核心设计理念 Hermes 的架构设计围绕三个核心理念： 确定性优先：在需要确定性的场景中，使用状态机而非自由对话 可解释性：每一步决策都有完整的推理链和置信度评分 渐进式自主：从辅助人类到自主执行的渐进路径 技术架构总览 ┌──────────────────────────────────────────────────────────┐ │ Interaction Layer │ │ REST API │ gRPC │ WebSocket │ SDK │ CLI │ IDE Plugin │ └─────────────────────┬────────────────────────────────────┘ │ ┌─────────────────────┴────────────────────────────────────┐ │ Orchestration Layer │ │ ┌────────────┐ ┌─────────────┐ ┌──────────────────┐ │ │ │ Planner │ │ Executor │ │ Reflector │ │ │ │ (规划引擎) │ │ (执行引擎) │ │ (反思引擎) │ │ │ └────────────┘ └─────────────┘ └──────────────────┘ │ │ ┌────────────┐ ┌─────────────┐ ┌──────────────────┐ │ │ │ Memory │ │ Knowledge │ │ Safety Guard │ │ │ │ (记忆) │ │ (知识库) │ │ (安全护栏) │ │ │ └────────────┘ └─────────────┘ └──────────────────┘ │ └─────────────────────┬────────────────────────────────────┘ │ ┌─────────────────────┴────────────────────────────────────┐ │ Execution Layer │ │ Code Sandbox │ API Gateway │ Data Pipeline │ ML Pipeline │ └─────────────────────┬────────────────────────────────────┘ │ ┌─────────────────────┴────────────────────────────────────┐ │ Infrastructure Layer │ │ Kubernetes │ Service Mesh │ Observability │ Secret Mgmt │ └──────────────────────────────────────────────────────────┘ 核心组件深度解析 1. Planner（规划引擎） Hermes 的规划引擎支持两种模式： ...

引言 Agent 的推理过程往往是漫长的等待——用户盯着加载动画，不知道 Agent 在做什么。流式响应把"等待结果"变成"实时观察思考"，是 Agent 用户体验的关键升级。2026年，三种流式协议各有优劣，选型不当会导致体验降级或工程复杂度爆炸。 一、三种协议对比 核心特性矩阵 特性 SSE WebSocket gRPC Stream 通信方向 服务器→客户端（单向） 双向 双向 底层协议 HTTP/1.1 或 HTTP/2 HTTP 升级 HTTP/2 数据格式 文本（text/event-stream） 文本/二进制 Protobuf（二进制） 自动重连 内置 需手动实现 需手动实现 浏览器支持 原生 EventSource 原生 WebSocket 需 gRPC-Web 代理/CDN兼容 优秀 良好 较差 连接数限制 浏览器6个/域名 无限制 无限制 序列化效率 低（文本） 中 高（Protobuf） 移动端友好 高 中 低 Agent 场景适配分析 Agent 流式需求频谱： 单向输出流 双向交互流 （LLM→用户） （用户↔Agent） │ │ │ ┌───────────┐ │ │ │ SSE 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ WebSocket │ │ │ │ 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ gRPC 最佳 │ │ │ └───────────┘ │ 简单聊天 ←─────────────→ 复杂多Agent 低延迟 ←─────────────→ 高吞吐 Web前端 ←─────────────→ 微服务后端 二、SSE 实现方案 2.1 服务端 from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() class AgentStreamEvent: """Agent 流式事件类型""" THINKING = "thinking" # Agent 思考中 TOOL_CALL = "tool_call" # 工具调用 TOOL_RESULT = "tool_result" # 工具结果 CONTENT = "content" # 内容输出 ERROR = "error" # 错误 DONE = "done" # 完成 async def agent_stream_generator( query: str, session_id: str ): """Agent SSE 流式生成器""" try: # 1. 发送思考状态 yield _format_sse(AgentStreamEvent.THINKING, { "message": "正在分析您的请求...", "session_id": session_id }) # 2. Agent 推理（流式 LLM 输出） async for chunk in agent.think_stream(query): if chunk.type == "tool_call": yield _format_sse(AgentStreamEvent.TOOL_CALL, { "tool": chunk.tool_name, "args": chunk.tool_args, "thinking": chunk.reasoning }) # 3. 工具执行 result = await agent.execute_tool(chunk.tool_call) yield _format_sse(AgentStreamEvent.TOOL_RESULT, { "tool": chunk.tool_name, "result": result.summary, "duration_ms": result.duration_ms }) elif chunk.type == "content": yield _format_sse(AgentStreamEvent.CONTENT, { "text": chunk.text, "tokens_so_far": chunk.token_count }) # 4. 完成 yield _format_sse(AgentStreamEvent.DONE, { "session_id": session_id, "total_tokens": agent.total_tokens, "duration_ms": agent.total_duration_ms }) except Exception as e: yield _format_sse(AgentStreamEvent.ERROR, { "message": str(e), "session_id": session_id }) def _format_sse(event_type: str, data: dict) -> str: return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @app.post("/api/agent/chat") async def chat(request: ChatRequest): return StreamingResponse( agent_stream_generator(request.query, request.session_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx 禁用缓冲 } ) 2.2 客户端 class AgentSSEClient { private eventSource: EventSource | null = null; private reconnectAttempts = 0; private maxReconnects = 3; connect(query: string, sessionId: string) { // 使用 fetch POST + ReadableStream（EventSource 仅支持 GET） fetch('/api/agent/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ query, sessionId }), }).then(response => { const reader = response.body!.getReader(); const decoder = new TextDecoder(); let buffer = ''; const read = () => { reader.read().then(({ done, value }) => { if (done) return; buffer += decoder.decode(value, { stream: true }); const events = buffer.split('\n\n'); buffer = events.pop() || ''; events.forEach(raw => this.handleEvent(raw)); read(); }); }; read(); }); } private handleEvent(raw: string) { const lines = raw.split('\n'); let event = 'message'; let data = ''; lines.forEach(line => { if (line.startsWith('event: ')) event = line.slice(7); if (line.startsWith('data: ')) data = line.slice(6); }); const parsed = JSON.parse(data); switch (event) { case 'thinking': this.onThinking?.(parsed); break; case 'tool_call': this.onToolCall?.(parsed); break; case 'tool_result': this.onToolResult?.(parsed); break; case 'content': this.onContent?.(parsed.text); break; case 'error': this.onError?.(parsed); break; case 'done': this.onDone?.(parsed); break; } } } 三、WebSocket 实现方案 3.1 服务端 from fastapi import WebSocket, WebSocketDisconnect class AgentConnectionManager: """Agent WebSocket 连接管理""" def __init__(self): self.active: dict[str, WebSocket] = {} # session_id → WebSocket self.agent_tasks: dict[str, asyncio.Task] = {} async def connect(self, ws: WebSocket, session_id: str): await ws.accept() self.active[session_id] = ws logger.info(f"WebSocket connected: {session_id}") async def disconnect(self, session_id: str): if session_id in self.active: del self.active[session_id] if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() del self.agent_tasks[session_id] async def handle_session(self, ws: WebSocket, session_id: str): """处理 WebSocket 会话""" await self.connect(ws, session_id) try: while True: # 接收客户端消息 message = await ws.receive_json() if message["type"] == "chat": # 启动 Agent 任务 task = asyncio.create_task( self._run_agent(ws, session_id, message["content"]) ) self.agent_tasks[session_id] = task elif message["type"] == "interrupt": # 用户中断当前 Agent 执行 if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() await ws.send_json({ "type": "interrupted", "session_id": session_id }) elif message["type"] == "feedback": # 用户实时反馈（Human-in-the-loop） await self._handle_feedback(session_id, message) except WebSocketDisconnect: await self.disconnect(session_id) async def _run_agent(self, ws: WebSocket, session_id: str, query: str): """运行 Agent 并通过 WebSocket 推送更新""" try: async for event in agent.run_stream(query): await ws.send_json({ "type": event.type, "data": event.data, "timestamp": time.time() }) except asyncio.CancelledError: logger.info(f"Agent task cancelled: {session_id}") except Exception as e: await ws.send_json({ "type": "error", "data": {"message": str(e)} }) manager = AgentConnectionManager() @app.websocket("/ws/agent/{session_id}") async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.handle_session(ws, session_id) 3.2 客户端 class AgentWSClient { private ws: WebSocket | null = null; private messageQueue: string[] = []; connect(sessionId: string) { this.ws = new WebSocket(`wss://api.example.com/ws/agent/${sessionId}`); this.ws.onopen = () => { // 发送排队消息 this.messageQueue.forEach(msg => this.ws!.send(msg)); this.messageQueue = []; }; this.ws.onmessage = (event) => { const data = JSON.parse(event.data); this.handleMessage(data); }; this.ws.onclose = () => { // 自动重连 setTimeout(() => this.connect(sessionId), 3000); }; } sendChat(content: string) { const msg = JSON.stringify({ type: 'chat', content }); if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(msg); } else { this.messageQueue.push(msg); } } interrupt() { this.ws?.send(JSON.stringify({ type: 'interrupt' })); } } 四、gRPC 流式方案 4.1 Proto 定义 // agent.proto syntax = "proto3"; service AgentService { // 服务端流式：Agent → 客户端 rpc ChatStream(ChatRequest) returns (stream ChatResponse); // 双向流式：支持实时交互 rpc ChatBidirectional(stream ChatMessage) returns (stream ChatResponse); } message ChatRequest { string session_id = 1; string query = 2; map<string, string> metadata = 3; } message ChatMessage { string session_id = 1; MessageType type = 2; // CHAT, INTERRUPT, FEEDBACK string content = 3; } message ChatResponse { ResponseType type = 1; // THINKING, TOOL_CALL, CONTENT, DONE, ERROR string session_id = 2; bytes data = 3; // JSON 编码的事件数据 int64 timestamp = 4; } enum MessageType { CHAT = 0; INTERRUPT = 1; FEEDBACK = 2; } enum ResponseType { THINKING = 0; TOOL_CALL = 1; TOOL_RESULT = 2; CONTENT = 3; DONE = 4; ERROR = 5; } 4.2 服务端实现 import grpc from concurrent import futures class AgentServicer(agent_pb2_grpc.AgentServiceServicer): def ChatStream(self, request, context): """服务端流式：逐条推送 Agent 事件""" try: for event in agent.run(request.query): response = agent_pb2.ChatResponse( type=self._map_event_type(event.type), session_id=request.session_id, data=json.dumps(event.data).encode(), timestamp=int(time.time()) ) yield response except Exception as e: yield agent_pb2.ChatResponse( type=agent_pb2.ERROR, data=json.dumps({"error": str(e)}).encode() ) def ChatBidirectional(self, request_iterator, context): """双向流式：支持中断和实时反馈""" session = None for message in request_iterator: if message.type == agent_pb2.CHAT: # 启动 Agent 执行 for event in agent.run(message.content): if not context.is_active(): break yield agent_pb2.ChatResponse( type=self._map_event_type(event.type), data=json.dumps(event.data).encode() ) elif message.type == agent_pb2.INTERRUPT: agent.interrupt() yield agent_pb2.ChatResponse( type=agent_pb2.DONE, data=json.dumps({"reason": "interrupted"}).encode() ) 五、性能基准测试 测试环境 服务端：4 vCPU / 16GB RAM / Python 3.12 / FastAPI 客户端：1000 并发连接 消息大小：平均 200 bytes / 消息 持续时间：5 分钟 结果对比 指标 SSE WebSocket gRPC Stream 最大并发连接 10,000 50,000 30,000 消息延迟 P50 12ms 8ms 5ms 消息延迟 P95 45ms 25ms 12ms 消息延迟 P99 120ms 60ms 30ms 吞吐量 (msg/s) 50,000 200,000 150,000 内存/连接 32KB 48KB 24KB CPU 利用率 (1k连接) 35% 28% 22% 带宽效率 基准 +15% +40% 六、选型决策树 ┌─────────────────┐ │ 是否需要双向通信？│ └────────┬────────┘ │ ┌──────────────┴──────────────┐ │ 否 │ 是 ▼ ▼ ┌────────────────┐ ┌──────────────────┐ │ 是否是微服务 │ │ 是否是浏览器前端？│ │ 内部通信？ │ └────────┬─────────┘ └───────┬────────┘ │ │ ┌───────┴───────┐ ┌───────┴───────┐ │ 否 │ 是 │ 是 │ 否 ▼ ▼ ▼ ▼ ┌──────────┐ ┌────────────┐ ┌──────┐ ┌────────┐ │ gRPC │ │ WebSocket │ │gRPC │ │ SSE │ │ Stream │ │ │ └──────┘ └────────┘ └──────────┘ └────────────┘ 场景推荐 场景 推荐协议 原因 Web 聊天界面 SSE 原生支持、简单、自动重连 实时协作编辑 WebSocket 双向低延迟 移动 App SSE 移动网络友好、自动重连 微服务间 Agent 通信 gRPC 高效序列化、强类型 多 Agent 系统 gRPC 流式 RPC 适配 Agent 通信 简单 LLM 问答 SSE 单向流足够、实现简单 Human-in-the-loop WebSocket 需要双向交互 大规模推送 SSE CDN 兼容、连接效率高 七、生产环境关键配置 Nginx SSE 配置 location /api/agent/chat { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; # 关键：禁用缓冲 proxy_cache off; proxy_read_timeout 300s; # 长连接超时 chunked_transfer_encoding on; } WebSocket 配置 location /ws/agent/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 86400; # 24小时 proxy_send_timeout 86400; } 八、流式架构 Checklist □ 协议选型基于通信方向和客户端类型 □ SSE 禁用代理缓冲（proxy_buffering off） □ WebSocket 实现心跳和重连机制 □ gRPC 配置 keepalive 和流控 □ 消息序列化使用高效格式（JSON/Protobuf） □ 背压机制防止慢客户端拖垮服务端 □ 连接超时和最大连接数限制 □ 流式错误不中断连接，通过事件传递 □ 客户端实现优雅降级（流式不可用时回退轮询） □ 监控流式连接的延迟和消息丢失率 结语 流式响应是 Agent 从"工具"到"伙伴"的关键体验升级。协议选型没有银弹：SSE 简单可靠适合 Web 场景，WebSocket 灵活双向适合交互场景，gRPC 高效强类型适合微服务场景。理解你的通信模式——是单向输出还是双向交互——然后选择最适合的工具。在 Agent 时代，好的流式架构让用户感觉 Agent 在"思考"，而不是在"卡住"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent 系统的三个核心维度——Prompt、工具、模型——任何一个的变更都可能引发连锁反应。传统软件的版本管理主要针对代码，而 Agent 还需要管理自然语言"代码"（Prompt）、动态加载的工具和外部模型版本。本文将构建完整的 Agent 版本管理体系。 一、Agent 版本的复杂性 变更类型与风险 ┌──────────────────────────────────────────────────────┐ │ Agent 变更类型与风险矩阵 │ ├──────────────┬──────────┬──────────┬─────────────────┤ │ 变更类型 │ 频率 │ 风险等级 │ 影响范围 │ ├──────────────┼──────────┼──────────┼─────────────────┤ │ Prompt 修改 │ 每周 │ 中-高 │ 输出质量/行为 │ │ 工具更新 │ 每月 │ 中 │ 工具调用/结果 │ │ 模型升级 │ 每季度 │ 高 │ 全局行为变化 │ │ System配置 │ 每周 │ 低-中 │ 性能/限制 │ │ Few-shot示例 │ 每月 │ 中 │ 输出风格/格式 │ │ 工作流变更 │ 每月 │ 高 │ 执行路径/延迟 │ └──────────────┴──────────┴──────────┴─────────────────┘ 版本组合的笛卡尔积问题 Agent 行为是 Prompt版本 × 工具版本 × 模型版本 的组合。如果三者各自有 3 个版本，理论上存在 27 种组合。版本管理的目标就是确保任意组合的行为可预测、可回滚。 ...

引言 2026年3月15日，OpenAI API 全球性宕机 47 分钟。那些没有降级方案的 Agent 应用全部显示"服务不可用"，而准备好容灾方案的产品几乎无感知地度过了这次故障。LLM 是 Agent 的核心，但它不是 100% 可靠的基础设施。本文将系统化构建 Agent 的多级降级体系。 一、故障分类与影响分析 1.1 LLM 服务故障类型 ┌──────────────────────────────────────────────────┐ │ LLM 故障分类 │ ├──────────────┬───────────┬───────────────────────┤ │ 故障类型 │ 持续时间 │ 影响范围 │ ├──────────────┼───────────┼───────────────────────┤ │ API 完全宕机 │ 5-60min │ 所有请求失败 │ │ 限流(429) │ 1-10min │ 超额请求失败 │ │ 超时 │ 30-120s │ 单请求受影响 │ │ 模型退版 │ 永久 │ 特定版本不可用 │ │ 区域故障 │ 5-30min │ 特定区域不可用 │ │ 质量退化 │ 未知 │ 输出质量下降 │ └──────────────┴───────────┴───────────────────────┘ 1.2 故障影响评估 @dataclass class FaultImpact: fault_type: str user_visible: bool # 用户是否感知 data_loss: bool # 是否丢数据 recovery_time: str # 恢复时间 business_impact: str # 低/中/高/严重 LLM_FAULT_IMPACTS = [ FaultImpact("API宕机", True, False, "5-60min", "严重"), FaultImpact("限流", True, False, "1-10min", "中"), FaultImpact("超时", True, False, "30-120s", "低"), FaultImpact("质量退化", False, False, "未知", "中"), ] 二、多级降级架构 ┌──────────────────────────────────────────────────────┐ │ 请求入口 │ │ ┌──────┴──────┐ │ │ │ 请求路由 │ │ │ └──────┬──────┘ │ │ │ │ │ ┌─────────────────────▼──────────────────────┐ │ │ │ Level 0: 主模型 (GPT-5) │ │ │ │ 延迟 < 2s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 1: 备用模型 (Claude Opus) │ │ │ │ 延迟 < 3s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 2: 降级模型 (GPT-5-mini) │ │ │ │ 延迟 < 1s, 功能受限 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 3: 缓存/预计算结果 │ │ │ │ 精度下降，但可用 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 4: 规则引擎/模板回复 │ │ │ │ 基本功能，无AI能力 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 5: 优雅错误页面 │ │ │ │ 引导用户重试或联系人工 │ │ │ └────────────────────────────────────────────┘ │ 三、各级降级实现 Level 0-1：模型切换 from enum import Enum import asyncio class ModelTier(Enum): PRIMARY = "gpt-5" FALLBACK = "claude-opus-4-2026" DEGRADED = "gpt-5-mini" EMERGENCY = "gpt-4o-mini" class LLMFailoverChain: """LLM 多级故障转移链""" def __init__(self): self.chain = [ {"model": "gpt-5", "timeout": 30, "retries": 2}, {"model": "claude-opus-4-2026", "timeout": 45, "retries": 1}, {"model": "gpt-5-mini", "timeout": 15, "retries": 2}, {"model": "gpt-4o-mini", "timeout": 10, "retries": 3}, ] self.circuit_breakers = {m["model"]: CircuitBreaker() for m in self.chain} self.health_checker = HealthChecker() async def invoke(self, messages: list, **kwargs) -> str: errors = [] for tier in self.chain: model = tier["model"] cb = self.circuit_breakers[model] # 检查熔断器状态 if not cb.can_call(): errors.append(f"{model}: circuit breaker open") continue # 检查健康状态 if not await self.health_checker.is_healthy(model): errors.append(f"{model}: unhealthy") continue try: response = await self._call_with_retry( model, messages, timeout=tier["timeout"], retries=tier["retries"], **kwargs ) cb.record_success() # 如果降级了，通知调用方 if model != self.chain[0]["model"]: logger.warning(f"Degraded to {model}") return response except (TimeoutError, RateLimitError) as e: cb.record_failure() errors.append(f"{model}: {e}") continue except Exception as e: cb.record_failure() errors.append(f"{model}: {e}") continue # 所有模型都失败，进入更深层的降级 raise AllModelsFailedError(errors) Level 2：功能降级 class DegradedMode: """降级模式：功能减少但核心可用""" DEGRADATION_RULES = { "gpt-5-mini": { "disable_tools": False, # 仍可用工具 "max_iterations": 3, # 减少迭代次数 "disable_multi_step": True, # 禁用多步推理 "simplified_prompt": True, # 简化 System Prompt "max_context_messages": 5, # 减少上下文 }, "gpt-4o-mini": { "disable_tools": True, # 禁用工具 "max_iterations": 1, # 单轮回复 "disable_multi_step": True, "simplified_prompt": True, "max_context_messages": 3, } } def apply(self, request: dict, model: str) -> dict: rules = self.DEGRADATION_RULES.get(model, {}) degraded_request = request.copy() if rules.get("simplified_prompt"): degraded_request["system"] = self._simplify_prompt(request.get("system", "")) if rules.get("disable_tools"): degraded_request.pop("tools", None) if rules.get("max_context_messages"): messages = degraded_request.get("messages", []) degraded_request["messages"] = messages[-rules["max_context_messages"]:] degraded_request["max_tokens"] = min( degraded_request.get("max_tokens", 4096), 1024 # 降级时限制输出长度 ) return degraded_request Level 3：缓存降级 class CacheFallback: """缓存降级：使用历史相似查询的结果""" async def try_cached_response(self, query: str) -> CachedResponse | None: # 1. 精确匹配 exact = await self.cache.get(query) if exact: return CachedResponse( content=exact, source="exact_cache", warning="This response is from cache due to AI service degradation." ) # 2. 语义匹配 similar = await self.semantic_cache.search(query, threshold=0.85) if similar: return CachedResponse( content=similar.response, source="semantic_cache", warning="This response is based on a similar historical query." ) # 3. 预计算回答（高频问题） precomputed = await self.precomputed_db.get(query) if precomputed: return CachedResponse( content=precomputed, source="precomputed", warning=None # 预计算结果质量有保证 ) return None Level 4：规则引擎降级 class RuleEngineFallback: """规则引擎：LLM 不可用时的最终保底""" def __init__(self): self.rules = self._load_rules() def _load_rules(self) -> list[Rule]: """加载预定义规则""" return [ # 意图匹配规则 Rule( pattern=r"(.*)价格(.*)", action=lambda m: f"我们的产品价格请参考：{self._get_pricing_table()}", intent="pricing_inquiry" ), Rule( pattern=r"(.*)退款(.*)", action=lambda m: f"退款申请已记录。您的退款将在3-5个工作日内处理。" f"工单号：{self._generate_ticket()}", intent="refund_request" ), Rule( pattern=r"(hello|hi|你好|嗨)", action=lambda m: "您好！我是智能助手。" "AI服务暂时受限，但我可以处理基础请求。请说明您需要什么帮助。", intent="greeting" ), ] def handle(self, user_input: str) -> str: for rule in self.rules: match = re.match(rule.pattern, user_input, re.IGNORECASE) if match: return rule.action(match) # 兜底回复 return ( "AI 服务暂时不可用，我无法处理您的复杂请求。\n" "您可以：\n" "1. 稍后重试\n" "2. 联系人工客服：400-xxx-xxxx\n" "3. 访问帮助中心：https://help.example.com" ) 四、降级决策器 class DegradationOrchestrator: """降级编排器：协调各级降级策略""" def __init__(self): self.llm_chain = LLMFailoverChain() self.cache = CacheFallback() self.rules = RuleEngineFallback() self.degraded_mode = DegradedMode() self.health_monitor = HealthMonitor() async def handle_request( self, messages: list, tools: list | None = None, user_id: str = "", priority: str = "normal" ) -> Response: # 检查系统健康状态 health = await self.health_monitor.get_status() # 根据健康状态选择策略 if health.llm_available: try: return await self.llm_chain.invoke(messages, tools=tools) except AllModelsFailedError: health.llm_available = False if health.cache_available: cached = await self.cache.try_cached_response( messages[-1]["content"] ) if cached: return Response( content=cached.content, degraded=True, source=cached.source, warning=cached.warning ) # 规则引擎兜底 rule_response = self.rules.handle(messages[-1]["content"]) return Response( content=rule_response, degraded=True, source="rule_engine", warning="AI 服务暂时不可用，正在使用规则引擎处理。" ) async def health_check_loop(self): """持续健康检查，自动恢复""" while True: await asyncio.sleep(30) # 每30秒检查一次 if not self.health_monitor.llm_available: # 尝试恢复 test_result = await self._test_llm_health() if test_result: self.health_monitor.llm_available = True logger.info("LLM service recovered!") await self._notify_recovery() 五、降级通知与用户体验 class DegradedUX: """降级时的用户体验设计""" MESSAGES = { "model_switch": { "inline": "（响应速度可能略有不同）", "banner": "⚠️ AI 服务正在切换到备用节点，响应可能略有延迟。" }, "cache_fallback": { "inline": "（以下为缓存回复）", "banner": "⚠️ AI 服务暂时繁忙，正在使用历史缓存回复您。" }, "rule_engine": { "inline": "（基础模式）", "banner": "⚠️ AI 服务暂时不可用，正在使用基础模式处理您的请求。" }, "full_failure": { "banner": "❌ AI 服务暂时不可用。您可以通过以下方式获取帮助：\n" "📞 客服热线：400-xxx-xxxx\n" "💬 在线客服：点击右下角\n" "📧 邮件：support@example.com\n" "我们正在紧急修复中，请稍后重试。" } } def render(self, response: Response) -> dict: if not response.degraded: return {"content": response.content, "banner": None} ux = self.MESSAGES.get(response.source, {}) return { "content": response.content, "banner": ux.get("banner"), "inline_note": ux.get("inline"), "retry_available": True, } 六、降级演练 class DegradationDrill: """降级演练：定期验证降级策略有效性""" async def run_drill(self): """模拟 LLM 故障，验证降级链""" results = [] # 测试 1: 主模型超时 with mock_llm_timeout(ModelTier.PRIMARY): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is 2+2?"}] ) results.append(DrillResult( test="primary_timeout", passed=result.degraded == True, response_source=result.source )) # 测试 2: 所有模型宕机 with mock_all_llm_down(): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "你好"}] ) results.append(DrillResult( test="all_llm_down", passed=result.source == "rule_engine", response_source=result.source )) # 测试 3: 缓存命中 with mock_all_llm_down(): # 先正常请求一次建立缓存 await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) # 再次请求相同问题 result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) results.append(DrillResult( test="cache_hit", passed=result.source in ("exact_cache", "semantic_cache"), response_source=result.source )) return DrillReport(results=results) 七、降级 Checklist □ 多 LLM Provider 热备，自动切换 < 5s □ 降级模型已配置，功能限制明确 □ 语义缓存层已部署，命中率 > 30% □ 高频问题预计算，离线生成 □ 规则引擎已覆盖核心业务意图 □ 降级通知 UX 已设计，用户可感知 □ 健康检查每 30s 执行，自动恢复 □ 降级演练每月执行，结果归档 □ 人工客服兜底渠道畅通 □ 降级期间数据不丢失（请求排队） 结语 降级策略的本质是"接受不完美，但永不不可用"。在 LLM 依赖度越来越高的今天，没有降级方案的 Agent 系统就像没有安全网的走钢丝。投资降级不是悲观——它是工程成熟度的标志。最好的降级是用户从未注意到的降级。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026 年，AI Agent 从实验性技术走向生产环境部署，标准化成为行业发展的迫切需求。当企业开始部署来自不同供应商的 Agent 系统，当 Agent 需要跨平台协作，当监管机构需要评估 Agent 安全性时——缺乏标准化的后果变得不可忽视。 2026 年是 AI Agent 标准化的"爆发年"。ISO、IEEE、W3C、ITU 等主要标准组织纷纷推出了 Agent 相关标准，覆盖互操作性、安全评估、性能基准和伦理框架等多个维度。本文将全面梳理这些进展。 一、为什么 Agent 标准化迫在眉睫 碎片化问题 2026 年的 Agent 生态面临严重的碎片化问题： 框架碎片化。 主流 Agent 框架（LangGraph、AutoGen、CrewAI、Semantic Kernel、LangChain）各有自己的架构概念、状态管理方式和工具调用协议。一个在 LangGraph 上开发的 Agent 无法直接在 AutoGen 上运行，反之亦然。 工具接口碎片化。 每个 Agent 框架都有自己的工具定义格式和调用方式。一个为 LangChain 开发的工具需要重写才能在 CrewAI 中使用。这种"工具孤岛"严重阻碍了生态发展。 评估标准缺失。 不同 Agent 框架使用不同的评估方法和基准测试，使得跨框架的性能对比几乎不可能。企业在选择 Agent 框架时缺乏客观依据。 安全标准空白。 Agent 的安全风险（越狱、工具滥用、信息泄露）与传统 AI 模型不同，但 2026 年之前没有专门针对 Agent 的安全标准。 标准化的潜在收益 据 McKinsey 估算，Agent 标准化可以： 降低 Agent 开发成本 40-60% 缩短企业 Agent 部署周期 50% 使 Agent 市场规模在 2027 年达到 $50B（无标准化的情景为 $20B） 减少 70% 的安全事件 二、ISO Agent 标准进展 ISO/IEC 42022：AI Agent 系统架构 ISO/IEC JTC 1/SC 42 在 2026 年 3 月发布了 ISO/IEC 42022《AI Agent 系统架构》标准草案。这是首个专门针对 AI Agent 的国际标准。 ...