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

核心原则：CI/CD的终极目标是让发布成为"无事件"——工程师可以自信地发布，用户不会感知到发布。自动化测试和质量门禁是实现这一目标的基础。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

引言#

CI/CD流水线全景#

代码提交与构建#

自动化测试#

环境管理#

灰度发布#

自动化回滚#

完整的CD流水线#

总结#

核心原则：CI/CD的终极目标是让发布成为"无事件"——工程师可以自信地发布，用户不会感知到发布。自动化测试和质量门禁是实现这一目标的基础。#

加入讨论#

📌 相关推荐

引言