架构设计

引言 容量规划是Agent系统运维中最容易被忽视却最关键的环节。一个容量规划不足的系统会在流量高峰时崩溃，而过度规划则会导致巨大的资源浪费。2026年，随着Agent系统规模的扩大，容量规划已从"拍脑袋估算"进化为基于数据驱动的科学决策过程。 容量规划流程 ┌──────────────────────────────────────────────────────────┐ │ 容量规划流程 │ │ │ │ Step 1: 需求预测 │ │ ┌────────────────────────────────────────────┐ │ │ │ 历史数据分析 → 增长趋势 → 流量预测 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 2: 压测验证 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 设计压测场景 → 执行压测 → 收集性能数据 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 3: 资源建模 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 建立资源消耗模型 → 计算所需资源 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 4: 容量决策 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 成本分析 → 容量方案 → 采购/扩容决策 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 5: 持续监控与调整 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 实时监控 → 对比预测 → 调整容量计划 │ │ │ └────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────┘ 需求预测 import numpy as np from sklearn.linear_model import LinearRegression class DemandForecaster: """需求预测器""" def __init__(self, historical_data: list): self.data = historical_data # [{"date": ..., "qps": ..., "sessions": ...}] def forecast( self, horizon_days: int = 30, confidence_interval: float = 0.95 ) -> dict: """预测未来需求""" # 准备训练数据 X = np.array(range(len(self.data))).reshape(-1, 1) y_qps = np.array([d["qps"] for d in self.data]) y_sessions = np.array([d["sessions"] for d in self.data]) # 训练模型 model_qps = LinearRegression() model_qps.fit(X, y_qps) model_sessions = LinearRegression() model_sessions.fit(X, y_sessions) # 预测 future_X = np.array(range( len(self.data), len(self.data) + horizon_days )).reshape(-1, 1) predicted_qps = model_qps.predict(future_X) predicted_sessions = model_sessions.predict(future_X) # 计算置信区间（简化版） residuals_qps = y_qps - model_qps.predict(X) std_qps = np.std(residuals_qps) z_score = 1.96 if confidence_interval == 0.95 else 1.645 forecast = { "horizon_days": horizon_days, "predicted_qps": predicted_qps.tolist(), "predicted_sessions": predicted_sessions.tolist(), "confidence_interval": { "lower_qps": (predicted_qps - z_score * std_qps).tolist(), "upper_qps": (predicted_qps + z_score * std_qps).tolist(), }, "peak_qps": float(np.max(predicted_qps)), "peak_sessions": float(np.max(predicted_sessions)), } return forecast def forecast_with_seasonality(self, horizon_days: int) -> dict: """考虑季节性的预测（如工作日vs周末）""" # 提取季节性模式 daily_pattern = self._extract_daily_pattern() weekly_pattern = self._extract_weekly_pattern() base_forecast = self.forecast(horizon_days, 0.95) # 应用季节性调整 adjusted = [] for i, qps in enumerate(base_forecast["predicted_qps"]): day_of_week = (len(self.data) + i) % 7 hour_of_day = (len(self.data) + i) % 24 seasonal_factor = ( daily_pattern[hour_of_day] * weekly_pattern[day_of_week] ) adjusted.append(qps * seasonal_factor) base_forecast["predicted_qps_seasonal"] = adjusted return base_forecast 压测方案设计 class CapacityTestPlan: """容量测试方案""" TEST_SCENARIOS = [ { "name": "steady_load", "description": "稳态负载测试", "qps": 100, "duration_minutes": 60, "concurrent_users": 500, }, { "name": "burst_load", "description": "突发负载测试", "qps": 500, "duration_minutes": 10, "concurrent_users": 2000, }, { "name": "ramp_up", "description": "逐步加压测试", "start_qps": 50, "end_qps": 1000, "step_qps": 50, "step_duration_minutes": 5, }, { "name": "spike_test", "description": "尖峰测试", "pattern": "spike", # 快速上升到峰值然后下降 "peak_qps": 2000, "spike_duration_minutes": 5, }, { "name": "soak_test", "description": "浸泡测试（长时间运行）", "qps": 200, "duration_hours": 24, }, ] async def run_capacity_tests(self) -> dict: """执行容量测试套件""" results = {} for scenario in self.TEST_SCENARIOS: logger.info(f"Running scenario: {scenario['name']}") result = await self._execute_test_scenario(scenario) results[scenario["name"]] = result # 如果系统已达极限，停止后续测试 if result["status"] == "system_overloaded": logger.warning(f"System overloaded at {scenario['name']}") break return self._analyze_capacity_results(results) def _analyze_capacity_results(self, results: dict) -> dict: """分析容量测试结果""" analysis = { "max_sustainable_qps": 0, "max_concurrent_users": 0, "bottleneck": None, "resource_utilization_at_max": {}, "recommendations": [], } for scenario_name, result in results.items(): if result["error_rate"] < 0.01: # 错误率<1%视为可持续 analysis["max_sustainable_qps"] = max( analysis["max_sustainable_qps"], result["achieved_qps"] ) analysis["max_concurrent_users"] = max( analysis["max_concurrent_users"], result["concurrent_users"] ) # 记录资源利用率 if result["achieved_qps"] > analysis["max_sustainable_qps"] * 0.9: analysis["resource_utilization_at_max"] = result["resource_utilization"] # 识别瓶颈 util = analysis["resource_utilization_at_max"] if util.get("gpu_utilization", 0) > 0.9: analysis["bottleneck"] = "GPU" analysis["recommendations"].append("Add more GPU nodes") elif util.get("cpu_utilization", 0) > 0.9: analysis["bottleneck"] = "CPU" analysis["recommendations"].append("Add more CPU nodes or optimize code") elif util.get("memory_utilization", 0) > 0.85: analysis["bottleneck"] = "Memory" analysis["recommendations"].append("Increase memory or optimize memory usage") return analysis 资源估算模型 class ResourceEstimator: """资源估算器""" # 基于压测数据的资源消耗基准 BENCHMARKS = { "requests_per_gpu": 50, # 每块GPU每秒处理的请求数 "requests_per_cpu": 10, # 每vCPU每秒处理的请求数（非LLM部分） "memory_per_session_mb": 10, # 每会话内存消耗 "storage_per_user_mb": 100, # 每用户存储消耗 } def estimate_resources( self, predicted_qps: float, predicted_sessions: int, growth_margin: float = 0.3 # 30%增长余量 ) -> dict: """估算所需资源""" # 1. GPU资源（LLM推理） required_qps_with_margin = predicted_qps * (1 + growth_margin) gpu_count = int(np.ceil( required_qps_with_margin / self.BENCHMARKS["requests_per_gpu"] )) # 2. CPU资源（路由、工具执行等） cpu_vcpus = int(np.ceil( required_qps_with_margin / self.BENCHMARKS["requests_per_cpu"] )) # 3. 内存资源 memory_gb = int(np.ceil( (predicted_sessions * self.BENCHMARKS["memory_per_session_mb"]) / 1024 )) + 8 # +8GB系统开销 # 4. 存储资源 storage_tb = (predicted_sessions * self.BENCHMARKS["storage_per_user_mb"]) / (1024 * 1024) estimate = { "compute": { "gpu": { "type": "A100-80GB", "count": gpu_count, "utilization_target": 0.75, # 目标利用率75% }, "cpu": { "vcpus": cpu_vcpus, "type": "8vCPU-16GB", "nodes": int(np.ceil(cpu_vcpus / 8)), } }, "memory": { "total_gb": memory_gb, "per_node_gb": 64, "nodes": int(np.ceil(memory_gb / 64)), }, "storage": { "total_tb": storage_tb * 1.5, # 1.5x用于复制和增长 "type": "SSD", }, "network": { "bandwidth_gbps": 10, "load_balancers": int(np.ceil(gpu_count / 8)), } } return estimate def estimate_cost( self, resources: dict, cloud_provider: str = "aws" ) -> dict: """估算成本""" PRICING = { "aws": { "a100_gpu_hour": 4.10, # A100每小时 "ec2_8vcpu_hour": 0.40, # 8vCPU实例 "memory_gb_month": 0.005, # $/GB/月 "storage_tb_month": 50, # $/TB/月 "network_gb": 0.09, # $/GB流量 }, "azure": { "a100_gpu_hour": 3.80, "vm_8vcpu_hour": 0.35, "memory_gb_month": 0.004, "storage_tb_month": 45, "network_gb": 0.08, } } pricing = PRICING[cloud_provider] # 计算月成本 gpu_cost = resources["compute"]["gpu"]["count"] * pricing["a100_gpu_hour"] * 24 * 30 cpu_cost = resources["compute"]["cpu"]["nodes"] * pricing["ec2_8vcpu_hour"] * 24 * 30 memory_cost = resources["memory"]["total_gb"] * pricing["memory_gb_month"] * 30 storage_cost = resources["storage"]["total_tb"] * pricing["storage_tb_month"] total_monthly = gpu_cost + cpu_cost + memory_cost + storage_cost return { "cloud_provider": cloud_provider, "resources": resources, "cost_breakdown": { "gpu": gpu_cost, "cpu": cpu_cost, "memory": memory_cost, "storage": storage_cost, }, "total_monthly_usd": total_monthly, "total_annual_usd": total_monthly * 12, "cost_per_request_usd": total_monthly / (resources["compute"]["gpu"]["count"] * self.BENCHMARKS["requests_per_gpu"] * 24 * 30), } 容量规划报告 class CapacityReport: """容量规划报告生成器""" def generate( self, forecast: dict, capacity_test: dict, resource_estimate: dict, cost_estimate: dict ) -> str: """生成容量规划报告""" report = f""" # Agent系统容量规划报告 **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')} --- ## 1. 需求预测 ### 未来{forecast["horizon_days"]}天预测 - **峰值QPS**: {forecast["peak_qps"]:.1f} - **峰值并发会话**: {forecast["peak_sessions"]:.0f} ### 预测曲线 QPS ^ | * | * | * | * | * |*_______ +————————> 时间 (天) 0 {forecast[“horizon_days”]//4} {forecast[“horizon_days”]//2} {forecast[“horizon_days”]*3//4} {forecast[“horizon_days”]} ...

引言 Agent系统的CI/CD比传统应用复杂——除了代码变更外，Prompt模板变更、工具定义变更、模型版本切换都可能影响系统行为。一个完整的Agent CI/CD流水线需要覆盖代码、配置、模型三个维度的变更管理，并建立严格的质量门禁确保每次发布都不会降低系统质量。 CI/CD流水线全景 ┌─────────────────────────────────────────────────────────────┐ │ Agent CI/CD Pipeline │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Code │ │ Build │ │ Test │ │ Deploy │ │ │ │ Commit │──▶│ & Push │──▶│ & QA │──▶│ to Prod │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Lint & │ │ Image │ │ Unit │ │ Staging │ │ │ │ Format │ │ Build │ │ Tests │ │ Deploy │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ ┌──────────┐ ┌──────────┐│ │ │ Integration│ │ Canary ││ │ │ Tests │ │ Deploy ││ │ └──────────┘ └──────────┘│ │ │ │ │ ┌──────────┐ │ │ │ GA Deploy│ │ │ └──────────┘ │ └─────────────────────────────────────────────────────────────┘ 代码提交与构建 # .github/workflows/ci.yml name: Agent CI Pipeline on: push: branches: [main, develop] pull_request: branches: [main] jobs: lint: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' - name: Install dependencies run: pip install flake8 black mypy - name: Lint with flake8 run: flake8 agent/ --max-line-length=120 --ignore=E203,W503 - name: Format check with black run: black --check agent/ - name: Type check with mypy run: mypy agent/ --ignore-missing-imports build: needs: lint runs-on: self-hosted # 需要Docker支持 steps: - uses: actions/checkout@v3 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v2 - name: Log in to Container Registry uses: docker/login-action@v2 with: registry: ${{ secrets.REGISTRY_URL }} username: ${{ secrets.REGISTRY_USER }} password: ${{ secrets.REGISTRY_PASSWORD }} - name: Build and Push Docker Image uses: docker/build-push-action@v4 with: context: . push: true tags: | ${{ secrets.REGISTRY_URL }}/agent-service:${{ github.sha }} ${{ secrets.REGISTRY_URL }}/agent-service:latest cache-from: type=gha cache-to: type=gha,mode=max - name: Scan Image for Vulnerabilities uses: aquasecurity/trivy-action@master with: image-ref: ${{ secrets.REGISTRY_URL }}/agent-service:${{ github.sha }} format: 'sarif' output: 'trivy-results.sarif' - name: Upload Trivy scan results uses: github/codeql-action/upload-sarif@v2 with: sarif_file: 'trivy-results.sarif' 自动化测试 test: needs: build runs-on: self-hosted services: redis: image: redis:7 ports: - 6379:6379 postgres: image: postgres:15 env: POSTGRES_PASSWORD: test ports: - 5432:5432 qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 steps: - uses: actions/checkout@v3 - name: Run Unit Tests run: | pytest tests/unit/ -v --cov=agent --cov-report=xml --junitxml=junit-unit.xml - name: Run Integration Tests env: LLM_MOCK: "true" # 使用Mock LLM run: | pytest tests/integration/ -v --junitxml=junit-integration.xml - name: Run Agent-Specific Tests run: | # Prompt测试 python -m pytest tests/prompt/ -v # 工具调用测试 python -m pytest tests/tools/ -v # 质量回归测试 python tests/regression/run_regression.py \ --baseline=baseline.json \ --report=regression-report.json - name: Upload Test Results if: always() uses: actions/upload-artifact@v3 with: name: test-results path: | coverage.xml junit-*.xml regression-report.json - name: Check Quality Gate run: | python scripts/check_quality_gate.py \ --coverage-report=coverage.xml \ --min-coverage=80 \ --regression-report=regression-report.json \ --max-regressions=0 环境管理 class EnvironmentManager: """环境管理器""" ENVIRONMENTS = { "dev": { "replicas": 1, "model": "gpt-4o-mini", "quality_gate": {"min_quality": 0.7}, }, "staging": { "replicas": 3, "model": "gpt-4o", "quality_gate": {"min_quality": 0.8}, }, "prod": { "replicas": 10, "model": "gpt-4o", "quality_gate": {"min_quality": 0.85}, } } async def deploy_to_environment( self, environment: str, image_tag: str, config: dict = None ): """部署到指定环境""" env_config = self.ENVIRONMENTS[environment] deploy_config = {**env_config, **(config or {})} # 1. 更新K8s Deployment await self.k8s_client.apply_deployment({ "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": f"agent-service-{environment}", "namespace": f"agent-{environment}" }, "spec": { "replicas": deploy_config["replicas"], "template": { "spec": { "containers": [{ "name": "agent", "image": f"{self.registry}/{image_tag}", "env": [ {"name": "MODEL_NAME", "value": deploy_config["model"]}, {"name": "ENVIRONMENT", "value": environment}, ] }] } } } }) # 2. 等待部署完成 await self._wait_for_rollout( f"agent-service-{environment}", timeout=300 ) # 3. 运行冒烟测试 await self._run_smoke_tests(environment) # 4. 运行质量门禁 quality_result = await self._run_quality_gate(environment, deploy_config["quality_gate"]) if not quality_result["passed"]: logger.error(f"Quality gate failed for {environment}") await self._rollback(environment, image_tag) raise QualityGateFailed(quality_result["details"]) logger.info(f"Successfully deployed to {environment}") 灰度发布 # Argo Rollouts配置 apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: agent-service spec: replicas: 10 strategy: canary: canaryService: agent-service-canary stableService: agent-service-stable # 分析阶段 analysis: templates: - templateName: agent-quality-analysis args: - name: service-name value: agent-service-canary # 渐进式发布 steps: - setWeight: 5 - pause: {duration: 10m} - setWeight: 20 - pause: {duration: 15m} - setWeight: 50 - pause: {duration: 20m} - setWeight: 100 # 流量路由 trafficRouting: istio: virtualService: name: agent-service-vs destinationRule: name: agent-service-dr 自动化回滚 class AutoRollbackManager: """自动回滚管理器""" async def monitor_and_rollback(self, rollout_name: str): """监控发布并自动回滚""" while True: # 获取Rollout状态 rollout = await self.k8s_client.get_rollout(rollout_name) if rollout["status"]["phase"] == "Degraded": logger.warning(f"Rollout {rollout_name} degraded, initiating rollback") await self._rollback(rollout_name) break # 检查质量指标 quality = await self._check_quality(rollout_name) if quality["error_rate"] > 0.05: logger.warning(f"Error rate {quality['error_rate']:.1%} > 5%, rolling back") await self._rollback(rollout_name) break if quality["quality_score"] < 0.8: logger.warning(f"Quality score {quality['quality_score']:.2f} < 0.8, rolling back") await self._rollback(rollout_name) break await asyncio.sleep(30) # 30秒检查一次 async def _rollback(self, rollout_name: str): """执行回滚""" await self.k8s_client.rollback_rollout( rollout_name, to_revision="previous" ) # 发送通知 await self.notifier.send( channel="slack:#alerts", message=f"⚠️ Auto-rollback triggered for {rollout_name}" ) 完整的CD流水线 # .github/workflows/cd.yml name: Agent CD Pipeline on: workflow_run: workflows: ["Agent CI Pipeline"] types: [completed] branches: [main] jobs: deploy-staging: if: ${{ github.event.workflow_run.conclusion == 'success' }} runs-on: self-hosted steps: - name: Deploy to Staging run: | python scripts/deploy.py \ --environment=staging \ --image-tag=${{ github.sha }} - name: Run Staging Tests run: | python scripts/run_e2e_tests.py --environment=staging - name: Notify Deployment uses: 8398a7/action-slack@v3 with: status: ${{ job.status }} text: "Deployed to Staging: ${{ github.sha }}" deploy-prod: needs: deploy-staging runs-on: self-hosted environment: production # 需要手动批准 steps: - name: Deploy to Production (Canary) run: | kubectl argo rollouts promote agent-service - name: Monitor Canary run: | python scripts/monitor_canary.py \ --rollout=agent-service \ --duration=30m \ --auto-rollback=true - name: Promote Canary if: success() run: | kubectl argo rollouts promote agent-service - name: Create GitHub Release uses: actions/create-release@v1 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} with: tag_name: v${{ github.run_number }} release_name: Release v${{ github.run_number }} body: | ## Changes ${{ steps.changelog.outputs.changelog }} ## Deployment - Staging: ✅ - Production: ✅ (Canary 100%) draft: false prerelease: false 总结 Agent CI/CD流水线的核心挑战在于三层面的变更管理：代码变更、配置变更（Prompt/工具）和模型变更。完整的流水线应该包括代码质量检查、自动化测试（单元测试+集成测试+回归测试）、多环境部署、灰度发布和自动回滚。质量门禁贯穿整个流水线，确保每次发布都不会降低系统质量。 ...

引言 LLM的本质是非确定性的——同样的输入可能产生不同的输出。这给Agent系统的测试带来了根本性挑战：如何验证Agent的"正确性"？如何检测"回归"？回放测试（Replay Testing）通过将真实请求重新执行并与历史结果对比，为Agent系统提供了一种实用且有效的验证手段。 2026年，回放测试已成为Agent系统质量保障的核心手段。本文系统介绍如何设计和实施Agent回放测试体系。 回放测试原理 ┌──────────────────────────────────────────────────────┐ │ 回放测试流程 │ │ │ │ 生产流量 ──▶ 录制 (Record) │ │ │ │ │ ▼ │ │ 测试数据集 (Test Set) │ │ │ │ │ ▼ │ │ 新版本 ──▶ 回放 (Replay) ──▶ 结果对比 ──▶ 报告 │ │ │ │ │ │ ▼ ▼ │ │ 差异分析 质量门禁 │ │ │ └──────────────────────────────────────────────────────┘ 录制生产流量 class ProductionRecorder: """生产流量录制器""" def __init__(self, storage_client): self.storage = storage_client self.sampling_rate = 0.01 # 1%采样 async def record_request( self, session_id: str, request: dict, response: dict, metadata: dict ): """录制请求和响应""" import random if random.random() > self.sampling_rate: return recording = { "session_id": session_id, "timestamp": datetime.now().isoformat(), "request": { "input": request["input"], "context": request.get("context", {}), "config": request.get("config", {}), }, "response": { "output": response["output"], "tool_calls": response.get("tool_calls", []), "tokens_used": response.get("usage", {}), "latency_ms": response.get("latency_ms"), }, "metadata": { "model": metadata.get("model"), "prompt_version": metadata.get("prompt_version"), "quality_score": metadata.get("quality_score"), "user_feedback": metadata.get("user_feedback"), }, "recording_version": "1.0" } # 存储到对象存储 key = f"recordings/{datetime.now().strftime('%Y%m%d')}/{session_id}.json" await self.storage.put(key, json.dumps(recording, ensure_ascii=False)) 回放执行 class ReplayExecutor: """回放执行器""" async def replay_test_set( self, test_set: list, config: dict ) -> dict: """回放测试集""" results = [] for i, test_case in enumerate(test_set): logger.info(f"Replaying test case {i+1}/{len(test_set)}") try: result = await self._replay_single(test_case, config) results.append(result) except Exception as e: logger.error(f"Replay failed for case {i+1}: {e}") results.append({ "test_case_id": test_case.get("id"), "status": "error", "error": str(e) }) # 汇总分析 return self._analyze_results(results) async def _replay_single( self, test_case: dict, config: dict ) -> dict: """回放单个测试用例""" # 使用录制时的配置（或覆盖） replay_config = {**test_case["request"]["config"], **config} # 执行请求 start = time.monotonic() response = await self.agent.process( test_case["request"]["input"], context=test_case["request"]["context"], config=replay_config ) latency_ms = (time.monotonic() - start) * 1000 # 对比结果 comparison = self._compare_responses( expected=test_case["response"], actual=response ) return { "test_case_id": test_case.get("id"), "status": "pass" if comparison["passed"] else "fail", "comparison": comparison, "actual_response": response, "latency_ms": latency_ms, } def _compare_responses(self, expected: dict, actual: dict) -> dict: """对比预期和实际响应""" comparison = { "passed": True, "checks": [] } # 检查1：工具调用是否相同 expected_tools = [t["name"] for t in expected.get("tool_calls", [])] actual_tools = [t["name"] for t in actual.get("tool_calls", [])] tools_match = set(expected_tools) == set(actual_tools) comparison["checks"].append({ "name": "tool_calls_match", "passed": tools_match, "expected": expected_tools, "actual": actual_tools }) if not tools_match: comparison["passed"] = False # 检查2：响应质量是否相似（语义相似度） similarity = self._compute_similarity( expected["output"], actual["output"] ) quality_ok = similarity > 0.85 # 85%相似度阈值 comparison["checks"].append({ "name": "response_quality", "passed": quality_ok, "similarity": similarity, "threshold": 0.85 }) if not quality_ok: comparison["passed"] = False # 检查3：Token消耗是否在合理范围 token_ratio = actual.get("usage", {}).get("total_tokens", 0) / \ max(expected.get("tokens_used", {}).get("total_tokens", 1), 1) token_ok = 0.5 < token_ratio < 2.0 # Token消耗在0.5x-2x之间 comparison["checks"].append({ "name": "token_consumption", "passed": token_ok, "ratio": token_ratio, "threshold": "0.5-2.0" }) return comparison 快照测试 class SnapshotTesting: """快照测试——保存首次运行结果为快照，后续运行对比快照""" def __init__(self, snapshot_dir: str): self.snapshot_dir = snapshot_dir os.makedirs(snapshot_dir, exist_ok=True) async def run_with_snapshot( self, test_name: str, test_fn: callable, update_snapshot: bool = False ) -> dict: """运行快照测试""" snapshot_file = os.path.join( self.snapshot_dir, f"{test_name}.snapshot.json" ) # 执行测试 actual = await test_fn() if update_snapshot: # 更新快照 with open(snapshot_file, "w") as f: json.dump(actual, f, indent=2, ensure_ascii=False) return {"status": "snapshot_updated", "result": actual} # 对比快照 if not os.path.exists(snapshot_file): raise FileNotFoundError( f"Snapshot not found: {snapshot_file}. " f"Run with update_snapshot=True to create." ) with open(snapshot_file) as f: expected = json.load(f) diff = self._deep_diff(expected, actual) if diff: return { "status": "fail", "diff": diff, "expected": expected, "actual": actual } else: return {"status": "pass", "result": actual} def _deep_diff(self, expected, actual, path: str = "") -> list: """深度对比，返回差异列表""" diffs = [] if isinstance(expected, dict) and isinstance(actual, dict): for key in set(list(expected.keys()) + list(actual.keys())): new_path = f"{path}.{key}" if path else key if key not in actual: diffs.append(f"Missing in actual: {new_path}") elif key not in expected: diffs.append(f"Extra in actual: {new_path}") else: diffs.extend( self._deep_diff(expected[key], actual[key], new_path) ) elif isinstance(expected, list) and isinstance(actual, list): if len(expected) != len(actual): diffs.append( f"List length mismatch at {path}: " f"expected {len(expected)}, actual {len(actual)}" ) else: for i, (e, a) in enumerate(zip(expected, actual)): diffs.extend(self._deep_diff(e, a, f"{path}[{i}]")) else: # 标量对比（对LLM输出用模糊匹配） if self._is_llm_output(path): similarity = self._compute_similarity(str(expected), str(actual)) if similarity < 0.9: diffs.append( f"Semantic difference at {path}: " f"similarity={similarity:.2f}" ) else: if expected != actual: diffs.append( f"Value mismatch at {path}: " f"expected={expected}, actual={actual}" ) return diffs 回归测试框架 class RegressionTestSuite: """回归测试套件""" def __init__(self): self.test_cases = [] self.quality_thresholds = { "response_similarity": 0.85, "tool_call_accuracy": 0.95, "latency_increase_max": 1.2, # 延迟最多增加20% "token_increase_max": 1.3, # Token最多增加30% } def add_test_case(self, test_case: dict): """添加测试用例""" self.test_cases.append(test_case) async def run_regression_test(self, version: str) -> dict: """运行回归测试""" results = { "version": version, "total_cases": len(self.test_cases), "passed": 0, "failed": 0, "regressions": [], "improvements": [], } for test_case in self.test_cases: # 获取基线结果 baseline = await self._get_baseline(test_case["id"]) # 执行测试 actual = await self._execute_test(test_case) # 对比 comparison = self._compare_with_baseline(baseline, actual) if comparison["is_regression"]: results["failed"] += 1 results["regressions"].append({ "test_case": test_case["id"], "regression_type": comparison["regression_type"], "details": comparison["details"] }) elif comparison["is_improvement"]: results["improvements"].append({ "test_case": test_case["id"], "improvement_type": comparison["improvement_type"], }) else: results["passed"] += 1 return results def _compare_with_baseline(self, baseline: dict, actual: dict) -> dict: """与基线对比""" issues = [] # 质量回归 if actual["quality_score"] < baseline["quality_score"] * 0.95: issues.append({ "type": "quality_regression", "baseline": baseline["quality_score"], "actual": actual["quality_score"], }) # 延迟回归 latency_ratio = actual["latency_ms"] / baseline["latency_ms"] if latency_ratio > self.quality_thresholds["latency_increase_max"]: issues.append({ "type": "latency_regression", "baseline": baseline["latency_ms"], "actual": actual["latency_ms"], "ratio": latency_ratio, }) # Token回归 token_ratio = actual["tokens_used"] / baseline["tokens_used"] if token_ratio > self.quality_thresholds["token_increase_max"]: issues.append({ "type": "token_regression", "baseline": baseline["tokens_used"], "actual": actual["tokens_used"], "ratio": token_ratio, }) return { "is_regression": len(issues) > 0, "is_improvement": actual["quality_score"] > baseline["quality_score"] * 1.05, "issues": issues, } CI/CD集成 # .github/workflows/regression-test.yml name: Agent Regression Test on: pull_request: branches: [main] jobs: regression-test: runs-on: self-hosted steps: - uses: actions/checkout@v3 - name: Setup Test Environment run: | docker-compose -f docker-compose-test.yml up -d sleep 30 # 等待服务就绪 - name: Run Regression Tests id: regression run: | python -m pytest tests/regression/ \ --baseline=baseline.json \ --output=regression-report.json \ --junitxml=junit.xml - name: Check Quality Gate run: | python scripts/check_quality_gate.py \ --report=regression-report.json \ --max-regressions=5 \ --min-quality-score=0.85 - name: Upload Test Results if: always() uses: actions/upload-artifact@v3 with: name: regression-test-results path: | regression-report.json junit.xml - name: Comment PR if: always() uses: actions/github-script@v6 with: script: | const report = require('./regression-report.json'); const comment = ` ## Regression Test Results - Total: ${report.total_cases} - Passed: ${report.passed} - Failed: ${report.failed} - Regressions: ${report.regressions.length} ${report.regressions.length > 0 ? '⚠️ Regressions detected!' : '✅ No regressions'} `; github.rest.issues.createComment({ issue_number: context.issue.number, body: comment }); 总结 回放测试为Agent系统提供了一种实用的验证手段——通过录制生产流量并在新版本上回放，可以有效检测功能回归和质量下降。快照测试通过保存首次运行结果作为基准，简化了测试用例的创建。回归测试套件则系统性地验证新版本在质量、延迟、Token消耗等维度上是否出现退化。 ...

引言 2026年，MoE（Mixture of Experts）架构已经成为大语言模型的主流选择。三大开源旗舰——DeepSeek V4、Qwen3.5和Llama 4——都采用了MoE架构，但实现方式各有不同。本文将深入对比这三者的MoE架构设计，分析其技术差异、性能表现和工程影响。 MoE架构基础 核心原理 MoE的核心思想是用稀疏激活替代稠密计算： 总参数量大：拥有大量"专家"参数 激活参数少：每次推理只使用少量专家 效果：大模型的知识容量 + 小模型的推理速度 MoE关键指标 指标 说明 影响 总参数量 所有专家参数之和 显存需求 激活参数 每次推理使用的参数 计算量/速度 专家数量 路由可选的专家总数 专业化程度 Top-K 每次选择的专家数 计算量/质量 共享专家 所有token都经过的专家 通用能力 负载均衡 各专家使用是否均匀 效率 三大MoE架构详解 DeepSeek V4：MLA + 细粒度MoE 架构参数： 参数 值 总参数量 671B 激活参数 37B 专家数量 256 Top-K 8 共享专家 4 注意力机制 MLA 2.0 上下文 256K 核心创新： 1. MLA 2.0（多头潜在注意力） DeepSeek V4的核心创新是MLA 2.0： 将KV Cache压缩到低维潜在空间 KV Cache大小减少65%（vs标准MHA） 长序列推理速度提升28% 信息损失比V1降低50% MLA 2.0的KV Cache对比： ...

引言 Agent系统的性能基准测试比传统Web应用复杂得多——响应延迟不仅取决于基础设施，还受LLM推理速度、工具调用延迟、Prompt长度等多重因素影响。没有经过充分基准测试的Agent系统，就像没有经过碰撞测试的自动驾驶汽车——上路后迟早会出事。 2026年，Agent性能测试已形成标准化的方法论。本文系统介绍如何对Agent系统进行全面、准确的性能基准测试。 性能测试维度 Agent性能测试矩阵 │ ├── 吞吐量测试 (Throughput) │ ├── 最大QPS │ ├── 可持续QPS │ └── QPS vs 延迟曲线 │ ├── 延迟测试 (Latency) │ ├── P50/P90/P99延迟 │ ├── 各阶段延迟分解 │ └── 长尾延迟分析 │ ├── 并发测试 (Concurrency) │ ├── 最大并发会话数 │ ├── 并发下质量保持 │ └── 资源竞争分析 │ └── 压力测试 (Stress) ├── 极限负载 ├── 故障恢复时间 └── 降级策略验证 测试环境搭建 # docker-compose-benchmark.yml version: '3.8' services: load-generator: image: agent/benchmark:latest environment: - TARGET_URL=http://agent-service:8080 - CONCURRENT_USERS=100 - TEST_DURATION=300s depends_on: - agent-service - llm-mock agent-service: image: agent/service:latest deploy: resources: limits: cpus: '4' memory: 8G llm-mock: image: agent/llm-mock:latest environment: - MOCK_LATENCY_MS=500 # 模拟LLM延迟 - MOCK_ERROR_RATE=0.01 prometheus: image: prom/prometheus:latest grafana: image: grafana/grafana:latest 吞吐量测试 import asyncio import time import statistics from dataclasses import dataclass @dataclass class BenchmarkConfig: """基准测试配置""" target_qps: int duration_seconds: int concurrent_requests: int test_cases: list warmup_seconds: int = 30 @dataclass class BenchmarkResult: """基准测试结果""" total_requests: int successful_requests: int failed_requests: int avg_latency_ms: float p50_latency_ms: float p90_latency_ms: float p99_latency_ms: float min_latency_ms: float max_latency_ms: float qps: float error_rate: float tokens_per_second: float class ThroughputBenchmark: """吞吐量基准测试""" async def run(self, config: BenchmarkConfig) -> BenchmarkResult: """运行吞吐量测试""" # 预热 await self._warmup(config.warmup_seconds) # 主测试 latencies = [] successes = 0 failures = 0 total_tokens = 0 start_time = time.monotonic() end_time = start_time + config.duration_seconds # 创建并发请求 tasks = [] for i in range(config.concurrent_requests): task = asyncio.create_task( self._request_worker( config, end_time, latencies, lambda s: nonlocal(successes) or successes++, lambda f: nonlocal(failures) or failures++, lambda t: nonlocal(total_tokens) or total_tokens += t ) ) tasks.append(task) # 等待完成 await asyncio.gather(*tasks) actual_duration = time.monotonic() - start_time # 计算结果 sorted_latencies = sorted(latencies) return BenchmarkResult( total_requests=len(latencies), successful_requests=successes, failed_requests=failures, avg_latency_ms=statistics.mean(latencies), p50_latency_ms=self._percentile(sorted_latencies, 0.5), p90_latency_ms=self._percentile(sorted_latencies, 0.9), p99_latency_ms=self._percentile(sorted_latencies, 0.99), min_latency_ms=min(latencies), max_latency_ms=max(latencies), qps=len(latencies) / actual_duration, error_rate=failures / len(latencies) if latencies else 0, tokens_per_second=total_tokens / actual_duration ) async def _request_worker( self, config: BenchmarkConfig, end_time: float, latencies: list, on_success: callable, on_failure: callable, on_tokens: callable ): """请求工作线程""" while time.monotonic() < end_time: test_case = random.choice(config.test_cases) start = time.monotonic() try: response = await self.client.request(test_case["input"]) latency = (time.monotonic() - start) * 1000 latencies.append(latency) on_success() if "usage" in response: on_tokens(response["usage"]["total_tokens"]) except Exception as e: latencies.append(30000) # 超时记录为30s on_failure() 延迟分解测试 class LatencyBreakdownBenchmark: """延迟分解测试""" async def measure_latency_breakdown( self, session_id: str, user_input: str ) -> dict: """测量各阶段延迟""" breakdown = { "total_ms": 0, "stages": {} } # 使用链路追踪获取各阶段延迟 trace = await self.tracer.get_trace_for_session(session_id) if trace: for span in trace.spans: stage_name = span.operation_name duration_ms = span.duration_ms breakdown["stages"][stage_name] = { "duration_ms": duration_ms, "percentage": 0, # 稍后计算 "service": span.process.service_name, } # 计算百分比 total = sum(s["duration_ms"] for s in breakdown["stages"].values()) breakdown["total_ms"] = total for stage in breakdown["stages"].values(): stage["percentage"] = stage["duration_ms"] / total if total > 0 else 0 return breakdown def print_breakdown(self, breakdown: dict): """打印延迟分解""" print(f"\n{'='*60}") print(f"Total Latency: {breakdown['total_ms']:.1f}ms") print(f"{'='*60}") print(f"{'Stage':<30} {'Latency(ms)':<15} {'Percentage':<10}") print(f"{'-'*60}") for stage_name, data in sorted( breakdown["stages"].items(), key=lambda x: x[1]["duration_ms"], reverse=True ): print( f"{stage_name:<30} " f"{data['duration_ms']:<15.1f} " f"{data['percentage']*100:<10.1f}%" ) 并发压力测试 class ConcurrencyBenchmark: """并发压力测试""" async def run_concurrency_test( self, max_concurrent: int, step: int = 10, hold_seconds: int = 60 ) -> dict: """逐步增加并发数，测试系统极限""" results = [] for concurrent in range(step, max_concurrent + 1, step): print(f"\nTesting with {concurrent} concurrent users...") config = BenchmarkConfig( target_qps=concurrent * 2, # 每人2 QPS duration_seconds=hold_seconds, concurrent_requests=concurrent, test_cases=self._get_test_cases() ) result = await self.benchmark.run(config) results.append({ "concurrent_users": concurrent, "qps": result.qps, "avg_latency_ms": result.avg_latency_ms, "p99_latency_ms": result.p99_latency_ms, "error_rate": result.error_rate, "tokens_per_second": result.tokens_per_second, }) # 如果错误率超过5%，停止测试 if result.error_rate > 0.05: print(f"Stopping: error rate {result.error_rate:.1%} > 5%") break return { "test_results": results, "max_sustainable_concurrent": self._find_max_sustainable(results), "performance_curve": self._generate_curve(results), } def _find_max_sustainable(self, results: list) -> int: """找到可持续的最大并发数""" for r in results: if r["error_rate"] > 0.01 or r["p99_latency_ms"] > 5000: return r["concurrent_users"] - 10 return results[-1]["concurrent_users"] if results else 0 测试结果解读 class BenchmarkReport: """基准测试报告生成器""" def generate_report(self, results: dict) -> str: """生成测试报告""" report = f""" # Agent性能基准测试报告 ## 测试概述 - 测试时间: {results["test_time"]} - 测试版本: {results["version"]} - 测试环境: {results["environment"]} ## 核心指标 ### 吞吐量 - 最大QPS: {results["max_qps"]} - 可持续QPS: {results["sustainable_qps"]} - QPS vs 并发曲线: [见图表] ### 延迟 - P50延迟: {results["p50_latency_ms"]}ms - P90延迟: {results["p90_latency_ms"]}ms - P99延迟: {results["p99_latency_ms"]}ms - 平均延迟: {results["avg_latency_ms"]}ms ### 并发 - 最大并发会话: {results["max_concurrent"]} - 并发下平均质量评分: {results["quality_at_max_concurrent"]} ### 资源消耗 - 单请求平均Token消耗: {results["avg_tokens"]} - GPU利用率: {results["gpu_utilization"]} - CPU利用率: {results["cpu_utilization"]} ## 延迟分解 {self._format_latency_breakdown(results["latency_breakdown"])} ## 瓶颈分析 {results["bottleneck_analysis"]} ## 优化建议 {results["optimization_recommendations"]} """ return report 持续基准测试 # .github/workflows/benchmark.yml name: Performance Benchmark on: push: branches: [main] pull_request: branches: [main] jobs: benchmark: runs-on: self-hosted # 需要稳定的硬件 steps: - uses: actions/checkout@v3 - name: Run Benchmark run: | docker-compose -f docker-compose-benchmark.yml up --abort-on-container-exit - name: Compare with Baseline run: | python scripts/compare_benchmark.py \ --current results.json \ --baseline baseline.json \ --threshold 0.1 # 允许10%回归 - name: Upload Results if: always() uses: actions/upload-artifact@v3 with: name: benchmark-results path: results/ 总结 Agent性能基准测试需要从吞吐量、延迟、并发、压力四个维度全面评估。延迟分解测试能够精准定位性能瓶颈——是LLM推理慢、工具调用慢还是向量检索慢。持续基准测试确保每次代码变更都不会引起性能回归。 ...

引言 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的跨服务传播让分布式追踪成为可能。智能采样策略确保在控制成本的同时不丢失关键信息。 ...