Agent

引言 2026年，Agent 不再是"全自动"或"全人工"的二元选择——而是光谱上的渐进式协作。从"全自动执行"到"每一步都确认"，中间有无数种可能。好的 Agent 设计让人类知道何时需要介入，何时可以放手。本文探讨人机协作的设计模式。 一、人机协作光谱 全自主 Agent 全人工控制 ◀─────────┬─────────┬─────────┬─────────▶ │ │ │ 全自动执行 需确认 需监督 需审批 (Auto) (Confirm) (Supervise) (Approve) 设计原则：让 Agent 做它能做好的，让人做只有人能做的 模式 自动化程度 人类参与点 适用场景 全自动 100% 事后审计 低风险、高频、标准化任务 需确认 80% 关键决策点 中风险、需合规 需监督 50% 持续监控 高风险、创造性任务 需审批 20% 每步审批 高风险、不可撤销 二、全自动模式设计 2.1 适用条件 全自动 Agent 只有在满足以下条件时才安全： class AutoModeSafetyCheck: """全自动模式安全检查""" SAFETY_RULES = [ # 1. 任务风险评级 ("task_risk_level", "low"), # 必须是低风险任务 ("output_reversibility", True), # 输出必须可撤销 ("cost_per_execution", 1.0), # 单次成本 < $1 ("data_access_level", "public"), # 仅访问公开数据 ("external_api_calls", "read_only"), # 外部调用只读 # 2. 质量保障 ("quality_score_threshold", 0.95), # 质量评分 > 95% ("error_rate_threshold", 0.01), # 错误率 < 1% ("test_coverage", 0.90), # 测试覆盖率 > 90% # 3. 监控 ("real_time_monitoring", True), # 实时监控 ("alert_on_anomaly", True), # 异常告警 ("automatic_rollback", True), # 自动回滚 ] def can_auto_mode( self, task: Task, agent: Agent ) -> SafetyAssessment: violations = [] # 检查任务风险 risk = self._assess_risk(task) if risk.level != "low": violations.append(f"Task risk level: {risk.level}") # 检查 Agent 历史表现 metrics = agent.get_performance_metrics(days=30) if metrics.error_rate > 0.01: violations.append(f"Error rate: {metrics.error_rate:.1%}") if metrics.avg_quality < 0.95: violations.append(f"Quality score: {metrics.avg_quality:.1%}") # 检查成本 estimated_cost = agent.estimate_cost(task) if estimated_cost > 1.0: violations.append(f"Estimated cost: ${estimated_cost:.2f}") return SafetyAssessment( can_auto=len(violations) == 0, violations=violations, recommended_mode="auto" if not violations else "confirm", confidence=1.0 - len(violations) * 0.2 ) 2.2 全自动 Agent 的护栏 class AutoModeGuardrails: """全自动模式的护栏""" def __init__(self): self.rules = self._load_guardrails() async def pre_check(self, task: Task) -> GuardrailResult: """执行前检查""" violations = [] # 1. 输入安全检查 if self._contains_sensitive_data(task.input): violations.append("Input contains sensitive data") # 2. 成本预估 estimated_cost = self._estimate_cost(task) if estimated_cost > self.rules.max_cost_per_run: violations.append(f"Estimated cost ${estimated_cost:.2f} > limit") # 3. 工具权限检查 required_tools = self._get_required_tools(task) unauthorized = [ t for t in required_tools if not self.rules.is_authorized(t, "auto") ] if unauthorized: violations.append(f"Unauthorized tools: {unauthorized}") return GuardrailResult( passed=len(violations) == 0, violations=violations ) async def post_check(self, output: str, actions: list) -> GuardrailResult: """执行后检查""" violations = [] # 1. 输出安全检查 safety = await self._check_output_safety(output) if not safety.safe: violations.append(f"Output safety: {safety.reason}") # 2. 行动审计 for action in actions: if action.type == "external_api" and action.method != "GET": violations.append(f"Non-read action executed: {action}") # 3. 成本检查 actual_cost = self._calculate_cost(actions) if actual_cost > self.rules.max_cost_per_run * 2: violations.append(f"Cost overrun: ${actual_cost:.2f}") return GuardrailResult( passed=len(violations) == 0, violations=violations, requires_escalation=len(violations) > 0 ) 三、需确认模式（Confirm） 3.1 确认点设计 class ConfirmationPoint: """确认点设计""" # 需要确认的情景 TRIGGERS = { "high_cost": lambda ctx: ctx.estimated_cost > 5.0, "external_api_write": lambda ctx: any( t.method != "GET" for t in ctx.tool_calls ), "irreversible_action": lambda ctx: any( t.tool in ["delete_file", "send_email", "publish_post"] for t in ctx.tool_calls ), "low_confidence": lambda ctx: ctx.confidence < 0.7, "new_tool": lambda ctx: any( t.tool not in ctx.agent.verified_tools for t in ctx.tool_calls ), } async def should_confirm(self, context: ExecutionContext) -> list[str]: """判断是否需要确认""" reasons = [] for trigger_name, trigger_fn in self.TRIGGERS.items(): if trigger_fn(context): reasons.append(trigger_name) return reasons def build_confirmation_ui( self, context: ExecutionContext, reasons: list[str] ) -> ConfirmationRequest: """构建确认 UI""" return ConfirmationRequest( title="Agent 需要确认", message=self._generate_message(reasons), plan=context.execution_plan, # Agent 的执行计划 estimated_cost=context.estimated_cost, estimated_time=context.estimated_time, risks=self._identify_risks(context), actions_preview=self._preview_actions(context.tool_calls), confirm_text="确认执行", cancel_text="取消", modify_text="修改计划", ) 3.2 确认 UI 实现 interface ConfirmationRequest { title: string; message: string; plan: ExecutionStep[]; estimated_cost: number; estimated_time: string; risks: string[]; actions_preview: ActionPreview[]; confirm_text: string; cancel_text: string; modify_text: string; } class AgentConfirmationDialog { render(request: ConfirmationRequest): JSX.Element { return ( <Card> <CardHeader> <Icon name="help-circle" /> <Title>{request.title}</Title> </CardHeader> <CardBody> {/* 原因说明 */} <Alert type="warning"> {request.message} </Alert> {/* 执行计划 */} <Section title="执行计划"> {request.plan.map((step, i) => ( <StepCard key={i} step={step} index={i} /> ))} </Section> {/* 风险评估 */} {request.risks.length > 0 && ( <Section title="⚠️ 风险提示"> <List items={request.risks} /> </Section> )} {/* 成本预估 */} <CostEstimate cost={request.estimated_cost} time={request.estimated_time} /> {/* 操作预览 */} <Section title="操作预览"> {request.actions_preview.map((action, i) => ( <ActionPreview key={i} action={action} /> ))} </Section> </CardBody> <CardFooter> <Button variant="outline" onClick={this.onCancel}> {request.cancel_text} </Button> <Button variant="outline" onClick={this.onModify}> {request.modify_text} </Button> <Button variant="primary" onClick={this.onConfirm}> {request.confirm_text} </Button> </CardFooter> </Card> ); } onConfirm = () => { this.props.onResponse({action: "confirm"}); }; onCancel = () => { this.props.onResponse({action: "cancel"}); }; onModify = () => { // 打开修改对话框 this.props.onResponse({ action: "modify", modifications: this.getModifications() }); }; } 四、需监督模式（Supervise） 4.1 实时监督界面 class SupervisedAgentUI: """需监督 Agent 的实时界面""" def render_live_view(self, session_id: str) -> str: """渲染实时监督界面""" # 获取 Agent 当前状态 state = self.agent.get_state(session_id) return f""" <div class="supervised-agent-ui"> <div class="agent-status"> <StatusBadge status="{state.status}" /> <span>当前步骤: {state.current_step}</span> <ProgressBar progress="{state.progress}" /> </div> <div class="reasoning-view"> <h4>🤔 Agent 正在思考</h4> <pre>{state.current_reasoning}</pre> </div> <div class="tool-execution-view"> <h4>🔧 工具执行</h4> {self._render_tool_execution(state.tool_history)} </div> <div class="controls"> <button onclick="pauseAgent()">⏸️ 暂停</button> <button onclick="resumeAgent()">▶️ 继续</button> <button onclick="stopAgent()">⏹️ 停止</button> <button onclick="provideFeedback()">💬 提供反馈</button> <button onclick="takeOver()">🤝 接管</button> </div> </div> """ def _render_tool_execution(self, history: list) -> str: html = "<ul class='tool-history'>" for item in history: status_icon = "✅" if item.success else "❌" html += f""" <li class='tool-item'> <span class='tool-name'>{item.tool_name}</span> <span class='tool-status'>{status_icon}</span> <pre class='tool-result'>{item.result_preview}</pre> </li> """ html += "</ul>" return html 4.2 监督模式的控制权转移 class ControlTransfer: """控制权转移管理""" async def request_control( self, from_entity: str, # "agent" or "human" to_entity: str, reason: str, context: dict ) -> ControlTransferResult: """请求控制权转移""" # 1. 检查是否可以转移 if not self._can_transfer(from_entity, to_entity): return ControlTransferResult( success=False, reason="Transfer not allowed in current state" ) # 2. 保存当前状态 snapshot = await self._take_snapshot(context["session_id"]) # 3. 转移控制权 self.current_controller = to_entity self.control_history.append({ "from": from_entity, "to": to_entity, "reason": reason, "timestamp": time.time(), "snapshot_id": snapshot.id }) # 4. 通知各方 await self._notify_control_change( from_entity, to_entity, reason ) return ControlTransferResult( success=True, snapshot=snapshot, instructions=self._get_instructions_for(to_entity) ) async def human_takeover( self, session_id: str, human_instructions: str ) -> str: """人类接管 Agent 执行""" # 请求控制权 result = await self.request_control( from_entity="agent", to_entity="human", reason="Human takeover", context={"session_id": session_id} ) if not result.success: raise ControlTransferError(result.reason) # 执行人类指令 response = await self.human_executor.execute( session_id=session_id, instructions=human_instructions, starting_from=result.snapshot ) # 可选：交还控制权给 Agent if response.hand_back_to_agent: await self.request_control( from_entity="human", to_entity="agent", reason="Task completed by human", context={"session_id": session_id} ) return response.output 五、需审批模式（Approve） 5.1 审批工作流 class ApprovalWorkflow: """审批工作流""" async def submit_for_approval( self, session_id: str, action: dict, priority: str = "normal" ) -> ApprovalRequest: """提交审批""" request = ApprovalRequest( id=str(uuid.uuid4()), session_id=session_id, action=action, priority=priority, submitted_by="agent", submitted_at=datetime.now(), status="pending", approvers=self._get_approvers(action), deadline=self._calculate_deadline(priority), ) # 保存审批请求 await self.approval_db.save(request) # 通知审批人 await self.notification_service.notify_approvers( request.approvers, f"Agent 请求审批: {request.id}", self._generate_approval_ui(request) ) # 设置超时处理 asyncio.create_task(self._handle_timeout(request)) return request async def handle_approval( self, request_id: str, approver: str, decision: str, # "approve" or "reject" comment: str = "" ) -> ApprovalResult: """处理审批""" request = await self.approval_db.get(request_id) # 检查审批人权限 if approver not in request.approvers: raise UnauthorizedApproverError(approver) # 记录审批 approval = Approval( request_id=request_id, approver=approver, decision=decision, comment=comment, timestamp=datetime.now() ) await self.approval_db.save_approval(approval) # 执行相应操作 if decision == "approve": return await self._execute_approved_action(request) else: return await self._handle_rejection(request, approval) async def _handle_timeout(self, request: ApprovalRequest): """处理审批超时""" await asyncio.sleep(request.deadline - time.time()) # 检查是否已审批 if request.status == "pending": # 根据策略处理 if request.priority == "urgent": # 紧急请求：自动升级 await self._escalate(request) else: # 普通请求：自动拒绝 await self.handle_approval( request.id, approver="system", decision="reject", comment="Approval timeout" ) 5.2 审批 UI class ApprovalUI { render(request: ApprovalRequest): JSX.Element { return ( <Card className="approval-request"> <CardHeader> <Icon name="clipboard-check" /> <Title>审批请求</Title> <Badge variant={request.priority}>{request.priority}</Badge> </CardHeader> <CardBody> <Section title="请求详情"> <Descriptions> <Description label="请求ID">{request.id}</Description> <Description label="提交时间"> {formatDateTime(request.submitted_at)} </Description> <Description label="截止时间"> {formatDateTime(request.deadline)} </Description> </Descriptions> </Section> <Section title="待审批操作"> <ActionDetail action={request.action} /> </Section> <Section title="风险分析"> <RiskAnalysis risk={request.risk_assessment} /> </Section> <Section title="审批意见"> <TextArea placeholder="请输入审批意见（可选）" value={this.state.comment} onChange={this.onCommentChange} /> </Section> </CardBody> <CardFooter> <Button variant="danger" onClick={this.onReject}> 拒绝 </Button> <Button variant="primary" onClick={this.onApprove}> 批准 </Button> </CardFooter> </Card> ); } onApprove = async () => { const result = await api.approveRequest( this.props.request.id, this.state.comment ); if (result.success) { toast.success("已批准"); this.props.onApproved(); } }; onReject = async () => { if (!this.state.comment) { toast.error("拒绝时必须填写意见"); return; } const result = await api.rejectRequest( this.props.request.id, this.state.comment ); if (result.success) { toast.success("已拒绝"); this.props.onRejected(); } }; } 六、渐进式自动化 6.1 信任度模型 class TrustAccumulationModel: """信任度积累模型——支持渐进式自动化""" def __init__(self): self.trust_factors = { "success_rate": 0.30, # 成功率权重 "quality_score": 0.25, # 质量评分权重 "consistency": 0.20, # 一致性权重 "user_feedback": 0.15, # 用户反馈权重 "expert_validation": 0.10, # 专家验证权重 } async def calculate_trust_score( self, agent_id: str, user_id: str, task_type: str ) -> TrustScore: """计算信任度评分""" # 获取历史表现 history = await self._get_agent_history( agent_id, user_id, task_type, days=30 ) if len(history) < 10: return TrustScore( score=0.0, level="new", automation_allowed=False, reason=" insufficient history" ) # 计算各因子得分 factors = {} factors["success_rate"] = history.success_rate factors["quality_score"] = history.avg_quality factors["consistency"] = 1.0 - history.variance factors["user_feedback"] = history.avg_user_rating / 5.0 factors["expert_validation"] = history.expert_approval_rate # 加权计算 score = sum( factors[factor] * weight for factor, weight in self.trust_factors.items() ) # 映射到自动化级别 if score >= 0.90: level, allowed = "full_auto", True elif score >= 0.75: level, allowed = "confirm", True elif score >= 0.50: level, allowed = "supervise", True else: level, allowed = "approve", False return TrustScore( score=score, level=level, automation_allowed=allowed, factors=factors, sample_size=len(history) ) 6.2 自动化级别升级 class AutomationLevelManager: """自动化级别管理""" LEVELS = ["approve", "supervise", "confirm", "auto"] async def try_upgrade_level( self, agent_id: str, user_id: str, task_type: str ) -> UpgradeResult: """尝试升级自动化级别""" current = await self._get_current_level(agent_id, user_id, task_type) trust = await self.trust_model.calculate_trust_score( agent_id, user_id, task_type ) if not trust.automation_allowed: return UpgradeResult( success=False, reason=trust.reason, current_level=current, suggested_level=current ) target_level = self._level_from_trust(trust.score) if self.LEVELS.index(target_level) <= self.LEVELS.index(current): return UpgradeResult( success=False, reason="Already at or above target level", current_level=current, suggested_level=current ) # 请求用户确认升级 confirmation = await self._request_upgrade_confirmation( user_id, current, target_level, trust ) if confirmation.approved: await self._set_level( agent_id, user_id, task_type, target_level ) return UpgradeResult( success=True, reason="User approved upgrade", current_level=current, suggested_level=target_level ) else: return UpgradeResult( success=False, reason="User declined upgrade", current_level=current, suggested_level=current ) 七、人机协作设计 Checklist □ 自动化级别与任务风险匹配 □ 全自动模式有完整护栏 □ 确认点在关键决策前触发 □ 监督模式提供实时状态视图 □ 审批工作流支持多级审批 □ 控制权可在人和 Agent 间转移 □ 信任度模型支持渐进式自动化 □ 所有人工介入点有完整上下文 □ 用户可随时暂停/继续/接管 □ 操作日志完整可追溯 结语 人机协作不是"人 vs 机器"的零和游戏，而是发挥各自优势的组合。Agent 擅长执行、搜索、计算；人类擅长判断、创意、伦理决策。好的协作设计让 Agent 知道何时该求助，让人类知道何时该放手。在 Agent 能力越来越强的2026年，最强大的不是完全自主的 Agent，而是最懂得与人类协作的 Agent。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 “你的 Agent 快吗？"——这个问题无法简单回答。Agent 的性能不是单一数字，而是延迟、吞吐量、成本、质量的四维空间。2026年，随着 AgentBench、SWE-bench 等标准化评测框架成熟，我们终于有了科学的 Agent 性能基准测试方法论。 一、四维性能模型 ┌──────────────────────────────────────────────┐ │ Agent 性能四维空间 │ ├──────────────┬───────────────────────────────┤ │ 延迟 (Latency) │ 首 Token 延迟 │ │ │ 完整响应延迟 │ │ │ P50/P95/P99 分布 │ ├──────────────┼───────────────────────────────┤ │ 吞吐 (Throughput)│ 请求/秒 │ │ │ 并发用户数 │ │ │ Token/秒 │ ├──────────────┼─────────────────────────────── │ 成本 (Cost) │ 单次请求成本 │ │ │ Token 效率 │ │ │ 月度总成本 │ ├──────────────┼───────────────────────────────┤ │ 质量 (Quality) │ 任务完成率 │ │ │ 输出准确率 │ │ │ 用户满意度 │ └──────────────┴───────────────────────────────┘ 关键洞察：四维之间存在 tradeoff - 提高质量通常增加延迟和成本 - 降低成本通常降低质量 - 提高吞吐通常增加延迟 二、延迟基准测试 2.1 延迟分解 class LatencyBreakdown: """Agent 延迟分解模型""" COMPONENTS = { "network_ingress": "API Gateway 到达延迟", "auth": "认证授权延迟", "queue": "排队等待延迟", "context_preparation": "上下文准备（历史压缩等）", "llm_first_token": "LLM 首 Token 延迟", "llm_streaming": "LLM 流式输出延迟", "tool_execution": "工具执行延迟", "tool_overhead": "工具调度开销", "state_persistence": "状态持久化延迟", "network_egress": "响应返回延迟", } @dataclass class LatencyMeasurement: component: str duration_ms: float percentage: float # 占总延迟百分比 def analyze(self, trace: list[dict]) -> list[LatencyMeasurement]: """从执行 trace 分析延迟分布""" total = sum(t["duration_ms"] for t in trace) return [ LatencyMeasurement( component=t["component"], duration_ms=t["duration_ms"], percentage=t["duration_ms"] / total * 100 ) for t in sorted(trace, key=lambda x: -x["duration_ms"]) ] # 典型 Agent 延迟分布 TYPICAL_BREAKDOWN = """ 组件 延迟(ms) 占比 ───────────────────────────────────────── llm_first_token 1200 40% llm_streaming 800 27% tool_execution 450 15% context_preparation 200 7% queue 150 5% state_persistence 100 3% auth 50 2% network 40 1% ───────────────────────────────────────── 总计 2990 100% 优化优先级：LLM 延迟占 67%，是首要优化目标 """ 2.2 延迟测试框架 class AgentLatencyBenchmark: """Agent 延迟基准测试""" TEST_SCENARIOS = [ BenchmarkScenario( name="simple_qa", description="简单问答（无工具）", query="What is 2+2?", expected_max_latency_ms=3000, tools=[], ), BenchmarkScenario( name="single_tool", description="单工具调用", query="Search for latest AI news", expected_max_latency_ms=8000, tools=["web_search"], ), BenchmarkScenario( name="multi_tool", description="多工具串联（3步）", query="Research and summarize quantum computing breakthroughs in 2026", expected_max_latency_ms=30000, tools=["web_search", "summarizer", "write_file"], ), BenchmarkScenario( name="complex_reasoning", description="复杂推理（5+步）", query="Analyze the competitive landscape of AI chip market", expected_max_latency_ms=60000, tools=["web_search", "data_analyzer", "chart_gen", "write_file"], ), ] async def run_benchmark( self, agent: Agent, scenarios: list[BenchmarkScenario] | None = None, iterations: int = 100 ) -> BenchmarkReport: scenarios = scenarios or self.TEST_SCENARIOS results = {} for scenario in scenarios: latencies = [] first_token_latencies = [] for _ in range(iterations): start = time.time() first_token_time = None async for chunk in agent.run_stream(scenario.query): if first_token_time is None: first_token_time = time.time() end = time.time() total_latency = (end - start) * 1000 first_token_latency = (first_token_time - start) * 1000 latencies.append(total_latency) first_token_latencies.append(first_token_latency) results[scenario.name] = LatencyResult( scenario=scenario.name, p50=np.percentile(latencies, 50), p95=np.percentile(latencies, 95), p99=np.percentile(latencies, 99), mean=np.mean(latencies), std=np.std(latencies), first_token_p50=np.percentile(first_token_latencies, 50), first_token_p95=np.percentile(first_token_latencies, 95), passed_p95=np.percentile(latencies, 95) < scenario.expected_max_latency_ms, ) return BenchmarkReport(results=results) 三、吞吐量基准测试 class ThroughputBenchmark: """吞吐量基准测试""" async def test_concurrent_users( self, agent: Agent, query: str, concurrent_users: list[int] = [1, 10, 50, 100, 200, 500] ) -> list[ThroughputResult]: results = [] for n_users in concurrent_users: print(f"Testing with {n_users} concurrent users...") # 创建并发请求 tasks = [ self._timed_request(agent, query, user_id=i) for i in range(n_users) ] start = time.time() responses = await asyncio.gather(*tasks, return_exceptions=True) total_time = time.time() - start # 统计 success_count = sum(1 for r in responses if not isinstance(r, Exception)) error_count = sum(1 for r in responses if isinstance(r, Exception)) result = ThroughputResult( concurrent_users=n_users, total_requests=n_users, successful_requests=success_count, failed_requests=error_count, total_time_s=total_time, requests_per_second=success_count / total_time, avg_latency_ms=np.mean([ r["latency_ms"] for r in responses if isinstance(r, dict) ]), p95_latency_ms=np.percentile([ r["latency_ms"] for r in responses if isinstance(r, dict) ], 95), error_rate=error_count / n_users, ) results.append(result) # 如果错误率 > 20%，停止加压 if result.error_rate > 0.2: print(f"Error rate {result.error_rate:.0%} > 20%, stopping") break return results async def find_max_throughput( self, agent: Agent, query: str, target_latency_p95_ms: float = 10000, target_error_rate: float = 0.01 ) -> int: """找到满足 SLA 的最大并发数""" # 二分搜索 low, high = 1, 1000 best = 1 while low <= high: mid = (low + high) // 2 results = await self.test_concurrent_users( agent, query, [mid] ) result = results[0] if (result.p95_latency_ms <= target_latency_p95_ms and result.error_rate <= target_error_rate): best = mid low = mid + 1 else: high = mid - 1 return best 四、成本效率基准 class CostEfficiencyBenchmark: """成本效率基准测试""" async def benchmark( self, agent: Agent, test_cases: list[TestCase] ) -> CostReport: results = [] for case in test_cases: start_cost = agent.total_cost response = await agent.run(case.input) cost = agent.total_cost - start_cost # 评估输出质量 quality = await self.judge.evaluate( case.input, response, case.criteria ) results.append(CostResult( test_id=case.id, input_tokens=agent.last_input_tokens, output_tokens=agent.last_output_tokens, total_tokens=agent.last_total_tokens, cost_usd=cost, quality_score=quality.score, cost_per_quality=cost / max(quality.score, 0.01), # 成本效率比 iterations=agent.iteration_count, )) return CostReport( results=results, avg_cost=np.mean([r.cost_usd for r in results]), avg_quality=np.mean([r.quality_score for r in results]), avg_cost_per_quality=np.mean([r.cost_per_quality for r in results]), total_cost=sum(r.cost_usd for r in results), cost_distribution=self._analyze_distribution( [r.cost_usd for r in results] ), ) def compare_models( self, models: list[str], test_cases: list[TestCase] ) -> ComparisonReport: """对比不同模型的成本效率""" model_results = {} for model in models: agent = Agent(llm=LLM(model=model)) report = self.benchmark(agent, test_cases) model_results[model] = report # 生成对比表 return ComparisonReport( models=model_results, best_cost=min(model_results.items(), key=lambda x: x[1].avg_cost), best_quality=max(model_results.items(), key=lambda x: x[1].avg_quality), best_efficiency=min( model_results.items(), key=lambda x: x[1].avg_cost_per_quality ), ) 五、质量基准测试 class QualityBenchmark: """Agent 输出质量基准测试""" BENCHMARK_SUITES = { "reasoning": ReasoningSuite(), # 推理能力 "coding": CodingSuite(), # 代码生成 "tool_use": ToolUseSuite(), # 工具使用 "safety": SafetySuite(), # 安全性 "instruction_follow": InstructionSuite(), # 指令遵循 "multilingual": MultilingualSuite(), # 多语言 } async def run_full_benchmark( self, agent: Agent, suites: list[str] | None = None ) -> FullBenchmarkReport: suites = suites or list(self.BENCHMARK_SUITES.keys()) results = {} for suite_name in suites: suite = self.BENCHMARK_SUITES[suite_name] suite_results = [] for test_case in suite.get_cases(): # 运行 Agent output = await agent.run(test_case.input) # 自动化评估 auto_score = await suite.evaluate( test_case, output ) # LLM-as-Judge 评估 judge_score = await self.judge.evaluate( test_case.input, output, test_case.criteria ) # 统计 suite_results.append(QualityResult( test_id=test_case.id, category=test_case.category, output_preview=output[:200], auto_score=auto_score, judge_score=judge_score.score, passed=judge_score.score >= test_case.min_score, duration_ms=test_case.duration_ms, )) results[suite_name] = SuiteResult( total=len(suite_results), passed=sum(1 for r in suite_results if r.passed), pass_rate=sum(1 for r in suite_results if r.passed) / len(suite_results), avg_score=np.mean([r.judge_score for r in suite_results]), results=suite_results, ) return FullBenchmarkReport( suites=results, overall_pass_rate=np.mean([ r.pass_rate for r in results.values() ]), timestamp=datetime.now(), ) 六、综合性能评分 class AgentPerformanceScore: """Agent 综合性能评分""" def calculate( self, latency: LatencyResult, throughput: ThroughputResult, cost: CostReport, quality: FullBenchmarkReport ) -> PerformanceScore: # 归一化评分（0-100） # 延迟分（越低越好，基准 30s = 0分, 1s = 100分） latency_score = max(0, min(100, 100 * (30 - latency.p95 / 1000) / 29 )) # 吞吐分（越高越好，基准 1 RPS = 0分, 100 RPS = 100分） throughput_score = max(0, min(100, 100 * throughput.requests_per_second / 100 )) # 成本分（越低越好，基准 $0.1/请求 = 0分, $0.001/请求 = 100分） cost_score = max(0, min(100, 100 * (0.1 - cost.avg_cost) / 0.099 )) # 质量分（越高越好） quality_score = quality.overall_pass_rate * 100 # 加权综合 weights = { "latency": 0.20, "throughput": 0.15, "cost": 0.25, "quality": 0.40, } overall = sum(score * weights[key] for key, score in [ ("latency", latency_score), ("throughput", throughput_score), ("cost", cost_score), ("quality", quality_score), ]) return PerformanceScore( overall=overall, latency=latency_score, throughput=throughput_score, cost=cost_score, quality=quality_score, grade=self._grade(overall), tradeoffs=self._analyze_tradeoffs( latency_score, throughput_score, cost_score, quality_score ), ) def _grade(self, score: float) -> str: if score >= 90: return "A+" if score >= 80: return "A" if score >= 70: return "B" if score >= 60: return "C" if score >= 50: return "D" return "F" 七、持续基准测试 # .github/workflows/agent-benchmark.yml name: Agent Performance Benchmark on: schedule: - cron: "0 2 * * 1" # 每周一凌晨2点 workflow_dispatch: # 手动触发 jobs: benchmark: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Run latency benchmark run: python benchmarks/latency_benchmark.py --output results/latency.json - name: Run throughput benchmark run: python benchmarks/throughput_benchmark.py --output results/throughput.json - name: Run cost benchmark run: python benchmarks/cost_benchmark.py --output results/cost.json - name: Run quality benchmark run: python benchmarks/quality_benchmark.py --output results/quality.json - name: Generate report run: python benchmarks/generate_report.py --input results/ --output report.md - name: Compare with baseline run: | python benchmarks/compare_baseline.py \ --current results/ \ --baseline benchmarks/baseline/ \ --threshold-latency 10 \ --threshold-cost 5 \ --threshold-quality 2 - name: Upload results uses: actions/upload-artifact@v4 with: name: benchmark-results path: results/ - name: Notify on regression if: failure() uses: ./.github/actions/slack-notify with: message: "Agent performance regression detected!" 八、基准测试 Checklist □ 四维基准测试覆盖（延迟/吞吐/成本/质量） □ 测试场景分级（简单/中等/复杂） □ 延迟测试包含首 Token 延迟 □ 吞吐测试找到最大并发数 □ 成本测试计算成本效率比 □ 质量测试使用标准化评测集 □ 持续基准测试（每周自动运行） □ 基线对比检测性能回归 □ SLA 定义明确（P95 延迟、错误率） □ 性能评分模型用于横向对比 结语 基准测试不是一次性的活动，而是持续的过程。Agent 的性能会随着 Prompt 修改、模型升级、工具变更而变化。建立持续的基准测试体系，让性能回归在 CI 阶段就被发现，而不是等到用户投诉。记住：没有测量就没有优化。在你开始优化 Agent 性能之前，先确保你能准确测量它。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

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

AutoGPT 的涅槃重生 2023 年，AutoGPT 以"让 AI 自己思考"的口号引爆了 AI 社区，成为 GitHub 上增长最快的项目之一。然而，早期的 AutoGPT 因成本高昂、循环执行、目标模糊等问题，逐渐淡出主流视野。2026 年，在 Significant Gravitas 团队的持续努力下，AutoGPT 迎来了真正的涅槃重生——从概念验证进化为可用的自主智能体平台。 2026 架构全景 从单体到微服务 2026 版 AutoGPT 彻底重写了架构，从单体应用转变为微服务架构： ┌──────────────────────────────────────────────────┐ │ AutoGPT Platform │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Server │ │ Frontend│ │ Market │ │ │ │ (API) │ │ (React) │ │ (Agent │ │ │ │ │ │ │ │ Store) │ │ │ └─────┬────┘ └──────────┘ └──────────┘ │ │ │ │ │ ┌─────┴────────────────────────────────────┐ │ │ │ Agent Runtime │ │ │ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │ │ │Planner │ │Executor│ │Critic │ │ │ │ │ └────────┘ └────────┘ └────────┘ │ │ │ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │ │ │Memory │ │Tools │ │Safety │ │ │ │ │ └────────┘ └────────┘ └────────┘ │ │ │ └──────────────────────────────────────────┘ │ │ │ │ ┌──────────────────────────────────────────┐ │ │ │ Infrastructure Layer │ │ │ │ PostgreSQL │ Redis │ S3 │ Docker │ K8s │ │ │ └──────────────────────────────────────────┘ │ └──────────────────────────────────────────────────┘ 核心改进 维度 2023 版本 2026 版本 架构 单体 Python 脚本 微服务 + 容器化 规划 纯 LLM 生成 HTN + LLM 混合规划 记忆 文件系统 向量数据库 + 图数据库 工具 10+ 基础工具 200+ 可扩展工具 安全 无 多层安全约束框架 成本控制 无 预算限制 + 成本追踪 执行模式 同步循环 异步事件驱动 部署 本地脚本 Docker/K8s 原生 新版核心组件 1. 智能规划器 2026 版 AutoGPT 引入了混合规划器，结合 HTN（层次任务网络）和 LLM 规划： ...

从 RAG 到 Agent：LlamaIndex 的范式跃迁 LlamaIndex 在 2023 年以"RAG 框架"闻名——它是构建检索增强生成应用最简单的方式。但创始人 Jerry Liu 很早就意识到：RAG 本质上是 Agent 的一个特例。到 2026 年，LlamaIndex 已经完成了从"RAG 框架"到"数据驱动 Agent 平台"的转型。 2026 架构演进 核心层次 ┌──────────────────────────────────────────────────┐ │ Agent Layer │ │ Workflows │ Data Agents │ Multi-Agent │ Tools │ ├──────────────────────────────────────────────────┤ │ Orchestration Layer │ │ Query Engine │ Router │ Planner │ Evaluator │ ├──────────────────────────────────────────────────┤ │ Data Layer │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ Ingestion│ │ Indexing │ │ Retrieval │ │ │ │ Pipeline │ │ (多索引) │ │ (混合检索) │ │ │ └──────────┘ └──────────┘ └──────────────────┘ │ ├──────────────────────────────────────────────────┤ │ Integration Layer │ │ 50+ LLM │ 30+ Vector DB │ 100+ Data Source │ └──────────────────────────────────────────────────┘ 版本对比 特性 LlamaIndex 0.10 (2024) LlamaIndex 1.0 (2026) 核心定位 RAG 框架 Agent 平台 Agent 支持 实验性 一等公民 Workflow 不支持 事件驱动 Workflow 多模态 基础 原生多模态 评估 离线评估 在线评估 + A/B 测试 部署 Python 脚本 LlamaCloud + 本地 性能 中等 优化 40-60% 核心新特性 1. Workflows：事件驱动编排 LlamaIndex 2026 的 Workflow 系统采用事件驱动模型，与 LangGraph 的状态机模型形成对比： ...

OpenClaw 2026：个人 AI 助手的新范式 OpenClaw（龙虾智能体）在 2026 年迎来了诞生以来最重要的一次架构升级。作为一个定位为"个人 AI 操作系统"的开源项目，OpenClaw 的核心理念是：让每个人拥有一个真正属于自己的、可深度定制的、跨设备运行的智能体。 2026 核心架构 总体架构 ┌─────────────────────────────────────────────────────────┐ │ 用户界面层 │ │ WebChat │ 微信 │ Discord │ Telegram │ 桌面客户端 │ └──────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────┴──────────────────────────────────┐ │ Gateway 网关层 │ │ 消息路由 │ 会话管理 │ 权限控制 │ 限流策略 │ 多模型切换 │ └──────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────┴──────────────────────────────────┐ │ Agent 核心层 │ │ ┌─────────┐ ┌─────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 推理引擎 │ │ 记忆系统 │ │ Skill 引擎│ │ 工具执行 │ │ │ │ (LLM) │ │(Memory) │ │ (Skills) │ │ (Tools) │ │ │ └─────────┘ └─────────┘ └──────────┘ └──────────┘ │ │ ┌─────────┐ ┌─────────┐ ┌──────────┐ │ │ │ 子Agent │ │ 定时任务 │ │ 节点管理 │ │ │ │(Subagent│ │ (Cron) │ │ (Nodes) │ │ │ └─────────┘ └─────────┘ └──────────┘ │ └──────────────────────┬──────────────────────────────────┘ │ ┌──────────────────────┴──────────────────────────────────┐ │ 存储与集成层 │ │ 文件系统 │ SQLite │ 向量库 │ MCP协议 │ API集成 │ └─────────────────────────────────────────────────────────┘ Skill 系统：插件化能力扩展 OpenClaw 2026 最显著的改进是 Skill 系统的成熟。每个 Skill 是一个独立的能力模块，通过标准化的接口与 Agent 核心交互： ...