Agent CI/CD设计：从代码到生产的完整流水线
引言 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/工具）和模型变更。完整的流水线应该包括代码质量检查、自动化测试（单元测试+集成测试+回归测试）、多环境部署、灰度发布和自动回滚。质量门禁贯穿整个流水线，确保每次发布都不会降低系统质量。 ...
