Claude Code 质量危机深度分析：从 Anthropic 事后报告看 AI 编程助手的可靠性工程

#!/usr/bin/env python3
"""
AI 编程助手端到端会话评测框架
模拟真实用户的多轮交互场景，检测渐进式质量退化
"""

import json
import time
from dataclasses import dataclass, field
from typing import Any
from enum import Enum


class EvalOutcome(Enum):
    PASS = "pass"
    FAIL = "fail"
    TIMEOUT = "timeout"
    INCONSISTENT = "inconsistent"


@dataclass
class SessionTestCase:
    """一个完整的用户会话测试用例"""
    name: str
    scenario: str  # 场景描述
    initial_context: dict  # 初始上下文
    turns: list  # 多轮对话步骤
    expected_outcome: dict  # 期望结果
    max_turns: int = 20
    timeout_seconds: int = 300


@dataclass
class EvalReport:
    """评测报告"""
    test_name: str
    outcome: EvalOutcome
    turns_completed: int
    details: str
    duration_seconds: float
    context_used: int = 0  # 使用的 token 数
    tools_called: list = field(default_factory=list)


class SessionEvaluator:
    """会话级评测器"""
    
    def __init__(self, client: Any):
        self.client = client
        self.history: list = []
    
    def run_session(self, test: SessionTestCase) -> EvalReport:
        """运行一个完整的会话测试"""
        start_time = time.time()
        self.history = []
        tools_called = []
        
        # 初始化上下文
        self.history.append({
            "role": "system",
            "content": test.initial_context.get("system_prompt", "")
        })
        
        # 加载初始文件内容
        for filename, content in test.initial_context.get("files", {}).items():
            self.history.append({
                "role": "system",
                "content": f"[File: {filename}]\\n{content}"
            })
        
        turn_count = 0
        for turn in test.turns:
            if turn_count >= test.max_turns:
                return EvalReport(
                    test_name=test.name,
                    outcome=EvalOutcome.TIMEOUT,
                    turns_completed=turn_count,
                    details=f"达到最大轮次限制 ({test.max_turns})",
                    duration_seconds=time.time() - start_time,
                )
            
            # 发送用户消息
            self.history.append({"role": "user", "content": turn["user_input"]})
            
            # 调用 AI 并检查响应
            try:
                response = self._call_ai_with_timeout(test.timeout_seconds)
                tools_called.extend(response.get("tools", []))
                
                # 验证响应质量
                quality_score = self._evaluate_response_quality(
                    response, turn.get("expected", {})
                )
                
                if quality_score < 0.5:
                    return EvalReport(
                        test_name=test.name,
                        outcome=EvalOutcome.FAIL,
                        turns_completed=turn_count,
                        details=f"第 {turn_count + 1} 轮响应质量过低 (score: {quality_score})",
                        duration_seconds=time.time() - start_time,
                        context_used=len(json.dumps(self.history)),
                        tools_called=tools_called,
                    )
                
                self.history.append({"role": "assistant", "content": response["content"]})
                
            except TimeoutError:
                return EvalReport(
                    test_name=test.name,
                    outcome=EvalOutcome.TIMEOUT,
                    turns_completed=turn_count,
                    details=f"第 {turn_count + 1} 轮超时",
                    duration_seconds=time.time() - start_time,
                    tools_called=tools_called,
                )
            
            turn_count += 1
        
        # 最终验证
        final_check = self._verify_final_outcome(test.expected_outcome)
        
        return EvalReport(
            test_name=test.name,
            outcome=EvalOutcome.PASS if final_check else EvalOutcome.FAIL,
            turns_completed=turn_count,
            details="通过" if final_check else "最终验证失败",
            duration_seconds=time.time() - start_time,
            context_used=len(json.dumps(self.history)),
            tools_called=tools_called,
        )
    
    def _call_ai_with_timeout(self, timeout: int) -> dict:
        """调用 AI 模型（含超时保护）"""
        # 实际实现中对接 Claude Code / Codex API
        pass
    
    def _evaluate_response_quality(self, response: dict, expected: dict) -> float:
        """评估单轮响应质量（0-1）"""
        score = 1.0
        
        # 检查输出格式
        if expected.get("format") and expected["format"] not in response.get("content", ""):
            score -= 0.3
        
        # 检查是否包含关键信息
        for key_info in expected.get("must_contain", []):
            if key_info not in response.get("content", ""):
                score -= 0.2
        
        return max(0, score)
    
    def _verify_final_outcome(self, expected: dict) -> bool:
        """验证会话最终结果"""
        # 检查最终文件内容是否符合预期
        pass


# 定义一个典型测试用例
edit_multiple_files_test = SessionTestCase(
    name="多文件编辑会话",
    scenario="用户要求同时修改多个相关文件",
    initial_context={
        "system_prompt": "你是一个编程助手。请帮助用户完成代码修改任务。",
        "files": {
            "src/main.py": "# Main application entry point\\nfrom utils import helper\\n\\ndef main():\\n    helper()",
            "src/utils.py": "# Utility functions\\ndef helper():\\n    print('Hello')",
        }
    },
    turns=[
        {
            "user_input": "把 utils.py 中的 helper 函数改成接收一个 name 参数",
            "expected": {"format": "python_code_block", "must_contain": ["name"]}
        },
        {
            "user_input": "同时更新 main.py 中的调用",
            "expected": {"must_contain": ["helper(", "name"]}
        },
        {
            "user_input": "再加一个 docstring 到 helper 函数",
            "expected": {"must_contain": ['"""', 'name']}
        }
    ],
    expected_outcome={
        "files_modified": ["src/main.py", "src/utils.py"],
        "no_syntax_errors": True
    }
)

# 运行测试
if __name__ == "__main__":
    evaluator = SessionEvaluator(client=None)  # 替换为实际 client
    report = evaluator.run_session(edit_multiple_files_test)
    print(f"测试: {report.test_name}")
    print(f"结果: {report.outcome.value}")
    print(f"轮次: {report.turns_completed}")
    print(f"耗时: {report.duration_seconds:.1f}s")
    print(f"详情: {report.details}")

#!/usr/bin/env python3
"""
AI 编程工具日常质量监控管道
每天自动运行，生成质量报告并发送告警
"""

import os
import json
import datetime
import sqlite3
from typing import Optional
from dataclasses import dataclass, asdict


@dataclass
class QualityMetric:
    """质量指标"""
    name: str
    value: float  # 0-100
    baseline: float  # 基线值
    threshold: float  # 告警阈值
    timestamp: str
    details: str = ""


class QualityMonitor:
    """质量监控器"""
    
    def __init__(self, db_path: str = "quality_monitor.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """初始化数据库"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS metrics (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    value REAL NOT NULL,
                    baseline REAL NOT NULL,
                    threshold REAL NOT NULL,
                    timestamp TEXT NOT NULL,
                    details TEXT
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_metrics_name_ts 
                ON metrics(name, timestamp)
            """)
    
    def record_metric(self, metric: QualityMetric):
        """记录质量指标"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO metrics (name, value, baseline, threshold, timestamp, details) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (metric.name, metric.value, metric.baseline, 
                 metric.threshold, metric.timestamp, metric.details)
            )
    
    def check_regression(self, metric: QualityMetric) -> Optional[str]:
        """检查是否发生退化"""
        deviation = metric.baseline - metric.value
        
        if deviation > metric.threshold:
            return (
                f"⚠️ {metric.name} 退化: "
                f"当前 {metric.value:.1f}% vs 基线 {metric.baseline:.1f}% "
                f"(偏差 {deviation:.1f}%, 阈值 {metric.threshold:.1f}%)"
            )
        return None
    
    def generate_daily_report(self) -> dict:
        """生成每日质量报告"""
        today = datetime.date.today().isoformat()
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT name, value, baseline, threshold, details "
                "FROM metrics WHERE date(timestamp) = ? ORDER BY name",
                (today,)
            )
            rows = cursor.fetchall()
        
        report = {
            "date": today,
            "metrics": [],
            "alerts": [],
            "summary": ""
        }
        
        for name, value, baseline, threshold, details in rows:
            metric = QualityMetric(
                name=name, value=value, baseline=baseline,
                threshold=threshold, timestamp=today, details=details
            )
            report["metrics"].append(asdict(metric))
            
            alert = self.check_regression(metric)
            if alert:
                report["alerts"].append(alert)
        
        # 生成摘要
        total = len(report["metrics"])
        alerts = len(report["alerts"])
        report["summary"] = (
            f"今日评测 {total} 项，{'全部正常' if alerts == 0 else f'{alerts} 项告警'}"
        )
        
        return report


# 模拟一次完整的日常监控
def run_daily_monitor():
    """运行日常监控"""
    monitor = QualityMonitor()
    now = datetime.datetime.now().isoformat()
    
    # 模拟评测结果
    metrics = [
        QualityMetric("单次代码生成准确率", 94.5, 95.0, 3.0, now),
        QualityMetric("多轮会话成功率", 89.2, 92.0, 5.0, now),  # 可能退化
        QualityMetric("工具调用成功率", 97.8, 98.0, 2.0, now),
        QualityMetric("多实例一致性", 99.5, 99.5, 1.0, now),
        QualityMetric("长会话稳定性", 91.0, 93.0, 4.0, now),
        QualityMetric("提示词版本一致性", 100.0, 100.0, 0.0, now),
    ]
    
    alerts = []
    for metric in metrics:
        monitor.record_metric(metric)
        alert = monitor.check_regression(metric)
        if alert:
            alerts.append(alert)
    
    report = monitor.generate_daily_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))
    
    if alerts:
        print("\\n=== 告警 ===")
        for alert in alerts:
            print(alert)


if __name__ == "__main__":
    run_daily_monitor()