架构设计

引言 Agent系统的故障排查比传统应用复杂得多——一个"回复质量下降"的问题可能涉及Prompt变化、模型版本更新、工具API变更、记忆检索质量下降等多个因素。没有系统化的排查方法论，工程师可能花费数小时甚至数天才能定位根因。 本文基于大量实战经验，提供一套系统化的Agent故障排查方法论。 故障排查金字塔 ┌─────────────┐ │ 用户反馈 │ │ "回复不对" │ └──────┬──────┘ │ ┌──────▼──────┐ │ 指标异常 │ │ 质量评分下降 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 链路追踪 │ │ 定位慢步骤 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 日志分析 │ │ 找到错误日志 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 根因定位 │ │ Prompt/模型 │ │ /数据/代码 │ └─────────────┘ 故障分类 Agent系统故障分类 │ ├── 功能故障 │ ├── 响应错误（幻觉、曲解意图） │ ├── 工具调用失败 │ ├── 路由错误 │ └── 超时/死循环 │ ├── 性能故障 │ ├── 响应变慢 │ ├── 吞吐量下降 │ └── 资源耗尽 │ ├── 质量故障 │ ├── 回复质量下降 │ ├── 用户满意度降低 │ └── Token消耗异常 │ └── 成本故障 ├── Token消耗突增 ├── 基础设施成本异常 └── API调用费用超标 排查流程 Step 1：收集症状 class SymptomCollector: """症状收集器""" async def collect(self, incident_id: str) -> dict: """收集故障症状""" symptoms = { "incident_id": incident_id, "reported_at": datetime.now().isoformat(), "reported_by": None, "description": None, "affected_users": [], "affected_services": [], "start_time": None, "error_rate": None, "quality_score": None, "recent_changes": [], } # 从多个数据源收集 symptoms["recent_deployments"] = await self._get_recent_deployments() symptoms["recent_config_changes"] = await self._get_recent_config_changes() symptoms["recent_model_updates"] = await self._get_recent_model_updates() symptoms["error_logs"] = await self._get_recent_errors() symptoms["metrics_anomalies"] = await self._get_metrics_anomalies() return symptoms Step 2：复现问题 class IssueReproducer: """问题复现器""" async def reproduce( self, session_id: str, user_input: str ) -> dict: """复现问题""" reproduction = { "original_session": session_id, "input": user_input, "attempts": [], "reproduced": False, "reproduction_rate": 0, } # 尝试复现3次 for i in range(3): try: result = await self.agent.process(user_input) reproduction["attempts"].append({ "attempt": i + 1, "response": result["response"], "quality_score": result.get("quality_score"), "full_trace": result.get("trace"), }) # 判断是否复现 if self._is_similar_issue(result, user_input): reproduction["reproduced"] = True except Exception as e: reproduction["attempts"].append({ "attempt": i + 1, "error": str(e), }) reproduction["reproduction_rate"] = sum( 1 for a in reproduction["attempts"] if a.get("error") or self._is_similar_issue(a, user_input) ) / len(reproduction["attempts"]) return reproduction Step 3：分析日志 class LogAnalyzer: """日志分析器""" async def analyze_for_incident( self, incident: dict, time_window_minutes: int = 60 ) -> dict: """分析故障相关日志""" analysis = { "error_patterns": [], "timeline": [], "affected_components": set(), "suspected_root_cause": None, } # 获取时间窗口内的日志 start = incident["start_time"] - timedelta(minutes=10) end = incident["start_time"] + timedelta(minutes=time_window_minutes) logs = await self.log_store.query( time_range=(start, end), filters={ "level": ["ERROR", "CRITICAL"], "service": incident.get("affected_services", []) } ) # 分析错误模式 error_counts = {} for log in logs: error_key = f"{log['service']}:{log['message'][:50]}" error_counts[error_key] = error_counts.get(error_key, 0) + 1 analysis["affected_components"].add(log["service"]) analysis["timeline"].append({ "timestamp": log["timestamp"], "service": log["service"], "message": log["message"], }) # 排序错误频率 analysis["error_patterns"] = sorted( [{"pattern": k, "count": v} for k, v in error_counts.items()], key=lambda x: x["count"], reverse=True ) # 推断根因 if analysis["error_patterns"]: top_error = analysis["error_patterns"][0] analysis["suspected_root_cause"] = { "type": "error_pattern", "pattern": top_error["pattern"], "frequency": top_error["count"], "confidence": min(top_error["count"] / len(logs), 1.0) } return analysis Step 4：链路追踪分析 class TraceAnalyzer: """Trace分析器""" async def find_slow_spans( self, trace_id: str, threshold_ms: int = 1000 ) -> list: """找到慢Span""" trace = await self.jaeger.get_trace(trace_id) spans = self._flatten_spans(trace) slow_spans = [ { "operation": span["operationName"], "service": span["process"]["serviceName"], "duration_ms": span["duration"], "tags": span.get("tags", {}), } for span in spans if span["duration"] > threshold_ms * 1000 # 转换为微秒 ] return sorted(slow_spans, key=lambda s: s["duration_ms"], reverse=True) async def find_error_spans(self, trace_id: str) -> list: """找到错误Span""" trace = await self.jaeger.get_trace(trace_id) spans = self._flatten_spans(trace) error_spans = [ { "operation": span["operationName"], "service": span["process"]["serviceName"], "error": span.get("tags", {}).get("error", True), "logs": span.get("logs", []), } for span in spans if span.get("tags", {}).get("error") ] return error_spans Step 5：根因分析 class RootCauseAnalyzer: """根因分析器""" async def analyze(self, incident: dict) -> dict: """分析根因""" hypotheses = [] # 假设1：最近的部署导致 recent_deploy = await self._check_recent_deployment(incident) if recent_deploy: hypotheses.append({ "hypothesis": "Recent deployment caused the issue", "evidence": recent_deploy, "confidence": 0.8, "verify_command": f"kubectl rollback deployment {recent_deploy['deployment']}" }) # 假设2：模型行为变化 model_change = await self._check_model_change(incident) if model_change: hypotheses.append({ "hypothesis": "Model behavior changed", "evidence": model_change, "confidence": 0.7, "verify_command": f"Compare outputs with previous model version" }) # 假设3：Prompt被修改 prompt_change = await self._check_prompt_change(incident) if prompt_change: hypotheses.append({ "hypothesis": "Prompt template was modified", "evidence": prompt_change, "confidence": 0.9, "verify_command": f"git diff {prompt_change['commit']} prompts/" }) # 假设4：工具API变更 tool_change = await self._check_tool_api_change(incident) if tool_change: hypotheses.append({ "hypothesis": "Tool API behavior changed", "evidence": tool_change, "confidence": 0.6, "verify_command": f"Test tool {tool_change['tool']} with previous inputs" }) # 按置信度排序 hypotheses.sort(key=lambda h: h["confidence"], reverse=True) return { "incident_id": incident["id"], "hypotheses": hypotheses, "recommended_first_check": hypotheses[0] if hypotheses else None, } 常见故障排查 故障1：响应质量下降 class QualityDropTroubleshooter: """回复质量下降排查""" CHECKLIST = [ { "name": "Check model version", "command": "kubectl get configmap agent-config -o jsonpath='{.data.model_version}'", "fix": "Rollback model version if recently changed" }, { "name": "Check prompt template", "command": "git log --oneline -10 prompts/", "fix": "Revert prompt changes if quality dropped after commit" }, { "name": "Check tool success rate", "command": "rate(agent_tool_calls_total{status='success'}[1h])", "fix": "Debug failing tools, may affect response quality" }, { "name": "Check memory retrieval quality", "command": "Analyze recall@5 for recent queries", "fix": "Reindex vector database if recall dropped" }, { "name": "Check for model drift", "command": "Compare quality scores across time", "fix": "Consider model retraining or fine-tuning" }, ] 故障2：Token消耗突增 class TokenSpikeTroubleshooter: """Token消耗突增排查""" async def diagnose(self) -> dict: diagnosis = { "possible_causes": [], "recommendations": [], } # 检查是否有循环 cycles = await self._check_for_cycles() if cycles: diagnosis["possible_causes"].append("Agent entering tool call loops") diagnosis["recommendations"].append("Enable cycle detection and break logic") # 检查Prompt是否变长 prompt_length = await self._get_avg_prompt_length() if prompt_length > self.baseline_prompt_length * 1.5: diagnosis["possible_causes"].append("Prompt length increased significantly") diagnosis["recommendations"].append("Optimize prompt template, use summarization") # 检查是否启用了缓存 cache_hit_rate = await self._get_cache_hit_rate() if cache_hit_rate < 0.3: diagnosis["possible_causes"].append("Cache hit rate too low") diagnosis["recommendations"].append("Investigate cache configuration") return diagnosis 排查工具箱 class TroubleshootingToolkit: """排障工具箱""" TOOLS = { "log_query": { "description": "Query logs with Loki", "example": 'logcli query \'{service="agent"} |~ "ERROR"\' --since=1h', }, "trace_query": { "description": "Find traces with Jaeger", "example": 'jaeger query --service=agent --lookback=1h --minDuration=1s', }, "metrics_query": { "description": "Query metrics with PromQL", "example": 'rate(agent_requests_total[5m])', }, "config_check": { "description": "Check K8s config", "example": 'kubectl get configmap agent-config -o yaml', }, "rollback": { "description": "Rollback deployment", "example": 'kubectl rollout undo deployment agent-service', }, "compare_versions": { "description": "Compare model/prompt versions", "example": 'ab compare --model-a=gpt-4o --model-b=gpt-4o-mini --samples=100', } } 总结 Agent系统故障排查需要系统化的方法论——从收集症状开始，通过复现问题、分析日志、追踪链路，最终定位根因。关键在于将模糊的"回复不对"拆解成可度量的指标异常，再通过对指标的分析定位到具体的组件和行为。 ...

引言 Agent系统的运维复杂度远超传统Web应用——LLM推理服务的GPU故障、工具调用的外部依赖故障、Token消耗突增导致的成本爆炸，这些都需要自动化运维系统能够及时发现并自动处理。2026年，成熟的Agent系统已经实现了从"手动运维"到"自愈+自动扩缩容"的跨越，将人工干预频率降低了90%以上。 自动化运维架构 ┌──────────────────────────────────────────────────────────┐ │ Agent自动化运维体系 │ │ │ │ ┌────────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 健康检测 │ │ 告警 │ │ 自愈 │ │ │ │ Health │ │ Alerts │ │ Self- │ │ │ │ Check │ │ │ │ healing │ │ │ └─────┬──────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────┐ │ │ │ 决策引擎 (Decision Engine) │ │ │ │ - 规则引擎 │ │ │ │ - ML预测模型 │ │ │ │ - 执行计划生成 │ │ │ └─────────────────┬───────────────────────────┘ │ │ │ │ │ ┌───────────┼───────────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │Auto- │ │Auto- │ │Auto- │ │ │ │Scaling │ │Remediation│ │Configuration│ │ │ │扩缩容 │ │修复 │ │配置 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ └──────────────────────────────────────────────────────────┘ 自愈机制 健康检查 class HealthChecker: """Agent系统健康检查""" CHECKS = { "llm_service": { "endpoint": "http://llm-service:8080/health", "timeout": 5, "expected_status": 200, "degraded_threshold_ms": 500, }, "tool_services": { "endpoint": "http://tool-service:8080/health", "timeout": 3, "expected_status": 200, }, "vector_db": { "endpoint": "http://qdrant:6333/health", "timeout": 5, }, "redis": { "command": "PING", "expected_response": "PONG", } } async def run_all_checks(self) -> dict: """运行所有健康检查""" results = {} for name, check in self.CHECKS.items(): try: result = await self._run_check(name, check) results[name] = result except Exception as e: results[name] = { "status": "unhealthy", "error": str(e), "timestamp": datetime.now().isoformat() } # 计算整体健康分 healthy_count = sum( 1 for r in results.values() if r["status"] == "healthy" ) health_score = healthy_count / len(results) return { "overall_health": health_score, "status": "healthy" if health_score > 0.8 else "degraded", "checks": results, "timestamp": datetime.now().isoformat() } async def _run_check(self, name: str, check: dict) -> dict: """运行单项检查""" start = time.monotonic() if "endpoint" in check: async with httpx.AsyncClient(timeout=check["timeout"]) as client: response = await client.get(check["endpoint"]) latency_ms = (time.monotonic() - start) * 1000 if response.status_code != check["expected_status"]: return { "status": "unhealthy", "actual_status": response.status_code, "expected_status": check["expected_status"] } status = "healthy" if latency_ms > check.get("degraded_threshold_ms", 1000): status = "degraded" return { "status": status, "latency_ms": latency_ms, "status_code": response.status_code } 自愈动作 class SelfHealingActions: """自愈动作库""" async def restart_service(self, service_name: str) -> dict: """重启服务""" logger.warning(f"Self-healing: restarting {service_name}") try: # 通过K8s API重启 await self.k8s_client.restart_deployment(service_name) # 等待就绪 await self._wait_for_ready(service_name, timeout=120) return { "action": "restart_service", "service": service_name, "success": True } except Exception as e: logger.error(f"Self-healing failed for {service_name}: {e}") return { "action": "restart_service", "service": service_name, "success": False, "error": str(e) } async def clear_cache(self, cache_type: str) -> dict: """清理缓存""" if cache_type == "redis": await self.redis_client.flushdb() elif cache_type == "vector": await self.vector_db.clear_cache() return {"action": "clear_cache", "type": cache_type, "success": True} async def scale_service( self, service_name: str, replicas: int ) -> dict: """扩缩容服务""" current = await self.k8s_client.get_deployment_replicas(service_name) if current == replicas: return {"action": "scale", "changed": False} await self.k8s_client.scale_deployment(service_name, replicas) return { "action": "scale", "service": service_name, "from": current, "to": replicas } async def failover_to_backup(self, service_name: str) -> dict: """故障转移到备用服务""" backup_config = self.backup_configs.get(service_name) if not backup_config: raise ValueError(f"No backup configured for {service_name}") # 切换流量到备用 await self.traffic_router.switch_to_backup( service_name, backup_config["endpoint"] ) return { "action": "failover", "service": service_name, "backup": backup_config["name"] } 自动扩缩容 预测性扩缩容 class PredictiveAutoScaler: """预测性自动扩缩容""" def __init__(self, k8s_client, metrics_client): self.k8s = k8s_client self.metrics = metrics_client self.prediction_model = None # 加载训练好的预测模型 async def run_scaling_loop(self): """扩缩容主循环""" while True: try: # 1. 收集当前指标 current_metrics = await self._collect_metrics() # 2. 预测未来负载 predicted_load = await self._predict_load( current_metrics, horizon_minutes=15 ) # 3. 计算目标副本数 target_replicas = self._calculate_target_replicas( predicted_load ) # 4. 执行扩缩容 await self._apply_scaling(target_replicas) # 5. 等待下一次评估 await asyncio.sleep(60) # 每分钟评估一次 except Exception as e: logger.error(f"Auto-scaling error: {e}") await asyncio.sleep(60) async def _predict_load(self, current: dict, horizon_minutes: int) -> dict: """预测未来负载""" # 使用历史数据 + 当前趋势预测 if self.prediction_model: features = self._extract_prediction_features(current) prediction = await self.prediction_model.predict(features) return prediction else: # 简单线性回归预测 trend = self._calculate_trend(current) predicted_qps = current["qps"] * (1 + trend * horizon_minutes / 60) return {"predicted_qps": max(0, predicted_qps)} def _calculate_target_replicas(self, predicted_load: dict) -> dict: """计算目标副本数""" targets = {} for service, config in self.service_configs.items(): predicted_qps = predicted_load.get("predicted_qps", 0) # 每个副本能处理的QPS qps_per_replica = config.get("qps_per_replica", 10) # 目标副本数（加20%安全余量） target = int(predicted_qps / qps_per_replica * 1.2) # 应用上下限 target = max(config["min_replicas"], target) target = min(config["max_replicas"], target) targets[service] = target return targets 扩缩容策略 class ScalingStrategy: """扩缩容策略""" STRATEGIES = { "conservative": { "scale_up_step": 1, # 每次只加1个副本 "scale_down_step": 1, # 每次只减1个副本 "scale_up_cooldown": 120, # 扩容冷却2分钟 "scale_down_cooldown": 300, # 缩容冷却5分钟 }, "aggressive": { "scale_up_step": 5, "scale_down_step": 2, "scale_up_cooldown": 30, "scale_down_cooldown": 180, }, "predictive": { "lookahead_minutes": 15, "safety_margin": 0.3, # 30%安全余量 } } async def execute_scaling( self, service: str, current: int, target: int, strategy: str = "conservative" ): """执行扩缩容""" config = self.STRATEGIES[strategy] if target > current: # 扩容 step = config["scale_up_step"] new_replicas = min(current + step, target) logger.info( f"Scaling up {service}: {current} -> {new_replicas}" ) await self.k8s.scale_deployment(service, new_replicas) elif target < current: # 缩容——更保守 step = config["scale_down_step"] new_replicas = max(current - step, target) # 检查是否有正在处理的请求 active_requests = await self._get_active_requests(service) if active_requests > 0: logger.info( f"Postponing scale down for {service}: " f"{active_requests} active requests" ) return logger.info( f"Scaling down {service}: {current} -> {new_replicas}" ) await self.k8s.scale_deployment(service, new_replicas) 故障预测 class FailurePredictor: """故障预测器""" async def predict_failures(self) -> list: """预测可能发生的故障""" predictions = [] # 1. 基于指标的预测 metrics_anomalies = await self._detect_metric_anomalies() for anomaly in metrics_anomalies: predictions.append({ "type": "metric_anomaly", "service": anomaly["service"], "probability": anomaly["probability"], "description": anomaly["description"], "recommended_action": anomaly["action"] }) # 2. 基于日志的预测 log_patterns = await self._analyze_log_patterns() for pattern in log_patterns: if pattern["risk_score"] > 0.7: predictions.append({ "type": "log_pattern", "pattern": pattern["pattern"], "probability": pattern["risk_score"], "description": f"Detected pattern: {pattern['description']}" }) # 3. 基于依赖健康的预测 dependency_health = await self._check_dependency_health() for dep in dependency_health: if dep["health_score"] < 0.5: predictions.append({ "type": "dependency", "dependency": dep["name"], "probability": 1 - dep["health_score"], "description": f"Dependency {dep['name']} is unhealthy" }) return sorted(predictions, key=lambda p: p["probability"], reverse=True) AIOps实践 class AIOpsEngine: """AIOps引擎""" async def analyze_incident(self, incident: dict) -> dict: """AI辅助事故分析""" # 1. 收集相关日志/指标/Trace context = await self._gather_incident_context(incident) # 2. 使用LLM分析根因 analysis = await self.llm.analyze( prompt=f""" Analyze this incident and identify the root cause: Incident: {incident} Context: {context} Provide: 1. Most likely root cause 2. Contributing factors 3. Recommended actions 4. Prevention measures """ ) # 3. 从历史事故中找相似案例 similar_incidents = await self._find_similar_incidents(incident) return { "incident_id": incident["id"], "root_cause_analysis": analysis["root_cause"], "recommended_actions": analysis["actions"], "similar_incidents": similar_incidents, "confidence": analysis["confidence"] } async def generate_runbook(self, incident_type: str) -> str: """自动生成Runbook""" return await self.llm.generate( prompt=f""" Generate a detailed runbook for handling {incident_type} incidents. Include: 1. Detection steps 2. Diagnosis procedures 3. Resolution steps 4. Verification steps 5. Post-mortem template """ ) 总结 Agent自动化运维的核心目标是"让系统自己管理自己"。自愈机制通过健康检查+修复动作的组合，能够自动处理80%以上的常见故障。预测性扩缩容通过提前预判负载变化，避免了响应式扩缩容的滞后性。AIOps则通过AI辅助事故分析和Runbook生成，显著提升了排障效率。 ...

引言 Agent系统的运营成本主要由三部分构成：LLM Token费用（60-70%）、基础设施成本（20-25%）和工具/API调用费用（5-15%）。随着用户规模增长，如果不做系统性的成本优化，每月的运营成本可能达到数十万甚至数百万美元。 2026年，Agent成本优化（Agent FinOps）已成为独立的技术领域。本文将从实际工程角度，系统讲解如何在不损害用户体验的前提下，将Agent系统的运营成本降低50-70%。 成本构成分析 class CostBreakdown: """Agent系统成本分解""" TYPICAL_DISTRIBUTION = { "llm_inference": { "proportion": 0.65, "sub_items": { "input_tokens": 0.35, "output_tokens": 0.55, "embedding": 0.10, } }, "infrastructure": { "proportion": 0.22, "sub_items": { "gpu_compute": 0.55, "cpu_compute": 0.20, "storage": 0.15, "network": 0.10, } }, "external_apis": { "proportion": 0.10, "sub_items": { "search_api": 0.40, "tool_apis": 0.35, "data_sources": 0.25, } }, "observability": { "proportion": 0.03, "sub_items": { "logging": 0.40, "tracing": 0.30, "monitoring": 0.30, } } } Token优化 Prompt压缩 class PromptOptimizer: """Prompt压缩与优化""" async def optimize_prompt(self, messages: list) -> list: """优化对话历史，减少Token消耗""" # 策略1：摘要旧消息 if len(messages) > 10: old_messages = messages[:-4] recent_messages = messages[-4:] summary = await self._summarize(old_messages) messages = [ {"role": "system", "content": f"Previous context: {summary}"} ] + recent_messages # 策略2：移除冗余系统提示 messages = self._deduplicate_system_messages(messages) # 策略3：压缩工具描述 messages = self._compress_tool_definitions(messages) return messages async def _summarize(self, messages: list) -> str: """使用小模型生成摘要""" summary_prompt = "Summarize the following conversation in 200 words:" for msg in messages: summary_prompt += f"\n{msg['role']}: {msg['content'][:200]}" response = await self.llm.generate( model="gpt-4o-mini", # 用小模型做摘要 prompt=summary_prompt, max_tokens=300 ) return response def _compress_tool_definitions(self, messages: list) -> list: """压缩工具定义""" for msg in messages: if msg["role"] == "system" and "tools" in msg.get("content", ""): # 只保留当前轮次需要的工具 needed_tools = self._get_relevant_tools(msg["content"]) msg["content"] = self._rewrite_tool_section(needed_tools) return messages 响应长度控制 class ResponseLengthController: """响应长度控制""" LENGTH_BUDGETS = { "simple_qa": {"max_tokens": 200, "avg_tokens": 100}, "explanation": {"max_tokens": 500, "avg_tokens": 300}, "code_generation": {"max_tokens": 2000, "avg_tokens": 800}, "analysis": {"max_tokens": 1500, "avg_tokens": 800}, "creative": {"max_tokens": 1000, "avg_tokens": 500}, } def get_token_budget(self, request: dict, route: dict) -> int: """获取Token预算""" task_type = route.get("task_type", "explanation") budget = self.LENGTH_BUDGETS.get(task_type, self.LENGTH_BUDGETS["explanation"]) # 根据用户tier调整 user_tier = request.get("user_tier", "free") if user_tier == "free": budget["max_tokens"] = int(budget["max_tokens"] * 0.7) return budget["max_tokens"] 缓存策略 多级缓存 class AgentCacheStack: """Agent多级缓存""" def __init__(self): self.layers = [ self._exact_match_cache, # L1: 精确匹配 self._semantic_cache, # L2: 语义缓存 self._tool_result_cache, # L3: 工具结果缓存 self._embedding_cache, # L4: 嵌入缓存 ] async def get_or_compute(self, request: dict) -> dict: """多级缓存查询""" # L1: 精确匹配 cache_key = self._generate_key(request) cached = await self._exact_match_cache.get(cache_key) if cached: self._record_cache_hit("L1_exact") return cached # L2: 语义相似查询 similar = await self._semantic_cache.find_similar( query=request["input"], threshold=0.95 ) if similar: self._record_cache_hit("L2_semantic") return similar # L3: 检查工具结果缓存 if request.get("tool_calls"): for tool_call in request["tool_calls"]: tool_result = await self._tool_result_cache.get( tool_call["name"], tool_call["params"] ) if tool_result: tool_call["cached_result"] = tool_result # 缓存未命中，执行计算 result = await self._execute(request) # 回填缓存 await self._fill_caches(request, result) return result async def _fill_caches(self, request: dict, result: dict): """回填各级缓存""" # L1 await self._exact_match_cache.set( self._generate_key(request), result, ttl=3600 # 1小时 ) # L2 await self._semantic_cache.store( query=request["input"], response=result, embedding=await self._embed(request["input"]), ttl=86400 # 24小时 ) # L3 if result.get("tool_results"): for tool_result in result["tool_results"]: await self._tool_result_cache.set( tool_result["tool_name"], tool_result["params"], tool_result["output"], ttl=1800 # 30分钟 ) 缓存命中率监控 class CacheMetrics: """缓存指标监控""" async def get_cache_report(self) -> dict: return { "l1_exact": { "hit_rate": await self._get_rate("cache_l1_hit", "cache_l1_miss"), "size": await self._get_size("exact_cache"), "eviction_rate": await self._get_rate("cache_l1_evict"), }, "l2_semantic": { "hit_rate": await self._get_rate("cache_l2_hit", "cache_l2_miss"), "size": await self._get_size("semantic_cache"), "avg_similarity": await self._get_avg("cache_l2_similarity"), }, "l3_tool": { "hit_rate": await self._get_rate("cache_l3_hit", "cache_l3_miss"), "savings_usd": await self._get_savings("tool_cache"), }, "total_savings": { "tokens_saved": await self._get_total("tokens_saved"), "cost_saved_usd": await self._get_total("cost_saved"), "latency_saved_ms": await self._get_total("latency_saved"), } } 模型选择优化 class ModelCostOptimizer: """模型成本优化器""" MODEL_COSTS = { "gpt-4o": {"input": 2.50, "output": 10.00}, # per 1M tokens "gpt-4o-mini": {"input": 0.15, "output": 0.60}, "claude-3.5-sonnet": {"input": 3.00, "output": 15.00}, "claude-3-haiku": {"input": 0.25, "output": 1.25}, "deepseek-v3": {"input": 0.14, "output": 0.28}, "local-llama-70b": {"input": 0.05, "output": 0.05}, # 自部署 } async def select_cost_optimal_model( self, request: dict, quality_threshold: float = 0.8 ) -> str: """选择成本最优模型""" # 估算各模型成本和质量 estimates = [] for model, cost in self.MODEL_COSTS.items(): estimated_tokens = self._estimate_tokens(request) total_cost = ( estimated_tokens["input"] * cost["input"] + estimated_tokens["output"] * cost["output"] ) / 1_000_000 quality = await self._estimate_quality(model, request) if quality >= quality_threshold: estimates.append({ "model": model, "cost": total_cost, "quality": quality, "value_ratio": quality / total_cost if total_cost > 0 else float('inf') }) # 选择性价比最高的 return max(estimates, key=lambda e: e["value_ratio"])["model"] def _estimate_tokens(self, request: dict) -> dict: """估算Token消耗""" input_tokens = len(request["input"]) // 4 # 粗略估算 output_tokens = min(input_tokens, 500) # 响应通常比输入短 return {"input": input_tokens, "output": output_tokens} 批处理优化 class BatchProcessor: """请求批处理——合并多个请求降低单次成本""" async def batch_llm_calls( self, requests: list, max_batch_size: int = 10, max_wait_ms: int = 100 ) -> list: """批处理LLM调用""" batch = [] results = {} for request in requests: batch.append(request) if len(batch) >= max_batch_size: batch_results = await self._process_batch(batch) results.update(batch_results) batch = [] # 处理剩余 if batch: batch_results = await self._process_batch(batch) results.update(batch_results) return [results[r["id"]] for r in requests] async def _process_batch(self, batch: list) -> dict: """处理批次""" # 合并多个请求为一个prompt combined_prompt = self._combine_prompts(batch) response = await self.llm.generate( model="gpt-4o-mini", prompt=combined_prompt, max_tokens=2000 ) # 解析并分配结果 return self._parse_batch_response(response, batch) 基础设施优化 class InfrastructureOptimizer: """基础设施成本优化""" async def optimize_gpu_utilization(self) -> dict: """优化GPU利用率""" current_util = await self._get_avg_gpu_utilization() recommendations = [] if current_util < 60: recommendations.append({ "action": "reduce_gpu_instances", "current": self.gpu_count, "recommended": max(1, int(self.gpu_count * 0.7)), "estimated_savings": (self.gpu_count - max(1, int(self.gpu_count * 0.7))) * 2000 # $2000/GPU/month }) # 建议使用Spot实例 recommendations.append({ "action": "use_spot_instances", "estimated_savings": self.gpu_count * 800, # 节省40% "risk": "可能被中断，需要检查点机制" }) # 建议模型量化 recommendations.append({ "action": "enable_model_quantization", "estimated_savings": self.gpu_count * 500, "quality_impact": "轻微下降（<2%）" }) return recommendations async def optimize_storage(self) -> dict: """优化存储成本""" # 向量数据库分片优化 vector_db_size = await self._get_vector_db_size() recommendations = [] # 冷热数据分离 recommendations.append({ "action": "tiered_storage", "hot_tier_gb": vector_db_size * 0.2, # 20%热数据 "cold_tier_gb": vector_db_size * 0.8, "estimated_savings": vector_db_size * 0.8 * 0.08 # 冷存储便宜90% }) return recommendations 成本看板 class CostDashboard: """成本看板""" async def generate_report(self, period: str = "monthly") -> dict: return { "period": period, "total_cost": await self._get_total_cost(period), "cost_by_category": { "llm": await self._get_category_cost("llm", period), "infrastructure": await self._get_category_cost("infra", period), "external_apis": await self._get_category_cost("apis", period), }, "cost_per_request": await self._get_cost_per_request(period), "cost_per_tenant": await self._get_cost_per_tenant(period), "optimization_savings": { "cache_savings": await self._get_cache_savings(period), "model_downgrade_savings": await self._get_downgrade_savings(period), "batch_savings": await self._get_batch_savings(period), }, "trend": await self._get_trend(period), "projected_next_month": await self._project_cost(), } 总结 Agent成本优化是一个系统工程，需要从Token、模型、缓存、批处理、基础设施多个层面协同发力。其中缓存策略是最立竿见影的优化手段——一个设计良好的多级缓存系统可以将LLM调用减少30-50%。模型选择优化通过让简单请求使用小模型，可以在不影响体验的前提下节省40-60%的Token成本。 ...

引言 一次Agent对话可能涉及路由决策、向量检索、工具调用、LLM推理等十几个步骤，跨越多个微服务。当用户反馈"Agent回复变慢了"时，如果没有链路追踪，定位瓶颈就像大海捞针。OpenTelemetry + Jaeger的组合为Agent系统提供了端到端的请求追踪能力，让每一次对话的完整路径都清晰可见。 OpenTelemetry架构 ┌──────────────────────────────────────────────────────────┐ │ OpenTelemetry Architecture │ │ │ │ Agent Services │ │ ┌─────────────────────────────────────────────┐ │ │ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │ │ │Router│ │ Tool │ │ LLM │ │Memory│ │ │ │ │ │Service│ │Service│ │Service│ │Service│ │ │ │ │ └──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘ │ │ │ │ │ │ │ │ │ │ │ │ ┌──▼─────────▼─────────▼─────────▼───┐ │ │ │ │ │ OpenTelemetry SDK (Instrument) │ │ │ │ │ └──────────────┬─────────────────────┘ │ │ │ └─────────────────┼───────────────────────┘ │ │ │ │ │ ┌────────▼────────┐ │ │ │ OTLP Exporter │ │ │ └────────┬────────┘ │ │ │ │ │ ┌────────▼────────┐ │ │ │ OTel Collector │ │ │ │ (接收+处理+导出) │ │ │ └────┬───────┬────┘ │ │ │ │ │ │ ┌──────▼┐ ┌──▼──────┐ │ │ │Jaeger │ │Prometheus│ │ │ │(Traces)│ │(Metrics) │ │ │ └───────┘ └─────────┘ │ └──────────────────────────────────────────────────────────┘ SDK初始化 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.instrumentation.grpc import GrpcInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor def setup_tracing(service_name: str, otlp_endpoint: str = "otel-collector:4317"): """初始化OpenTelemetry追踪""" resource = Resource.create({ "service.name": service_name, "service.version": "2.0.0", "deployment.environment": "production", }) provider = TracerProvider(resource=resource) # OTLP导出器 exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True) processor = BatchSpanProcessor( exporter, max_queue_size=2048, max_export_batch_size=512, export_timeout_millis=30000, ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 自动注入HTTP/gRPC调用 GrpcInstrumentor().instrument() HTTPXClientInstrumentor().instrument() return trace.get_tracer(service_name) Agent Span设计 class AgentTracer: """Agent专用追踪器""" def __init__(self, tracer): self.tracer = tracer async def trace_request( self, session_id: str, user_input: str, handler: callable ): """追踪完整请求链路""" with self.tracer.start_as_current_span( "agent.request", attributes={ "session.id": session_id, "input.length": len(user_input), "input.language": self._detect_language(user_input), } ) as request_span: try: result = await handler() request_span.set_attribute( "response.length", len(result.get("response", "")) ) request_span.set_attribute( "response.quality_score", result.get("quality_score", 0) ) request_span.set_status(trace.StatusCode.OK) return result except Exception as e: request_span.record_exception(e) request_span.set_status( trace.Status(trace.StatusCode.ERROR, str(e)) ) raise async def trace_routing( self, user_input: str, routing_fn: callable ): """追踪路由决策""" with self.tracer.start_as_current_span( "agent.route", attributes={"input.preview": user_input[:100]} ) as span: decision = await routing_fn() span.set_attributes({ "route.model": decision.get("model", ""), "route.tools": ",".join(decision.get("tools", [])), "route.confidence": decision.get("confidence", 0), "route.reason": decision.get("reason", ""), }) return decision async def trace_tool_call( self, tool_name: str, params: dict, executor: callable ): """追踪工具调用""" with self.tracer.start_as_current_span( f"tool.{tool_name}", attributes={ "tool.name": tool_name, "tool.params_hash": hashlib.md5( json.dumps(params, sort_keys=True).encode() ).hexdigest()[:8], } ) as span: start = time.monotonic() try: result = await executor() latency_ms = (time.monotonic() - start) * 1000 span.set_attributes({ "tool.latency_ms": latency_ms, "tool.success": True, "tool.result_size": len(str(result)), }) return result except Exception as e: span.set_attributes({ "tool.success": False, "tool.error": str(e)[:200], }) span.record_exception(e) raise async def trace_llm_call( self, model: str, prompt: str, generator: callable ): """追踪LLM推理""" with self.tracer.start_as_current_span( f"llm.{model}", attributes={ "llm.model": model, "llm.prompt_length": len(prompt), } ) as span: start = time.monotonic() result = await generator() latency_ms = (time.monotonic() - start) * 1000 span.set_attributes({ "llm.latency_ms": latency_ms, "llm.prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0), "llm.completion_tokens": result.get("usage", {}).get("completion_tokens", 0), "llm.total_tokens": result.get("usage", {}).get("total_tokens", 0), "llm.finish_reason": result.get("finish_reason", ""), }) return result Span层级示例 agent.request (session=abc123) [2000ms] ├── agent.route [15ms] │ ├── embedding.generate [8ms] │ └── similarity.match [5ms] ├── memory.retrieve [50ms] │ ├── vector.search [30ms] │ └── context.assemble [15ms] ├── tool.search [800ms] │ ├── http.get [600ms] │ └── result.parse [50ms] ├── tool.calculator [20ms] ├── llm.gpt-4o [1100ms] │ ├── prompt.build [5ms] │ ├── api.call [1050ms] │ └── response.parse [30ms] └── response.format [15ms] 上下文传播 from opentelemetry.propagate import inject, extract from opentelemetry.trace import get_current_span class TraceContextPropagator: """跨服务Trace上下文传播""" @staticmethod def inject_to_headers(headers: dict = None) -> dict: """注入trace context到HTTP头""" headers = headers or {} inject(headers) return headers @staticmethod def extract_from_headers(headers: dict): """从HTTP头提取trace context""" return extract(headers) @staticmethod def get_current_trace_id() -> str: """获取当前trace ID""" span = get_current_span() if span and span.is_recording(): return format(span.get_span_context().trace_id, "032x") return "" @staticmethod def get_current_span_id() -> str: """获取当前span ID""" span = get_current_span() if span and span.is_recording(): return format(span.get_span_context().span_id, "016x") return "" # 在gRPC metadata中传播 class GrpcTraceInterceptor: """gRPC trace拦截器""" async def intercept(self, method, request, context): # 从metadata提取trace context metadata = dict(context.invocation_metadata()) trace_context = TraceContextPropagator.extract_from_headers(metadata) tracer = trace.get_tracer(__name__) with tracer.start_as_current_span( f"grpc.{method}", context=trace_context, ) as span: # 注入trace context到响应metadata response_metadata = TraceContextPropagator.inject_to_headers() context.set_trailing_metadata( [(k, v) for k, v in response_metadata.items()] ) return await method(request, context) Jaeger查询与分析 class JaegerAnalyzer: """Jaeger数据分析器""" def __init__(self, jaeger_url: str): self.url = jaeger_url async def find_slow_traces( self, service: str, min_duration_ms: int = 2000, limit: int = 20 ) -> list: """查找慢trace""" async with httpx.AsyncClient() as client: response = await client.get( f"{self.url}/api/traces", params={ "service": service, "limit": limit, "minDuration": f"{min_duration_ms}ms", "lookback": "1h", } ) return response.json()["data"] async def analyze_bottleneck( self, trace_id: str ) -> dict: """分析trace瓶颈""" trace = await self._get_trace(trace_id) spans = self._flatten_spans(trace) # 找到最耗时的span slowest = max(spans, key=lambda s: s["duration"]) # 分析Span层级 tree = self._build_span_tree(spans) # 找到关键路径 critical_path = self._find_critical_path(tree) return { "trace_id": trace_id, "total_duration_ms": tree["duration"], "slowest_span": { "name": slowest["operationName"], "duration_ms": slowest["duration"], "service": slowest["process"]["serviceName"], }, "critical_path": [ { "span": s["operationName"], "duration_ms": s["duration"], "service": s["process"]["serviceName"], } for s in critical_path ], "span_count": len(spans), } 性能影响控制 class TracingPerformanceGuard: """追踪性能守护——控制追踪开销""" def __init__(self): self.sampling_rates = { "fast_path": 0.05, # <500ms的请求5%采样 "normal": 0.2, # 500ms-2s的请求20%采样 "slow": 1.0, # >2s的请求100%采样 "error": 1.0, # 错误请求100%采样 } def should_trace(self, estimated_duration_ms: int = 0) -> bool: """决定是否追踪""" if estimated_duration_ms > 2000: rate = self.sampling_rates["slow"] elif estimated_duration_ms > 500: rate = self.sampling_rates["normal"] else: rate = self.sampling_rates["fast_path"] return random.random() < rate @contextmanager def conditional_span(self, tracer, name: str, should_trace: bool): """条件性创建span""" if should_trace: with tracer.start_as_current_span(name) as span: yield span else: yield None 总结 OpenTelemetry为Agent系统提供了统一的、标准化的链路追踪能力。精心设计的Span层级让每一次Agent对话的完整路径都清晰可见——从路由决策到工具调用，从记忆检索到LLM推理。Jaeger的可视化让性能瓶颈一目了然，基于trace的分析能够将排障效率提升数倍。 ...

引言 Agent系统的日志不仅是排障工具，更是质量改进和安全审计的数据基础。一次Agent对话可能涉及路由决策、记忆检索、工具调用、LLM推理等多个步骤，跨越多个微服务。如何在分布式环境中建立完整的日志链路，是Agent系统可观测性的核心挑战。 日志架构全景 ┌──────────────────────────────────────────────────────┐ │ 日志数据流 │ │ │ │ Agent Services │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │Router│ │Tool │ │LLM │ │Memory│ │ │ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ │ │ ┌──────────────────────────────┐ │ │ │ Fluent Bit (采集) │ │ │ └──────────┬───────────────────┘ │ │ │ │ │ ┌───────┼───────┐ │ │ ▼ ▼ │ │ ┌──────┐ ┌──────────┐ │ │ │ Loki │ │Elasticsearch│ │ │ │(日志) │ │ (全文搜索) │ │ │ └──────┘ └──────────┘ │ │ │ │ │ │ └───────┬───────┘ │ │ ▼ │ │ ┌──────────────┐ │ │ │ Grafana │ │ │ │ (可视化) │ │ │ └──────────────┘ │ └──────────────────────────────────────────────────────┘ 结构化日志标准 import structlog from datetime import datetime import uuid # 结构化日志配置 structlog.configure( processors=[ structlog.stdlib.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer() ], wrapper_class=structlog.stdlib.BoundLogger, logger_factory=structlog.stdlib.LoggerFactory(), ) logger = structlog.get_logger() class AgentLogger: """Agent专用日志器""" @staticmethod def log_request( session_id: str, request_id: str, user_input: str, route_decision: dict ): """记录请求日志""" logger.info( "agent_request_received", session_id=session_id, request_id=request_id, user_input_length=len(user_input), user_input_preview=user_input[:100], route_model=route_decision.get("model"), route_tools=route_decision.get("tools"), timestamp=datetime.now().isoformat() ) @staticmethod def log_tool_call( session_id: str, request_id: str, tool_name: str, params: dict, result: dict, latency_ms: float, success: bool ): """记录工具调用日志""" logger.info( "tool_call_completed", session_id=session_id, request_id=request_id, tool_name=tool_name, params_hash=hash(str(sorted(params.items()))), result_size=len(str(result)), latency_ms=latency_ms, success=success, error=result.get("error") if not success else None ) @staticmethod def log_llm_call( session_id: str, request_id: str, model: str, prompt_tokens: int, completion_tokens: int, latency_ms: float, quality_score: float = None ): """记录LLM调用日志""" logger.info( "llm_call_completed", session_id=session_id, request_id=request_id, model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, total_tokens=prompt_tokens + completion_tokens, latency_ms=latency_ms, quality_score=quality_score ) @staticmethod def log_agent_decision( session_id: str, decision_type: str, reasoning: str, action: str, confidence: float ): """记录Agent决策日志——用于审计和改进""" logger.info( "agent_decision", session_id=session_id, decision_type=decision_type, # route, tool_select, terminate reasoning=reasoning[:500], # 截断推理过程 action=action, confidence=confidence ) Trace ID传播 from opentelemetry import trace from opentelemetry.propagate import inject, extract class TracingMiddleware: """分布式追踪中间件""" async def __call__(self, request, call_next): # 提取或生成trace context context = extract(request.headers) tracer = trace.get_tracer(__name__) with tracer.start_as_current_span( "agent_request", context=context, attributes={ "session_id": request.session_id, "user_id": request.user_id, } ) as span: # 在请求上下文中注入trace信息 headers = {} inject(headers) request.trace_context = headers try: response = await call_next(request) span.set_attribute("response.status", "success") span.set_attribute( "response.latency_ms", response.latency_ms ) return response except Exception as e: span.record_exception(e) span.set_status(trace.Status(trace.StatusCode.ERROR)) raise # 在服务间调用时传播Trace ID class ServiceClient: """带Trace传播的服务客户端""" async def call_service( self, service: str, method: str, data: dict, trace_context: dict = None ) -> dict: """调用其他微服务""" headers = { "Content-Type": "application/json", } # 注入trace context if trace_context: headers.update(trace_context) else: inject(headers) # 从当前context注入 # 记录出站调用 tracer = trace.get_tracer(__name__) with tracer.start_as_current_span( f"call_{service}", attributes={ "peer.service": service, "http.method": method, } ) as span: response = await self.http_client.post( f"http://{service}/{method}", json=data, headers=headers ) span.set_attribute("http.status_code", response.status_code) return response.json() 日志关联查询 class LogCorrelator: """日志关联查询器""" async def get_session_timeline( self, session_id: str ) -> list: """获取会话的完整事件时间线""" # 从多个数据源查询 logs = await self.loki.query( f'{{session_id="{session_id}"}} | json' ) traces = await self.jaeger.get_traces( tags={"session_id": session_id} ) metrics = await self.prometheus.query_range( query=f'agent_session_metrics{{session_id="{session_id}"}}', start=..., end=... ) # 合并并按时间排序 events = [] for log in logs: events.append({ "type": "log", "timestamp": log["timestamp"], "level": log["level"], "message": log["message"], **log }) for trace in traces: for span in trace.spans: events.append({ "type": "trace", "timestamp": span.start_time, "span_name": span.name, "duration_ms": span.duration_ms, "service": span.service, **span.tags }) events.sort(key=lambda e: e["timestamp"]) return events 日志采样策略 class LogSampler: """日志采样器——在保证可观测性的前提下控制日志量""" SAMPLING_RULES = { # 正常请求：10%采样 "normal": {"rate": 0.1, "level": "INFO"}, # 错误请求：100%记录 "error": {"rate": 1.0, "level": "ERROR"}, # 慢请求（>5s）：100%记录 "slow": {"rate": 1.0, "level": "INFO", "min_latency_ms": 5000}, # 工具调用失败：100%记录 "tool_failure": {"rate": 1.0, "level": "WARN"}, # 安全相关：100%记录 "security": {"rate": 1.0, "level": "INFO"}, # 高价值用户：50%采样 "enterprise": {"rate": 0.5, "level": "INFO"}, } def should_log( self, log_type: str, request: dict, response: dict = None ) -> tuple: """判断是否需要记录日志""" # 优先级判断 if response and response.get("error"): rule = self.SAMPLING_RULES["error"] elif response and response.get("latency_ms", 0) > 5000: rule = self.SAMPLING_RULES["slow"] elif request.get("user_tier") == "enterprise": rule = self.SAMPLING_RULES["enterprise"] else: rule = self.SAMPLING_RULES["normal"] import random if random.random() < rule["rate"]: return True, rule["level"] return False, None 日志分析 class LogAnalyzer: """日志分析器""" async def analyze_session(self, session_id: str) -> dict: """分析单个会话日志""" events = await self.log_store.get_session_events(session_id) analysis = { "session_id": session_id, "total_steps": len(events), "tool_calls": [], "llm_calls": [], "errors": [], "total_tokens": 0, "total_latency_ms": 0, "quality_indicators": {}, } for event in events: if event["type"] == "tool_call": analysis["tool_calls"].append({ "tool": event["tool_name"], "latency_ms": event["latency_ms"], "success": event["success"] }) if not event["success"]: analysis["errors"].append(event) elif event["type"] == "llm_call": analysis["llm_calls"].append({ "model": event["model"], "tokens": event["total_tokens"], "latency_ms": event["latency_ms"] }) analysis["total_tokens"] += event["total_tokens"] analysis["total_latency_ms"] += event.get("latency_ms", 0) return analysis async def detect_anomalies( self, time_window_hours: int = 1 ) -> list: """检测日志异常模式""" anomalies = [] # 1. 突发错误聚集 error_clusters = await self._find_error_clusters(time_window_hours) for cluster in error_clusters: anomalies.append({ "type": "error_cluster", "service": cluster["service"], "error_count": cluster["count"], "time_range": cluster["range"] }) # 2. 异常Token消耗 token_outliers = await self._find_token_outliers(time_window_hours) for outlier in token_outliers: anomalies.append({ "type": "token_anomaly", "session_id": outlier["session_id"], "tokens": outlier["tokens"], "expected": outlier["expected"] }) # 3. 工具调用模式异常 tool_anomalies = await self._find_tool_anomalies(time_window_hours) anomalies.extend(tool_anomalies) return anomalies 日志保留策略 日志类型 热存储（SSD） 温存储（HDD） 冷存储（S3） 错误日志 7天 30天 180天 安全审计 30天 180天 2年 正常请求 3天 14天 90天 Trace数据 3天 7天 30天 指标数据 7天 90天 365天 总结 Agent系统的日志架构需要在"详细度"和"成本"之间取得平衡。结构化日志是一切的基础——没有结构化，就无法进行有效的查询和分析。Trace ID的跨服务传播让分布式追踪成为可能。智能采样策略确保在控制成本的同时不丢失关键信息。 ...

引言 Agent系统的监控告警比传统应用复杂一个量级——除了需要监控CPU、内存、延迟等基础设施指标外，还需要监控Token消耗、工具成功率、幻觉率、安全违规率等Agent特有指标。一个没有完善监控的Agent系统就像盲飞——出了问题不知道哪里出了问题，没出问题不知道什么时候会出问题。 2026年，Prometheus + Grafana + AlertManager已成为Agent监控告警的事实标准，但Agent系统需要在此基础上构建专门的监控体系。 指标体系设计 四层指标架构 ┌─────────────────────────────────────────────────────┐ │ Layer 4: Business Metrics │ │ 用户满意度、任务完成率、对话质量评分 │ ├─────────────────────────────────────────────────────┤ │ Layer 3: Agent Metrics │ │ Token消耗、工具调用成功率、幻觉率、安全违规率 │ ├─────────────────────────────────────────────────────┤ │ Layer 2: Application Metrics │ │ 请求QPS、响应延迟、错误率、并发会话数 │ ├─────────────────────────────────────────────────────┤ │ Layer 1: Infrastructure Metrics │ │ CPU、内存、GPU利用率、磁盘IO、网络流量 │ └─────────────────────────────────────────────────────┘ Agent核心指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # ===== Layer 3: Agent特有指标 ===== # Token消耗 token_usage = Counter( "agent_token_total", "Total tokens consumed", ["tenant_id", "model", "type"] # type: input/output ) # 工具调用 tool_calls = Counter( "agent_tool_calls_total", "Total tool calls", ["tool_name", "status"] # status: success/failed/timeout ) tool_latency = Histogram( "agent_tool_latency_seconds", "Tool execution latency", ["tool_name"], buckets=[0.1, 0.5, 1, 5, 10, 30, 60] ) # Agent质量 hallucination_rate = Gauge( "agent_hallucination_rate", "Hallucination rate (rolling 1h)", ["model"] ) safety_violations = Counter( "agent_safety_violations_total", "Safety violations detected", ["type", "severity"] ) # 循环检测 cycle_detections = Counter( "agent_cycle_detections_total", "Cycle detections", ["cycle_type", "resolution"] # resolution: broken/escalated ) # 会话指标 active_sessions = Gauge( "agent_active_sessions", "Active sessions", ["tenant_id"] ) session_duration = Histogram( "agent_session_duration_seconds", "Session duration", buckets=[10, 30, 60, 120, 300, 600, 1200] ) # 路由决策 routing_decisions = Counter( "agent_routing_decisions_total", "Routing decisions", ["source_model", "target_model", "reason"] ) Prometheus配置 # prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s rule_files: - "agent_alerts.yml" scrape_configs: # Agent应用指标 - job_name: "agent-service" metrics_path: /metrics static_configs: - targets: ["agent-service:9090"] labels: service: "agent" # LLM推理服务 - job_name: "llm-inference" metrics_path: /metrics static_configs: - targets: ["llm-inference:9090"] # 工具执行服务 - job_name: "tool-executor" metrics_path: /metrics static_configs: - targets: ["tool-executor-0:9090", "tool-executor-1:9090"] 告警规则 # agent_alerts.yml groups: - name: agent_infra_alerts rules: # 高错误率 - alert: AgentHighErrorRate expr: | sum(rate(agent_requests_total{status="error"}[5m])) by (service) / sum(rate(agent_requests_total[5m])) by (service) > 0.05 for: 2m labels: severity: critical team: agent-platform annotations: summary: "Agent error rate > 5%" description: "{{ $labels.service }} error rate is {{ $value | humanizePercentage }}" # P99延迟过高 - alert: AgentHighLatency expr: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m]) ) > 5 for: 5m labels: severity: warning annotations: summary: "Agent P99 latency > 5s" - name: agent_quality_alerts rules: # 幻觉率过高 - alert: AgentHighHallucination expr: agent_hallucination_rate > 0.05 for: 10m labels: severity: warning team: agent-quality annotations: summary: "Hallucination rate > 5% for {{ $labels.model }}" # 安全违规 - alert: AgentSafetyViolation expr: increase(agent_safety_violations_total[1h]) > 0 labels: severity: critical team: security annotations: summary: "Safety violation detected" # 循环检测频繁 - alert: AgentFrequentCycles expr: | increase(agent_cycle_detections_total[1h]) > 10 for: 5m labels: severity: warning annotations: summary: "Frequent cycle detections (>10/hour)" - name: agent_cost_alerts rules: # Token消耗异常 - alert: AgentTokenSpike expr: | rate(agent_token_total[5m]) > 2 * avg_over_time(rate(agent_token_total[5m])[1h:5m]) for: 10m labels: severity: warning team: agent-platform annotations: summary: "Token consumption spiked 2x above average" # 工具调用失败率 - alert: ToolFailureRate expr: | sum(rate(agent_tool_calls_total{status="failed"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name) > 0.1 for: 5m labels: severity: warning annotations: summary: "Tool {{ $labels.tool_name }} failure rate > 10%" 告警路由与通知 class AlertRouter: """告警路由器""" ROUTING_RULES = { "critical": { "channels": ["pagerduty", "slack:#oncall", "sms"], "escalation_delay_min": 5, "escalation_target": "team-lead", }, "warning": { "channels": ["slack:#alerts"], "escalation_delay_min": 30, "escalation_target": "secondary-oncall", }, "info": { "channels": ["slack:#monitoring"], "escalation_delay_min": None, "escalation_target": None, } } async def handle_alert(self, alert: dict): """处理告警""" severity = alert["labels"]["severity"] rule = self.ROUTING_RULES[severity] # 告警去重 if await self._is_duplicate(alert): logger.debug(f"Duplicate alert suppressed: {alert['fingerprint']}") return # 告警分组 group_key = self._get_group_key(alert) group = await self._get_or_create_group(group_key) group.add_alert(alert) # 发送通知 for channel in rule["channels"]: await self._send_notification(channel, group) # 设置升级定时器 if rule["escalation_delay_min"]: asyncio.create_task( self._schedule_escalation( group, rule["escalation_delay_min"], rule["escalation_target"] ) ) 告警治理 class AlertGovernance: """告警治理——防止告警风暴""" def __init__(self): self.suppression_rules = [] self.rate_limits = {} def should_send(self, alert: dict) -> bool: """判断告警是否应该发送""" # 1. 维护窗口抑制 if self._in_maintenance_window(alert): return False # 2. 依赖抑制——如果上游告警活跃，抑制下游 if self._suppressed_by_dependency(alert): return False # 3. 频率限制——同一告警5分钟内只发一次 alert_key = alert["fingerprint"] if alert_key in self.rate_limits: last_sent = self.rate_limits[alert_key] if (datetime.now() - last_sent).total_seconds() < 300: return False # 4. 告警噪音评分 noise_score = self._calculate_noise_score(alert) if noise_score < 0.3: return False self.rate_limits[alert_key] = datetime.now() return True Grafana仪表板 { "dashboard": { "title": "Agent System Overview", "panels": [ { "title": "Request Rate & Error Rate", "targets": [ { "expr": "sum(rate(agent_requests_total[5m]))", "legendFormat": "QPS" }, { "expr": "sum(rate(agent_requests_total{status=\"error\"}[5m])) / sum(rate(agent_requests_total[5m]))", "legendFormat": "Error Rate" } ] }, { "title": "Token Consumption by Model", "targets": [ { "expr": "sum(rate(agent_token_total[5m])) by (model)", "legendFormat": "{{model}}" } ] }, { "title": "Tool Success Rate", "targets": [ { "expr": "sum(rate(agent_tool_calls_total{status=\"success\"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name)", "legendFormat": "{{tool_name}}" } ] }, { "title": "Active Sessions & Concurrency", "targets": [ { "expr": "agent_active_sessions", "legendFormat": "{{tenant_id}}" } ] } ] } } SLI/SLO定义 # Agent系统SLO定义 slo: availability: target: 99.9% window: 30d query: | 1 - (sum(rate(agent_requests_total{status="error"}[5m])) / sum(rate(agent_requests_total[5m]))) latency_p99: target: 2000ms window: 7d query: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m])) quality_score: target: 0.85 window: 7d query: | avg(agent_response_quality_score) safety: target: 99.99% window: 30d query: | 1 - (increase(agent_safety_violations_total[30d]) / sum(increase(agent_requests_total[30d]))) 总结 Agent系统的监控告警需要覆盖从基础设施到业务质量的四个层次。指标设计要全面但不过载，告警规则要精准且有层次，通知路由要高效且不产生噪音。告警治理是长期工作——定期回顾告警有效性，淘汰无用告警，优化有用告警。 ...

引言 路由是Agent系统的"大脑"——它决定了每个请求由哪个模型处理、调用哪些工具、使用什么记忆策略。一个优秀的路由架构可以在不增加硬件资源的前提下，将Agent系统的整体性能提升30-50%，同时显著降低运营成本。 2026年，Agent路由已从简单的规则匹配进化到基于语义理解和强化学习的智能路由，成为Agent系统核心竞争力之一。 路由架构演进 阶段1：固定路由 阶段2：规则路由 阶段3：语义路由 ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 所有请求 │ │ 规则匹配 │ │ 语义理解 │ │ ↓ │ │ ↓ │ │ ↓ │ │ 同一模型 │ │ 多模型分发 │ │ 意图分类 │ │ 同一工具 │ │ 工具选择 │ │ 智能选模型 │ └──────────┘ └──────────┘ └──────────┘ 阶段4：学习型路由 ┌──────────┐ │ 历史数据 │ │ ↓ │ │ RL优化 │ │ ↓ │ │ 动态路由 │ └──────────┘ 规则路由 class RuleBasedRouter: """基于规则的路由器""" def __init__(self): self.rules = [] def add_rule(self, condition: callable, target: dict, priority: int = 0): """添加路由规则""" self.rules.append({ "condition": condition, "target": target, "priority": priority }) self.rules.sort(key=lambda r: r["priority"], reverse=True) async def route(self, request: dict) -> dict: """路由请求""" for rule in self.rules: if rule["condition"](request): return rule["target"] # 默认路由 return {"model": "gpt-4o-mini", "tools": ["search"]} def setup_default_rules(self): """设置默认规则集""" # 代码相关 → 代码模型 self.add_rule( condition=lambda r: "code" in r["input"].lower(), target={"model": "deepseek-coder", "tools": ["code_exec"]}, priority=10 ) # 数学相关 → 数学增强模型 self.add_rule( condition=lambda r: any(w in r["input"].lower() for w in ["计算", "求解", "公式"]), target={"model": "gpt-4o", "tools": ["calculator"]}, priority=8 ) # 简单问候 → 小模型 self.add_rule( condition=lambda r: len(r["input"]) < 20, target={"model": "gpt-4o-mini", "tools": []}, priority=5 ) 语义路由 class SemanticRouter: """基于语义理解的路由器""" def __init__(self, embedding_model, vector_store): self.embedder = embedding_model self.vectors = vector_store self.routes = {} async def register_route( self, name: str, description: str, examples: list, target: dict ): """注册语义路由""" # 为路由描述和示例生成嵌入 texts = [description] + examples embeddings = await self.embedder.embed_batch(texts) self.routes[name] = { "description": description, "examples": examples, "embeddings": embeddings, "target": target } async def route(self, user_input: str) -> dict: """语义路由""" # 生成输入嵌入 input_embedding = await self.embedder.embed(user_input) # 计算与每个路由的相似度 scores = {} for name, route in self.routes.items(): # 使用所有嵌入的最大相似度 similarities = [ self._cosine_similarity(input_embedding, emb) for emb in route["embeddings"] ] scores[name] = max(similarities) # 选择最匹配的路由 best_route = max(scores, key=scores.get) best_score = scores[best_route] # 如果置信度不足，使用默认路由 if best_score < 0.7: logger.info( f"Low confidence route: {best_route} ({best_score:.2f}), " f"using default" ) return self.routes.get("default", {}).get("target", {}) return self.routes[best_route]["target"] @staticmethod def _cosine_similarity(a, b) -> float: return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) 学习型路由 class LearningRouter: """基于历史数据的学习型路由器""" def __init__(self, feature_extractor, model_registry): self.features = feature_extractor self.models = model_registry self.routing_history = [] self.performance_predictor = None # 训练好的ML模型 async def route(self, request: dict) -> dict: """智能路由——预测最佳模型和工具组合""" # 提取请求特征 features = self._extract_features(request) # 预测每个候选模型的性能 candidates = await self._get_candidates(features) predictions = [] for candidate in candidates: perf = await self._predict_performance(features, candidate) predictions.append({ "candidate": candidate, "predicted_quality": perf["quality"], "predicted_latency": perf["latency"], "predicted_cost": perf["cost"], "score": self._compute_score(perf) }) # 选择综合得分最高的 best = max(predictions, key=lambda p: p["score"]) return best["candidate"] def _extract_features(self, request: dict) -> dict: """提取请求特征""" return { "input_length": len(request["input"]), "language": self._detect_language(request["input"]), "complexity_score": self._estimate_complexity(request["input"]), "has_code": "```" in request["input"], "has_math": any(c in request["input"] for c in ["∑", "∫", "√"]), "requires_search": self._needs_search(request["input"]), "conversation_depth": request.get("turn_count", 0), "user_tier": request.get("user_tier", "free"), } def _compute_score(self, perf: dict) -> float: """综合评分""" # 质量权重0.5，延迟权重0.3，成本权重0.2 return ( perf["predicted_quality"] * 0.5 + (1 - min(perf["predicted_latency"] / 5000, 1)) * 0.3 + (1 - min(perf["predicted_cost"] / 0.1, 1)) * 0.2 ) async def record_outcome( self, request: dict, route_decision: dict, outcome: dict ): """记录路由结果，用于持续学习""" self.routing_history.append({ "features": self._extract_features(request), "decision": route_decision, "outcome": outcome, # quality_score, latency, cost "timestamp": datetime.now() }) # 定期重训练 if len(self.routing_history) % 1000 == 0: await self._retrain() 多模型调度 class MultiModelScheduler: """多模型调度器""" def __init__(self): self.models = { "gpt-4o": { "strength": ["reasoning", "code", "creative"], "cost_per_1k": 0.015, "avg_latency_ms": 1500, "context_window": 128000, "max_concurrent": 10, }, "gpt-4o-mini": { "strength": ["simple_qa", "summarization"], "cost_per_1k": 0.0003, "avg_latency_ms": 500, "context_window": 128000, "max_concurrent": 50, }, "claude-3.5-sonnet": { "strength": ["analysis", "writing", "long_context"], "cost_per_1k": 0.008, "avg_latency_ms": 1200, "context_window": 200000, "max_concurrent": 20, }, "deepseek-coder-v2": { "strength": ["code", "debug"], "cost_per_1k": 0.0014, "avg_latency_ms": 800, "context_window": 128000, "max_concurrent": 30, }, } self.model_loads = {m: 0 for m in self.models} async def select_model(self, request: dict) -> str: """选择最优模型""" scores = {} for model_name, config in self.models.items(): # 能力匹配分 capability_score = self._capability_match( request, config["strength"] ) # 负载分（负载越低越好） load_ratio = self.model_loads[model_name] / config["max_concurrent"] load_score = 1 - load_ratio # 成本分 estimated_tokens = len(request["input"]) // 4 cost = estimated_tokens * config["cost_per_1k"] cost_score = 1 - min(cost / 0.5, 1) # 综合得分 scores[model_name] = ( capability_score * 0.5 + load_score * 0.3 + cost_score * 0.2 ) return max(scores, key=scores.get) 级联路由 class CascadingRouter: """级联路由——先试便宜模型，不够再升级""" CASCADE = [ {"model": "gpt-4o-mini", "confidence_threshold": 0.85}, {"model": "gpt-4o", "confidence_threshold": 0.75}, {"model": "gpt-4o", "confidence_threshold": 0.0}, # 最终兜底 ] async def route_with_cascade(self, request: dict) -> dict: """级联路由""" total_cost = 0 for stage in self.CASCADE: model = stage["model"] threshold = stage["confidence_threshold"] response = await self._call_model(model, request) total_cost += response["cost"] if response["confidence"] >= threshold: return { **response, "model_used": model, "total_cost": total_cost, "cascade_stages": 1 } # 将低置信度响应作为上下文传给下一级 request["previous_attempt"] = response return { **response, "model_used": model, "total_cost": total_cost, "cascade_stages": len(self.CASCADE) } 路由监控指标 指标 说明 目标 路由准确率 路由决策被后续验证为正确的比例 >90% 模型利用率 各模型的负载均衡度 偏差<20% 平均路由延迟 路由决策耗时 <50ms 级联触发率 需要升级模型的比例 <15% 成本节省率 相比全用最强模型的节省比例 >40% 总结 Agent路由架构是系统性能和成本的杠杆点。语义路由实现了基于意图的精准分发，学习型路由通过历史数据持续优化决策，多模型调度在能力、负载和成本之间找到最优平衡，级联路由则提供了"先省后花"的成本优化思路。 ...

引言 Agent系统面临独特的流量挑战：LLM推理的QPS受GPU数量硬性限制、工具调用可能触发外部API限流、突发的复杂查询可能导致单请求消耗数十倍于平均值的资源。传统的固定阈值限流无法应对这些挑战，Agent系统需要更智能的限流和熔断机制。 2026年，自适应限流已成为Agent系统的标配——系统根据实时负载和资源利用率动态调整限流策略，在保护系统的同时最大化吞吐量。 Agent系统的流量特征 传统Web应用流量 Agent系统流量 │ │ │ ╱╲ │ ╱╲ │ ╱ ╲ ╱╲ │ ╱ ╲╱╲╱╲ │ ╱ ╲__╱ ╲___ │ ╱ ╲___ │________________ │________________ 时间 时间 相对均匀的请求量 突发性强、长尾明显 每请求资源消耗相近 单请求资源消耗差异巨大（10-100x） 令牌桶限流 基础令牌桶 import asyncio import time from dataclasses import dataclass @dataclass class TokenBucket: """令牌桶限流器""" capacity: float # 桶容量 refill_rate: float # 每秒补充令牌数 tokens: float = 0 # 当前令牌数 last_refill: float = 0 # 上次补充时间 def __post_init__(self): self.tokens = self.capacity self.last_refill = time.monotonic() async def acquire(self, tokens: int = 1) -> bool: """获取令牌""" while True: self._refill() if self.tokens >= tokens: self.tokens -= tokens return True # 令牌不足，等待补充 wait_time = (tokens - self.tokens) / self.refill_rate await asyncio.sleep(min(wait_time, 0.1)) def _refill(self): """补充令牌""" now = time.monotonic() elapsed = now - self.last_refill self.tokens = min( self.capacity, self.tokens + elapsed * self.refill_rate ) self.last_refill = now def try_acquire(self, tokens: int = 1) -> bool: """非阻塞获取令牌""" self._refill() if self.tokens >= tokens: self.tokens -= tokens return True return False 多维度限流 class AgentRateLimiter: """Agent多维度限流器""" def __init__(self): # 不同维度的限流器 self.limiters = { "requests_per_minute": TokenBucket(capacity=100, refill_rate=100/60), "tokens_per_minute": TokenBucket(capacity=50000, refill_rate=50000/60), "tool_calls_per_minute": TokenBucket(capacity=50, refill_rate=50/60), "concurrent_sessions": SemaphoreLimiter(max_concurrent=20), "gpu_concurrent": SemaphoreLimiter(max_concurrent=4), } async def check_request(self, request: dict) -> dict: """检查请求是否被允许""" required = { "requests_per_minute": 1, "concurrent_sessions": 1, } # 根据请求类型添加额外限制 if request.get("estimated_tokens"): required["tokens_per_minute"] = request["estimated_tokens"] if request.get("tool_calls"): required["tool_calls_per_minute"] = len(request["tool_calls"]) if request.get("requires_gpu"): required["gpu_concurrent"] = 1 # 检查所有维度 for dim, amount in required.items(): limiter = self.limiters[dim] if not limiter.try_acquire(amount): return { "allowed": False, "limited_dimension": dim, "retry_after_ms": self._get_retry_after(dim, amount), "message": f"Rate limit exceeded: {dim}" } return {"allowed": True} def _get_retry_after(self, dimension: str, amount: int) -> int: """计算重试等待时间""" limiter = self.limiters[dimension] if hasattr(limiter, 'refill_rate'): return int((amount - limiter.tokens) / limiter.refill_rate * 1000) return 1000 # 默认1秒 自适应限流 class AdaptiveRateLimiter: """自适应限流器——基于系统负载动态调整""" def __init__(self, config: dict): self.min_limit = config.get("min_limit", 10) self.max_limit = config.get("max_limit", 1000) self.current_limit = config.get("initial_limit", 100) # AIMD参数 self.additive_increase = config.get("additive_increase", 1) self.multiplicative_decrease = config.get("multiplicative_decrease", 0.5) # 系统指标 self.metrics = { "cpu_utilization": 0, "memory_utilization": 0, "gpu_utilization": 0, "p99_latency_ms": 0, "error_rate": 0, } # 调整周期 self.adjustment_interval = 10 # 秒 self.window_requests = 0 self.window_errors = 0 async def should_allow(self, request: dict) -> bool: """判断是否允许请求""" current_rps = self.window_requests / self.adjustment_interval if current_rps >= self.current_limit: return False self.window_requests += 1 return True async def adjust_limits(self): """定期调整限流阈值""" while True: await asyncio.sleep(self.adjustment_interval) # 计算系统健康度 health_score = self._calculate_health() if health_score > 0.8: # 系统健康，增加限流（AIMD的AI） self.current_limit = min( self.current_limit + self.additive_increase, self.max_limit ) logger.info( f"Increasing rate limit to {self.current_limit} " f"(health: {health_score:.2f})" ) elif health_score < 0.5: # 系统不健康，减少限流（AIMD的MD） self.current_limit = max( int(self.current_limit * self.multiplicative_decrease), self.min_limit ) logger.warning( f"Decreasing rate limit to {self.current_limit} " f"(health: {health_score:.2f})" ) # 重置窗口 self.window_requests = 0 self.window_errors = 0 def _calculate_health(self) -> float: """计算系统健康度（0-1）""" weights = { "cpu_utilization": 0.2, "memory_utilization": 0.15, "gpu_utilization": 0.25, "p99_latency_ms": 0.25, "error_rate": 0.15, } thresholds = { "cpu_utilization": 0.8, "memory_utilization": 0.85, "gpu_utilization": 0.9, "p99_latency_ms": 2000, "error_rate": 0.05, } score = 1.0 for metric, weight in weights.items(): value = self.metrics[metric] threshold = thresholds[metric] ratio = value / threshold if threshold > 0 else 0 if ratio > 1: # 超过阈值，扣分 score -= weight * min(ratio - 1, 1) return max(0, score) 熔断器设计 class CircuitBreaker: """熔断器""" class State: CLOSED = "closed" # 正常工作 OPEN = "open" # 熔断，拒绝请求 HALF_OPEN = "half_open" # 半开，试探恢复 def __init__( self, failure_threshold: int = 10, failure_rate_threshold: float = 0.5, recovery_timeout: float = 30.0, half_open_max_calls: int = 3 ): self.state = self.State.CLOSED self.failure_threshold = failure_threshold self.failure_rate_threshold = failure_rate_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls self.failure_count = 0 self.success_count = 0 self.total_count = 0 self.last_failure_time = None self.half_open_calls = 0 async def call(self, func, *args, **kwargs): """通过熔断器调用函数""" if self.state == self.State.OPEN: if self._should_attempt_recovery(): self.state = self.State.HALF_OPEN self.half_open_calls = 0 logger.info("Circuit breaker entering half-open state") else: raise CircuitBreakerOpenError( f"Circuit breaker is open. " f"Retry after {self._recovery_remaining():.0f}s" ) if self.state == self.State.HALF_OPEN: if self.half_open_calls >= self.half_open_max_calls: raise CircuitBreakerOpenError( "Half-open: max probe calls reached" ) self.half_open_calls += 1 try: result = await func(*args, **kwargs) await self._on_success() return result except Exception as e: await self._on_failure() raise async def _on_success(self): self.success_count += 1 self.total_count += 1 if self.state == self.State.HALF_OPEN: if self.half_open_calls >= self.half_open_max_calls: self.state = self.State.CLOSED self.failure_count = 0 logger.info("Circuit breaker recovered, closing") async def _on_failure(self): self.failure_count += 1 self.total_count += 1 self.last_failure_time = time.monotonic() if self.state == self.State.HALF_OPEN: self.state = self.State.OPEN logger.warning("Circuit breaker re-opened from half-open") elif self.state == self.State.CLOSED: failure_rate = self.failure_count / max(self.total_count, 1) if (self.failure_count >= self.failure_threshold or failure_rate >= self.failure_rate_threshold): self.state = self.State.OPEN logger.error( f"Circuit breaker opened: " f"{self.failure_count}/{self.total_count} failures " f"({failure_rate:.1%})" ) 多级熔断 class MultiLevelCircuitBreaker: """多级熔断器""" def __init__(self): self.breakers = { "llm_inference": CircuitBreaker( failure_threshold=5, failure_rate_threshold=0.3, recovery_timeout=30 ), "tool_search": CircuitBreaker( failure_threshold=10, failure_rate_threshold=0.5, recovery_timeout=15 ), "vector_db": CircuitBreaker( failure_threshold=8, failure_rate_threshold=0.4, recovery_timeout=10 ), "external_api": CircuitBreaker( failure_threshold=3, failure_rate_threshold=0.2, recovery_timeout=60 ), } async def call_with_fallback( self, primary: callable, fallbacks: list, circuit_key: str ): """带降级的熔断调用""" breaker = self.breakers.get(circuit_key) try: if breaker: return await breaker.call(primary) return await primary() except CircuitBreakerOpenError: # 主路径熔断，尝试降级 for i, fallback in enumerate(fallbacks): try: logger.info(f"Trying fallback {i+1}/{len(fallbacks)}") return await fallback() except Exception as e: logger.warning(f"Fallback {i+1} failed: {e}") continue # 所有降级都失败 raise AllFallbacksFailedError( f"All fallbacks failed for {circuit_key}" ) 降级策略 class DegradationStrategy: """Agent降级策略""" LEVELS = { "normal": { "model": "gpt-4o", "max_tools": 10, "max_context": 128000, "streaming": True, }, "degraded_1": { "model": "gpt-4o-mini", # 降级到小模型 "max_tools": 5, "max_context": 32000, "streaming": True, }, "degraded_2": { "model": "gpt-4o-mini", "max_tools": 2, # 仅保留核心工具 "max_context": 8000, "streaming": False, }, "emergency": { "model": "cached-response", # 使用缓存或模板回复 "max_tools": 0, "max_context": 1000, "streaming": False, } } def get_current_level(self, system_load: float) -> str: """根据系统负载获取降级级别""" if system_load < 0.7: return "normal" elif system_load < 0.85: return "degraded_1" elif system_load < 0.95: return "degraded_2" else: return "emergency" 总结 Agent系统的限流与熔断需要从三个层面协同工作：令牌桶实现精确的多维度限流，自适应算法根据系统健康度动态调整阈值，熔断器在故障时快速切断流量并支持降级恢复。关键是在系统保护和用户体验之间找到平衡——过度保护会浪费资源，保护不足会导致雪崩。 ...

引言 Agent系统最令人头疼的故障模式之一就是无限循环——Agent反复调用同一个工具、在两个状态间来回切换、或者陷入"思考但不行动"的死循环。这类问题不仅浪费Token和计算资源，还可能导致用户长时间等待无响应。 2026年，随着Agent自主能力的增强（如AutoGPT式的自主规划），循环检测和超时控制变得更加关键。一个能够自主决策的Agent，如果不能有效检测和打破循环，其危害性远大于传统软件的死循环。 循环类型分析 Agent系统中的四种典型循环 类型1：工具调用循环 类型2：状态转移循环 ┌─────────────────┐ ┌──────────────────┐ │ Agent ──▶ Tool A│ │ State A ──▶State B│ │ ▲ │ │ │ ▲ │ │ │ └──────┘ │ │ └─────────┘ │ │ (反复调用同一工具)│ │ (状态来回切换) │ └─────────────────┘ └──────────────────┘ 类型3：推理循环 类型4：工具链循环 ┌─────────────────┐ ┌──────────────────┐ │ Think ──▶ Think │ │ Tool A ──▶Tool B │ │ ▲ │ │ │ ▲ │ │ │ └─────────┘ │ │ └─────────┘ │ │ (反复思考不行动) │ │ (工具间互相触发) │ └─────────────────┘ └──────────────────┘ 循环检测算法 基于状态指纹的检测 import hashlib from collections import defaultdict from dataclasses import dataclass @dataclass class CycleDetector: """基于状态指纹的循环检测器""" max_history: int = 50 # 保留最近50步 cycle_threshold: int = 3 # 重复出现3次判定为循环 def __init__(self): self.state_history: list = [] self.fingerprint_counts: dict = defaultdict(int) def record_state(self, state: dict) -> bool: """记录状态，返回是否检测到循环""" # 生成状态指纹 fingerprint = self._generate_fingerprint(state) self.state_history.append({ "fingerprint": fingerprint, "state": state, "timestamp": datetime.now() }) # 限制历史长度 if len(self.state_history) > self.max_history: old = self.state_history.pop(0) self.fingerprint_counts[old["fingerprint"]] -= 1 self.fingerprint_counts[fingerprint] += 1 # 检测循环 if self.fingerprint_counts[fingerprint] >= self.cycle_threshold: return True # 检测模式循环（A-B-A-B模式） if self._detect_pattern_cycle(): return True return False def _generate_fingerprint(self, state: dict) -> str: """生成状态指纹""" # 提取关键状态信息 key_info = { "intent": state.get("intent"), "active_tool": state.get("active_tool"), "tool_params_hash": hashlib.md5( json.dumps(state.get("tool_params", {}), sort_keys=True).encode() ).hexdigest()[:8], "fsm_state": state.get("fsm_state"), "pending_actions": state.get("pending_actions", []) } fingerprint_str = json.dumps(key_info, sort_keys=True) return hashlib.md5(fingerprint_str.encode()).hexdigest() def _detect_pattern_cycle(self) -> bool: """检测模式循环（如A-B-A-B）""" if len(self.state_history) < 4: return False # 检查最近的状态是否形成周期模式 recent = [s["fingerprint"] for s in self.state_history[-6:]] # 尝试不同的周期长度 for period in range(2, 4): if len(recent) >= period * 2: pattern = recent[-period:] previous = recent[-period*2:-period] if pattern == previous: return True return False 基于行为序列的检测 class BehaviorSequenceAnalyzer: """基于行为序列的循环检测""" def __init__(self): self.action_sequences = [] def analyze(self, actions: list) -> dict: """分析行为序列""" result = { "has_cycle": False, "cycle_type": None, "cycle_length": 0, "suggestion": None } # 使用Floyd算法检测环 cycle = self._floyd_cycle_detection(actions) if cycle: result["has_cycle"] = True result["cycle_length"] = len(cycle) result["cycle_type"] = self._classify_cycle(cycle) result["suggestion"] = self._suggest_break_strategy(cycle) return result def _floyd_cycle_detection(self, sequence: list) -> list: """Floyd环检测算法""" if len(sequence) < 2: return None # 快慢指针 slow = 0 fast = 0 while True: slow = (slow + 1) % len(sequence) fast = (fast + 2) % len(sequence) if sequence[slow] == sequence[fast]: # 找到环，确定环的起始和长度 start = 0 ptr1 = start ptr2 = slow while ptr1 != ptr2: ptr1 = (ptr1 + 1) % len(sequence) ptr2 = (ptr2 + 1) % len(sequence) # 提取环 cycle_start = ptr1 cycle = [sequence[cycle_start]] ptr = (cycle_start + 1) % len(sequence) while ptr != cycle_start: cycle.append(sequence[ptr]) ptr = (ptr + 1) % len(sequence) return cycle if fast == 0: return None # 无环 def _classify_cycle(self, cycle: list) -> str: """分类循环类型""" tools_in_cycle = [a for a in cycle if a.get("type") == "tool_call"] thoughts_in_cycle = [a for a in cycle if a.get("type") == "thought"] if len(tools_in_cycle) == len(cycle): return "tool_cycle" elif len(thoughts_in_cycle) == len(cycle): return "reasoning_cycle" else: return "mixed_cycle" def _suggest_break_strategy(self, cycle: list) -> str: """建议打破循环的策略""" cycle_type = self._classify_cycle(cycle) strategies = { "tool_cycle": "尝试用不同参数调用工具，或切换到替代工具", "reasoning_cycle": "强制进入执行阶段，或向用户请求澄清", "mixed_cycle": "重置上下文窗口，或回退到上一个成功状态" } return strategies.get(cycle_type, "终止当前任务并重试") 多级超时控制 class MultiLevelTimeout: """多级超时控制体系""" LEVELS = { "tool_call": { "default": 30, # 单次工具调用 "search": 15, # 搜索类工具 "code_exec": 60, # 代码执行 "file_io": 10, # 文件操作 }, "step": { "understanding": 10, # 意图理解 "planning": 15, # 规划 "retrieving": 10, # 检索 "executing": 120, # 工具执行 "generating": 30, # 响应生成 }, "session": { "max_duration": 600, # 单次会话最大10分钟 "idle_timeout": 120, # 空闲2分钟超时 }, "workflow": { "max_steps": 50, # 工作流最大步骤数 "max_tool_calls": 20, # 最大工具调用次数 "max_tokens": 100000, # 最大Token消耗 } } def __init__(self): self.active_timers = {} async def with_timeout( self, level: str, operation: str, coro: asyncio.coroutines ): """带超时执行协程""" timeout = self.LEVELS.get(level, {}).get(operation, 30) try: result = await asyncio.wait_for(coro, timeout=timeout) return result except asyncio.TimeoutError: logger.warning( f"Timeout at {level}.{operation} after {timeout}s" ) await self._handle_timeout(level, operation) raise async def _handle_timeout(self, level: str, operation: str): """超时处理策略""" if level == "tool_call": # 记录工具超时，可能降级 await self._record_tool_timeout(operation) elif level == "step": # 步骤超时，尝试跳过或降级 await self._try_degrade_step(operation) elif level == "session": # 会话超时，优雅终止 await self._graceful_session_end() elif level == "workflow": # 工作流超限，强制终止 await self._force_terminate() 循环打破与自恢复 class CycleBreaker: """循环打破器""" def __init__(self, cycle_detector, timeout_controller): self.detector = cycle_detector self.timeout = timeout_controller async def monitor_and_break( self, agent_session, check_interval: float = 1.0 ): """持续监控并打破循环""" while not agent_session.is_complete(): current_state = agent_session.get_state() if self.detector.record_state(current_state): logger.warning("Cycle detected, initiating break sequence") await self._break_cycle(agent_session) return True await asyncio.sleep(check_interval) return False async def _break_cycle(self, agent_session): """执行循环打破策略""" # 策略1：注入扰动——修改Agent的上下文 perturbation = { "system_message": ( "你似乎陷入了循环。请尝试不同的方法，" "或者明确说明你无法完成此任务。" ), "temperature_boost": 0.3, # 提高温度增加随机性 "disable_repeated_tool": True # 禁用循环工具 } await agent_session.inject_perturbation(perturbation) # 策略2：重置到上一个健康状态 healthy_state = agent_session.get_last_healthy_state() if healthy_state: await agent_session.restore_state(healthy_state) # 策略3：降级为简化流程 await agent_session.switch_to_degraded_mode( simplified_tools=True, max_steps=5 ) # 策略4：最终兜底——请求人工介入 if not await agent_session.try_recover(): await agent_session.escalate_to_human( reason="无法自动打破循环", context=agent_session.get_debug_context() ) 生产实践：超时配置矩阵 场景 工具超时 步骤超时 会话超时 最大步数 最大Token 简单问答 10s 15s 60s 5 5K 工具调用 30s 120s 300s 15 30K 复杂分析 60s 300s 600s 30 80K 自主任务 120s 600s 1800s 50 200K 批处理 300s 1800s 7200s 100 500K 监控与告警 class CycleMonitor: """循环监控器""" async def collect_metrics(self) -> dict: return { "cycle_detected_rate": await self._get_rate("cycle_detected"), "timeout_rate_by_level": { "tool": await self._get_rate("timeout.tool_call"), "step": await self._get_rate("timeout.step"), "session": await self._get_rate("timeout.session"), }, "avg_steps_to_cycle": await self._get_avg("steps_before_cycle"), "break_success_rate": await self._get_rate("cycle_break_success"), "human_escalation_rate": await self._get_rate("human_escalation"), } 总结 循环检测和超时控制是Agent系统鲁棒性的基石。基于状态指纹的检测能够高效识别精确循环，基于行为序列的分析能够发现模式循环。多级超时体系确保任何级别的异常都有对应的兜底机制。当检测到循环时，系统应按照"注入扰动→重置状态→降级模式→人工介入"的顺序尝试恢复。 ...

引言 Agent SaaS平台在2026年面临的核心挑战之一是多租户架构设计。不同租户的Agent可能使用不同的模型、不同的工具集、不同的Prompt模板，且对性能、安全性和成本的要求差异巨大。如何在共享基础设施上实现高效的资源隔离和公平的成本分摊，是Agent平台架构师必须解决的问题。 多租户隔离模型 三种隔离级别 隔离程度 ──────────────────────────────────▶ 强 ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ 共享模式 │ │ 混合模式 │ │ 独占模式 │ │ │ │ │ │ │ │ 共享所有资源 │ │ 共享计算资源 │ │ 独立资源栈 │ │ 逻辑隔离数据 │ │ 隔离存储资源 │ │ 物理隔离 │ │ │ │ │ │ │ │ 成本最低 │ │ 平衡 │ │ 隔离最强 │ │ 隔离最弱 │ │ │ │ 成本最高 │ └──────────────┘ └──────────────┘ └──────────────┘ from enum import Enum class IsolationLevel(Enum): SHARED = "shared" # 共享模式：所有租户共享同一Agent实例 HYBRID = "hybrid" # 混合模式：共享计算，隔离存储 DEDICATED = "dedicated" # 独占模式：每个租户独立资源栈 class TenantConfig: """租户配置""" def __init__( self, tenant_id: str, tier: str, # free, pro, enterprise isolation: IsolationLevel, quota: dict, custom_config: dict = None ): self.tenant_id = tenant_id self.tier = tier self.isolation = isolation self.quota = quota self.custom_config = custom_config or {} # 根据tier设置默认配额 if not quota: self.quota = self._default_quota(tier) @staticmethod def _default_quota(tier: str) -> dict: defaults = { "free": { "max_sessions": 10, "max_concurrent": 2, "max_tokens_per_day": 100000, "max_tools": 5, "max_memory_mb": 256, "rate_limit_rpm": 20, # 每分钟请求数 }, "pro": { "max_sessions": 100, "max_concurrent": 10, "max_tokens_per_day": 2000000, "max_tools": 20, "max_memory_mb": 2048, "rate_limit_rpm": 200, }, "enterprise": { "max_sessions": -1, # 无限 "max_concurrent": 100, "max_tokens_per_day": 50000000, "max_tools": -1, "max_memory_mb": 32768, "rate_limit_rpm": 2000, } } return defaults.get(tier, defaults["free"]) 资源隔离实现 计算资源隔离 class TenantResourceManager: """租户资源管理器""" def __init__(self, k8s_client): self.k8s = k8s_client self.tenant_pools = {} # tenant_id -> resource pool async def get_or_create_pool( self, tenant: TenantConfig ) -> str: """获取或创建租户资源池""" if tenant.tenant_id in self.tenant_pools: return self.tenant_pools[tenant.tenant_id] if tenant.isolation == IsolationLevel.DEDICATED: # 独占模式：创建独立namespace和资源 namespace = await self._create_dedicated_namespace(tenant) await self._deploy_dedicated_resources(tenant, namespace) self.tenant_pools[tenant.tenant_id] = namespace elif tenant.isolation == IsolationLevel.HYBRID: # 混合模式：使用共享namespace但设置ResourceQuota await self._apply_resource_quota(tenant) self.tenant_pools[tenant.tenant_id] = "shared" else: # 共享模式：仅通过应用层隔离 self.tenant_pools[tenant.tenant_id] = "shared" return self.tenant_pools[tenant.tenant_id] async def _apply_resource_quota(self, tenant: TenantConfig): """应用K8s ResourceQuota""" quota_yaml = { "apiVersion": "v1", "kind": "ResourceQuota", "metadata": { "name": f"quota-{tenant.tenant_id}", "namespace": "agent-shared" }, "spec": { "hard": { "requests.cpu": f"{tenant.quota['max_cpu']}", "requests.memory": f"{tenant.quota['max_memory_mb']}Mi", "pods": str(tenant.quota["max_pods"]), } } } await self.k8s.apply_resource(quota_yaml) class TenantRateLimiter: """租户级限流器""" def __init__(self, redis_client): self.redis = redis_client async def check_and_consume( self, tenant_id: str, resource: str, # "api_call", "token", "tool_exec" amount: int = 1 ) -> bool: """检查配额并消费""" # 滑动窗口限流 key = f"quota:{tenant_id}:{resource}:{datetime.now().strftime('%Y%m%d%H%M')}" pipe = self.redis.pipeline() pipe.incr(key, amount) pipe.expire(key, 3600) # 1小时TTL results = await pipe.execute() current_usage = results[0] limit = await self._get_limit(tenant_id, resource) if current_usage > limit: # 回滚消费 await self.redis.decr(key, amount) return False return True 数据隔离 class TenantDataIsolation: """租户数据隔离管理""" def __init__(self, db_client): self.db = db_client async def execute_for_tenant( self, tenant_id: str, query: str, params: tuple = None ): """在租户上下文中执行查询""" # 方式1：Row-Level Security (PostgreSQL RLS) await self.db.execute( f"SET app.current_tenant = '{tenant_id}'" ) try: result = await self.db.fetch(query, *(params or ())) return result finally: await self.db.execute("RESET app.current_tenant") async def get_vector_store_for_tenant( self, tenant_id: str, collection_name: str ): """获取租户专属的向量存储""" # 使用租户ID作为namespace前缀 namespaced_collection = f"tenant_{tenant_id}_{collection_name}" return VectorStore( collection=namespaced_collection, metadata_filter={"tenant_id": tenant_id} # 双重保障 ) 成本分摊模型 class CostAllocator: """成本分摊器——精确追踪每租户的资源消耗""" # 2026年典型成本基准（美元） COST_RATES = { "llm_token_input": 0.00001, # per token "llm_token_output": 0.00003, # per token "embedding_token": 0.0000001, # per token "vector_search": 0.0001, # per 1k queries "tool_execution": 0.001, # per execution "memory_storage_gb_month": 0.10, "gpu_hour": 2.50, # per GPU hour "cpu_hour": 0.05, } def __init__(self, metrics_store): self.metrics = metrics_store async def record_usage( self, tenant_id: str, resource: str, amount: float, session_id: str = None ): """记录资源使用""" cost = amount * self.COST_RATES.get(resource, 0) await self.metrics.insert({ "tenant_id": tenant_id, "resource": resource, "amount": amount, "cost": cost, "session_id": session_id, "timestamp": datetime.now() }) async def calculate_bill( self, tenant_id: str, period_start: datetime, period_end: datetime ) -> dict: """计算租户账单""" usage = await self.metrics.aggregate( tenant_id=tenant_id, start=period_start, end=period_end ) bill = { "tenant_id": tenant_id, "period": f"{period_start.date()} to {period_end.date()}", "items": [], "total": 0 } for resource, amount in usage.items(): rate = self.COST_RATES.get(resource, 0) cost = amount * rate bill["items"].append({ "resource": resource, "amount": amount, "rate": rate, "cost": round(cost, 4) }) bill["total"] += cost # 应用tier折扣 tier = await self._get_tenant_tier(tenant_id) discount = {"free": 0, "pro": 0.1, "enterprise": 0.25}.get(tier, 0) bill["discount"] = round(bill["total"] * discount, 2) bill["final_total"] = round(bill["total"] - bill["discount"], 2) return bill 实时成本监控 class RealtimeCostMonitor: """实时成本监控与告警""" async def monitor_tenant(self, tenant_id: str): """监控租户实时成本""" while True: daily_cost = await self._get_daily_cost(tenant_id) budget = await self._get_budget(tenant_id) utilization = daily_cost / budget if budget > 0 else 0 if utilization > 0.9: await self._alert( tenant_id=tenant_id, level="critical", message=f"Budget at {utilization:.0%}: ${daily_cost:.2f}/${budget:.2f}" ) # 触发降级或限流 if utilization > 1.0: await self._throttle_tenant(tenant_id) elif utilization > 0.7: await self._alert( tenant_id=tenant_id, level="warning", message=f"Budget at {utilization:.0%}" ) await asyncio.sleep(60) # 每分钟检查 租户级配置管理 class TenantConfigManager: """租户配置管理器""" async def get_agent_config(self, tenant_id: str) -> dict: """获取租户专属的Agent配置""" base_config = { "model": "gpt-4o-mini", "temperature": 0.7, "max_tokens": 4096, "tools": ["search", "calculator"], "system_prompt": "You are a helpful assistant.", "safety_level": "standard" } # 合并租户自定义配置 tenant_overrides = await self._load_tenant_overrides(tenant_id) config = {**base_config, **tenant_overrides} # 应用tier限制 tier = await self._get_tier(tenant_id) if tier == "free": config["model"] = "gpt-4o-mini" # 限制免费用户使用小模型 config["max_tokens"] = min(config["max_tokens"], 2048) return config async def validate_config_change( self, tenant_id: str, new_config: dict ) -> dict: """验证租户配置变更""" tier = await self._get_tier(tenant_id) tier_limits = self.TIER_LIMITS[tier] errors = [] # 检查模型权限 if new_config.get("model") not in tier_limits["allowed_models"]: errors.append(f"Model {new_config['model']} not available for {tier} tier") # 检查工具数量 if len(new_config.get("tools", [])) > tier_limits["max_tools"]: errors.append(f"Too many tools for {tier} tier") # 检查安全级别 if new_config.get("safety_level") == "none" and tier != "enterprise": errors.append("Safety level 'none' requires enterprise tier") return {"valid": len(errors) == 0, "errors": errors} 安全边界设计 class TenantSecurityBoundary: """租户安全边界""" async def enforce_boundary(self, tenant_id: str, request: dict): """执行安全边界检查""" # 1. 防止跨租户数据访问 if request.get("target_tenant") and request["target_tenant"] != tenant_id: raise SecurityViolation("Cross-tenant access denied") # 2. 工具白名单检查 allowed_tools = await self._get_allowed_tools(tenant_id) for tool in request.get("tools", []): if tool not in allowed_tools: raise SecurityViolation(f"Tool '{tool}' not allowed for tenant") # 3. 出站请求域名白名单 if request.get("api_endpoints"): allowed_domains = await self._get_allowed_domains(tenant_id) for endpoint in request["api_endpoints"]: domain = urllib.parse.urlparse(endpoint).hostname if domain not in allowed_domains: raise SecurityViolation(f"Domain '{domain}' not allowed") # 4. 敏感操作审计 if request.get("action") in ["file_write", "code_exec", "network_access"]: await self._audit_log(tenant_id, request) 总结 Agent多租户架构的核心是在资源共享与租户隔离之间找到平衡点。共享模式成本最低但隔离最弱，独占模式隔离最强但成本最高，混合模式是大多数SaaS平台的最佳选择。无论选择哪种模式，都必须建立完善的配额管理、成本分摊和安全边界机制。 ...