From 078f928f750776d542a3e17e6972a2659ee00927 Mon Sep 17 00:00:00 2001 From: ZhuJW <1421267742@qq.com> Date: Fri, 13 Mar 2026 15:59:01 +0800 Subject: [PATCH] fix --- .env | 28 ++--- docker-compose.yml | 20 +--- example_usage.py | 167 ------------------------------ generated_output/README.md | 188 --------------------------------- test_import.py | 207 ------------------------------------- 5 files changed, 10 insertions(+), 600 deletions(-) delete mode 100644 example_usage.py delete mode 100644 generated_output/README.md delete mode 100644 test_import.py diff --git a/.env b/.env index 57be582..db5147f 100644 --- a/.env +++ b/.env @@ -2,10 +2,16 @@ # 复制此文件为 .env 并填入实际值 # ==================== DashScope 配置 ==================== -# DashScope API Key (通义千问) -# 请替换为您的真实 API Key +# DashScope API Key (阿里百炼 Qwen 模型) # 获取地址：https://dashscope.console.aliyun.com/ -DASHSCOPE_API_KEY=sk-616332b2afa94699b4572d0fe6ac370a +DASHSCOPE_API_KEY=sk-your-actual-api-key-here + +# 禁用 LiteLLM 的远程模型成本映射（避免启动时联网超时） +DISABLE_LITELLM_MODEL_COST_MAP=True +LITELLM_LOCAL_MODEL_COST_MAP=True + +# 注意：请将 DASHSCOPE_API_KEY 替换为您的真实 API Key +# 不要将 .env 文件提交到版本控制系统 # ==================== 服务配置 ==================== HOST=0.0.0.0 @@ -23,19 +29,3 @@ STREAM_CLEANUP_INTERVAL=300 # 任务超时时间（秒） TASK_TIMEOUT=3600 - -# =========================================== -# 多智能体系统环境变量配置 -# =========================================== - -# DashScope API Key (通义千问 - Qwen3.5-flash) -# 获取地址：https://dashscope.console.aliyun.com/ -# 注意：请确保您的账户有足够的额度 -DASHSCOPE_API_KEY=sk-616332b2afa94699b4572d0fe6ac370a - -# =========================================== -# 使用说明： -# 1. 将上方的 DASHSCOPE_API_KEY 替换为您的真实 API Key -# 2. 保存此文件后运行 python main.py 启动服务 -# 3. 访问 http://localhost:8000/test-ui 使用测试界面 -# =========================================== diff --git a/docker-compose.yml b/docker-compose.yml index 2a06410..6400577 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,7 +11,7 @@ services: - "8000:8000" environment: # DashScope API Key（必需） - - DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY:-your_api_key_here} + - DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY:sk-616332b2afa94699b4572d0fe6ac370a} # 可选配置 - HOST=0.0.0.0 - PORT=8000 @@ -41,24 +41,6 @@ services: retries: 3 start_period: 10s - # Nginx 反向代理（可选，用于生产环境） - nginx: - image: nginx:alpine - container_name: multi-agent-nginx - restart: unless-stopped - ports: - - "80:80" - - "443:443" - volumes: - - ./nginx.conf:/etc/nginx/nginx.conf:ro - - ./ssl:/etc/nginx/ssl:ro # SSL 证书目录（如果需要 HTTPS） - depends_on: - - multi-agent-system - networks: - - agent-network - profiles: - - production # 仅在生产环境启动 - networks: agent-network: driver: bridge diff --git a/example_usage.py b/example_usage.py deleted file mode 100644 index b2713e2..0000000 --- a/example_usage.py +++ /dev/null @@ -1,167 +0,0 @@ -""" -示例使用脚本 -演示如何使用多智能体系统生成代码和文档 -""" - -import asyncio -import json -from pathlib import Path -from datetime import datetime - -from crew_factory import CrewFactory, run_multi_agent_task -from stream_manager import stream_manager - - -async def save_generated_content(task_id: str, output_dir: str = "generated_output"): - """ - 订阅 SSE 流并保存生成的内容到文件 - - Args: - task_id: 任务 ID - output_dir: 输出目录 - """ - output_path = Path(output_dir) - output_path.mkdir(exist_ok=True) - - # 创建时间戳目录 - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - task_output_dir = output_path / f"task_{timestamp}" - task_output_dir.mkdir() - - print(f"📁 生成内容将保存到：{task_output_dir}") - - # 保存的文件 - prd_file = task_output_dir / "PRD_产品需求文档.md" - qa_file = task_output_dir / "QA_测试计划.md" - dev_file = task_output_dir / "Dev_技术方案.md" - final_file = task_output_dir / "Final_交付报告.md" - - content_buffer = { - "ProductManager": [], - "QAEngineer": [], - "SoftwareDeveloper": [], - "Coordinator": [] - } - - print(f"\n🚀 开始订阅任务流：{task_id}\n") - - # 获取流 - stream = await stream_manager.get_stream(task_id) - if not stream: - print("❌ 未找到流") - return - - try: - while not stream.is_closed or not stream.queue.empty(): - try: - event = await asyncio.wait_for(stream.get(), timeout=5.0) - if not event: - break - - # 打印事件 - agent = event.agent - content = event.content - event_type = event.event_type - - print(f"[{agent}] {event_type}: {content[:100]}...") - - # 累积内容（简单示例，实际应该解析完整内容） - if event_type == "output" and agent in content_buffer: - content_buffer[agent].append(content) - - # 检测任务完成 - if event_type == "end": - print("\n✅ 任务完成！正在保存文件...\n") - - # 保存各角色的输出 - for role, contents in content_buffer.items(): - if contents: - filename = None - if role == "ProductManager": - filename = prd_file - elif role == "QAEngineer": - filename = qa_file - elif role == "SoftwareDeveloper": - filename = dev_file - elif role == "Coordinator": - filename = final_file - - if filename: - with open(filename, 'w', encoding='utf-8') as f: - f.write(f"# {role} 输出\n\n") - f.write(f"生成时间：{datetime.now().isoformat()}\n\n") - f.write('\n'.join(contents)) - print(f"✓ 已保存：{filename}") - - # 保存完整的事件日志 - log_file = task_output_dir / "events_log.json" - print(f"✓ 已保存完整日志：{log_file}") - - break - - except asyncio.TimeoutError: - if stream.is_closed: - break - continue - - except Exception as e: - print(f"❌ 错误：{e}") - import traceback - traceback.print_exc() - - -async def main(): - """主函数""" - print("=" * 60) - print("多智能体系统 - 代码和文档生成示例") - print("=" * 60) - - # 用户需求 - user_requirement = """ - 开发一个简单的在线待办事项应用（Todo App），包含以下功能： - 1. 用户可以注册和登录 - 2. 创建、编辑、删除待办事项 - 3. 标记事项为完成/未完成 - 4. 按优先级和截止日期排序 - 5. 基本的搜索和过滤功能 - - 技术栈要求： - - 后端：Python FastAPI - - 数据库：SQLite - - 前端：简单的 HTML/CSS/JavaScript - """ - - print(f"\n📝 用户需求：{user_requirement[:200]}...\n") - print("⏳ 启动多智能体系统，请稍候...\n") - - try: - # 启动任务 - task_id = await run_multi_agent_task( - user_requirement=user_requirement, - skip_confirmation=True - ) - - print(f"✅ 任务已启动，Task ID: {task_id}\n") - - # 订阅并保存生成的内容 - await save_generated_content(task_id) - - print("\n" + "=" * 60) - print("✨ 生成完成！请查看 generated_output/ 目录") - print("=" * 60) - - except Exception as e: - print(f"\n❌ 执行失败：{e}") - import traceback - traceback.print_exc() - - -if __name__ == "__main__": - # 加载环境变量 - try: - from dotenv import load_dotenv - load_dotenv() - except ImportError: - pass - - asyncio.run(main()) diff --git a/generated_output/README.md b/generated_output/README.md deleted file mode 100644 index e50a245..0000000 --- a/generated_output/README.md +++ /dev/null @@ -1,188 +0,0 @@ -# 生成的代码和文档 - -本目录包含多智能体系统自动生成的所有产出物。 - -## 📁 目录结构 - -每次任务执行后，会在 `task_YYYYMMDD_HHMMSS/` 子目录中生成以下文件： - -``` -task_20260313_140000/ -├── PRD_产品需求文档.md # 产品经理输出的需求文档 -├── QA_测试计划.md # QA 工程师输出的测试计划 -├── Dev_技术方案.md # 软件工程师输出的技术方案 -├── Final_交付报告.md # 协调员输出的最终交付报告 -└── events_log.json # 完整的事件日志（JSON 格式） -``` - -## 📄 文档说明 - -### 1. PRD_产品需求文档.md - -**内容包含**： -- 项目概述（背景、目标用户、核心价值） -- 功能需求列表（P0/P1/P2优先级） -- 用户故事和用例 -- 验收标准 -- 风险评估和缓解措施 - -**示例片段**： -```markdown -# 产品需求文档 - -## 1. 项目概述 -### 1.1 项目背景 -随着...的发展，需要一个...系统 - -### 1.2 目标用户 -- 主要用户群体：... -- 次要用户群体：... - -## 2. 功能需求 -### P0 - 核心功能 -1. 用户注册与登录 -2. CRUD 操作 -... -``` - -### 2. QA_测试计划.md - -**内容包含**： -- 测试策略（单元测试、集成测试、E2E 测试） -- 详细测试用例 -- 性能测试方案 -- 自动化测试建议 - -**示例片段**： -```markdown -# 测试计划 - -## 1. 测试策略 -### 1.1 单元测试 -- 覆盖核心业务逻辑 -- 目标覆盖率：80%+ - -## 2. 测试用例 -### TC-001: 用户注册 -**前置条件**: 无 -**步骤**: -1. 访问注册页面 -2. 填写表单 -... -``` - -### 3. Dev_技术方案.md - -**内容包含**： -- 系统架构设计 -- 技术栈选择及理由 -- 数据库 Schema 设计 -- API 接口定义 -- 核心代码实现 -- 部署方案 - -**示例片段**： -```markdown -# 技术方案 - -## 1. 架构设计 -### 1.1 整体架构 -采用前后端分离的 RESTful 架构 - -### 1.2 技术栈 -- 后端：FastAPI + SQLAlchemy -- 数据库：SQLite/PostgreSQL -- 前端：Vue.js/React - -## 2. 数据库设计 -### User 表 -| 字段 | 类型 | 说明 | -|------|------|------| -| id | INTEGER | 主键 | -| username | VARCHAR(50) | 用户名 | -... -``` - -### 4. Final_交付报告.md - -**内容包含**： -- 交付摘要 -- 一致性检查（PRD↔测试计划↔技术方案） -- 质量评估（完整性、可行性评分） -- 风险提示 -- 后续行动建议 - -**示例片段**： -```markdown -# 最终交付报告 - -## 1. 交付摘要 -本项目已完成以下交付物： -- ✓ PRD 文档（版本 1.0） -- ✓ 测试计划（版本 1.0） -- ✓ 技术方案（版本 1.0） - -## 2. 质量评估 -### 完整性评分：8.5/10 -优点： -- 需求描述清晰 -- 测试覆盖全面 - -改进点： -- 部分边界情况未考虑 -... -``` - -## 🔍 如何查看 - -### Windows 用户 -```powershell -# 打开最新生成的目录 -explorer (Get-ChildItem . -Directory | Sort-Object LastWriteTime -Descending | Select-Object -First 1).FullName -``` - -### Mac/Linux 用户 -```bash -# Mac -open $(ls -td task_* | head -n1) - -# Linux -xdg-open $(ls -td task_* | head -n1) -``` - -### 通用方法 -直接在文件管理器中浏览本目录，找到对应时间戳的文件夹。 - -## 💾 文件格式说明 - -- **Markdown (.md)**: 可用任何文本编辑器或 Markdown 阅读器打开 - - 推荐工具：VS Code、Typora、Obsidian -- **JSON (.json)**: 结构化事件日志，可用于程序处理 - - 可用浏览器、文本编辑器或 JSON 查看器打开 - -## 📊 文件大小参考 - -典型任务的输出文件大小： -- PRD 文档：10-30 KB -- 测试计划：15-40 KB -- 技术方案：20-50 KB -- 交付报告：10-25 KB -- 事件日志：5-15 KB - -## ⚠️ 注意事项 - -1. **及时备份**: 生成的文件存储在本地，请定期备份重要文档 -2. **版本管理**: 建议将生成的文档纳入 Git 版本控制 -3. **敏感信息**: 注意不要泄露 API Key 等敏感信息 -4. **磁盘空间**: 长期运行会产生大量文件，定期清理旧文件 - -## 🎯 使用建议 - -1. **审查生成内容**: AI 生成的内容可能有误，务必人工审查 -2. **迭代优化**: 根据实际反馈调整需求描述，重新生成 -3. **团队协作**: 将生成的文档作为讨论基础，团队共同完善 -4. **知识沉淀**: 将优秀实践固化到需求模板中 - ---- - -**开始使用**: 运行 `python ../example_usage.py` 或访问 http://localhost:8000/test-ui diff --git a/test_import.py b/test_import.py deleted file mode 100644 index 0104475..0000000 --- a/test_import.py +++ /dev/null @@ -1,207 +0,0 @@ -""" -Quick Test Script -Verifies all modules can be imported and initialized correctly -""" - -import sys - - -def test_imports(): - """Test all module imports""" - print("=" * 60) - print("Testing Module Imports...") - print("=" * 60) - - try: - from agents_config import create_agents, TASK_TEMPLATES, QWEN_MODEL_CONFIG - print("[OK] agents_config") - except Exception as e: - print(f"[FAIL] agents_config - {e}") - return False - - try: - from stream_manager import stream_manager, StreamEvent, TaskStreamQueue - print("[OK] stream_manager") - except Exception as e: - print(f"[FAIL] stream_manager - {e}") - return False - - try: - from crew_factory import CrewFactory, SSECrewExecutor, CrewExecutionLogger - print("[OK] crew_factory") - except Exception as e: - print(f"[FAIL] crew_factory - {e}") - return False - - try: - from main import app - print("[OK] main") - except Exception as e: - print(f"[FAIL] main - {e}") - return False - - return True - - -def test_agents(): - """Test Agent creation""" - print("\n" + "=" * 60) - print("Testing Agent Creation...") - print("=" * 60) - - try: - from agents_config import create_agents, get_product_manager_agent - # Test creating a single agent without LLM configuration - agent = get_product_manager_agent() - - print(f"[OK] Agent structure created:") - print(f" - Role: {agent.role}") - print(f" - Goal: {agent.goal[:50]}...") - print(f"\n[NOTE] Full agent initialization requires DASHSCOPE_API_KEY") - print(f" Set environment variable before running the server.") - - return True - except Exception as e: - # This is expected if API key is not configured - error_msg = str(e) - if "API_KEY" in error_msg or "provider" in error_msg.lower(): - print(f"[SKIP] Agent creation skipped (API key not configured)") - print(f" Set DASHSCOPE_API_KEY environment variable to enable.") - return True # Not a failure, just needs configuration - else: - print(f"[FAIL] Agent creation failed: {e}") - return False - - -def test_stream_manager(): - """Test stream manager""" - print("\n" + "=" * 60) - print("Testing Stream Manager...") - print("=" * 60) - - try: - import asyncio - from stream_manager import stream_manager, StreamEvent - - async def test(): - # Create test stream - task_id = "test-123" - await stream_manager.create_stream(task_id) - - # Publish test event - await stream_manager.publish_event( - task_id=task_id, - event_type="test", - agent="TestAgent", - content="This is a test event" - ) - - # Get stream - stream = await stream_manager.get_stream(task_id) - if stream: - event = await stream.get() - if event: - print(f"[OK] Event format: {event.to_dict()}") - await stream_manager.close_stream(task_id) - return True - - return False - - result = asyncio.get_event_loop().run_until_complete(test()) - if result: - print("[OK] Stream manager test passed") - return True - else: - print("[FAIL] Stream manager test failed") - return False - - except Exception as e: - print(f"[FAIL] Stream manager test failed: {e}") - return False - - -def test_api_endpoints(): - """Test API endpoint registration""" - print("\n" + "=" * 60) - print("Testing API Endpoints...") - print("=" * 60) - - try: - from main import app - - routes = [route.path for route in app.routes] - - required_endpoints = [ - "/api/run_task", - "/api/stream/{task_id}", - "/api/task/{task_id}/status", - "/api/streams", - "/health", - "/test-ui" - ] - - print(f"[OK] Registered endpoints ({len(routes)} total):") - for endpoint in required_endpoints: - found = False - for r in routes: - if endpoint.split('{')[0].rstrip('/') in r: - print(f" [OK] {endpoint}") - found = True - break - if not found: - print(f" [?] {endpoint} (may use different format)") - - return True - except Exception as e: - print(f"[FAIL] API endpoint test failed: {e}") - return False - - -def main(): - """Run all tests""" - print("\n" + "=" * 60) - print("Multi-Agent System Module Test") - print("=" * 60 + "\n") - - results = [] - - # Test imports - results.append(("Module Imports", test_imports())) - - # Test Agent creation - results.append(("Agent Creation", test_agents())) - - # Test stream manager - results.append(("Stream Manager", test_stream_manager())) - - # Test API endpoints - results.append(("API Endpoints", test_api_endpoints())) - - # Summary - print("\n" + "=" * 60) - print("Test Summary") - print("=" * 60) - - for name, passed in results: - status = "[OK]" if passed else "[FAIL]" - print(f"{status} - {name}") - - all_passed = all(result[1] for result in results) - - print("\n" + "=" * 60) - if all_passed: - print("[SUCCESS] All tests passed! System is ready.") - print("\nTo start the server:") - print(" python main.py") - print("\nTest UI:") - print(" http://localhost:8000/test-ui") - print("\nAPI Documentation:") - print(" http://localhost:8000/docs") - else: - print("[FAILURE] Some tests failed. Please check error messages.") - sys.exit(1) - print("=" * 60) - - -if __name__ == "__main__": - main()