Files
sanguo_moziplus_v3/docs/design/01-design-v0.1.md
T
claude_dev 37579ebdcd refactor: 重新组织文档结构
- docs/PRD.md - 产品需求文档(移到根目录)
- docs/design/01-design-v0.1.md - 系统设计文档(移到子目录)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-29 08:25:00 +08:00

751 lines
21 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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 桥接器接口
```python
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**:
```python
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**:
```python
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 任务模型
```python
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 编排器实现
```python
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)
```python
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 表
```sql
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 表
```sql
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 表
```sql
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 模板
```markdown
# {项目名称}
## 工作模式
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)
```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
```python
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 命令行接口
```bash
# 委派任务
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 日志策略
```python
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
**审核状态**: 待审核