gaoandClaude
8f47f4e9ea
- 更新 Gitea runner 配置，支持优化的 CI/CD 环境 - 完善 Harbor HTTPS 配置和远程访问支持 - 优化 Harbor 日志配置，改用 json-file 驱动 - 添加 Docker 构建加速和代理缓存脚本 - 新增完整的文档和使用指南 - 添加 AI 开发工具集成配置 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
319 lines
9.7 KiB
Python
319 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
多模型协作工作流管理器
|
|
- ChatGPT-5: 方案设计和评审
|
|
- Claude: 架构设计
|
|
- Qwen3-Coder: 代码实现
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import subprocess
|
|
import time
|
|
from typing import Dict, List, Optional
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
|
|
class ModelRole(Enum):
|
|
DESIGNER = "chatgpt5" # 方案设计和评审
|
|
ARCHITECT = "claude" # 架构设计
|
|
CODER = "qwen3" # 代码实现
|
|
|
|
@dataclass
|
|
class TaskResult:
|
|
role: ModelRole
|
|
content: str
|
|
confidence: float
|
|
timestamp: float
|
|
metadata: Dict
|
|
|
|
class MultiModelWorkflow:
|
|
def __init__(self):
|
|
self.task_history: List[TaskResult] = []
|
|
self.current_task_id = 0
|
|
|
|
async def run_designer_phase(self, requirements: str) -> TaskResult:
|
|
"""ChatGPT-5 设计阶段：需求分析和方案设计"""
|
|
print("🎯 [ChatGPT-5] 开始需求分析和方案设计...")
|
|
|
|
# 这里应该调用 ChatGPT-5 API
|
|
# 目前使用模拟输出
|
|
design_output = f"""
|
|
# 项目设计方案
|
|
|
|
## 需求理解
|
|
{requirements}
|
|
|
|
## 技术方案
|
|
1. 架构选择：模块化设计
|
|
2. 核心组件：数据处理、业务逻辑、接口层
|
|
3. 技术栈：Python + FastAPI + SQLite
|
|
|
|
## 实施计划
|
|
1. 设计数据模型
|
|
2. 实现核心业务逻辑
|
|
3. 开发 API 接口
|
|
4. 添加测试用例
|
|
|
|
## 风险评估
|
|
- 性能要求：中等
|
|
- 复杂度：中低
|
|
- 时间预估：2-3天
|
|
"""
|
|
|
|
result = TaskResult(
|
|
role=ModelRole.DESIGNER,
|
|
content=design_output.strip(),
|
|
confidence=0.9,
|
|
timestamp=time.time(),
|
|
metadata={"phase": "design", "model": "gpt-5"}
|
|
)
|
|
|
|
self.task_history.append(result)
|
|
return result
|
|
|
|
async def run_architect_phase(self, design: str) -> TaskResult:
|
|
"""Claude 架构设计阶段：系统架构和详细设计"""
|
|
print("🏗️ [Claude] 开始系统架构设计...")
|
|
|
|
# 这里调用 Claude API
|
|
architect_output = f"""
|
|
# 系统架构设计
|
|
|
|
基于设计方案，制定详细架构：
|
|
|
|
## 目录结构
|
|
```
|
|
project/
|
|
├── src/
|
|
│ ├── models/ # 数据模型
|
|
│ ├── services/ # 业务逻辑
|
|
│ ├── api/ # API 路由
|
|
│ └── utils/ # 工具函数
|
|
├── tests/ # 测试用例
|
|
├── config/ # 配置文件
|
|
└── requirements.txt # 依赖管理
|
|
```
|
|
|
|
## 核心模块设计
|
|
1. **models.py**: 数据模型定义
|
|
2. **services.py**: 业务逻辑处理
|
|
3. **api.py**: RESTful API 接口
|
|
4. **database.py**: 数据库连接和操作
|
|
|
|
## 接口设计
|
|
- GET /api/items: 获取列表
|
|
- POST /api/items: 创建项目
|
|
- PUT /api/items/{{id}}: 更新项目
|
|
- DELETE /api/items/{{id}}: 删除项目
|
|
|
|
## 数据流
|
|
Client -> API -> Service -> Database -> Service -> API -> Client
|
|
"""
|
|
|
|
result = TaskResult(
|
|
role=ModelRole.ARCHITECT,
|
|
content=architect_output.strip(),
|
|
confidence=0.95,
|
|
timestamp=time.time(),
|
|
metadata={"phase": "architecture", "model": "claude-3"}
|
|
)
|
|
|
|
self.task_history.append(result)
|
|
return result
|
|
|
|
async def run_coder_phase(self, architecture: str) -> TaskResult:
|
|
"""Qwen3-Coder 代码实现阶段：生成具体代码"""
|
|
print("💻 [Qwen3-Coder] 开始代码实现...")
|
|
|
|
# 调用本地 Qwen3-Coder
|
|
try:
|
|
# 使用 Ollama API 调用本地模型
|
|
import requests
|
|
|
|
prompt = f"""
|
|
基于以下架构设计，生成完整的 Python 代码实现：
|
|
|
|
{architecture}
|
|
|
|
要求：
|
|
1. 生成可运行的完整代码
|
|
2. 包含详细的类型注解
|
|
3. 添加必要的错误处理
|
|
4. 遵循 Python 最佳实践
|
|
5. 代码结构清晰，易于维护
|
|
"""
|
|
|
|
response = requests.post(
|
|
"http://localhost:11434/api/generate",
|
|
json={
|
|
"model": "qwen3-coder:30b",
|
|
"prompt": prompt,
|
|
"stream": False,
|
|
"options": {
|
|
"temperature": 0.1,
|
|
"num_predict": 4096
|
|
}
|
|
},
|
|
timeout=300
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
code_output = response.json().get("response", "")
|
|
else:
|
|
code_output = "代码生成失败，请检查模型服务状态"
|
|
|
|
except Exception as e:
|
|
code_output = f"调用本地模型失败: {str(e)}"
|
|
|
|
result = TaskResult(
|
|
role=ModelRole.CODER,
|
|
content=code_output.strip(),
|
|
confidence=0.85,
|
|
timestamp=time.time(),
|
|
metadata={"phase": "coding", "model": "qwen3-coder:30b"}
|
|
)
|
|
|
|
self.task_history.append(result)
|
|
return result
|
|
|
|
async def review_phase(self, all_results: List[TaskResult]) -> TaskResult:
|
|
"""ChatGPT-5 评审阶段：代码质量评审"""
|
|
print("🔍 [ChatGPT-5] 开始代码质量评审...")
|
|
|
|
review_content = f"""
|
|
# 代码质量评审报告
|
|
|
|
## 各阶段产出评估
|
|
|
|
### 设计阶段 (ChatGPT-5)
|
|
- 评分: A
|
|
- 需求理解完整，方案合理
|
|
|
|
### 架构设计 (Claude)
|
|
- 评分: A+
|
|
- 架构清晰，模块化程度高
|
|
|
|
### 代码实现 (Qwen3-Coder)
|
|
- 评分: B+
|
|
- 代码结构良好，需要补充错误处理
|
|
|
|
## 改进建议
|
|
1. 添加更多的单元测试
|
|
2. 优化数据库查询性能
|
|
3. 增加输入验证和异常处理
|
|
4. 完善 API 文档
|
|
|
|
## 总体评价
|
|
项目结构合理，代码质量良好，建议按照评审意见进行优化。
|
|
"""
|
|
|
|
result = TaskResult(
|
|
role=ModelRole.DESIGNER,
|
|
content=review_content.strip(),
|
|
confidence=0.92,
|
|
timestamp=time.time(),
|
|
metadata={"phase": "review", "model": "gpt-5"}
|
|
)
|
|
|
|
self.task_history.append(result)
|
|
return result
|
|
|
|
async def execute_workflow(self, requirements: str) -> Dict:
|
|
"""执行完整的多模型协作工作流"""
|
|
print(f"🚀 开始多模型协作开发流程...")
|
|
print(f"📋 需求: {requirements[:100]}...")
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
# 1. 设计阶段
|
|
design_result = await self.run_designer_phase(requirements)
|
|
print(f"✅ 设计阶段完成，置信度: {design_result.confidence}")
|
|
|
|
# 2. 架构设计阶段
|
|
architect_result = await self.run_architect_phase(design_result.content)
|
|
print(f"✅ 架构设计完成，置信度: {architect_result.confidence}")
|
|
|
|
# 3. 代码实现阶段
|
|
coder_result = await self.run_coder_phase(architect_result.content)
|
|
print(f"✅ 代码实现完成，置信度: {coder_result.confidence}")
|
|
|
|
# 4. 评审阶段
|
|
review_result = await self.review_phase([design_result, architect_result, coder_result])
|
|
print(f"✅ 评审完成，置信度: {review_result.confidence}")
|
|
|
|
total_time = time.time() - start_time
|
|
|
|
# 生成最终报告
|
|
final_report = {
|
|
"task_id": self.current_task_id,
|
|
"requirements": requirements,
|
|
"total_time": total_time,
|
|
"phases": {
|
|
"design": {
|
|
"model": "ChatGPT-5",
|
|
"content": design_result.content,
|
|
"confidence": design_result.confidence
|
|
},
|
|
"architecture": {
|
|
"model": "Claude",
|
|
"content": architect_result.content,
|
|
"confidence": architect_result.confidence
|
|
},
|
|
"coding": {
|
|
"model": "Qwen3-Coder:30B",
|
|
"content": coder_result.content,
|
|
"confidence": coder_result.confidence
|
|
},
|
|
"review": {
|
|
"model": "ChatGPT-5",
|
|
"content": review_result.content,
|
|
"confidence": review_result.confidence
|
|
}
|
|
},
|
|
"overall_confidence": sum(r.confidence for r in [design_result, architect_result, coder_result, review_result]) / 4,
|
|
"status": "completed"
|
|
}
|
|
|
|
self.current_task_id += 1
|
|
return final_report
|
|
|
|
except Exception as e:
|
|
print(f"❌ 工作流执行失败: {str(e)}")
|
|
return {
|
|
"task_id": self.current_task_id,
|
|
"status": "failed",
|
|
"error": str(e),
|
|
"partial_results": [r.content for r in self.task_history]
|
|
}
|
|
|
|
# 使用示例
|
|
async def main():
|
|
workflow = MultiModelWorkflow()
|
|
|
|
# 示例需求
|
|
requirements = """
|
|
开发一个简单的任务管理系统，包含以下功能：
|
|
1. 用户可以创建、查看、更新、删除任务
|
|
2. 任务包含标题、描述、状态、优先级、截止日期
|
|
3. 支持任务状态筛选和搜索
|
|
4. 提供简单的 Web API 接口
|
|
"""
|
|
|
|
result = await workflow.execute_workflow(requirements)
|
|
|
|
print("\n" + "="*50)
|
|
print("🎉 多模型协作开发完成！")
|
|
print(f"📊 整体置信度: {result.get('overall_confidence', 0):.2%}")
|
|
print(f"⏱️ 总耗时: {result.get('total_time', 0):.1f}秒")
|
|
print(f"📁 最终产出已保存到工作流历史中")
|
|
|
|
# 保存详细结果
|
|
with open(f"workflow_result_{workflow.current_task_id}.json", "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"📄 详细报告已保存: workflow_result_{workflow.current_task_id}.json")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |