【系统架构设计百科】康威定律与逆康威定律：组织架构决定系统架构

# 组织结构
departments:
  - name: 用户中心部
    teams:
      - user-service
      - auth-service
    reports_to: CTO

  - name: 订单中心部
    teams:
      - order-service
      - inventory-service
    reports_to: CTO

  - name: 支付中心部
    teams:
      - payment-service
      - settlement-service
    reports_to: CFO

# 团队依赖度评估工具

from dataclasses import dataclass
from typing import List, Dict, Set


@dataclass
class Team:
    name: str
    owned_services: List[str]
    members: List[str]


@dataclass
class ServiceDependency:
    source: str
    target: str
    call_frequency: str  # high / medium / low
    change_coupling: float  # 0.0 - 1.0


def calculate_team_dependency_score(
    teams: List[Team],
    dependencies: List[ServiceDependency],
) -> Dict[str, Dict[str, float]]:
    """计算团队间依赖度得分。

    得分越高，说明两个团队之间的耦合越紧，
    组织架构调整的优先级越高。
    """
    service_to_team: Dict[str, str] = {}
    for team in teams:
        for svc in team.owned_services:
            service_to_team[svc] = team.name

    frequency_weight = {"high": 3.0, "medium": 1.5, "low": 0.5}
    team_scores: Dict[str, Dict[str, float]] = {}

    for dep in dependencies:
        src_team = service_to_team.get(dep.source)
        tgt_team = service_to_team.get(dep.target)

        if src_team and tgt_team and src_team != tgt_team:
            score = (
                frequency_weight.get(dep.call_frequency, 1.0)
                * (1 + dep.change_coupling)
            )
            team_scores.setdefault(src_team, {})
            team_scores[src_team][tgt_team] = (
                team_scores[src_team].get(tgt_team, 0) + score
            )

    return team_scores


def suggest_reorg(
    scores: Dict[str, Dict[str, float]],
    threshold: float = 5.0,
) -> List[str]:
    """基于依赖度得分建议组织调整。"""
    suggestions = []
    for src, targets in scores.items():
        for tgt, score in targets.items():
            if score >= threshold:
                suggestions.append(
                    f"建议合并或紧密对齐团队 [{src}] 和 [{tgt}]，"
                    f"依赖度得分：{score:.1f}"
                )
    return suggestions


# 使用示例
if __name__ == "__main__":
    teams = [
        Team("用户团队", ["user-svc", "auth-svc"], ["张三", "李四"]),
        Team("订单团队", ["order-svc", "cart-svc"], ["王五", "赵六"]),
        Team("支付团队", ["pay-svc", "settle-svc"], ["钱七", "孙八"]),
    ]

    deps = [
        ServiceDependency("order-svc", "user-svc", "high", 0.3),
        ServiceDependency("order-svc", "pay-svc", "high", 0.8),
        ServiceDependency("cart-svc", "user-svc", "medium", 0.2),
        ServiceDependency("pay-svc", "user-svc", "low", 0.1),
    ]

    scores = calculate_team_dependency_score(teams, deps)
    for suggestion in suggest_reorg(scores):
        print(suggestion)

// 流对齐团队的职责边界定义示例

package team

// StreamAlignedTeam 流对齐团队定义
type StreamAlignedTeam struct {
    Name           string          `json:"name"`
    BusinessDomain string          `json:"business_domain"`
    ValueStream    string          `json:"value_stream"`
    OwnedServices  []Service       `json:"owned_services"`
    Capabilities   []Capability    `json:"capabilities"`
    Metrics        []BusinessMetric `json:"metrics"`
}

// Service 团队拥有的服务
type Service struct {
    Name        string `json:"name"`
    Repository  string `json:"repository"`
    Language    string `json:"language"`
    DeployTarget string `json:"deploy_target"`
}

// Capability 团队具备的能力
type Capability struct {
    Type  string `json:"type"`  // frontend, backend, testing, ops, data
    Level string `json:"level"` // expert, proficient, learning
}

// BusinessMetric 业务度量指标
type BusinessMetric struct {
    Name   string  `json:"name"`
    Target float64 `json:"target"`
    Unit   string  `json:"unit"`
}

// CanDeliverIndependently 检查团队是否能独立交付
func (t *StreamAlignedTeam) CanDeliverIndependently() bool {
    requiredCaps := map[string]bool{
        "frontend": false,
        "backend":  false,
        "testing":  false,
        "ops":      false,
    }
    for _, cap := range t.Capabilities {
        if _, ok := requiredCaps[cap.Type]; ok {
            if cap.Level == "expert" || cap.Level == "proficient" {
                requiredCaps[cap.Type] = true
            }
        }
    }
    for _, met := range requiredCaps {
        if !met {
            return false
        }
    }
    return true
}

# 团队认知负荷评估模型

from dataclasses import dataclass, field
from enum import Enum
from typing import List


class LoadType(Enum):
    INTRINSIC = "intrinsic"      # 内在负荷
    EXTRANEOUS = "extraneous"    # 外在负荷
    GERMANE = "germane"          # 关联负荷


@dataclass
class CognitiveLoadItem:
    description: str
    load_type: LoadType
    score: int  # 1-10

    def __post_init__(self):
        if not 1 <= self.score <= 10:
            raise ValueError("认知负荷评分必须在 1-10 之间")


@dataclass
class TeamCognitiveProfile:
    team_name: str
    team_size: int
    items: List[CognitiveLoadItem] = field(default_factory=list)

    @property
    def total_load(self) -> int:
        return sum(item.score for item in self.items)

    @property
    def load_per_person(self) -> float:
        if self.team_size == 0:
            return 0.0
        return self.total_load / self.team_size

    @property
    def extraneous_load(self) -> int:
        return sum(
            item.score
            for item in self.items
            if item.load_type == LoadType.EXTRANEOUS
        )

    @property
    def extraneous_ratio(self) -> float:
        if self.total_load == 0:
            return 0.0
        return self.extraneous_load / self.total_load

    def health_report(self) -> str:
        lines = [f"团队：{self.team_name}（{self.team_size} 人）"]
        lines.append(f"总认知负荷：{self.total_load}")
        lines.append(f"人均认知负荷：{self.load_per_person:.1f}")
        lines.append(f"外在负荷占比：{self.extraneous_ratio:.0%}")

        if self.load_per_person > 15:
            lines.append("警告：人均认知负荷过高，建议拆分团队或减少职责范围")
        if self.extraneous_ratio > 0.4:
            lines.append("警告：外在负荷占比过高，建议投入平台建设降低环境复杂度")

        return "\n".join(lines)


# 使用示例
profile = TeamCognitiveProfile(
    team_name="订单团队",
    team_size=7,
    items=[
        CognitiveLoadItem("订单状态机逻辑", LoadType.INTRINSIC, 8),
        CognitiveLoadItem("促销规则引擎", LoadType.INTRINSIC, 7),
        CognitiveLoadItem("复杂的 CI/CD 配置", LoadType.EXTRANEOUS, 6),
        CognitiveLoadItem("手动数据库迁移流程", LoadType.EXTRANEOUS, 5),
        CognitiveLoadItem("学习领域驱动设计", LoadType.GERMANE, 4),
        CognitiveLoadItem("跨服务链路追踪工具", LoadType.EXTRANEOUS, 3),
    ],
)
print(profile.health_report())

# team-cognitive-load-assessment.yaml
# 团队认知负荷季度评估模板

assessment:
  team: order-team
  quarter: 2026-Q2
  assessor: tech-lead

  domains_owned:
    - name: 订单核心流程
      complexity: high
      description: 包含状态机、幂等性、分布式事务
    - name: 购物车
      complexity: medium
      description: 会话管理、库存预占

  technologies_maintained:
    - name: Java 21 / Spring Boot 3
      proficiency: expert
    - name: PostgreSQL
      proficiency: expert
    - name: Redis
      proficiency: proficient
    - name: Kafka
      proficiency: learning
    - name: Kubernetes
      proficiency: learning

  external_dependencies:
    - team: payment-team
      interaction_mode: x-as-a-service
      friction: low
    - team: user-team
      interaction_mode: collaboration
      friction: medium
    - team: platform-team
      interaction_mode: x-as-a-service
      friction: high
      note: 平台文档不完善，经常需要找人问

  cognitive_load_scores:
    intrinsic: 7    # 业务复杂度
    extraneous: 6   # 环境复杂度
    germane: 4      # 学习投入
    total: 17
    per_person: 2.4  # 17 / 7 人

  actions:
    - 推动平台团队完善文档，降低外在负荷
    - 安排 Kafka 培训，将 learning 提升为 proficient
    - 考虑将购物车拆分给独立团队

// Amazon 内部服务接口规范示例（简化版）

/**
 * 每个团队的服务必须通过明确定义的接口暴露功能。
 * 接口定义必须满足以下要求：
 * 1. 版本化——支持向后兼容
 * 2. 自描述——不需要阅读源码即可理解
 * 3. 可外部化——任何接口都可能被暴露给外部
 */
public interface OrderService {

    /**
     * 创建订单。
     *
     * @param request 包含商品列表、收货地址、支付方式
     * @return 订单创建结果，包含订单 ID 和预估配送时间
     * @throws InvalidRequestException 请求参数不合法
     * @throws InventoryException 库存不足
     */
    CreateOrderResponse createOrder(CreateOrderRequest request)
        throws InvalidRequestException, InventoryException;

    /**
     * 查询订单状态。
     *
     * @param orderId 订单 ID
     * @return 订单当前状态
     * @throws OrderNotFoundException 订单不存在
     */
    OrderStatus getOrderStatus(String orderId)
        throws OrderNotFoundException;

    /**
     * 取消订单。
     * 只有在订单未发货前才能取消。
     *
     * @param orderId 订单 ID
     * @param reason 取消原因
     * @return 取消结果
     * @throws OrderNotFoundException 订单不存在
     * @throws OrderNotCancellableException 订单已发货，无法取消
     */
    CancelOrderResponse cancelOrder(String orderId, String reason)
        throws OrderNotFoundException, OrderNotCancellableException;
}

# Netflix 内部平台服务清单（简化版）

platform_services:
  deployment:
    - name: Spinnaker
      description: 持续交付平台，支持多云部署
      self_service: true
    - name: Nebula
      description: 构建工具（基于 Gradle）
      self_service: true

  resilience:
    - name: Hystrix
      description: 熔断器库
      self_service: true
    - name: Chaos Monkey
      description: 混沌工程工具
      self_service: true

  observability:
    - name: Atlas
      description: 时间序列指标平台
      self_service: true
    - name: Mantis
      description: 实时流处理平台
      self_service: true
    - name: Edgar
      description: 分布式追踪系统
      self_service: true

  data:
    - name: EVCache
      description: 分布式缓存（基于 Memcached）
      self_service: true
    - name: Cassandra / CockroachDB
      description: 分布式数据库
      self_service: true

  security:
    - name: Repokid
      description: 最小权限自动管理
      self_service: true
    - name: Security Monkey
      description: 安全合规自动检查
      self_service: true

// 逆康威策略实施检查清单

interface ConwayAssessmentItem {
  category: string;
  question: string;
  weight: number;
  answer: boolean | null;
}

const assessmentChecklist: ConwayAssessmentItem[] = [
  // 架构与组织对齐
  {
    category: "架构对齐",
    question: "每个微服务是否由且仅由一个团队拥有？",
    weight: 10,
    answer: null,
  },
  {
    category: "架构对齐",
    question: "团队边界是否与限界上下文（Bounded Context）一致？",
    weight: 9,
    answer: null,
  },
  {
    category: "架构对齐",
    question: "跨团队的服务依赖是否有明确的接口契约？",
    weight: 8,
    answer: null,
  },

  // 团队自治
  {
    category: "团队自治",
    question: "团队是否能独立完成从编码到部署的全流程？",
    weight: 9,
    answer: null,
  },
  {
    category: "团队自治",
    question: "团队是否拥有自己的数据存储，不与其他团队共享数据库？",
    weight: 8,
    answer: null,
  },
  {
    category: "团队自治",
    question: "团队是否可以独立做技术选型决策？",
    weight: 6,
    answer: null,
  },

  // 认知负荷
  {
    category: "认知负荷",
    question: "团队负责的领域数量是否不超过 2-3 个？",
    weight: 7,
    answer: null,
  },
  {
    category: "认知负荷",
    question: "是否有平台团队承担基础设施的复杂性？",
    weight: 8,
    answer: null,
  },
  {
    category: "认知负荷",
    question: "团队成员是否需要了解其他团队服务的内部实现？",
    weight: 7,
    answer: null,
  },

  // 沟通结构
  {
    category: "沟通结构",
    question: "高耦合服务的团队之间是否有高带宽沟通通道？",
    weight: 8,
    answer: null,
  },
  {
    category: "沟通结构",
    question: "松耦合服务的团队之间是否通过 API 而非口头沟通协调？",
    weight: 7,
    answer: null,
  },
  {
    category: "沟通结构",
    question: "是否定期审视组织结构与系统架构的一致性？",
    weight: 6,
    answer: null,
  },
];

function calculateScore(checklist: ConwayAssessmentItem[]): {
  score: number;
  maxScore: number;
  percentage: number;
  riskAreas: string[];
} {
  let score = 0;
  let maxScore = 0;
  const riskAreas: string[] = [];

  for (const item of checklist) {
    maxScore += item.weight;
    if (item.answer === true) {
      score += item.weight;
    } else if (item.answer === false) {
      riskAreas.push(`[${item.category}] ${item.question}`);
    }
  }

  return {
    score,
    maxScore,
    percentage: maxScore > 0 ? (score / maxScore) * 100 : 0,
    riskAreas,
  };
}

# 组织-架构对齐度量工具

from dataclasses import dataclass
from typing import List, Tuple, Set


@dataclass
class DeploymentRecord:
    service: str
    team: str
    timestamp: str
    success: bool


@dataclass
class CrossTeamPR:
    pr_id: str
    source_team: str
    target_repo_team: str
    merge_time_hours: float


def deployment_independence_ratio(
    records: List[DeploymentRecord],
) -> float:
    """部署独立性比率。

    计算不需要跨团队协调的部署占总部署的比例。
    目标值：> 0.9（90% 以上的部署应该是独立的）。
    """
    if not records:
        return 0.0

    # 按天分组，检查同一天是否有多个团队部署
    from collections import defaultdict
    daily: defaultdict[str, Set[str]] = defaultdict(set)
    for r in records:
        day = r.timestamp[:10]  # YYYY-MM-DD
        daily[day].add(r.team)

    independent_days = sum(
        1 for teams in daily.values() if len(teams) == 1
    )
    return independent_days / len(daily)


def cross_team_pr_ratio(
    prs: List[CrossTeamPR],
    total_prs: int,
) -> float:
    """跨团队 PR 比率。

    计算需要修改其他团队代码仓库的 PR 占总 PR 的比例。
    目标值：< 0.05（低于 5%）。
    """
    if total_prs == 0:
        return 0.0
    cross = sum(1 for pr in prs if pr.source_team != pr.target_repo_team)
    return cross / total_prs


def cross_team_pr_merge_time(prs: List[CrossTeamPR]) -> float:
    """跨团队 PR 平均合并时间（小时）。

    目标值：< 24 小时。
    如果远高于团队内 PR 的合并时间，说明跨团队协调成本过高。
    """
    cross_prs = [
        pr for pr in prs if pr.source_team != pr.target_repo_team
    ]
    if not cross_prs:
        return 0.0
    return sum(pr.merge_time_hours for pr in cross_prs) / len(cross_prs)


def team_coupling_matrix(
    dependencies: List[Tuple[str, str]],
    teams: List[str],
) -> List[List[int]]:
    """团队耦合矩阵。

    生成一个 N x N 的矩阵，表示团队间的依赖次数。
    对角线上的数字应该远大于非对角线上的数字。
    """
    team_idx = {t: i for i, t in enumerate(teams)}
    n = len(teams)
    matrix = [[0] * n for _ in range(n)]

    for src, tgt in dependencies:
        if src in team_idx and tgt in team_idx:
            matrix[team_idx[src]][team_idx[tgt]] += 1

    return matrix

package governance

import "time"

// AlignmentReview 架构-组织对齐审查
type AlignmentReview struct {
    ReviewDate     time.Time          `json:"review_date"`
    Reviewer       string             `json:"reviewer"`
    TeamSnapshots  []TeamSnapshot     `json:"team_snapshots"`
    ServiceMap     []ServiceOwnership `json:"service_map"`
    Findings       []Finding          `json:"findings"`
    Actions        []Action           `json:"actions"`
}

// TeamSnapshot 团队快照
type TeamSnapshot struct {
    Name              string   `json:"name"`
    Size              int      `json:"size"`
    Type              string   `json:"type"` // stream-aligned, enabling, platform, complicated-subsystem
    OwnedServices     []string `json:"owned_services"`
    CognitiveLoadScore int     `json:"cognitive_load_score"`
}

// ServiceOwnership 服务归属
type ServiceOwnership struct {
    ServiceName string   `json:"service_name"`
    OwnerTeam   string   `json:"owner_team"`
    Consumers   []string `json:"consumers"`
    APIStability string  `json:"api_stability"` // stable, evolving, experimental
}

// Finding 审查发现
type Finding struct {
    Severity    string `json:"severity"` // critical, warning, info
    Category    string `json:"category"`
    Description string `json:"description"`
    Impact      string `json:"impact"`
}

// Action 整改行动
type Action struct {
    Description string    `json:"description"`
    Owner       string    `json:"owner"`
    DueDate     time.Time `json:"due_date"`
    Status      string    `json:"status"` // pending, in-progress, done
}

// RunQuarterlyReview 执行季度对齐审查
func RunQuarterlyReview(review *AlignmentReview) []Finding {
    var findings []Finding

    // 检查 1：是否存在没有明确归属的服务
    for _, svc := range review.ServiceMap {
        if svc.OwnerTeam == "" {
            findings = append(findings, Finding{
                Severity:    "critical",
                Category:    "所有权",
                Description: "服务 " + svc.ServiceName + " 没有明确的归属团队",
                Impact:      "无人负责意味着无人维护，技术债务会快速积累",
            })
        }
    }

    // 检查 2：是否存在认知负荷过高的团队
    for _, team := range review.TeamSnapshots {
        if team.CognitiveLoadScore > 20 {
            findings = append(findings, Finding{
                Severity:    "warning",
                Category:    "认知负荷",
                Description: "团队 " + team.Name + " 的认知负荷评分过高",
                Impact:      "可能导致交付速度下降和缺陷率上升",
            })
        }
        if len(team.OwnedServices) > 5 {
            findings = append(findings, Finding{
                Severity:    "warning",
                Category:    "服务数量",
                Description: "团队 " + team.Name + " 拥有超过 5 个服务",
                Impact:      "团队注意力分散，建议拆分或转移部分服务",
            })
        }
    }

    // 检查 3：是否存在一个服务被多个团队修改
    serviceModifiers := make(map[string][]string)
    for _, team := range review.TeamSnapshots {
        for _, svc := range team.OwnedServices {
            serviceModifiers[svc] = append(serviceModifiers[svc], team.Name)
        }
    }
    for svc, teams := range serviceModifiers {
        if len(teams) > 1 {
            findings = append(findings, Finding{
                Severity:    "critical",
                Category:    "所有权冲突",
                Description: "服务 " + svc + " 被多个团队声称拥有",
                Impact:      "共享拥有权会导致责任不清和变更冲突",
            })
        }
    }

    return findings
}

# team-api.yaml —— 订单团队的团队 API

team:
  name: 订单团队
  type: stream-aligned
  mission: >
    让用户能够顺畅地完成从加入购物车到收到商品的全流程，
    持续优化订单转化率和用户满意度。

contact:
  slack_channel: "#team-order"
  email: order-team@example.com
  on_call_rotation: PagerDuty - Order Team

services_owned:
  - name: order-service
    repo: github.com/example/order-service
    api_docs: https://api-docs.internal/order
    sla: 99.95% 可用性，P99 延迟 < 200ms

  - name: cart-service
    repo: github.com/example/cart-service
    api_docs: https://api-docs.internal/cart
    sla: 99.9% 可用性，P99 延迟 < 100ms

interaction_modes:
  - with: payment-team
    mode: x-as-a-service
    description: 通过支付 API 发起支付请求
    contract: https://api-docs.internal/payment/v2

  - with: user-team
    mode: x-as-a-service
    description: 通过用户 API 查询用户信息
    contract: https://api-docs.internal/user/v3

  - with: platform-team
    mode: x-as-a-service
    description: 使用 CI/CD 平台、监控平台、日志平台

  - with: sre-enabling-team
    mode: facilitating
    description: SRE 赋能团队协助我们建立 SLO 体系
    duration: 2026-Q1 至 2026-Q2

work_in_progress:
  current_focus:
    - 订单状态机重构
    - 支持预售订单类型
  upcoming:
    - 国际化订单支持
    - 订单履约与物流对接

preferred_communication:
  sync: 每周二下午技术交流会
  async: Slack 频道优先，非紧急问题 24 小时内响应
  code_review: 跨团队 PR 请提前在 Slack 沟通背景

versioning:
  api_versioning: URL 路径版本（/v1、/v2）
  breaking_changes: 至少提前 2 个迭代通知消费方
  deprecation_policy: 旧版本在新版本发布后保留 3 个月

# ODR-007：将购物车功能从订单团队拆分为独立团队

## 状态
已批准（2026-03-15）

## 背景
订单团队当前负责订单核心流程和购物车两个领域。
随着购物车功能的复杂度增长（社交分享、智能推荐、库存预占），
团队的认知负荷评分已达到 19 分（目标上限 15 分）。

## 决策
成立独立的购物车团队（流对齐团队），从订单团队拆分出购物车相关服务。

## 影响
- 订单团队认知负荷降至 12 分
- 购物车团队初始认知负荷约 8 分
- 两个团队之间通过事件驱动的方式交互（订单创建事件）
- 需要从订单团队调配 3 名熟悉购物车业务的工程师

## 风险
- 拆分初期可能出现职责模糊
- 需要定义清晰的事件契约

## 审查时间
2026-06-15（拆分后 3 个月进行效果评估）