分层的艺术：从 OSI 七层到 Agent OS 五层

#!/usr/bin/env python3
"""
Code Review Agent - 分层架构示例
"""

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any, Protocol
from enum import Enum
import asyncio
import hashlib
from datetime import datetime

# ==================== 领域模型 ====================

class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

@dataclass
class CodeChange:
    file_path: str
    diff: str
    language: str
    additions: int
    deletions: int

@dataclass
class ReviewComment:
    file_path: str
    line_number: int
    message: str
    severity: Severity
    category: str  # "security", "performance", "style", etc.
    suggestion: Optional[str] = None

@dataclass
class ReviewReport:
    pr_id: str
    summary: str
    comments: List[ReviewComment]
    statistics: Dict[str, int]
    generated_at: datetime = field(default_factory=datetime.now)

# ==================== 第5层：接口层 ====================

@dataclass
class PRWebhookPayload:
    """GitHub PR Webhook 数据结构"""
    action: str
    pr_number: int
    repository: str
    head_sha: str
    diff_url: str
    author: str

class InterfaceLayer:
    """接口层：处理多种输入协议"""
    
    def __init__(self, orchestrator: "OrchestrationLayer"):
        self.orchestrator = orchestrator
    
    async def handle_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """处理 GitHub Webhook"""
        pr_payload = self._parse_webhook(payload)
        
        # 构建上下文
        context = ReviewContext(
            pr_id=f"{pr_payload.repository}#{pr_payload.pr_number}",
            repository=pr_payload.repository,
            head_sha=pr_payload.head_sha,
            diff_url=pr_payload.diff_url,
            author=pr_payload.author,
            timestamp=datetime.now()
        )
        
        # 调用编排层
        report = await self.orchestrator.review(context)
        
        return self._format_response(report)
    
    async def handle_cli(self, repo: str, pr_number: int) -> str:
        """处理 CLI 输入"""
        context = ReviewContext(
            pr_id=f"{repo}#{pr_number}",
            repository=repo,
            head_sha="",  # 将通过 API 获取
            diff_url=f"https://api.github.com/repos/{repo}/pulls/{pr_number}",
            author="cli_user",
            timestamp=datetime.now()
        )
        
        report = await self.orchestrator.review(context)
        return self._format_markdown(report)
    
    def _parse_webhook(self, payload: Dict) -> PRWebhookPayload:
        """解析并验证 Webhook 数据"""
        return PRWebhookPayload(
            action=payload.get("action", ""),
            pr_number=payload["pull_request"]["number"],
            repository=payload["repository"]["full_name"],
            head_sha=payload["pull_request"]["head"]["sha"],
            diff_url=payload["pull_request"]["diff_url"],
            author=payload["pull_request"]["user"]["login"]
        )
    
    def _format_response(self, report: ReviewReport) -> Dict[str, Any]:
        return {
            "pr_id": report.pr_id,
            "summary": report.summary,
            "comment_count": len(report.comments),
            "critical_issues": report.statistics.get("critical", 0)
        }
    
    def _format_markdown(self, report: ReviewReport) -> str:
        lines = [
            f"# Code Review Report: {report.pr_id}",
            "",
            f"**Generated:** {report.generated_at}",
            "",
            "## Summary",
            report.summary,
            "",
            "## Issues Found",
        ]
        
        for comment in sorted(report.comments, key=lambda c: c.severity.value):
            lines.extend([
                f"### [{comment.severity.value.upper()}] {comment.category}",
                f"**File:** `{comment.file_path}`:{comment.line_number}",
                f"{comment.message}",
            ])
            if comment.suggestion:
                lines.extend([
                    "**Suggestion:**",
                    f"```\n{comment.suggestion}\n```"
                ])
            lines.append("")
        
        return "\n".join(lines)

# ==================== 第4层：编排层 ====================

@dataclass
class ReviewContext:
    """贯穿审查流程的上下文"""
    pr_id: str
    repository: str
    head_sha: str
    diff_url: str
    author: str
    timestamp: datetime
    changes: List[CodeChange] = field(default_factory=list)
    history_reviews: List[Dict] = field(default_factory=list)

class OrchestrationLayer:
    """编排层：任务分解与调度"""
    
    def __init__(
        self,
        cognition: "CognitionLayer",
        memory: "MemoryLayer",
        execution: "ExecutionLayer"
    ):
        self.cognition = cognition
        self.memory = memory
        self.execution = execution
    
    async def review(self, context: ReviewContext) -> ReviewReport:
        """编排完整的代码审查流程"""
        
        # Phase 1: 获取代码变更（执行层）
        context.changes = await self.execution.fetch_diff(context.diff_url)
        
        # Phase 2: 检索历史审查（记忆层）
        context.history_reviews = await self.memory.get_similar_reviews(
            context.repository, context.author
        )
        
        # Phase 3: 并发分析每个文件
        file_tasks = [
            self._analyze_file(change, context)
            for change in context.changes
        ]
        file_results = await asyncio.gather(*file_tasks)
        
        # Phase 4: 聚合所有评论
        all_comments = []
        for comments in file_results:
            all_comments.extend(comments)
        
        # Phase 5: 生成总结（认知层）
        summary = await self.cognition.generate_summary(
            context.changes, all_comments
        )
        
        # Phase 6: 记录本次审查（记忆层）
        await self.memory.store_review(context, all_comments)
        
        # 构建报告
        return ReviewReport(
            pr_id=context.pr_id,
            summary=summary,
            comments=all_comments,
            statistics=self._calculate_stats(all_comments)
        )
    
    async def _analyze_file(
        self,
        change: CodeChange,
        context: ReviewContext
    ) -> List[ReviewComment]:
        """分析单个文件"""
        
        # 并行执行多种分析
        analysis_tasks = [
            self.cognition.analyze_security(change),
            self.cognition.analyze_performance(change),
            self.cognition.analyze_style(change, context.history_reviews),
        ]
        
        results = await asyncio.gather(*analysis_tasks)
        
        # 合并所有发现
        all_comments = []
        for comments in results:
            all_comments.extend(comments)
        
        return all_comments
    
    def _calculate_stats(self, comments: List[ReviewComment]) -> Dict[str, int]:
        stats = {"total": len(comments)}
        for sev in Severity:
            stats[sev.value] = sum(1 for c in comments if c.severity == sev)
        return stats

# ==================== 第3层：认知层 ====================

class LLMProvider(Protocol):
    """LLM 提供商抽象"""
    async def analyze(self, prompt: str, context: Dict) -> str:
        ...

class CognitionLayer:
    """认知层：代码分析与推理"""
    
    def __init__(self, llm: LLMProvider):
        self.llm = llm
        self.security_patterns = self._load_security_patterns()
    
    async def analyze_security(self, change: CodeChange) -> List[ReviewComment]:
        """安全分析"""
        prompt = f"""
        Analyze the following {change.language} code for security issues:
        
        Look for: SQL injection, XSS, path traversal, hardcoded secrets, etc.
        Return findings as JSON array with file_path, line, severity, message.
        """
        
        response = await self.llm.analyze(prompt, {"type": "security"})
        return self._parse_llm_response(response, change.file_path)
    
    async def analyze_performance(self, change: CodeChange) -> List[ReviewComment]:
        """性能分析"""
        prompt = f"""
        Analyze this {change.language} code for performance issues:
        N+1 queries, inefficient loops, memory leaks, etc.
        
        """
        
        response = await self.llm.analyze(prompt, {"type": "performance"})
        return self._parse_llm_response(response, change.file_path)
    
    async def analyze_style(
        self,
        change: CodeChange,
        history: List[Dict]
    ) -> List[ReviewComment]:
        """风格与最佳实践分析"""
        
        # 结合团队历史风格偏好
        style_context = self._extract_team_patterns(history)
        
        prompt = f"""
        Review this code against team style guidelines:
        {style_context}
        
        """
        
        response = await self.llm.analyze(prompt, {"type": "style"})
        return self._parse_llm_response(response, change.file_path)
    
    async def generate_summary(
        self,
        changes: List[CodeChange],
        comments: List[ReviewComment]
    ) -> str:
        """生成审查总结"""
        
        critical_count = sum(1 for c in comments if c.severity == Severity.CRITICAL)
        high_count = sum(1 for c in comments if c.severity == Severity.HIGH)
        
        lines = [
            f"Reviewed {len(changes)} files with {len(comments)} comments.",
        ]
        
        if critical_count > 0:
            lines.append(f"⚠️ Found {critical_count} critical security issues that need immediate attention.")
        if high_count > 0:
            lines.append(f"⚡ Found {high_count} high-priority issues.")
        
        lines.append("Overall code quality assessment based on changes.")
        
        return " ".join(lines)
    
    def _load_security_patterns(self) -> List[str]:
        # 加载已知的安全模式
        return ["eval(", "exec(", "subprocess.call", "os.system"]
    
    def _extract_team_patterns(self, history: List[Dict]) -> str:
        # 从历史审查中提取团队风格偏好
        return "Team prefers explicit error handling over exceptions."
    
    def _parse_llm_response(self, response: str, file_path: str) -> List[ReviewComment]:
        # 解析 LLM 响应为结构化评论
        # 简化实现，实际应使用结构化输出
        return []

# ==================== 第2层：记忆层 ====================

class VectorStore(Protocol):
    async def search(self, query: str, top_k: int) -> List[Dict]:
        ...
    async def upsert(self, documents: List[Dict]) -> None:
        ...

class MemoryLayer:
    """记忆层：审查历史存储与检索"""
    
    def __init__(self, vector_store: VectorStore):
        self.vector_store = vector_store
        self.cache: Dict[str, Any] = {}
    
    async def get_similar_reviews(
        self,
        repository: str,
        author: str
    ) -> List[Dict]:
        """检索相似的审查记录"""
        
        cache_key = f"{repository}:{author}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        # 向量检索相关审查
        query = f"code review patterns for {repository} by {author}"
        results = await self.vector_store.search(query, top_k=5)
        
        self.cache[cache_key] = results
        return results
    
    async def store_review(
        self,
        context: ReviewContext,
        comments: List[ReviewComment]
    ) -> None:
        """存储审查记录供未来检索"""
        
        # 构建文档
        document = {
            "id": hashlib.md5(context.pr_id.encode()).hexdigest(),
            "pr_id": context.pr_id,
            "repository": context.repository,
            "author": context.author,
            "comment_count": len(comments),
            "categories": list(set(c.category for c in comments)),
            "timestamp": context.timestamp.isoformat(),
            "content": self._summarize_for_embedding(comments)
        }
        
        await self.vector_store.upsert([document])
    
    def _summarize_for_embedding(self, comments: List[ReviewComment]) -> str:
        """生成用于向量化的摘要"""
        categories = {}
        for c in comments:
            categories.setdefault(c.category, []).append(c.message)
        
        parts = []
        for cat, msgs in categories.items():
            parts.append(f"{cat}: {len(msgs)} issues")
        
        return "; ".join(parts)

# ==================== 第1层：执行层 ====================

class GitHubAPI(Protocol):
    async def get_diff(self, url: str) -> str:
        ...
    async def post_comment(self, pr_id: str, comment: str) -> None:
        ...

class ExecutionLayer:
    """执行层：外部交互与工具执行"""
    
    def __init__(self, github: GitHubAPI, llm_client: Any):
        self.github = github
        self.llm_client = llm_client
    
    async def fetch_diff(self, diff_url: str) -> List[CodeChange]:
        """获取 PR 的 diff 内容"""
        
        diff_content = await self.github.get_diff(diff_url)
        
        # 解析 diff 为 CodeChange 对象
        changes = []
        for file_diff in self._split_by_file(diff_content):
            change = self._parse_file_diff(file_diff)
            changes.append(change)
        
        return changes
    
    async def post_comments(self, pr_id: str, comments: List[ReviewComment]) -> None:
        """将评论发布到 GitHub"""
        for comment in comments:
            formatted = self._format_github_comment(comment)
            await self.github.post_comment(pr_id, formatted)
    
    def _split_by_file(self, diff_content: str) -> List[str]:
        """按文件分割 diff"""
        # 简化实现
        return diff_content.split("diff --git")[1:]
    
    def _parse_file_diff(self, file_diff: str) -> CodeChange:
        """解析单个文件的 diff"""
        lines = file_diff.split("\n")
        
        # 提取文件路径
        file_path = lines[0].split(" b/")[-1] if " b/" in lines[0] else "unknown"
        
        # 统计变更
        additions = sum(1 for l in lines if l.startswith("+"))
        deletions = sum(1 for l in lines if l.startswith("-"))
        
        # 检测语言
        language = self._detect_language(file_path)
        
        return CodeChange(
            file_path=file_path,
            diff=file_diff,
            language=language,
            additions=additions,
            deletions=deletions
        )
    
    def _detect_language(self, file_path: str) -> str:
        """根据文件扩展名检测语言"""
        ext = file_path.split(".")[-1] if "." in file_path else ""
        mapping = {
            "py": "python",
            "js": "javascript",
            "ts": "typescript",
            "go": "go",
            "rs": "rust",
            "java": "java"
        }
        return mapping.get(ext, "text")
    
    def _format_github_comment(self, comment: ReviewComment) -> str:
        """格式化为 GitHub 评论格式"""
        emoji_map = {
            Severity.CRITICAL: "🚨",
            Severity.HIGH: "⚠️",
            Severity.MEDIUM: "💡",
            Severity.LOW: "ℹ️",
            Severity.INFO: "📝"
        }
        
        emoji = emoji_map.get(comment.severity, "💬")
        return f"{emoji} **[{comment.category.upper()}]** {comment.message}"

# ==================== 系统组装 ====================

class CodeReviewAgent:
    """完整的代码审查 Agent"""
    
    def __init__(
        self,
        github_api: GitHubAPI,
        llm_provider: LLMProvider,
        vector_store: VectorStore
    ):
        # 从底层开始构建
        self.execution = ExecutionLayer(github_api, llm_provider)
        self.memory = MemoryLayer(vector_store)
        self.cognition = CognitionLayer(llm_provider)
        self.orchestration = OrchestrationLayer(
            self.cognition, self.memory, self.execution
        )
        self.interface = InterfaceLayer(self.orchestration)
    
    async def review_pr(self, repo: str, pr_number: int) -> str:
        """CLI 入口"""
        return await self.interface.handle_cli(repo, pr_number)
    
    async def handle_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Webhook 入口"""
        return await self.interface.handle_webhook(payload)

# ==================== 使用示例 ====================

async def main():
    """演示如何使用分层架构的代码审查 Agent"""
    
    # 初始化各层依赖（使用模拟实现）
    class MockGitHubAPI:
        async def get_diff(self, url: str) -> str:
            return """diff --git a/src/main.py b/src/main.py
new file mode 100644
index 0000000..1234567
--- /dev/null
+++ b/src/main.py
@@ -0,0 +1,10 @@
+import os
+
+def process_user_input(user_data):
+    query = f"SELECT * FROM users WHERE id = {user_data['id']}"
+    os.system(f"mysql -e '{query}'")
+    return eval(user_data['expression'])
+"""
    
    class MockLLMProvider:
        async def analyze(self, prompt: str, context: Dict) -> str:
            # 模拟 LLM 响应
            return "[]"
    
    class MockVectorStore:
        async def search(self, query: str, top_k: int) -> List[Dict]:
            return []
        async def upsert(self, documents: List[Dict]) -> None:
            pass
    
    # 构建 Agent
    agent = CodeReviewAgent(
        github_api=MockGitHubAPI(),
        llm_provider=MockLLMProvider(),
        vector_store=MockVectorStore()
    )
    
    # 执行审查
    report = await agent.review_pr("myorg/myrepo", 123)
    print(report)

if __name__ == "__main__":
    asyncio.run(main())