refactor: 将设计文档移到 docs 目录

- 移动 design/01-design-v0.1.md → docs/01-design-v0.1.md
- 删除空的 design 目录

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-29 08:21:47 +08:00
parent 37ba76f92e
commit d811a5e212
+750
View File
@@ -0,0 +1,750 @@
# 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
**审核状态**: 待审核