架构设计

引言 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平台的最佳选择。无论选择哪种模式，都必须建立完善的配额管理、成本分摊和安全边界机制。 ...

引言 Agent系统的发布比传统应用复杂得多——一个Prompt的微调可能导致Agent行为完全改变，一个工具的版本升级可能影响所有依赖它的Agent。传统的"停机发布"在Agent系统中不可接受，而简单的"滚动更新"也无法满足Agent系统对质量保障的高要求。 2026年，金丝雀发布 + 自动回滚已成为Agent系统的标准发布实践，但Agent系统的灰度发布有其独特的挑战和解决方案。 Agent发布的特殊性 维度 传统应用 Agent系统 变更类型 代码逻辑 Prompt/模型/工具/代码 质量评估 单元测试+集成测试 需要LLM评估+人工审核 回滚速度 秒级 秒级（代码）/分钟级（模型） 影响范围 功能正确性 对话质量、安全性、成本 监控指标 错误率、延迟 +质量评分、Token消耗、用户满意度 灰度发布策略 策略一：金丝雀发布 class CanaryReleaseManager: """金丝雀发布管理器""" def __init__(self, traffic_router, metrics_collector): self.router = traffic_router self.metrics = metrics_collector async def canary_deploy( self, new_version: str, stages: list = None ) -> bool: """渐进式金丝雀发布""" if stages is None: stages = [ {"traffic_percent": 5, "duration_minutes": 10}, {"traffic_percent": 20, "duration_minutes": 15}, {"traffic_percent": 50, "duration_minutes": 20}, {"traffic_percent": 100, "duration_minutes": 30}, ] baseline = await self.metrics.get_baseline() for stage in stages: # 调整流量分配 await self.router.set_traffic_split({ "stable": 100 - stage["traffic_percent"], "canary": stage["traffic_percent"] }) logger.info( f"Canary stage: {stage['traffic_percent']}% traffic " f"for {stage['duration_minutes']}min" ) # 等待观察期 await asyncio.sleep(stage["duration_minutes"] * 60) # 评估金丝雀指标 canary_metrics = await self.metrics.collect("canary") evaluation = self._evaluate(baseline, canary_metrics) if evaluation["action"] == "rollback": logger.warning( f"Canary failed at {stage['traffic_percent']}%: " f"{evaluation['reason']}" ) await self._rollback() return False elif evaluation["action"] == "hold": logger.info(f"Pausing canary: {evaluation['reason']}") await self._notify_human(evaluation) await self._wait_for_approval() # 所有阶段通过，完成发布 await self.router.promote_canary() return True def _evaluate(self, baseline: dict, canary: dict) -> dict: """评估金丝雀健康度""" checks = [ self._check_error_rate(baseline, canary), self._check_latency(baseline, canary), self._check_quality_score(baseline, canary), self._check_cost(baseline, canary), self._check_safety(baseline, canary), ] for check in checks: if check["status"] == "fail": return {"action": "rollback", "reason": check["reason"]} if check["status"] == "warn": return {"action": "hold", "reason": check["reason"]} return {"action": "proceed", "reason": "All checks passed"} def _check_quality_score(self, baseline: dict, canary: dict) -> dict: """质量评分检查——Agent特有的评估维度""" quality_drop = baseline["quality_score"] - canary["quality_score"] if quality_drop > 0.1: # 质量下降超过10% return { "status": "fail", "reason": f"Quality dropped {quality_drop:.1%}" } elif quality_drop > 0.05: return { "status": "warn", "reason": f"Quality dropped {quality_drop:.1%}, review needed" } return {"status": "pass"} def _check_safety(self, baseline: dict, canary: dict) -> dict: """安全检查——检测有害输出""" safety_violation_rate = canary.get("safety_violation_rate", 0) if safety_violation_rate > 0.001: # 0.1%安全违规 return { "status": "fail", "reason": f"Safety violation rate: {safety_violation_rate:.3%}" } return {"status": "pass"} 策略二：蓝绿部署 # K8s蓝绿部署配置 apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: agent-service spec: replicas: 10 strategy: blueGreen: activeService: agent-service-active previewService: agent-service-preview autoPromotionEnabled: false # 手动确认 scaleDownDelaySeconds: 30 prePromotionAnalysis: templates: - templateName: agent-quality-check args: - name: service-name value: agent-service-preview selector: matchLabels: app: agent-service template: metadata: labels: app: agent-service spec: containers: - name: agent image: agent/service:{{ .Values.version }} env: - name: AGENT_VERSION value: "{{ .Values.version }}" - name: MODEL_ENDPOINT value: "http://llm-service:8080/v1" 策略三：流量镜像 流量镜像是Agent系统特别适合的灰度策略——将生产流量复制一份到新版本，不影响真实用户： ...

引言 Agent系统本质上是一个工作流编排系统——理解意图、检索知识、调用工具、评估结果、生成回复，每一步都是工作流中的一个节点。选择合适的工作流引擎，直接决定了Agent系统的可靠性、可观测性和开发效率。 2026年，工作流引擎领域已形成了清晰的格局。Temporal凭借其强大的状态管理和重试机制成为Agent系统的热门选择，Airflow在数据处理管道中依然占有一席之地，而自研引擎则在对性能和灵活性有极致要求的场景中仍有市场。 Agent工作流的特殊需求 与传统数据处理工作流不同，Agent工作流有其独特的需求特征： 需求维度 传统工作流 Agent工作流 执行时长 分钟到小时 秒到分钟 分支复杂度 低（线性DAG） 高（动态分支、循环） 人机交互 罕见 频繁（澄清、确认） 失败处理 重试或告警 重新规划、降级策略 状态管理 简单 复杂（对话历史、中间结果） 实时性 批处理 实时或近实时 动态性 静态DAG 运行时动态生成 三大方案深度对比 Temporal：Agent工作流的最佳搭档 from temporalio import workflow, activity from datetime import timedelta @activity.defn async def understand_intent(user_input: str) -> dict: """意图理解活动""" # 调用LLM进行意图分类 result = await llm_client.classify(user_input) return { "intent": result.intent, "confidence": result.confidence, "entities": result.entities } @activity.defn async def retrieve_memory(query: str, top_k: int = 5) -> list: """记忆检索活动""" memories = await vector_db.search(query, top_k=top_k) return [{"content": m.text, "score": m.score} for m in memories] @activity.defn async def execute_tool(tool_name: str, params: dict) -> dict: """工具执行活动""" tool = tool_registry.get(tool_name) result = await tool.run(**params) return result @activity.defn async def generate_response(prompt: str, context: dict) -> str: """响应生成活动""" response = await llm_client.generate(prompt, **context) return response @workflow.defn class AgentWorkflow: """Agent主工作流""" @workflow.run async def run(self, user_input: str) -> str: # Step 1: 意图理解 intent_result = await workflow.execute_activity( understand_intent, user_input, start_to_close_timeout=timedelta(seconds=10), retry_policy=RetryPolicy( initial_interval=timedelta(seconds=1), maximum_interval=timedelta(seconds=10), maximum_attempts=3 ) ) # 需要澄清时，等待用户输入 if intent_result["confidence"] < 0.6: clarification = await workflow.wait_for_signal( "user_clarification", timeout=timedelta(minutes=5) ) intent_result = await workflow.execute_activity( understand_intent, clarification, start_to_close_timeout=timedelta(seconds=10) ) # Step 2: 并行检索记忆和执行工具 memory_task = workflow.execute_activity( retrieve_memory, user_input, start_to_close_timeout=timedelta(seconds=5) ) tool_task = workflow.execute_activity( execute_tool, intent_result["intent"], intent_result["entities"], start_to_close_timeout=timedelta(seconds=30) ) memories, tool_results = await asyncio.gather(memory_task, tool_task) # Step 3: 生成响应 prompt = build_prompt(user_input, memories, tool_results) response = await workflow.execute_activity( generate_response, prompt, {"intent": intent_result["intent"]}, start_to_close_timeout=timedelta(seconds=15) ) return response Temporal的优势在Agent场景中极为突出： ...

引言 Agent系统的扩展性挑战与传统Web应用截然不同。LLM推理是GPU密集型操作，工具执行可能是CPU或IO密集型，而向量检索则是内存密集型。这意味着简单的"加机器"策略无法有效解决Agent系统的扩展问题。 2026年，K8s + GPU Operator已成为Agent系统部署的事实标准，但如何高效利用集群资源仍然是工程团队面临的核心挑战。 扩展维度分析 Agent系统需要在多个维度上独立扩展： ┌─────────────────────────────────────────────────────────┐ │ Agent系统扩展维度 │ ├─────────────┬──────────────┬──────────────┬────────────┤ │ 并发会话数 │ 推理吞吐量 │ 工具执行并发 │ 记忆检索延迟│ │ (CPU/Mem) │ (GPU) │ (CPU/IO) │ (RAM/SSD) │ ├─────────────┼──────────────┼──────────────┼────────────┤ │ 水平扩展 │ GPU水平扩展 │ 水平扩展 │ 分片+副本 │ │ +Stateless │ +模型并行 │ +无状态 │ +读副本 │ └─────────────┴──────────────┴──────────────┴────────────┘ 从单机到集群的演进路径 Phase 1：单机优化 在扩展之前，先榨干单机性能： import asyncio import uvicorn from concurrent.futures import ThreadPoolExecutor class SingleNodeAgent: """单机Agent——最大化单节点利用率""" def __init__(self): # CPU密集型任务（工具执行） self.cpu_pool = ThreadPoolExecutor( max_workers=8, thread_name_prefix="tool-exec" ) # IO密集型任务（网络请求） self.io_pool = ThreadPoolExecutor( max_workers=32, thread_name_prefix="io-op" ) # LLM推理使用GPU，通过信号量控制并发 self.llm_semaphore = asyncio.Semaphore(4) async def process_request(self, user_input: str) -> str: # 并行执行独立任务 memory_task = asyncio.create_task(self._retrieve_memory(user_input)) tool_task = asyncio.create_task(self._execute_tools(user_input)) memory = await memory_task tool_results = await tool_task # LLM推理（GPU受限） async with self.llm_semaphore: response = await self._llm_inference(user_input, memory, tool_results) return response Phase 2：水平拆分 将不同负载特征的服务拆分到不同节点： ...

引言 Agent的状态管理是系统设计中最容易被忽视却又最关键的环节。一个Agent在执行任务时，可能经历"理解意图→检索记忆→调用工具→评估结果→生成回复"等多个阶段，每个阶段都有不同的状态和转移条件。状态管理不当会导致上下文丢失、重复执行、死循环等严重问题。 2026年，随着Agent系统复杂度的指数级增长，系统化的状态管理架构已成为生产部署的必备条件。 Agent状态的三个层次 第一层：会话状态（Session State） 会话状态是最基础的状态层，管理单次用户交互的上下文： from dataclasses import dataclass, field from datetime import datetime from enum import Enum class SessionStatus(Enum): ACTIVE = "active" PAUSED = "paused" COMPLETED = "completed" FAILED = "failed" TIMEOUT = "timeout" @dataclass class SessionState: """会话状态——管理单次交互的完整生命周期""" session_id: str user_id: str status: SessionStatus created_at: datetime updated_at: datetime message_history: list = field(default_factory=list) active_tools: list = field(default_factory=list) pending_actions: list = field(default_factory=list) context_window: dict = field(default_factory=dict) metadata: dict = field(default_factory=dict) def add_message(self, role: str, content: str): self.message_history.append({ "role": role, "content": content, "timestamp": datetime.now().isoformat() }) self.updated_at = datetime.now() def is_expired(self, ttl_seconds: int = 3600) -> bool: elapsed = (datetime.now() - self.updated_at).total_seconds() return elapsed > ttl_seconds 第二层：工作流状态（Workflow State） 工作流状态管理Agent执行复杂多步骤任务时的进度： from enum import Enum class WorkflowStepStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" RETRYING = "retrying" @dataclass class WorkflowStep: step_id: str step_name: str status: WorkflowStepStatus dependencies: list # 前置步骤ID inputs: dict outputs: dict retry_count: int = 0 max_retries: int = 3 started_at: datetime = None completed_at: datetime = None @dataclass class WorkflowState: """工作流状态——管理多步骤任务的执行进度""" workflow_id: str session_id: str steps: dict # step_id -> WorkflowStep current_step: str context: dict # 跨步骤共享的上下文 def get_ready_steps(self) -> list: """获取可执行的步骤（依赖已完成）""" ready = [] for step_id, step in self.steps.items(): if step.status != WorkflowStepStatus.PENDING: continue deps_satisfied = all( self.steps[dep].status == WorkflowStepStatus.SUCCESS for dep in step.dependencies ) if deps_satisfied: ready.append(step) return ready def is_complete(self) -> bool: return all( s.status in [WorkflowStepStatus.SUCCESS, WorkflowStepStatus.SKIPPED] for s in self.steps.values() ) 第三层：持久状态（Persistent State） 持久状态跨越会话边界，包括用户偏好、长期记忆和已学习的模式： ...

引言：为什么Agent系统需要事件驱动？ 传统的同步HTTP调用在Agent系统中面临严峻挑战：当一个复杂任务涉及数十次工具调用、记忆检索和多轮对话时，同步调用会导致漫长的等待时间和服务间的强耦合。更关键的是，Agent系统天然需要处理异步、长时运行的任务——一次网页搜索可能耗时数秒，一次代码执行可能需要更长时间。 事件驱动架构通过解耦服务间的时间依赖，让Agent系统能够优雅地处理异步性和并发性。 事件驱动架构的核心模式 发布-订阅模式 ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Tool │───▶│ Message │───▶│ Memory │ │ Service │ │ Bus │ │ Service │ └─────────────┘ │ │ └─────────────┘ │ ┌─────────┐ │ ┌─────────────┐ ┌─────────────┐ │ │ Topic │ │───▶│ LLM │ │ Router │───▶│ │ Router │ │ │ Inference │ │ Service │ │ └─────────┘ │ └─────────────┘ └─────────────┘ └─────────────┘ ▲ ┌────┴────┐ │Event │ │Producer │ └─────────┘ 事件溯源模式 在Agent系统中，事件溯源（Event Sourcing）能够完整记录Agent的思考和决策过程： ...