sanguo_moziplus_v3/docs/design/01-design-v0.1.md

sanguo_moziplus_v3 设计文档 v0.1

项目名称: sanguo_moziplus_v3 版本: v0.1 创建日期: 2026-06-29 状态: Draft

---

1. 架构概述

1.1 系统架构图

┌─────────────────────────────────────────────────────────────────┐
│                        Claude Code CLI                          │
│                     (用户交互层)                                  │
└─────────────────────────────┬───────────────────────────────────┘
                              │
┌─────────────────────────────▼───────────────────────────────────┐
│                      CLAUDE.md (规则层)                          │
│  - 协作规范                                                       │
│  - 角色分工                                                       │
│  - Git 规范                                                      │
└─────────────────────────────┬───────────────────────────────────┘
                              │
┌─────────────────────────────▼───────────────────────────────────┐
│                   Superpowers (能力层)                           │
│  - Planning Skill                                                │
│  - Review Skill                                                   │
│  - Debug Skill                                                    │
└─────────────────────────────┬───────────────────────────────────┘
                              │
┌─────────────────────────────▼───────────────────────────────────┐
│                      CCB (通信层)                                │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐              │
│  │Bridge   │  │Bridge   │  │Bridge   │  │Bridge   │              │
│  │Codex    │  │Gemini   │  │GPT      │  │Custom   │              │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘              │
└───────┼────────────┼────────────┼────────────┼───────────────────┘
        │            │            │            │
        ▼            ▼            ▼            ▼
    ┌───────┐   ┌───────┐   ┌───────┐   ┌───────┐
    │Codex  │   │Gemini │   │GPT    │   │Custom │
    │API    │   │API    │   │API    │   │API    │
    └───────┘   └───────┘   └───────┘   └───────┘

┌─────────────────────────────────────────────────────────────────┐
│                    任务存储层 (SQLite)                           │
│  - tasks (任务表)                                                │
│  - collaborations (协作记录表)                                   │
│  - costs (成本表)                                                │
└─────────────────────────────────────────────────────────────────┘

1.2 核心设计原则

1. 分层解耦: 规则层、能力层、通信层清晰分离
2. 可扩展: 桥接器插件化，支持新增 AI 模型
3. 可配置: 通过配置文件定义角色和协作模式
4. 成本优先: Claude 只做决策，体力活交给便宜 AI

---

2. 核心组件设计

2.1 CCB 桥接器 (Bridge)

2.1.1 桥接器接口

class BridgeAgent(ABC):
    """桥接器抽象基类"""

    @abstractmethod
    async def ask(self, prompt: str, context: dict) -> BridgeResponse:
        """向 AI 发送请求"""
        pass

    @abstractmethod
    def get_cost(self) -> TokenCost:
        """获取成本统计"""
        pass

    @abstractmethod
    def get_capabilities(self) -> List[str]:
        """获取能力列表"""
        pass

2.1.2 具体实现

Codex Bridge:

class CodexBridge(BridgeAgent):
    def __init__(self, api_key: str, endpoint: str):
        self.api_key = api_key
        self.endpoint = endpoint

    async def ask(self, prompt: str, context: dict) -> BridgeResponse:
        # 调用 Codex API
        response = await self._call_codex(prompt, context)
        return BridgeResponse(
            content=response.content,
            tokens_used=response.tokens,
            cost=response.tokens * CODEX_PRICE
        )

Gemini Bridge:

class GeminiBridge(BridgeAgent):
    def __init__(self, api_key: str, model: str = "gemini-pro"):
        self.api_key = api_key
        self.model = model

    async def ask(self, prompt: str, context: dict) -> BridgeResponse:
        # 调用 Gemini API
        response = await self._call_gemini(prompt, context)
        return BridgeResponse(
            content=response.content,
            tokens_used=response.tokens,
            cost=response.tokens * GEMINI_PRICE
        )

2.2 任务编排器 (Orchestrator)

2.2.1 任务模型

from enum import Enum
from datetime import datetime
from typing import Optional, List, Dict, Any

class TaskType(Enum):
    PLAN = "plan"           # 规划任务
    CODE_BACKEND = "code_backend"   # 后端编码
    CODE_FRONTEND = "code_frontend" # 前端编码
    REVIEW = "review"       # 代码审查
    VERIFY = "verify"       # 验收测试

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

class AgentRole(Enum):
    CLAUDE = "claude"       # PM / 架构师
    CODEX = "codex"         # 后端开发
    GEMINI = "gemini"       # 前端开发
    GPT = "gpt"             # 通用助手

@dataclass
class TokenCost:
    model: str
    input_tokens: int
    output_tokens: int
    total_cost: float

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens

@dataclass
class Task:
    id: str
    type: TaskType
    title: str
    description: str
    assigned_to: AgentRole
    status: TaskStatus = TaskStatus.PENDING
    dependencies: List[str] = field(default_factory=list)
    input: Dict[str, Any] = field(default_factory=dict)
    output: Dict[str, Any] = field(default_factory=dict)
    cost: Optional[TokenCost] = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
    error_message: Optional[str] = None

    def can_execute(self, completed_tasks: set) -> bool:
        """检查任务是否可以执行（依赖已完成）"""
        return all(dep in completed_tasks for dep in self.dependencies)

2.2.2 编排器实现

class TaskOrchestrator:
    """任务编排器"""

    def __init__(self, db_path: str):
        self.db = sqlite3.connect(db_path)
        self.bridges: Dict[AgentRole, BridgeAgent] = {}
        self._init_bridges()

    def _init_bridges(self):
        """初始化桥接器"""
        config = self._load_config()
        if config.get("codex", {}).get("enabled"):
            self.bridges[AgentRole.CODEX] = CodexBridge(...)
        if config.get("gemini", {}).get("enabled"):
            self.bridges[AgentRole.GEMINI] = GeminiBridge(...)

    async def create_task(self, task: Task) -> str:
        """创建任务"""
        await self.db.execute(
            "INSERT INTO tasks ...",
            task.to_dict()
        )
        return task.id

    async def execute_task(self, task_id: str) -> Task:
        """执行任务"""
        task = await self._get_task(task_id)

        # 检查依赖
        if not task.can_execute(await self._get_completed_tasks()):
            raise TaskBlockedError("Dependencies not met")

        # 获取桥接器
        bridge = self.bridges.get(task.assigned_to)
        if not bridge:
            raise BridgeNotFoundError(f"No bridge for {task.assigned_to}")

        # 执行任务
        task.status = TaskStatus.IN_PROGRESS
        await self._update_task(task)

        try:
            response = await bridge.ask(
                prompt=task.description,
                context=task.input
            )

            task.output = response.content
            task.cost = response.cost
            task.status = TaskStatus.COMPLETED
            task.completed_at = datetime.now()

        except Exception as e:
            task.status = TaskStatus.FAILED
            task.error_message = str(e)

        await self._update_task(task)
        return task

2.3 成本追踪器 (CostTracker)

class CostTracker:
    """成本追踪器"""

    async def record_cost(self, task_id: str, cost: TokenCost):
        """记录成本"""
        await self.db.execute(
            "INSERT INTO costs (task_id, model, input_tokens, output_tokens, total_cost) VALUES (?, ?, ?, ?, ?)",
            (task_id, cost.model, cost.input_tokens, cost.output_tokens, cost.total_cost)
        )

    async def get_cost_report(self, project_id: str) -> CostReport:
        """获取成本报告"""
        rows = await self.db.execute(
            "SELECT model, SUM(input_tokens), SUM(output_tokens), SUM(total_cost) FROM costs WHERE project_id = ? GROUP BY model",
            (project_id,)
        )

        return CostReport(
            by_model=[{r[0]: {"tokens": r[1]+r[2], "cost": r[3]}} for r in rows],
            total=sum(r[3] for r in rows)
        )

---

3. 数据库设计

3.1 表结构

tasks 表

CREATE TABLE tasks (
    id TEXT PRIMARY KEY,
    type TEXT NOT NULL,
    title TEXT NOT NULL,
    description TEXT NOT NULL,
    assigned_to TEXT NOT NULL,
    status TEXT NOT NULL,
    dependencies TEXT,  -- JSON array
    input TEXT,         -- JSON object
    output TEXT,        -- JSON object
    cost_model TEXT,
    cost_input_tokens INTEGER,
    cost_output_tokens INTEGER,
    cost_total REAL,
    created_at TEXT NOT NULL,
    completed_at TEXT,
    error_message TEXT
);

CREATE INDEX idx_tasks_status ON tasks(status);
CREATE INDEX idx_tasks_assigned ON tasks(assigned_to);

costs 表

CREATE TABLE costs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id TEXT NOT NULL,
    model TEXT NOT NULL,
    input_tokens INTEGER NOT NULL,
    output_tokens INTEGER NOT NULL,
    total_cost REAL NOT NULL,
    recorded_at TEXT NOT NULL,
    FOREIGN KEY (task_id) REFERENCES tasks(id)
);

CREATE INDEX idx_costs_model ON costs(model);
CREATE INDEX idx_costs_task ON costs(task_id);

collaborations 表

CREATE TABLE collaborations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    project_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    from_agent TEXT NOT NULL,
    to_agent TEXT NOT NULL,
    message TEXT NOT NULL,
    created_at TEXT NOT NULL,
    FOREIGN KEY (task_id) REFERENCES tasks(id)
);

---

4. 协作模式设计

4.1 串行审查模式 (Superpowers)

┌─────────────┐
│ Implementer │
│  (Codex)    │
└──────┬──────┘
       │ task_completed
       ▼
┌─────────────┐
│Spec Reviewer│
│  (Gemini)   │
└──────┬──────┘
       │ spec_passed
       ▼
┌──────────────────┐
│ Code Quality      │
│   Reviewer        │
│   (Gemini)        │
└──────┬───────────┘
       │ quality_passed
       ▼
┌─────────────┐
│   Next Task │
└─────────────┘

实现要点:

• Implementer 完成任务后报告状态
• Spec Reviewer 先检查实现与需求一致性
• 通过后再进行 Code Quality Review
• 任一环节失败则返回 Implementer

4.2 对抗辩论模式 (TradingAgents)

┌──────────────────┐
│ 4 Analysts (并行) │
└────┬────────┬────┘
     │        │
     ▼        ▼
┌─────────┐ ┌─────────┐
│  Bull   │ │  Bear   │
└────┬────┘ └────┬────┘
     │           │
     └─────┬─────┘
           ▼
    ┌─────────┐
    │Research │
    │ Manager │
    └────┬────┘
         ▼
    ┌─────────┐
    │ Trader  │
    └─────────┘

实现要点:

• 多个 Analyst 并行分析
• Bull 和 Bear 交替辩论（最多 N 轮）
• Research Manager 裁决并综合
• 最终由 Trader 做决策

4.3 三方共识模式 (oh-my-claudecode)

┌─────────┐
│ Planner │
└────┬────┘
     │ plan_ready
     ▼
┌───────────┐
│ Architect │
└────┬──────┘
     │ architecture_approved
     ▼
┌─────────┐
│ Critic  │
└────┬────┘
     │ consensus_reached
     ▼
┌─────────────┐
│   Execute   │
└─────────────┘

实现要点:

• Planner 先做规划
• Architect 审查架构
• Critic 提出批判
• 三方达成共识后才执行

---

5. 配置设计

5.1 CLAUDE.md 模板

# {项目名称}

## 工作模式

Superpowers + AI 协作

## 角色分工

### Claude (我) — 架构师 / 项目经理
- 需求分析、架构设计、任务拆分
- 使用 Superpowers 进行规划、审查、调试
- 代码审核、最终验收、Git 提交管理
- **绝对不亲自编写代码**

### Codex — 后端开发
- 服务端代码、API、数据库、Migration
- 单元测试、集成测试
- 通过 `ask codex "..."` 调用

### Gemini — 前端开发
- 前端组件、页面、样式、交互逻辑
- 代码审查、安全审计
- 通过 `ask gemini "..."` 调用

## 协作规范

1. Claude 不写代码，所有编码任务委派给 Codex/Gemini
2. 后端任务 → Codex，前端任务 → Gemini
3. 代码审查 → Gemini
4. 验收测试 → Claude

## Git 规范

- 分支命名: `feature/<task-name>`
- 提交格式: `<type>: <description>`
- 合并策略: Squash merge

5.2 配置文件 (config.yaml)

# 项目配置
project:
  name: "sanguo_moziplus_v3"
  description: "多 AI 桥接协作系统"

# 桥接器配置
bridges:
  codex:
    enabled: true
    endpoint: "http://localhost:8080"
    api_key_env: "CODEX_API_KEY"
    model: "codex-4"
    max_tokens: 4000

  gemini:
    enabled: true
    endpoint: "http://localhost:8081"
    api_key_env: "GEMINI_API_KEY"
    model: "gemini-pro"
    max_tokens: 4000

  gpt:
    enabled: false
    endpoint: "https://api.openai.com/v1"
    api_key_env: "OPENAI_API_KEY"
    model: "gpt-4"
    max_tokens: 4000

# 协作模式配置
collaboration:
  default_mode: "sequential"  # sequential, debate, consensus
  max_retries: 3
  timeout_seconds: 300

# 成本配置
cost:
  tracking_enabled: true
  alert_threshold: 10.0  # 美元
  models:
    claude:
      input_price: 0.003  # per 1k tokens
      output_price: 0.015
    codex:
      input_price: 0.0001
      output_price: 0.0002
    gemini:
      input_price: 0.0001
      output_price: 0.0001
    gpt:
      input_price: 0.03
      output_price: 0.06

# 数据库配置
database:
  path: "~/.moziplus-v3/tasks.db"
  backup_enabled: true
  backup_interval_hours: 24

---

6. API 设计

6.1 Claude Code 集成 API

class CCBClient:
    """CCB 客户端，供 Claude Code 调用"""

    async def ask(self, agent: str, task: str, context: dict = None) -> dict:
        """
        委派任务给指定 AI

        Args:
            agent: AI 名称 (codex, gemini, gpt)
            task: 任务描述
            context: 上下文信息

        Returns:
            {
                "task_id": "...",
                "status": "pending",
                "estimated_cost": 0.01
            }
        """
        pass

    async def status(self, task_id: str) -> dict:
        """查询任务状态"""
        pass

    async def cost(self, project_id: str = None) -> dict:
        """获取成本统计"""
        pass

    async def config(self, key: str = None, value: Any = None) -> dict:
        """配置管理"""
        pass

6.2 命令行接口

# 委派任务
moziplus ask codex "实现用户认证 API"

# 查询状态
moziplus status <task_id>

# 成本报告
moziplus cost

# 配置管理
moziplus config set collaboration.default_mode debate

---

7. 实现计划

7.1 Phase 1: 桥接器实现 (Week 1)

• 实现 BridgeAgent 抽象类
• 实现 CodexBridge
• 实现 GeminiBridge
• 实现 GPTBridge
• 单元测试

7.2 Phase 2: 任务编排器 (Week 1-2)

• 实现任务模型
• 实现数据库层
• 实现任务编排器
• 实现成本追踪器
• 集成测试

7.3 Phase 3: Claude Code 集成 (Week 2)

• 实现 CCBClient
• 实现命令行接口
• 实现 CLAUDE.md 加载
• 端到端测试

7.4 Phase 4: 协作模式 (Week 3)

• 实现串行审查模式
• 实现对抗辩论模式
• 实现三方共识模式
• 模式测试

---

8. 技术栈

组件	技术	说明
语言	Python 3.11+	主要开发语言
数据库	SQLite	任务存储
异步	asyncio	异步任务处理
配置	PyYAML	配置文件解析
CLI	Click / Typer	命令行接口
测试	pytest	单元测试和集成测试
API 调用	httpx	HTTP 客户端

---

9. 部署架构

┌─────────────────────────────────────────────────┐
│              开发环境                              │
│  ~/.openclaw/sanguo_projects/sanguo_moziplus_v3  │
└─────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────┐
│              安装环境                              │
│  ~/.sanguo_projects/sanguo_moziplus_v3           │
└─────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────┐
│              数据目录                              │
│  ~/.moziplus-v3/                                 │
│    - tasks.db                                    │
│    - config.yaml                                 │
│    - logs/                                       │
└─────────────────────────────────────────────────┘

---

10. 与 v2 的集成

10.1 兼容性设计

• v3 可独立运行
• 可选择性集成 v2 的黑板架构
• 共享数据库 schema（部分表）

10.2 迁移路径

v2 用户 → v3 迁移步骤：
1. 安装 v3
2. 配置桥接器
3. 更新 CLAUDE.md
4. 运行迁移脚本
5. 验证功能

---

11. 测试策略

11.1 单元测试

• 桥接器测试（Mock API）
• 任务模型测试
• 成本计算测试

11.2 集成测试

• 端到端任务流程
• 多 AI 协作场景
• 成本追踪验证

11.3 压力测试

• 并发任务处理
• 大量任务调度
• 长时间运行稳定性

---

12. 监控和日志

12.1 日志策略

import logging

logger = logging.getLogger("moziplus-v3")

# 任务日志
logger.info(f"Task {task_id} assigned to {agent}")

# 成本日志
logger.info(f"Cost: {model} {tokens} tokens = ${cost}")

# 错误日志
logger.error(f"Task {task_id} failed: {error}")

12.2 监控指标

• 任务成功率
• 平均任务耗时
• 成本趋势
• 桥接器可用性

---

文档版本: v0.1 最后更新: 2026-06-29 作者: Claude Dev 审核状态: 待审核