模块六-架构演进与团队协作 | 第40讲：团队 AI 协作模式 - Code Review 工作流、AI 编码规范的推广策略

开场：评审没变快，变的是「该由谁看什么」

许多团队在引入 AI 辅助编码后的第一个误区，是把 Code Review 当成「多了一道自动 linter」。结果是：机器人报了一堆格式问题，架构师仍在深夜人肉看业务边界，人类注意力没有重新配置，整体效率提升有限。第二个误区是走向另一个极端——迷信「AI 全绿即可合并」，把架构一致性与业务不变量交给概率模型，短期快、长期脆。

本讲把协作模式拆成可落地的四类范式，并给出 PR 工作流在 AI 时代的重排：机器先扫什么、人必须盯什么；如何把 AGENTS.md（或等价「人机协约」）从个人习惯推广为组织标准；以及如何用指标衡量协作是否真的变好。最后提供 CodeSentinel 侧 TeamMetrics、CollaborationDashboard 与 StandardAdoptionTracker 的 Python 骨架，便于与 FastAPI 与现有审核流水线对接。目标只有一个：让 AI 吃掉重复劳动，让人专注于不可替代的判断。

---

全局视角：人机分工与采纳飞轮

下图描述推荐的 PR 流水线：提交后先运行静态与策略门禁，再进行 AI 自动评审（模式、安全、性能提示），随后人类聚焦架构与业务不变量；合并后指标回写团队看板，驱动规范采纳与培训。

团队采纳 AGENTS.md 的四阶段可用下图与业务方沟通预期：从冠军用户到跨团队治理，每一步的「完成定义」不同。

---

核心原理：四种协作模式与工作流重塑

1. AI-First：AI 生成，人类审核

适用于脚手架、重复 CRUD、测试数据构造。关键是生成后必须有针对「边界条件与错误路径」的人类检查清单，否则容易得到「演示级正确」的代码。

2. Human-First + AI Review：人类主写，AI 辅助找茬

适用于核心业务与计费、权限、合规路径。人类保持设计主导，AI 作为第二意见：模式反模式、潜在注入、性能陷阱、文档字符串缺失等。

3. Pair Programming with AI：交互式协作

在 IDE 内多轮对话迭代，适合探索 API 设计或算法草稿。风险是上下文漂移——需要阶段性「压缩」为明确接口与测试，再进入 PR。

4. AI Architecture Advisor：咨询式使用

在 RFC/ADR 阶段引入，对备选方案做利弊罗列与风险提醒；不替代决策责任人。产出应链接到 ADR 与适应度函数（参见第38讲）。

5. Code Review 分工表（建议）

维度	适合 AI 初审	必须人类把关
语法/风格/明显 bug 模式	✓	边界案例
安全模式（注入/密钥）	✓ 启发式	威胁建模与数据流
性能热点	✓ 静态提示	业务负载与 SLO
架构边界与模块依赖	辅助	✓
业务规则与不变量	辅助	✓

6. AGENTS.md 推广策略（四阶段详解）

阶段1：冠军采纳——选择 1–2 名高影响力开发者，在真实项目上试用，沉淀「项目级提示词与禁区」样例。
阶段2：团队采纳——工作坊 + PR 模板中强制勾选「是否遵循 AGENTS.md」；CodeSentinel 记录违规次数。
阶段3：组织标准化——将通用规则上收为组织仓库，子项目通过引用或生成同步；CI 校验 AGENTS.md 存在性与版本。
阶段4：跨团队治理——共享规则包、审计字段统一（谁生成、谁审核），与合规报表打通（第41讲）。

7. 效果度量

评审耗时下降需控制变量（变更规模、文件类型）。缺陷逃逸率（合并后 N 天内关联缺陷 / PR 数）更可靠。开发者满意度可用季度问卷 + 可选匿名 NPS。架构合规提升可用适应度加权分或分层违规计数趋势（与第38讲联动）。

---

代码实战：TeamMetrics、CollaborationDashboard、StandardAdoptionTracker

以下代码为标准库实现，演示如何从「原始事件日志」聚合团队指标；生产环境可替换为 PostgreSQL 查询层。

team_metrics.py

# team_metrics.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional, Tuple


@dataclass
class ReviewEvent:
    pr_id: str
    author: str
    reviewer: Optional[str]
    merged_at: datetime
    review_minutes: float
    lines_changed: int
    ai_findings: int
    post_merge_defects: int


class TeamMetrics:
    """从 ReviewEvent 列表计算团队与个人的协作指标。"""

    def __init__(self, events: List[ReviewEvent]) -> None:
        self.events = events

    def avg_review_minutes(self) -> float:
        if not self.events:
            return 0.0
        return round(sum(e.review_minutes for e in self.events) / len(self.events), 2)

    def defect_escape_rate(self) -> float:
        merged = [e for e in self.events if e.merged_at]
        if not merged:
            return 0.0
        with_defect = sum(1 for e in merged if e.post_merge_defects > 0)
        return round(with_defect / len(merged), 4)

    def ai_signal_density(self) -> float:
        """每千行变更对应的 AI 发现问题数，用于观察自动化覆盖强度。"""
        total_lines = sum(max(e.lines_changed, 1) for e in self.events)
        findings = sum(e.ai_findings for e in self.events)
        return round(1000.0 * findings / total_lines, 4)

    def per_developer_load(self) -> Dict[str, Tuple[int, float]]:
        """作者 -> (PR 数, 平均评审时长)"""
        buckets: Dict[str, List[ReviewEvent]] = {}
        for e in self.events:
            buckets.setdefault(e.author, []).append(e)
        out: Dict[str, Tuple[int, float]] = {}
        for author, evs in buckets.items():
            avg_r = sum(x.review_minutes for x in evs) / len(evs)
            out[author] = (len(evs), round(avg_r, 2))
        return out

collaboration_dashboard.py

# collaboration_dashboard.py
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List

from team_metrics import ReviewEvent, TeamMetrics


@dataclass
class CollaborationDashboard:
    """团队看板数据模型：总览 + 排行榜 + 原始事件子集。"""

    team_summary: Dict[str, Any] = field(default_factory=dict)
    leaderboard: List[Dict[str, Any]] = field(default_factory=list)
    recent_events: List[Dict[str, Any]] = field(default_factory=list)

    @classmethod
    def build(cls, events: List[ReviewEvent], top_n: int = 5) -> "CollaborationDashboard":
        tm = TeamMetrics(events)
        summary = {
            "avg_review_minutes": tm.avg_review_minutes(),
            "defect_escape_rate": tm.defect_escape_rate(),
            "ai_signal_density": tm.ai_signal_density(),
            "pr_count": len(events),
        }
        per = tm.per_developer_load()
        board = sorted(
            (
                {
                    "author": a,
                    "pr_count": cnt,
                    "avg_review_minutes": avg_min,
                }
                for a, (cnt, avg_min) in per.items()
            ),
            key=lambda x: x["pr_count"],
            reverse=True,
        )
        # ReviewEvent 含 datetime，需可序列化
        serial_recent: List[Dict[str, Any]] = []
        for e in events[-10:]:
            serial_recent.append(
                {
                    "pr_id": e.pr_id,
                    "author": e.author,
                    "reviewer": e.reviewer,
                    "merged_at": e.merged_at.isoformat(),
                    "review_minutes": e.review_minutes,
                    "lines_changed": e.lines_changed,
                    "ai_findings": e.ai_findings,
                    "post_merge_defects": e.post_merge_defects,
                }
            )
        return cls(team_summary=summary, leaderboard=board[:top_n], recent_events=serial_recent)

standard_adoption.py

# standard_adoption.py
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional


@dataclass
class RepoComplianceSnapshot:
    repo: str
    ts: datetime
    has_agents_md: bool
    agents_version: Optional[str]
    pr_template_links_agents: bool
    ci_enforces_presence: bool


@dataclass
class StandardAdoptionTracker:
    """
    追踪 AGENTS.md 推广阶段完成情况。
    phase: 1 champion -> 4 org governance
    """

    snapshots: List[RepoComplianceSnapshot] = field(default_factory=list)

    def record(self, snap: RepoComplianceSnapshot) -> None:
        self.snapshots.append(snap)

    def latest_per_repo(self) -> Dict[str, RepoComplianceSnapshot]:
        latest: Dict[str, RepoComplianceSnapshot] = {}
        for s in sorted(self.snapshots, key=lambda x: x.ts):
            latest[s.repo] = s
        return latest

    def adoption_score(self, repo: str) -> float:
        """0-100 简易得分：四项布尔权重各 25。"""
        m = self.latest_per_repo()
        s = m.get(repo)
        if not s:
            return 0.0
        flags = [
            s.has_agents_md,
            bool(s.agents_version),
            s.pr_template_links_agents,
            s.ci_enforces_presence,
        ]
        return round(100.0 * sum(1 for f in flags if f) / len(flags), 2)

    def phase_guess(self, repo: str) -> int:
        s = self.latest_per_repo().get(repo)
        if not s:
            return 0
        score = self.adoption_score(repo)
        if score < 25:
            return 1
        if score < 50:
            return 2
        if score < 75:
            return 3
        return 4

collab_demo.py

# collab_demo.py
from __future__ import annotations

from datetime import datetime, timedelta, timezone

from collaboration_dashboard import CollaborationDashboard
from standard_adoption import RepoComplianceSnapshot, StandardAdoptionTracker
from team_metrics import ReviewEvent


def main() -> None:
    now = datetime.now(timezone.utc)
    events = [
        ReviewEvent(
            pr_id="101",
            author="alice",
            reviewer="bob",
            merged_at=now,
            review_minutes=35,
            lines_changed=220,
            ai_findings=4,
            post_merge_defects=0,
        ),
        ReviewEvent(
            pr_id="102",
            author="bob",
            reviewer="carol",
            merged_at=now - timedelta(days=1),
            review_minutes=55,
            lines_changed=800,
            ai_findings=9,
            post_merge_defects=1,
        ),
    ]
    dash = CollaborationDashboard.build(events)
    print("summary:", dash.team_summary)
    print("leaderboard:", dash.leaderboard)

    tracker = StandardAdoptionTracker()
    tracker.record(
        RepoComplianceSnapshot(
            repo="codesentinel",
            ts=now,
            has_agents_md=True,
            agents_version="2025.03.1",
            pr_template_links_agents=True,
            ci_enforces_presence=False,
        )
    )
    print("adoption score:", tracker.adoption_score("codesentinel"))
    print("phase guess:", tracker.phase_guess("codesentinel"))


if __name__ == "__main__":
    main()

最佳实践分享机制（产品化建议）：在 CodeSentinel 增加「优秀评审片段」匿名转载（脱敏后），把高价值人类评论与 AI 提示并列展示，形成可学习的组织记忆，比单纯排行榜更能改变行为。

---

生产环境实战

1. 与 Git 平台 Webhook 集成：在 PR opened/synchronized 时写入 ReviewEvent 初稿，合并与缺陷关联时回填 post_merge_defects。
2. 防止指标游戏化：排行榜以「架构合规提升」「缺陷逃逸下降」为主，不仅仅看 PR 数量。
3. AGENTS.md 版本化：与容器镜像 tag 或 pyproject 版本对齐，便于审计「当时遵循的是哪版规范」。
4. 多时区团队：review_minutes 应用工作日窗口归一，避免误判。

---

本讲小结（Mermaid mindmap）

---

指标看板线框（Mermaid）

---

思考题

1. 你的业务里，哪些不变量绝对不能交给 AI 初审通过即默认安全？如何写成可执行检查？
2. 「评审变快」是否可能来自 AI 漏报？你会配什么对照指标？
3. 阶段4跨团队治理时，如何平衡统一标准与前端/数据/算法各栈差异？

---

下一讲预告

第41讲：AI 生成代码的治理框架——从开发、预提交、PR、CI/CD、预发、到线上的全链路策略；审计轨迹与合规报表；CodeSentinel PolicyEngine 与 ComplianceReporter 实战。

---

协作模式不是口号，而是流水线与度量刻出来的习惯。