Agent 版本管理：Prompt/工具/模型的灰度发布

┌──────────────────────────────────────────────────────┐
│            Agent 变更类型与风险矩阵                    │
├──────────────┬──────────┬──────────┬─────────────────┤
│   变更类型    │ 频率     │ 风险等级  │   影响范围      │
├──────────────┼──────────┼──────────┼─────────────────┤
│ Prompt 修改  │ 每周     │ 中-高    │ 输出质量/行为   │
│ 工具更新     │ 每月     │ 中       │ 工具调用/结果   │
│ 模型升级     │ 每季度   │ 高       │ 全局行为变化    │
│ System配置   │ 每周     │ 低-中    │ 性能/限制       │
│ Few-shot示例 │ 每月     │ 中       │ 输出风格/格式   │
│ 工作流变更   │ 每月     │ 高       │ 执行路径/延迟   │
└──────────────┴──────────┴──────────┴─────────────────┘

from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class VersionStatus(Enum):
    DRAFT = "draft"
    STAGING = "staging"
    CANARY = "canary"      # 灰度中
    LIVE = "live"           # 全量发布
    ROLLED_BACK = "rolled_back"
    ARCHIVED = "archived"

@dataclass
class AgentVersion:
    version_id: str             # 如 "v1.3.2"
    agent_name: str
    prompt_version: str         # 引用 PromptVersion
    tools_version: str          # 引用 ToolsVersion
    model_config_version: str   # 引用 ModelConfigVersion
    workflow_version: str       # 引用 WorkflowVersion
    status: VersionStatus
    created_at: datetime
    rollout_percentage: float   # 灰度比例 0-100
    parent_version: str | None  # 上一个版本
    changelog: str
    metrics: dict               # 质量指标快照


@dataclass
class PromptVersion:
    version_id: str             # "prompt-v3.1"
    system_prompt: str
    few_shot_examples: list[dict]
    temperature: float
    max_tokens: int
    output_format: str | None
    parent_version: str | None
    diff: str                   # 与上一版本的 diff
    created_at: datetime
    author: str
    review_status: str          # pending/approved/rejected


@dataclass
class ToolsVersion:
    version_id: str
    tools: list[ToolConfig]     # 工具列表及配置
    added: list[str]            # 新增工具
    removed: list[str]          # 移除工具
    modified: list[str]         # 修改工具
    parent_version: str | None


@dataclass
class ModelConfigVersion:
    version_id: str
    primary_model: str          # 模型名称+版本
    fallback_model: str
    parameters: dict            # temperature, top_p, etc.
    parent_version: str | None

class AgentVersionRepository:
    """Agent 版本仓库"""
    
    def __init__(self, db, storage):
        self.db = db
        self.storage = storage  # 对象存储（存 Prompt 大文本等）
    
    async def create_version(
        self,
        agent_name: str,
        changes: VersionChanges,
        parent_version: str | None = None
    ) -> AgentVersion:
        """创建新版本"""
        
        # 1. 解析变更
        prompt_v = await self._version_prompt(changes.prompt, parent_version)
        tools_v = await self._version_tools(changes.tools, parent_version)
        model_v = await self._version_model(changes.model_config, parent_version)
        workflow_v = await self._version_workflow(changes.workflow, parent_version)
        
        # 2. 生成版本号
        new_version_id = self._bump_version(parent_version, changes.severity)
        
        # 3. 计算差异
        diffs = await self._compute_diffs(parent_version, changes)
        
        # 4. 创建版本记录
        version = AgentVersion(
            version_id=new_version_id,
            agent_name=agent_name,
            prompt_version=prompt_v,
            tools_version=tools_v,
            model_config_version=model_v,
            workflow_version=workflow_v,
            status=VersionStatus.DRAFT,
            created_at=datetime.now(),
            rollout_percentage=0,
            parent_version=parent_version,
            changelog=changes.description,
            metrics={}
        )
        
        await self.db.save(version)
        return version
    
    def _bump_version(self, parent: str | None, severity: str) -> str:
        """语义化版本号"""
        if not parent:
            return "v1.0.0"
        
        major, minor, patch = parent.lstrip("v").split(".")
        
        if severity == "major":  # 模型变更、工作流重构
            return f"v{int(major)+1}.0.0"
        elif severity == "minor":  # 新增工具、Prompt 大改
            return f"v{major}.{int(minor)+1}.0"
        else:  # patch: 小修改
            return f"v{major}.{int(minor)}.{int(patch)+1}"


class PromptDiffCalculator:
    """Prompt 差异计算器"""
    
    def calculate(self, old: str, new: str) -> PromptDiff:
        lines_old = old.splitlines()
        lines_new = new.splitlines()
        
        import difflib
        diff = list(difflib.unified_diff(
            lines_old, lines_new,
            lineterm='',
            n=3
        ))
        
        added = sum(1 for l in diff if l.startswith('+') and not l.startswith('+++'))
        removed = sum(1 for l in diff if l.startswith('-') and not l.startswith('---'))
        
        return PromptDiff(
            diff_text='\n'.join(diff),
            lines_added=added,
            lines_removed=removed,
            change_ratio=(added + removed) / max(len(lines_old), 1)
        )

class RolloutEngine:
    """灰度发布引擎"""
    
    ROLLOUT_STAGES = [
        {"percentage": 0, "duration_minutes": 0, "label": "init"},
        {"percentage": 1, "duration_minutes": 30, "label": "canary"},
        {"percentage": 5, "duration_minutes": 60, "label": "early"},
        {"percentage": 25, "duration_minutes": 120, "label": "mid"},
        {"percentage": 50, "duration_minutes": 240, "label": "late"},
        {"percentage": 100, "duration_minutes": 0, "label": "complete"},
    ]
    
    def __init__(self, db, metrics_collector, alert_manager):
        self.db = db
        self.metrics = metrics_collector
        self.alerts = alert_manager
    
    async def start_rollout(
        self,
        version_id: str,
        auto_rollback: bool = True,
        success_threshold: float = 0.95,
        error_rate_threshold: float = 0.05
    ) -> RolloutSession:
        """启动灰度发布"""
        
        session = RolloutSession(
            version_id=version_id,
            stage_index=0,
            auto_rollback=auto_rollback,
            success_threshold=success_threshold,
            error_rate_threshold=error_rate_threshold,
            started_at=datetime.now(),
            baseline_metrics=await self._capture_baseline(version_id)
        )
        
        await self._advance_stage(session)
        return session
    
    async def _advance_stage(self, session: RolloutSession):
        """推进到下一灰度阶段"""
        next_stage = self.ROLLOUT_STAGES[session.stage_index]
        
        # 更新流量分配
        await self._set_traffic_split(
            session.version_id,
            next_stage["percentage"]
        )
        
        session.current_stage = next_stage["label"]
        session.stage_started_at = datetime.now()
        
        logger.info(
            f"Rollout {session.version_id}: "
            f"stage={next_stage['label']}, "
            f"traffic={next_stage['percentage']}%"
        )
        
        # 启动监控
        asyncio.create_task(self._monitor_stage(session, next_stage))
    
    async def _monitor_stage(self, session: RolloutSession, stage: dict):
        """监控灰度阶段指标"""
        await asyncio.sleep(stage["duration_minutes"] * 60)
        
        # 收集指标
        current_metrics = await self._collect_metrics(session.version_id)
        baseline = session.baseline_metrics
        
        # 评估
        assessment = self._assess_metrics(current_metrics, baseline, session)
        
        if assessment.should_rollback:
            await self._rollback(session, assessment.reason)
        elif assessment.should_proceed:
            session.stage_index += 1
            if session.stage_index < len(self.ROLLOUT_STAGES):
                await self._advance_stage(session)
            else:
                await self._complete(session)
        else:
            # 指标不确定，等待更多数据
            await asyncio.sleep(600)  # 再等10分钟
            await self._monitor_stage(session, stage)
    
    def _assess_metrics(
        self,
        current: dict,
        baseline: dict,
        session: RolloutSession
    ) -> RolloutAssessment:
        """评估灰度指标"""
        
        # 错误率检查
        if current["error_rate"] > session.error_rate_threshold:
            return RolloutAssessment(
                should_rollback=True,
                reason=f"Error rate {current['error_rate']:.1%} > threshold {session.error_rate_threshold:.1%}"
            )
        
        # 延迟检查
        if current["p95_latency"] > baseline["p95_latency"] * 1.5:
            return RolloutAssessment(
                should_rollback=True,
                reason=f"P95 latency degraded: {current['p95_latency']:.0f}ms vs {baseline['p95_latency']:.0f}ms"
            )
        
        # 质量评分检查（通过 LLM-as-Judge）
        if current["quality_score"] < baseline["quality_score"] * 0.9:
            return RolloutAssessment(
                should_rollback=True,
                reason=f"Quality score dropped: {current['quality_score']:.2f} vs {baseline['quality_score']:.2f}"
            )
        
        # 成本检查
        if current["cost_per_request"] > baseline["cost_per_request"] * 2:
            return RolloutAssessment(
                should_rollback=False,
                should_proceed=False,
                reason="Cost increased significantly, manual review recommended"
            )
        
        return RolloutAssessment(should_proceed=True)
    
    async def _set_traffic_split(self, version_id: str, percentage: float):
        """设置流量分配"""
        # 方案1: 基于用户ID的哈希分桶（确保同一用户体验一致）
        await self.redis.set(
            f"rollout:{version_id}:percentage",
            percentage
        )
        
    async def _rollback(self, session: RolloutSession, reason: str):
        """回滚"""
        logger.warning(f"Rolling back {session.version_id}: {reason}")
        
        await self._set_traffic_split(session.version_id, 0)
        
        await self.db.update_version(
            session.version_id,
            status=VersionStatus.ROLLED_BACK,
            rollback_reason=reason
        )
        
        await self.alerts.notify(
            level="critical",
            title=f"Agent version {session.version_id} rolled back",
            message=reason
        )


class TrafficRouter:
    """流量路由器：根据灰度比例分配版本"""
    
    async def route(self, agent_name: str, user_id: str) -> str:
        """返回用户应该使用的版本ID"""
        # 获取当前灰度配置
        rollouts = await self.db.get_active_rollouts(agent_name)
        
        if not rollouts:
            return await self.db.get_live_version(agent_name)
        
        # 基于用户ID哈希分桶
        bucket = hash(user_id) % 10000 / 100  # 0-99.99
        
        for rollout in rollouts:
            if bucket < rollout.rollout_percentage:
                return rollout.version_id
        
        # 默认返回上一稳定版
        return await self.db.get_previous_live(agent_name)

class PromptManager:
    """Prompt 生命周期管理"""
    
    async def update_prompt(
        self,
        agent_name: str,
        new_prompt: str,
        author: str,
        review_required: bool = True
    ) -> PromptVersion:
        """创建 Prompt 新版本"""
        
        # 获取当前版本
        current = await self.repo.get_current_prompt(agent_name)
        
        # 计算差异
        diff = self.diff_calculator.calculate(
            current.system_prompt if current else "",
            new_prompt
        )
        
        # 创建新版本
        version = PromptVersion(
            version_id=self._generate_id(),
            system_prompt=new_prompt,
            few_shot_examples=[],  # 可单独版本化
            temperature=0.3,
            max_tokens=4096,
            parent_version=current.version_id if current else None,
            diff=diff.diff_text,
            created_at=datetime.now(),
            author=author,
            review_status="pending" if review_required else "approved"
        )
        
        await self.repo.save(version)
        
        if review_required:
            await self._request_review(version)
        
        return version
    
    async def _request_review(self, version: PromptVersion):
        """请求 Prompt 审核"""
        # 自动化测试
        test_results = await self._run_prompt_tests(version)
        
        # 人工审核通知
        await self.notify_reviewers(
            version=version,
            test_results=test_results,
            diff=version.diff
        )
    
    async def _run_prompt_tests(self, version: PromptVersion) -> TestReport:
        """对新 Prompt 运行测试套件"""
        results = []
        for test_case in self.test_suite:
            response = await self.llm.invoke(
                system=version.system_prompt,
                messages=[{"role": "user", "content": test_case.input}],
                temperature=version.temperature
            )
            score = await self.judge.evaluate(
                test_case.input, response.content, test_case.criteria
            )
            results.append(TestResult(
                test_id=test_case.id,
                score=score.score,
                passed=score.score >= test_case.min_score
            ))
        
        return TestReport(
            total=len(results),
            passed=sum(1 for r in results if r.passed),
            results=results
        )

# .github/workflows/agent-release.yml
name: Agent Version Release

on:
  push:
    branches: [main]
    paths:
      - "agents/*/prompt.txt"
      - "agents/*/tools.yaml"
      - "agents/*/config.yaml"

jobs:
  version:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # 获取完整历史用于 diff
      
      - name: Detect changes
        id: changes
        run: |
          python scripts/detect_agent_changes.py >> $GITHUB_OUTPUT
      
      - name: Create version
        if: steps.changes.outputs.has_changes == 'true'
        run: |
          python scripts/create_version.py \
            --agent ${{ steps.changes.outputs.agent }} \
            --changes ${{ steps.changes.outputs.changes }}
      
      - name: Run evaluation tests
        run: |
          python scripts/eval_version.py \
            --version ${{ steps.changes.outputs.version }} \
            --test-suite regression
      
      - name: Request review
        if: steps.changes.outputs.severity != 'patch'
        uses: actions/github-script@v7
        with:
          script: |
            github.rest.pulls.createReviewRequest({
              pull_number: context.issue.number,
              reviewers: ['ai-safety-lead', 'agent-team-lead']
            })
      
      - name: Start canary rollout
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          python scripts/start_rollout.py \
            --version ${{ steps.changes.outputs.version }} \
            --canary-percentage 1 \
            --auto-rollback true

class RollbackManager:
    """回滚管理器"""
    
    async def rollback(
        self,
        agent_name: str,
        target_version: str | None = None,
        reason: str = ""
    ) -> RollbackResult:
        """回滚到指定版本（默认回退到上一稳定版）"""
        
        if target_version is None:
            target_version = await self._get_previous_stable(agent_name)
        
        current = await self.repo.get_live_version(agent_name)
        
        # 1. 立即切换流量
        await self.router.set_live(agent_name, target_version)
        
        # 2. 标记当前版本为已回滚
        await self.repo.update_status(
            current.version_id,
            VersionStatus.ROLLED_BACK,
            rollback_reason=reason,
            rollback_target=target_version
        )
        
        # 3. 验证回滚后系统正常
        health = await self._health_check(agent_name, target_version)
        if not health.healthy:
            # 紧急回退到规则引擎模式
            await self.router.enable_fallback_mode(agent_name)
            await self.alerts.critical(
                f"Rollback failed for {agent_name}, entering fallback mode"
            )
        
        return RollbackResult(
            agent=agent_name,
            from_version=current.version_id,
            to_version=target_version,
            reason=reason,
            healthy=health.healthy
        )
    
    async def instant_rollback(self, agent_name: str):
        """秒级回滚：仅切换流量，不做验证"""
        prev = await self._get_previous_stable(agent_name)
        await self.router.set_live(agent_name, prev)
        logger.warning(f"Instant rollback: {agent_name} → {prev}")

□ 所有 Agent 组件（Prompt/工具/模型/工作流）版本化
□ 语义化版本号规则
□ 版本变更必须通过自动化测试
□ 灰度发布分6阶段（1%→5%→25%→50%→100%）
□ 灰度阶段自动监控关键指标
□ 异常指标触发自动回滚
□ Prompt 变更需要 Code Review
□ 模型变更需要回归测试
□ 版本历史可追溯，支持 diff 查看
□ 秒级回滚能力已验证
□ 灰度期间用户体验一致性保证
□ 定期进行回滚演练