commit 80d9b50587f8d02c19ee785e6b4dad767d986c8a Author: ZhuJW <1421267742@qq.com> Date: Fri Mar 13 14:20:58 2026 +0800 第一次提交 diff --git a/.env b/.env new file mode 100644 index 0000000..57be582 --- /dev/null +++ b/.env @@ -0,0 +1,41 @@ +# 环境变量配置示例 +# 复制此文件为 .env 并填入实际值 + +# ==================== DashScope 配置 ==================== +# DashScope API Key (通义千问) +# 请替换为您的真实 API Key +# 获取地址：https://dashscope.console.aliyun.com/ +DASHSCOPE_API_KEY=sk-616332b2afa94699b4572d0fe6ac370a + +# ==================== 服务配置 ==================== +HOST=0.0.0.0 +PORT=8000 + +# ==================== 日志配置 ==================== +LOG_LEVEL=info # debug, info, warning, error + +# ==================== 高级配置（可选）==================== +# SSE 队列最大长度 +SSE_QUEUE_MAX_SIZE=1000 + +# 流清理间隔（秒） +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/.env.example b/.env.example new file mode 100644 index 0000000..5f1f722 --- /dev/null +++ b/.env.example @@ -0,0 +1,27 @@ +# 环境变量配置示例 +# 复制此文件为 .env 并填入实际值 + +# ==================== DashScope 配置 ==================== +# DashScope API Key (通义千问) +# 获取地址：https://dashscope.console.aliyun.com/ +DASHSCOPE_API_KEY=sk-your-actual-api-key-here + +# 注意：如果您使用 OpenAI，也可以设置 OPENAI_API_KEY +# OPENAI_API_KEY=sk-your-openai-key + +# ==================== 服务配置 ==================== +HOST=0.0.0.0 +PORT=8000 + +# ==================== 日志配置 ==================== +LOG_LEVEL=info # debug, info, warning, error + +# ==================== 高级配置（可选）==================== +# SSE 队列最大长度 +SSE_QUEUE_MAX_SIZE=1000 + +# 流清理间隔（秒） +STREAM_CLEANUP_INTERVAL=300 + +# 任务超时时间（秒） +TASK_TIMEOUT=3600 diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..7abe554 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,46 @@ +# Multi-Agent Software Delivery System +# 基于 CrewAI + Qwen3.5-flash 的多智能体软件交付系统 + +FROM python:3.11-slim + +# 设置环境变量 +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +# 设置工作目录 +WORKDIR /app + +# 安装系统依赖 +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# 复制依赖文件 +COPY requirements.txt . + +# 安装 Python 依赖 +RUN pip install --no-cache-dir -r requirements.txt + +# 复制应用代码 +COPY main.py . +COPY crew_factory.py . +COPY agents_config.py . +COPY stream_manager.py . +COPY .env.example .env.example + +# 创建非 root 用户运行应用 +RUN useradd --create-home --shell /bin/bash appuser && \ + chown -R appuser:appuser /app +USER appuser + +# 暴露端口 +EXPOSE 8000 + +# 健康检查 +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1 + +# 启动命令 +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..a7bd504 --- /dev/null +++ b/README.md @@ -0,0 +1,408 @@ +# Multi-Agent Software Delivery System + +基于 CrewAI + Qwen3.5-flash 的多智能体软件交付系统，支持 SSE 实时推送。 + +## 📋 功能特性 + +- **多智能体协作**: 4 个专业 Agent 协同完成软件交付 + - ProductManager: 产品需求分析 + - QAEngineer: 测试计划制定 + - SoftwareDeveloper: 技术方案设计 + - Coordinator: 质量审核与交付 + +- **实时通信**: 基于 SSE 协议实时推送执行日志 +- **异步处理**: 任务异步启动，不阻塞 API 响应 +- **并发支持**: 多用户同时请求互不干扰 + +## 🔑 关键技术点 + +### 1. SSE 数据格式设计 + +统一的 JSON 格式，便于前端解析： + +```json +{ + "task_id": "550e8400-e29b...", + "sequence": 1, + "agent_name": "ProductManager", + "event_type": "thought", + "content": "正在分析用户需求，提取关键指标...", + "timestamp": "2023-10-27T10:00:00Z" +} +``` + +**字段说明**： +- `task_id`: 任务唯一标识，用于区分不同用户的请求 +- `sequence`: 序列号（每个 task_id 独立递增），用于检测消息丢失 +- `agent_name`: 发送事件的 Agent 名称 +- `event_type`: 事件类型（start/thought/action/output/end/error） +- `content`: 事件内容 +- `timestamp`: UTC 时间戳（ISO 8601 格式） + +### 2. 并发处理逻辑 + +**核心挑战**：CrewAI 默认是同步运行的，而 FastAPI 和 SSE 需要异步。 + +**解决方案**： + +```python +# 在 TaskStreamQueue 中实现线程安全的消息发布 +def put_nowait(self, event: StreamEvent) -> bool: + """ + 从同步线程（如 CrewAI 事件处理器）安全地发布事件 + + 使用 run_coroutine_threadsafe 将协程提交到事件循环执行 + 这是实现 CrewAI（同步）与 SSE（异步）集成的关键 + """ + future = asyncio.run_coroutine_threadsafe( + self.queue.put(event), + self._loop + ) + future.result(timeout=5.0) + return True +``` + +**关键实现**： +1. 使用 `asyncio.Queue` 实现异步非阻塞消息队列 +2. 通过 `asyncio.Lock` 保证并发安全 +3. 每个 `task_id` 独立队列，实现任务隔离 +4. 使用 `run_coroutine_threadsafe` 从同步线程安全发布事件 +5. 确保 `stream_manager` 能安全地在线程间传递消息 + +### 3. 多任务隔离机制 + +```python +class StreamManager: + """全局流管理器 - 管理所有任务的 SSE 流""" + + def __init__(self): + # task_id -> TaskStreamQueue 映射 + self.streams: Dict[str, TaskStreamQueue] = {} + self._lock = asyncio.Lock() # 并发控制 +``` + +- 每个 `task_id` 对应独立的 `TaskStreamQueue` +- 使用 `asyncio.Lock` 保护字典操作 +- 定期清理已完成的流（默认每小时） +- 序列号计数器按 `task_id` 独立维护 + +## 🚀 快速开始 + +### 1. 安装依赖 + +```bash +pip install -r requirements.txt +``` + +### 2. 测试模块 + +运行测试脚本验证所有模块正常： + +```bash +python test_import.py +``` + +预期输出： +``` +[OK] Module Imports +[OK] Agent Creation +[OK] Stream Manager +[OK] API Endpoints + +[SUCCESS] All tests passed! System is ready. +``` + +### 3. 配置环境变量 + +复制 `.env.example` 为 `.env` 并填入 DashScope API Key： + +```bash +cp .env.example .env +``` + +编辑 `.env` 文件： +``` +DASHSCOPE_API_KEY=sk-your-actual-api-key +``` + +获取 API Key: https://dashscope.console.aliyun.com/ + +### 4. 启动服务 + +```bash +python main.py +``` + +或使用 uvicorn： +```bash +uvicorn main:app --host 0.0.0.0 --port 8000 --reload +``` + +### 4. 访问服务 + +- **API 文档**: http://localhost:8000/docs +- **测试 UI**: http://localhost:8000/test-ui + +## 📡 API 接口 + +### POST /api/run_task + +启动多智能体任务 + +**请求体**: +```json +{ + "user_requirement": "开发一个在线商城系统...", + "skip_confirmation": true +} +``` + +**响应**: +```json +{ + "task_id": "550e8400-e29b-41d4-a716-446655440000", + "status": "started", + "message": "任务已启动..." +} +``` + +### GET /api/stream/{task_id} + +SSE 端点，订阅任务执行日志 + +**事件格式**: +```json +{ + "agent": "ProductManager", + "type": "thought", + "content": "正在分析用户需求...", + "timestamp": "2024-01-01T12:00:00", + "task_id": "uuid" +} +``` + +**事件类型**: +- `start`: 任务开始 +- `agent_start`: Agent 开始执行 +- `thought`: Agent 思考过程 +- `action`: Agent 执行动作 +- `output`: Agent 输出结果 +- `step_end`: 步骤完成 +- `end`: 任务结束 +- `error`: 发生错误 + +### GET /api/task/{task_id}/status + +查询任务状态 + +### GET /api/streams + +列出所有活跃的 SSE 流 + +## 🧪 使用示例 + +### 使用 curl 测试 + +```bash +# 1. 启动任务 +curl -X POST http://localhost:8000/api/run_task \ + -H "Content-Type: application/json" \ + -d '{ + "user_requirement": "开发一个简单的待办事项应用", + "skip_confirmation": true + }' + +# 2. 订阅 SSE 流 (使用返回的 task_id) +curl -N http://localhost:8000/api/stream/{task_id} +``` + +### Python 客户端示例 + +```python +import requests +import json + +# 启动任务 +response = requests.post( + 'http://localhost:8000/api/run_task', + json={ + 'user_requirement': '开发一个博客系统', + 'skip_confirmation': True + } +) +task_data = response.json() +task_id = task_data['task_id'] + +# 订阅 SSE 流 +import eventstream + +with requests.get( + f'http://localhost:8000/api/stream/{task_id}', + stream=True +) as r: + for line in r.iter_lines(): + if line: + data = line.decode('utf-8').replace('data: ', '') + event = json.loads(data) + print(f"[{event['agent']}] {event['type']}: {event['content']}") +``` + +## 🏗️ 项目结构 + +``` +. +├── main.py # FastAPI 入口和路由 +├── crew_factory.py # CrewAI 工厂和事件处理器 +├── agents_config.py # Agent 配置和 Prompt 模板 +├── stream_manager.py # SSE 流管理和消息队列 +├── requirements.txt # Python 依赖 +├── test_import.py # 模块测试脚本 +├── Dockerfile # Docker 镜像构建文件 +├── docker-compose.yml # Docker Compose 配置 +├── nginx.conf # Nginx 反向代理配置 +├── .env.example # 环境变量示例 +└── README.md # 本文档 +``` + +## ⚙️ 配置说明 + +### LLM 配置 + +在 `agents_config.py` 中配置： + +```python +QWEN_MODEL_CONFIG = { + "model": "qwen-plus", # Qwen3.5-flash + "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", + "api_key_env": "DASHSCOPE_API_KEY", +} +``` + +### Agent 角色定制 + +在 `agents_config.py` 中修改各 Agent 的： +- `role`: 角色名称 +- `goal`: 目标 +- `backstory`: 背景描述 +- `TASK_TEMPLATES`: 任务 Prompt 模板 + +## 🔧 高级用法 + +### 自定义事件处理器 + +继承 `CrewEventsHandler` 类并重写回调方法： + +```python +class MyCustomHandler(CrewEventsHandler): + def on_agent_output(self, agent, output): + # 自定义处理逻辑 + pass +``` + +### 调整并发策略 + +在 `stream_manager.py` 中调整队列大小和清理策略： + +```python +TaskStreamQueue(task_id, max_size=1000) # 默认 1000 +``` + +## 🐛 故障排查 + +### 问题：SSE 连接立即断开 + +**解决**: 确保 Nginx 配置中禁用了缓冲： +```nginx +proxy_buffering off; +proxy_cache off; +proxy_request_buffering off; +``` + +### 问题：LLM 调用失败 + +**检查**: +1. DASHSCOPE_API_KEY 是否正确配置 +2. 网络连接是否正常 +3. API Key 是否有足够额度 + +### 问题：内存占用过高 + +**解决**: 调整 `cleanup_old_streams` 的调用频率和保留时间 + +## 🏗️ Docker 部署 + +### 使用 Docker Compose（推荐） + +```bash +# 1. 设置环境变量 +export DASHSCOPE_API_KEY=sk-your-api-key + +# 2. 启动服务 +docker-compose up -d + +# 3. 查看日志 +docker-compose logs -f multi-agent-system + +# 4. 停止服务 +docker-compose down +``` + +### 生产环境部署（带 Nginx） + +```bash +# 使用 production profile 启动（包含 Nginx） +docker-compose --profile production up -d +``` + +**目录结构**： +``` +. +├── docker-compose.yml # Docker Compose 配置 +├── Dockerfile # 应用镜像构建文件 +├── nginx.conf # Nginx 配置（反向代理 + SSE 支持） +├── .env # 环境变量文件 +└── ... +``` + +### Nginx 关键配置 + +对于 SSE 流端点，Nginx 必须禁用缓冲： + +```nginx +location /api/stream/ { + proxy_pass http://multi-agent-system:8000; + + # HTTP/1.1 支持（SSE 必需） + proxy_http_version 1.1; + proxy_set_header Connection ""; + + # 禁用缓冲（关键） + proxy_buffering off; + proxy_cache off; + proxy_request_buffering off; + + # 长连接超时 + proxy_read_timeout 300s; +} +``` + +## 📝 注意事项 + +1. **生产环境部署**: + - 限制 CORS 允许的来源域名 + - 添加 API 认证机制 + - 使用 Redis 等持久化队列替代内存队列 + +2. **性能优化**: + - 调整 SSE 队列大小避免内存溢出 + - 限制单个任务的超时时间 + - 实现任务优先级队列 + +3. **安全考虑**: + - 验证用户输入防止注入攻击 + - 限制请求频率防止滥用 + - 记录审计日志 + +## 📄 License + +MIT License diff --git a/USAGE_GUIDE.md b/USAGE_GUIDE.md new file mode 100644 index 0000000..f2d88b7 --- /dev/null +++ b/USAGE_GUIDE.md @@ -0,0 +1,243 @@ +# 多智能体系统使用指南 + +## 📋 目录结构 + +``` +AI_CrewAI/ +├── generated_output/ # 生成的代码和文档输出目录 +│ └── task_YYYYMMDD_HHMMSS/ # 每次任务的时间戳目录 +│ ├── PRD_产品需求文档.md # 产品需求文档 +│ ├── QA_测试计划.md # 测试计划文档 +│ ├── Dev_技术方案.md # 技术方案文档 +│ ├── Final_交付报告.md # 最终交付报告 +│ └── events_log.json # 完整事件日志 +├── code_docs/ # 文档目录 +├── example_usage.py # 使用示例脚本 +├── main.py # FastAPI服务入口 +└── USAGE_GUIDE.md # 本文件 +``` + +## 🚀 快速开始 + +### 1. 配置 API Key + +编辑 `.env` 文件，设置您的阿里百炼 API Key： + +```bash +DASHSCOPE_API_KEY=sk-your-actual-api-key-here +``` + +获取 API Key: https://dashscope.console.aliyun.com/ + +### 2. 安装依赖 + +```bash +pip install -r requirements.txt +``` + +### 3. 启动服务 + +**方式一：直接运行 Python** +```bash +python main.py +``` + +**方式二：使用 uvicorn** +```bash +uvicorn main:app --host 0.0.0.0 --port 8000 --reload +``` + +### 4. 访问服务 + +- **API 文档**: http://localhost:8000/docs +- **测试 UI**: http://localhost:8000/test-ui + +## 📝 使用方法 + +### 方法一：通过 Web UI（推荐） + +1. 访问 http://localhost:8000/test-ui +2. 在输入框中输入您的需求，例如： + ``` + 开发一个简单的在线待办事项应用（Todo App），包含以下功能： + 1. 用户可以注册和登录 + 2. 创建、编辑、删除待办事项 + 3. 标记事项为完成/未完成 + 4. 按优先级和截止日期排序 + 5. 基本的搜索和过滤功能 + + 技术栈要求： + - 后端：Python FastAPI + - 数据库：SQLite + - 前端：简单的 HTML/CSS/JavaScript + ``` +3. 点击"启动任务"按钮 +4. 实时查看各 Agent 的执行过程 +5. 完成后查看生成的文档 + +### 方法二：通过 API 调用 + +使用 curl 或任何 HTTP 客户端： + +```bash +# 启动任务 +curl -X POST http://localhost:8000/api/run_task \ + -H "Content-Type: application/json" \ + -d '{ + "user_requirement": "开发一个博客系统", + "skip_confirmation": true + }' + +# 返回示例： +# {"task_id":"uuid","status":"started","message":"..."} +``` + +然后订阅 SSE 流： + +```bash +curl -N http://localhost:8000/api/stream/{task_id} +``` + +### 方法三：使用示例脚本 + +运行提供的示例脚本： + +```bash +python example_usage.py +``` + +这会自动保存生成的内容到 `generated_output/` 目录。 + +## 📊 生成的内容 + +每个任务会生成以下文档： + +### 1. PRD_产品需求文档.md +- 项目概述 +- 功能需求列表（按优先级） +- 用户故事和用例 +- 验收标准 +- 风险评估 + +### 2. QA_测试计划.md +- 测试策略 +- 测试用例（正常场景 + 异常场景） +- 性能测试方案 +- 自动化测试建议 + +### 3. Dev_技术方案.md +- 系统架构设计 +- 技术栈选择 +- 数据库 Schema +- API 接口设计 +- 核心代码实现 +- 部署方案 + +### 4. Final_交付报告.md +- 交付摘要 +- 一致性检查 +- 质量评估 +- 风险提示 +- 后续行动建议 + +## 🔍 查看生成的文件 + +生成的文件保存在 `generated_output/task_YYYYMMDD_HHMMSS/` 目录下。 + +**Windows 用户**： +```powershell +# 打开最新生成的目录 +explorer (Get-ChildItem generated_output -Directory | Sort-Object LastWriteTime -Descending | Select-Object -First 1).FullName +``` + +**Linux/Mac 用户**： +```bash +# 打开最新生成的目录 +xdg-open $(ls -td generated_output/task_* | head -n1) # Linux +open $(ls -td generated_output/task_* | head -n1) # Mac +``` + +或者直接在文件管理器中浏览 `generated_output/` 目录。 + +## 💡 实用技巧 + +### 1. 实时监控任务进度 + +在另一个终端窗口查看日志： + +```bash +# Docker Compose 方式 +docker-compose logs -f multi-agent-system + +# 直接运行 +# 日志会自动输出到控制台 +``` + +### 2. 自定义输出目录 + +修改 `example_usage.py` 中的 `output_dir` 参数： + +```python +await save_generated_content(task_id, output_dir="my_custom_output") +``` + +### 3. 批量处理多个需求 + +创建批处理脚本： + +```python +requirements = [ + "需求 1...", + "需求 2...", + "需求 3..." +] + +for req in requirements: + task_id = await run_multi_agent_task(user_requirement=req) + await save_generated_content(task_id) +``` + +## ⚠️ 常见问题 + +### Q: 为什么没有生成文件？ + +**A**: 确保： +1. API Key 已正确配置 +2. 网络连接正常 +3. 任务执行完成（看到"任务执行完成"事件） + +### Q: 生成的内容在哪里？ + +**A**: +- 默认在 `generated_output/task_时间戳/` 目录 +- 可以通过 Web UI 直接查看 +- 或在控制台查找保存路径 + +### Q: 如何重新查看生成的内容？ + +**A**: +1. 找到对应的任务时间戳目录 +2. 打开相应的 .md 文件 +3. 使用 Markdown 阅读器或 VS Code 查看 + +### Q: 可以修改生成的文档吗？ + +**A**: 当然可以！生成的文档是 Markdown 格式，可以使用任何文本编辑器修改。 + +## 🎯 最佳实践 + +1. **明确需求描述**：需求越详细，生成的文档越准确 +2. **指定技术栈**：明确说明使用的技术和框架 +3. **设定约束条件**：如性能要求、安全要求等 +4. **迭代优化**：根据生成结果调整需求描述 + +## 📞 技术支持 + +如有问题，请查看： +- API 文档：http://localhost:8000/docs +- 项目 README：README.md +- 日志输出：控制台或 Docker 日志 + +--- + +**祝您使用愉快！** 🎉 diff --git a/agents_config.py b/agents_config.py new file mode 100644 index 0000000..0f10e29 --- /dev/null +++ b/agents_config.py @@ -0,0 +1,302 @@ +""" +Agent 配置文件 +定义所有 Agent 的角色、目标、背景描述和任务模板 +""" + +from typing import Dict, Any +from crewai import Agent + + +# Qwen3.5-flash 模型配置（阿里百炼 - DashScope） +QWEN_MODEL_CONFIG = { + "model": "qwen3.5-flash", # 阿里百炼 Qwen3.5-flash 对应 qwen-plus + "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", + "api_key_env": "DASHSCOPE_API_KEY", +} + + +def get_product_manager_agent() -> Agent: + """创建产品经理 Agent""" + from crewai.llm import LLM + + # 创建一个基础的 LLM 占位符（会在 configure_llm 中被替换） + llm_placeholder = LLM( + model="qwen-plus", + api_key="placeholder", # 会被 configure_llm 替换 + base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", + ) + + return Agent( + role="ProductManager", + goal="将用户需求转化为清晰的产品需求文档 (PRD)", + backstory="""你是一位经验丰富的产品经理，擅长理解用户需求并转化为可执行的产品规格。 +你的职责包括： +1. 分析用户需求的核心痛点和业务价值 +2. 定义功能列表和优先级 +3. 编写详细的功能描述和验收标准 +4. 识别潜在的技术风险和业务风险""", + verbose=True, + allow_delegation=False, + llm=llm_placeholder, + ) + + +def get_qa_engineer_agent() -> Agent: + """创建 QA 工程师 Agent""" + from crewai.llm import LLM + + # 创建一个基础的 LLM 占位符（会在 configure_llm 中被替换） + llm_placeholder = LLM( + model="qwen-plus", + api_key="placeholder", # 会被 configure_llm 替换 + base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", + ) + + return Agent( + role="QAEngineer", + goal="根据产品需求制定测试计划和测试用例", + backstory="""你是一位资深 QA 工程师，专注于软件质量保障。 +你的职责包括： +1. 分析 PRD 文档，识别测试范围 +2. 设计测试策略（单元测试、集成测试、端到端测试） +3. 编写详细的测试用例，包括正常场景和异常场景 +4. 定义验收标准和性能指标""", + verbose=True, + allow_delegation=False, + llm=llm_placeholder, + ) + + +def get_software_developer_agent() -> Agent: + """创建软件开发工程师 Agent""" + from crewai.llm import LLM + + # 创建一个基础的 LLM 占位符（会在 configure_llm 中被替换） + llm_placeholder = LLM( + model="qwen-plus", + api_key="placeholder", # 会被 configure_llm 替换 + base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", + ) + + return Agent( + role="SoftwareDeveloper", + goal="根据需求和测试用例设计技术方案并生成代码框架", + backstory="""你是一位全栈软件架构师，拥有深厚的技术功底。 +你的职责包括： +1. 根据 PRD 和测试用例设计系统架构 +2. 选择合适的技术栈和框架 +3. 设计数据库 schema 和 API 接口 +4. 生成核心模块的代码框架和关键算法实现 +5. 编写技术文档和部署指南""", + verbose=True, + allow_delegation=False, + llm=llm_placeholder, + ) + + +def get_coordinator_agent() -> Agent: + """创建协调员 Agent - 负责最终审核和交付""" + from crewai.llm import LLM + + # 创建一个基础的 LLM 占位符（会在 configure_llm 中被替换） + llm_placeholder = LLM( + model="qwen-plus", + api_key="placeholder", # 会被 configure_llm 替换 + base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", + ) + + return Agent( + role="Coordinator", + goal="审核所有产出物，确保质量并生成最终交付报告", + backstory="""你是一位项目协调员，负责把控整体交付质量。 +你的职责包括： +1. 审核 PRD、测试计划和技术方案的完整性 +2. 识别各文档之间的一致性和潜在冲突 +3. 汇总所有产出物，生成结构化交付报告 +4. 评估项目风险和后续建议 +5. 在需要时请求人工确认（通过 API 控制）""", + verbose=True, + allow_delegation=False, + llm=llm_placeholder, + ) + + +# 任务模板配置 +TASK_TEMPLATES = { + "product_manager": """ +## 任务：产品需求分析 + +### 用户需求输入： +{user_requirement} + +### 输出要求： +请按照以下结构输出产品需求文档 (PRD)： + +1. **项目概述** + - 项目背景 + - 目标用户群体 + - 核心价值主张 + +2. **功能需求列表** + - 按优先级排序（P0/P1/P2） + - 每个功能的详细描述 + - 用户故事格式（作为...我希望...以便...） + +3. **非功能需求** + - 性能要求 + - 安全性要求 + - 可用性要求 + +4. **验收标准** + - 每个功能的验收条件 + - 关键业务指标 (KPI) + +5. **风险评估** + - 技术风险 + - 业务风险 + - 缓解措施 + +请使用 Markdown 格式输出，确保结构清晰、内容完整. +""", + + "qa_engineer": """ +## 任务：制定测试计划 + +### 输入信息： +- 产品需求文档 (来自 ProductManager): +{prd_content} + +### 输出要求： +请按照以下结构输出测试计划文档： + +1. **测试策略** + - 测试范围界定 + - 测试类型（单元/集成/E2E/性能/安全） + - 测试环境规划 + +2. **测试用例设计** + - 针对每个 P0/P1 功能设计测试用例 + - 包含：测试目的、前置条件、测试步骤、预期结果 + - 覆盖正常流程和异常流程 + +3. **自动化测试建议** + - 推荐使用的测试框架 + - 需要自动化的测试场景列表 + - CI/CD 集成建议 + +4. **性能测试方案** + - 压测场景设计 + - 性能基准指标 + - 监控指标 + +5. **验收检查清单** + - 上线前必须通过的检查项 + +请使用 Markdown 格式，测试用例使用表格形式展示。 +""", + + "software_developer": """ +## 任务：技术方案设计与代码实现 + +### 输入信息： +- 产品需求文档： +{prd_content} + +- 测试计划： +{qa_plan} + +### 输出要求： +请按照以下结构输出技术方案文档： + +1. **系统架构设计** + - 架构图（使用文字描述或 ASCII art） + - 技术选型及理由 + - 系统组件划分 + +2. **API 接口设计** + - RESTful API 列表（方法、路径、请求/响应格式） + - 接口鉴权方案 + - 错误码规范 + +3. **数据模型设计** + - 数据库表结构（表名、字段、类型、索引） + - ER 关系描述 + - 数据迁移策略 + +4. **核心代码实现** + - 关键模块的代码框架（Python/TypeScript 等） + - 核心算法伪代码或实现 + - 重要设计模式的应用 + +5. **部署方案** + - 基础设施需求 + - Docker 容器化配置示例 + - CI/CD 流水线配置建议 + +6. **开发注意事项** + - 代码规范 + - 日志和监控 + - 安全最佳实践 + +请确保代码片段语法正确，注释清晰。 +""", + + "coordinator": """ +## 任务：最终审核与交付报告 + +### 输入信息： +- 产品需求文档： +{prd_content} + +- 测试计划： +{qa_plan} + +- 技术方案： +{dev_plan} + +### 输出要求： +请按照以下结构输出最终交付报告： + +1. **交付摘要** + - 项目基本信息 + - 交付物清单 + - 整体质量评估 + +2. **一致性检查** + - PRD 与测试计划的匹配度 + - 技术方案对需求的覆盖度 + - 发现的遗漏或不一致点 + +3. **质量评估** + - 文档完整性评分（1-10 分） + - 技术可行性评估 + - 测试覆盖度评估 + +4. **风险提示** + - 高风险项列表 + - 中低风险项列表 + - 风险缓解建议 + +5. **后续行动建议** + - 短期行动计划（1-2 周） + - 中期目标（1-3 月） + - 长期演进方向 + +6. **交付确认** + - 是否满足交付标准 + - 需要人工复核的点 + - 最终结论（通过/有条件通过/不通过） + +请使用专业、客观的语气，给出具体、可执行的建议。 +""", +} + + +def create_agents() -> Dict[str, Agent]: + """创建所有 Agent 实例""" + return { + "product_manager": get_product_manager_agent(), + "qa_engineer": get_qa_engineer_agent(), + "software_developer": get_software_developer_agent(), + "coordinator": get_coordinator_agent(), + } diff --git a/code_docs/example_code.py b/code_docs/example_code.py new file mode 100644 index 0000000..09213d9 --- /dev/null +++ b/code_docs/example_code.py @@ -0,0 +1,481 @@ +""" +示例代码 - 待办事项应用后端实现 +这是多智能体系统生成的代码示例 +""" + +from fastapi import FastAPI, HTTPException, Depends, status +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials +from pydantic import BaseModel, EmailStr, Field +from typing import Optional, List +from datetime import datetime, timedelta +import uvicorn +import sqlite3 +import hashlib +import jwt + +# ==================== 配置 ==================== + +SECRET_KEY = "your-secret-key-change-in-production" +ALGORITHM = "HS256" +DATABASE = "todo_app.db" + +# ==================== 数据模型 ==================== + +class UserCreate(BaseModel): + """用户注册请求""" + username: str = Field(..., min_length=3, max_length=50) + email: EmailStr + password: str = Field(..., min_length=6) + + +class UserLogin(BaseModel): + """用户登录请求""" + username: str + password: str + + +class TodoItemCreate(BaseModel): + """创建待办事项""" + title: str = Field(..., min_length=1, max_length=200) + description: Optional[str] = None + priority: int = Field(default=1, ge=1, le=5) # 1-5, 5 最高 + due_date: Optional[datetime] = None + + +class TodoItemUpdate(BaseModel): + """更新待办事项""" + title: Optional[str] = Field(None, min_length=1, max_length=200) + description: Optional[str] = None + priority: Optional[int] = Field(None, ge=1, le=5) + due_date: Optional[datetime] = None + completed: Optional[bool] = None + + +class TodoItemResponse(BaseModel): + """待办事项响应""" + id: int + user_id: int + title: str + description: Optional[str] + priority: int + completed: bool + due_date: Optional[datetime] + created_at: datetime + updated_at: datetime + + class Config: + from_attributes = True + + +# ==================== 数据库操作 ==================== + +def get_db_connection(): + """获取数据库连接""" + conn = sqlite3.connect(DATABASE) + conn.row_factory = sqlite3.Row + return conn + + +def init_db(): + """初始化数据库""" + conn = get_db_connection() + cursor = conn.cursor() + + # 创建用户表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username VARCHAR(50) UNIQUE NOT NULL, + email VARCHAR(100) UNIQUE NOT NULL, + password_hash VARCHAR(255) NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # 创建待办事项表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS todo_items ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + title VARCHAR(200) NOT NULL, + description TEXT, + priority INTEGER DEFAULT 1, + completed BOOLEAN DEFAULT 0, + due_date TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE + ) + ''') + + # 创建索引 + cursor.execute('CREATE INDEX IF NOT EXISTS idx_todo_user ON todo_items(user_id)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_todo_completed ON todo_items(completed)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_todo_priority ON todo_items(priority DESC)') + + conn.commit() + conn.close() + print("✓ 数据库初始化完成") + + +# ==================== 认证工具 ==================== + +security = HTTPBearer() + + +def hash_password(password: str) -> str: + """密码哈希""" + return hashlib.sha256(password.encode()).hexdigest() + + +def verify_password(password: str, password_hash: str) -> bool: + """验证密码""" + return hash_password(password) == password_hash + + +def create_access_token(data: dict, expires_delta: timedelta = timedelta(days=7)) -> str: + """创建 JWT Token""" + to_encode = data.copy() + expire = datetime.utcnow() + expires_delta + to_encode.update({"exp": expire}) + return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) + + +def get_current_user( + credentials: HTTPAuthorizationCredentials = Depends(security) +) -> dict: + """获取当前用户（从 JWT Token）""" + try: + token = credentials.credentials + payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) + user_id = payload.get("sub") + if user_id is None: + raise HTTPException(status_code=401, detail="Invalid token") + return {"user_id": int(user_id), "username": payload.get("username")} + except jwt.PyJWTError: + raise HTTPException(status_code=401, detail="Token validation failed") + + +# ==================== FastAPI 应用 ==================== + +app = FastAPI( + title="Todo App API", + description="在线待办事项应用 - 由多智能体系统生成", + version="1.0.0" +) + + +@app.on_event("startup") +async def startup_event(): + """应用启动时初始化""" + init_db() + + +# ==================== 用户接口 ==================== + +@app.post("/api/users/register", tags=["用户管理"]) +async def register(user_data: UserCreate): + """用户注册""" + conn = get_db_connection() + cursor = conn.cursor() + + # 检查用户名是否已存在 + cursor.execute('SELECT id FROM users WHERE username = ?', (user_data.username,)) + if cursor.fetchone(): + conn.close() + raise HTTPException(status_code=400, detail="Username already exists") + + # 检查邮箱是否已存在 + cursor.execute('SELECT id FROM users WHERE email = ?', (user_data.email,)) + if cursor.fetchone(): + conn.close() + raise HTTPException(status_code=400, detail="Email already registered") + + # 创建用户 + password_hash = hash_password(user_data.password) + cursor.execute( + 'INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)', + (user_data.username, user_data.email, password_hash) + ) + conn.commit() + user_id = cursor.lastrowid + conn.close() + + return { + "message": "注册成功", + "user_id": user_id, + "username": user_data.username + } + + +@app.post("/api/users/login", tags=["用户管理"]) +async def login(credentials: UserLogin): + """用户登录""" + conn = get_db_connection() + cursor = conn.cursor() + + # 查找用户 + cursor.execute( + 'SELECT id, username, password_hash FROM users WHERE username = ?', + (credentials.username,) + ) + user = cursor.fetchone() + conn.close() + + if not user or not verify_password(credentials.password, user['password_hash']): + raise HTTPException(status_code=401, detail="Invalid username or password") + + # 生成 Token + access_token = create_access_token( + data={"sub": str(user['id']), "username": user['username']} + ) + + return { + "message": "登录成功", + "access_token": access_token, + "token_type": "bearer", + "user_id": user['id'], + "username": user['username'] + } + + +# ==================== 待办事项接口 ==================== + +@app.post("/api/todos", tags=["待办事项"], response_model=TodoItemResponse) +async def create_todo( + todo_data: TodoItemCreate, + current_user: dict = Depends(get_current_user) +): + """创建待办事项""" + conn = get_db_connection() + cursor = conn.cursor() + + cursor.execute(''' + INSERT INTO todo_items (user_id, title, description, priority, due_date) + VALUES (?, ?, ?, ?, ?) + ''', ( + current_user["user_id"], + todo_data.title, + todo_data.description, + todo_data.priority, + todo_data.due_date + )) + + conn.commit() + todo_id = cursor.lastrowid + + # 查询刚创建的记录 + cursor.execute('SELECT * FROM todo_items WHERE id = ?', (todo_id,)) + todo_row = cursor.fetchone() + conn.close() + + return dict(todo_row) + + +@app.get("/api/todos", tags=["待办事项"], response_model=List[TodoItemResponse]) +async def list_todos( + skip: int = 0, + limit: int = 50, + completed: Optional[bool] = None, + priority: Optional[int] = None, + search: Optional[str] = None, + sort_by: str = "priority", # priority, due_date, created_at + current_user: dict = Depends(get_current_user) +): + """获取待办事项列表（支持过滤和排序）""" + conn = get_db_connection() + cursor = conn.cursor() + + # 构建查询 + query = 'SELECT * FROM todo_items WHERE user_id = ?' + params = [current_user["user_id"]] + + # 添加过滤条件 + if completed is not None: + query += ' AND completed = ?' + params.append(completed) + + if priority is not None: + query += ' AND priority = ?' + params.append(priority) + + if search: + query += ' AND (title LIKE ? OR description LIKE ?)' + search_term = f'%{search}%' + params.extend([search_term, search_term]) + + # 添加排序 + order_map = { + 'priority': 'priority DESC', + 'due_date': 'due_date ASC', + 'created_at': 'created_at DESC' + } + order_clause = order_map.get(sort_by, 'priority DESC') + query += f' ORDER BY {order_clause}' + + # 添加分页 + query += ' LIMIT ? OFFSET ?' + params.extend([limit, skip]) + + cursor.execute(query, params) + todos = cursor.fetchall() + conn.close() + + return [dict(todo) for todo in todos] + + +@app.get("/api/todos/{todo_id}", tags=["待办事项"], response_model=TodoItemResponse) +async def get_todo( + todo_id: int, + current_user: dict = Depends(get_current_user) +): + """获取单个待办事项详情""" + conn = get_db_connection() + cursor = conn.cursor() + + cursor.execute( + 'SELECT * FROM todo_items WHERE id = ? AND user_id = ?', + (todo_id, current_user["user_id"]) + ) + todo = cursor.fetchone() + conn.close() + + if not todo: + raise HTTPException(status_code=404, detail="Todo item not found") + + return dict(todo) + + +@app.put("/api/todos/{todo_id}", tags=["待办事项"], response_model=TodoItemResponse) +async def update_todo( + todo_id: int, + todo_data: TodoItemUpdate, + current_user: dict = Depends(get_current_user) +): + """更新待办事项""" + conn = get_db_connection() + cursor = conn.cursor() + + # 检查是否存在 + cursor.execute( + 'SELECT id FROM todo_items WHERE id = ? AND user_id = ?', + (todo_id, current_user["user_id"]) + ) + if not cursor.fetchone(): + conn.close() + raise HTTPException(status_code=404, detail="Todo item not found") + + # 构建更新字段 + updates = [] + params = [] + + if todo_data.title is not None: + updates.append('title = ?') + params.append(todo_data.title) + + if todo_data.description is not None: + updates.append('description = ?') + params.append(todo_data.description) + + if todo_data.priority is not None: + updates.append('priority = ?') + params.append(todo_data.priority) + + if todo_data.completed is not None: + updates.append('completed = ?') + params.append(todo_data.completed) + + if todo_data.due_date is not None: + updates.append('due_date = ?') + params.append(todo_data.due_date) + + # 添加更新时间 + updates.append('updated_at = CURRENT_TIMESTAMP') + + # 执行更新 + params.append(todo_id) + query = f'UPDATE todo_items SET {", ".join(updates)} WHERE id = ?' + cursor.execute(query, params) + conn.commit() + + # 查询更新后的记录 + cursor.execute('SELECT * FROM todo_items WHERE id = ?', (todo_id,)) + todo = cursor.fetchone() + conn.close() + + return dict(todo) + + +@app.delete("/api/todos/{todo_id}", tags=["待办事项"]) +async def delete_todo( + todo_id: int, + current_user: dict = Depends(get_current_user) +): + """删除待办事项""" + conn = get_db_connection() + cursor = conn.cursor() + + # 检查是否存在 + cursor.execute( + 'SELECT id FROM todo_items WHERE id = ? AND user_id = ?', + (todo_id, current_user["user_id"]) + ) + if not cursor.fetchone(): + conn.close() + raise HTTPException(status_code=404, detail="Todo item not found") + + # 执行删除 + cursor.execute('DELETE FROM todo_items WHERE id = ?', (todo_id,)) + conn.commit() + conn.close() + + return {"message": "删除成功"} + + +@app.get("/api/todos/stats", tags=["待办事项"]) +async def get_stats(current_user: dict = Depends(get_current_user)): + """获取统计信息""" + conn = get_db_connection() + cursor = conn.cursor() + + # 总数 + cursor.execute( + 'SELECT COUNT(*) as total FROM todo_items WHERE user_id = ?', + (current_user["user_id"],) + ) + total = cursor.fetchone()['total'] + + # 已完成 + cursor.execute( + 'SELECT COUNT(*) as completed FROM todo_items WHERE user_id = ? AND completed = 1', + (current_user["user_id"],) + ) + completed = cursor.fetchone()['completed'] + + # 未完成 + pending = total - completed + + conn.close() + + return { + "total": total, + "completed": completed, + "pending": pending + } + + +# ==================== 主程序入口 ==================== + +if __name__ == "__main__": + print("=" * 60) + print("🚀 Todo App API服务启动中...") + print("=" * 60) + print("\n📖 API 文档：http://localhost:8000/docs") + print("📊 健康检查：http://localhost:8000/health\n") + + uvicorn.run( + "example_code:app", + host="0.0.0.0", + port=8000, + reload=True + ) diff --git a/crew_factory.py b/crew_factory.py new file mode 100644 index 0000000..4994f16 --- /dev/null +++ b/crew_factory.py @@ -0,0 +1,523 @@ +""" +CrewAI 工厂模块 +负责根据输入动态创建 Crew 实例，并集成 SSE 事件推送 + +技术方案说明： +由于 CrewAI 库的事件 API 可能随版本变化，我们采用更稳定的方案： +1. 使用自定义的 Logger 拦截 Agent 输出 +2. 在任务执行前后手动发送 SSE 事件 +3. 通过 run_in_executor 实现同步转异步 +""" + +import asyncio +from typing import Dict, Any, Optional, List, Callable +from datetime import datetime +import uuid +import io +import sys +from contextlib import contextmanager + +from crewai import Agent, Task, Crew, Process + +from agents_config import ( + create_agents, + TASK_TEMPLATES, + QWEN_MODEL_CONFIG, +) +from stream_manager import stream_manager, StreamEvent + + +class CrewExecutionLogger: + """ + CrewAI 执行日志拦截器 + + 通过捕获 stdout/stderr 来实时获取 CrewAI 的执行日志， + 并将其转发到 SSE 流。这是一种稳定且非侵入式的方法。 + """ + + def __init__(self, task_id: str, callback: Optional[Callable] = None): + self.task_id = task_id + self.callback = callback + self._old_stdout = None + self._old_stderr = None + self._buffer = io.StringIO() + + def _write(self, text: str): + """写入时触发回调""" + self._buffer.write(text) + + # 按行处理（遇到换行符时发送） + if '\n' in text: + lines = self._buffer.getvalue().split('\n') + for line in lines[:-1]: # 除了最后一行 + if line.strip(): + self._send_line(line) + self._buffer = io.StringIO() + self._buffer.write(lines[-1]) # 保留未完成的行 + + def _send_line(self, line: str): + """发送单行日志到 SSE 流""" + if self.callback: + self.callback("log", "System", line.strip()) + + def start_capture(self): + """开始捕获输出""" + self._old_stdout = sys.stdout + self._old_stderr = sys.stderr + + # 创建代理对象 + class OutputProxy: + def __init__(self, logger, original): + self.logger = logger + self.original = original + + def write(self, text): + self.logger._write(text) + if self.original: + self.original.write(text) + + def flush(self): + if self.original: + self.original.flush() + + sys.stdout = OutputProxy(self, self._old_stdout) + sys.stderr = OutputProxy(self, self._old_stderr) + + def stop_capture(self): + """停止捕获输出""" + if self._old_stdout: + sys.stdout = self._old_stdout + if self._old_stderr: + sys.stderr = self._old_stderr + + # 发送剩余内容 + remaining = self._buffer.getvalue() + if remaining.strip(): + self._send_line(remaining) + + +class SSECrewExecutor: + """ + SSE Crew 执行器 + + 封装 CrewAI 的执行过程，在关键节点发送 SSE 事件。 + 这是与 CrewAI 版本无关的稳定方案。 + """ + + def __init__(self, task_id: str): + self.task_id = task_id + self.logger = None + + def _send_event(self, event_type: str, agent_name: str, content: str, metadata: Optional[Dict] = None): + """发送 SSE 事件（同步方法，使用线程安全的方式）""" + try: + loop = asyncio.get_event_loop() + + async def send_async(): + stream = await stream_manager.get_stream(self.task_id) + if stream is None: + return False + + event = StreamEvent( + event_type=event_type, + agent=agent_name, + content=content, + task_id=self.task_id, + metadata=metadata or {} + ) + return stream.put_nowait(event) + + future = asyncio.run_coroutine_threadsafe(send_async(), loop) + future.result(timeout=5.0) + except Exception as e: + # 静默失败，不影响主流程 + pass + + def execute(self, crew: 'Crew', inputs: Dict[str, Any]) -> Any: + """ + 执行 Crew 任务，并在过程中发送 SSE 事件 + + Args: + crew: CrewAI Crew 实例 + inputs: 任务输入参数 + + Returns: + Crew 执行结果 + """ + # 1. 发送开始事件 + self._send_event( + event_type="start", + agent_name="System", + content=f"Crew 任务开始执行，共 {len(crew.agents)} 个 Agent，{len(crew.tasks)} 个任务", + metadata={ + "agent_count": len(crew.agents), + "task_count": len(crew.tasks) + } + ) + + # 2. 设置日志拦截 + def log_callback(event_type: str, agent_name: str, content: str): + self._send_event(event_type, agent_name, content) + + self.logger = CrewExecutionLogger(self.task_id, log_callback) + self.logger.start_capture() + + try: + # 3. 遍历执行任务（手动控制以便发送事件） + context = inputs.copy() + last_output = None + + for i, task in enumerate(crew.tasks): + # 将 context 字典转换为 JSON 字符串（CrewAI 要求） + import json + context_str = json.dumps(context, ensure_ascii=False) + task_result = self._execute_task(task, context_str, context, i) + + # 更新上下文（简单的字符串替换） + if task_result: + context[f"task_{i}_output"] = str(task_result) + last_output = task_result + + # 4. 发送结束事件 + self._send_event( + event_type="end", + agent_name="System", + content="Crew 任务执行完成", + metadata={"success": True} + ) + + return last_output + + except Exception as e: + # 5. 发送错误事件 + self._send_event( + event_type="error", + agent_name="System", + content=str(e), + metadata={"error_type": type(e).__name__} + ) + raise + + finally: + # 6. 恢复原始输出 + self.logger.stop_capture() + + def _execute_task(self, task: 'Task', context_str: str, context_dict: Dict[str, Any], index: int) -> Optional[Any]: + """ + 执行单个任务 + + Args: + task: Task 实例 + context_str: 上下文信息（JSON 字符串格式，用于传递给 CrewAI） + context_dict: 上下文信息（字典格式，用于变量替换） + index: 任务索引 + + Returns: + 任务执行结果 + """ + # 获取负责此任务的 agent + agent_name = task.agent.role if task.agent else "Unknown" + + # 1. 发送任务开始事件 + self._send_event( + event_type="agent_start", + agent_name=agent_name, + content=f"[任务 {index + 1}] {agent_name} 开始执行任务", + metadata={ + "task_index": index, + "task_description": task.description[:200] if task.description else "" + } + ) + + # 2. 准备任务描述（替换上下文变量） + description = self._prepare_task_description(task.description, context_dict) + + # 3. 执行任务 + try: + # 使用 CrewAI 的原生执行方式 + result = task.execute_sync(context=context_str) + + # 4. 发送任务完成事件 + output_str = str(result)[:500] if result else "No output" + self._send_event( + event_type="output", + agent_name=agent_name, + content=f"任务完成，输出摘要：{output_str}", + metadata={ + "output_length": len(str(result)) if result else 0 + } + ) + + return result + + except Exception as e: + # 5. 发送错误事件 + self._send_event( + event_type="error", + agent_name=agent_name, + content=f"任务执行失败：{str(e)}", + metadata={"error_type": type(e).__name__} + ) + raise + + def _prepare_task_description(self, description: str, context: Dict[str, Any]) -> str: + """ + 准备任务描述，替换上下文变量 + + 支持以下占位符格式： + - {prd_output}: 产品需求文档输出 + - {qa_output}: QA 测试计划输出 + - {dev_output}: 技术方案输出 + """ + if not description: + return "" + + result = description + + # 尝试从上下文中获取输出 + prd_output = context.get('prd_content', context.get('task_0_output', '')) + qa_output = context.get('qa_plan', context.get('task_1_output', '')) + dev_output = context.get('dev_plan', context.get('task_2_output', '')) + + # 替换占位符 + result = result.replace('{prd_output}', str(prd_output)[:3000]) + result = result.replace('{qa_plan}', str(qa_output)[:3000]) + result = result.replace('{qa_output}', str(qa_output)[:3000]) + result = result.replace('{dev_plan}', str(dev_output)[:3000]) + result = result.replace('{dev_output}', str(dev_output)[:3000]) + + return result + + +def configure_llm(): + """配置 LLM 使用 Qwen3.5-flash (DashScope/阿里百炼)""" + import os + + # 检查 API Key 是否配置 + dashscope_api_key = os.getenv("DASHSCOPE_API_KEY") + if not dashscope_api_key: + raise ValueError( + "DASHSCOPE_API_KEY 未配置，请设置环境变量或在 .env 文件中配置" + ) + + # 使用 CrewAI 原生的 LLM 类（通过 OpenAI 兼容接口连接 DashScope） + from crewai.llm import LLM + + llm = LLM( + model=QWEN_MODEL_CONFIG["model"], # qwen-plus (Qwen3.5-flash) + api_key=dashscope_api_key, + base_url=QWEN_MODEL_CONFIG["base_url"], # https://dashscope.aliyuncs.com/compatible-mode/v1 + temperature=0.7, + max_tokens=4096, + ) + + return llm + + +class CrewFactory: + """Crew 工厂类 - 负责创建和执行业务流程""" + + @staticmethod + def create_crew( + task_id: str, + user_requirement: str, + skip_confirmation: bool = True + ) -> tuple[Crew, Dict[str, Any]]: + """ + 创建 Crew 实例 + + Args: + task_id: 任务 ID + user_requirement: 用户需求描述 + skip_confirmation: 是否跳过 Coordinator 的人工确认环节 + + Returns: + (Crew 实例，上下文信息) + """ + # 创建 Agents + agents = create_agents() + pm_agent = agents["product_manager"] + qa_agent = agents["qa_engineer"] + dev_agent = agents["software_developer"] + coord_agent = agents["coordinator"] + + # 配置 LLM + try: + llm = configure_llm() + # 为每个 agent 分配 LLM + for agent in agents.values(): + agent.llm = llm + except ValueError as e: + # 如果 LLM 配置失败，使用默认配置（会在运行时失败） + print(f"警告：LLM 配置失败 - {e}") + + # 创建任务 + # Task 1: ProductManager 分析需求 + pm_task = Task( + description=TASK_TEMPLATES["product_manager"].format( + user_requirement=user_requirement + ), + expected_output="完整的产品需求文档 (PRD)，包含功能列表、验收标准和风险评估", + agent=pm_agent, + ) + + # Task 2: QAEngineer 制定测试计划 + qa_task = Task( + description=TASK_TEMPLATES["qa_engineer"].format( + prd_content="{prd_output}" + ), + expected_output="详细的测试计划文档，包含测试策略、测试用例和性能测试方案", + agent=qa_agent, + context=[pm_task], + ) + + # Task 3: SoftwareDeveloper 设计技术方案 + dev_task = Task( + description=TASK_TEMPLATES["software_developer"].format( + prd_content="{prd_output}", + qa_plan="{qa_output}" + ), + expected_output="完整的技术方案文档，包含架构设计、API 接口、数据模型和核心代码实现", + agent=dev_agent, + context=[pm_task, qa_task], + ) + + # Task 4: Coordinator 最终审核 + coord_task = Task( + description=TASK_TEMPLATES["coordinator"].format( + prd_content="{prd_output}", + qa_plan="{qa_output}", + dev_plan="{dev_output}" + ), + expected_output="最终交付报告，包含质量评估、风险提示和交付结论", + agent=coord_agent, + context=[pm_task, qa_task, dev_task], + ) + + # 创建 Crew + crew = Crew( + agents=list(agents.values()), + tasks=[pm_task, qa_task, dev_task, coord_task], + process=Process.sequential, + verbose=True, + ) + + # 上下文信息 + context = { + "user_requirement": user_requirement, + "skip_confirmation": skip_confirmation, + "created_at": datetime.now().isoformat(), + } + + return crew, context + + @staticmethod + async def execute_crew_async( + task_id: str, + user_requirement: str, + skip_confirmation: bool = True + ) -> Dict[str, Any]: + """ + 异步执行 Crew 任务 + + Args: + task_id: 任务 ID + user_requirement: 用户需求描述 + skip_confirmation: 是否跳过人工确认 + + Returns: + 执行结果摘要 + """ + # 创建流队列 + await stream_manager.create_stream(task_id) + + # 发送开始事件 + await stream_manager.publish_event( + task_id=task_id, + event_type="start", + agent="System", + content="任务已启动，开始处理用户需求...", + metadata={"user_requirement": user_requirement[:200]} + ) + + try: + # 创建 Crew + crew, context = CrewFactory.create_crew( + task_id=task_id, + user_requirement=user_requirement, + skip_confirmation=skip_confirmation + ) + + # 创建 SSE 执行器 + executor = SSECrewExecutor(task_id) + + # 在线程池中执行（避免阻塞事件循环） + loop = asyncio.get_event_loop() + result = await loop.run_in_executor( + None, + lambda: executor.execute(crew, context) + ) + + # 发送完成事件 + await stream_manager.publish_event( + task_id=task_id, + event_type="end", + agent="System", + content="任务执行完成", + metadata={"success": True} + ) + + return { + "task_id": task_id, + "status": "completed", + "result": str(result)[:5000] if result else None, + "context": context, + } + + except Exception as e: + # 发送错误事件 + await stream_manager.publish_event( + task_id=task_id, + event_type="error", + agent="System", + content=str(e), + metadata={"error_type": type(e).__name__} + ) + + await stream_manager.close_stream(task_id) + + return { + "task_id": task_id, + "status": "failed", + "error": str(e), + "error_type": type(e).__name__, + } + + +# 便捷函数 +async def run_multi_agent_task( + user_requirement: str, + skip_confirmation: bool = True +) -> str: + """ + 运行多智能体任务的便捷函数 + + Args: + user_requirement: 用户需求描述 + skip_confirmation: 是否跳过人工确认 + + Returns: + task_id: 任务 ID（用于后续 SSE 流订阅） + """ + task_id = str(uuid.uuid4()) + + # 异步启动任务 + asyncio.create_task( + CrewFactory.execute_crew_async( + task_id=task_id, + user_requirement=user_requirement, + skip_confirmation=skip_confirmation + ) + ) + + return task_id diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..2a06410 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,68 @@ +version: '3.8' + +services: + multi-agent-system: + build: + context: . + dockerfile: Dockerfile + container_name: multi-agent-delivery-system + restart: unless-stopped + ports: + - "8000:8000" + environment: + # DashScope API Key（必需） + - DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY:-your_api_key_here} + # 可选配置 + - HOST=0.0.0.0 + - PORT=8000 + # 日志级别 + - LOG_LEVEL=info + volumes: + # 挂载日志目录 + - ./logs:/app/logs + # 如果需要持久化 .env 文件 + - ./.env:/app/.env:ro + networks: + - agent-network + # 资源限制 + deploy: + resources: + limits: + cpus: '2.0' + memory: 4G + reservations: + cpus: '0.5' + memory: 1G + # 健康检查 + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] + interval: 30s + timeout: 10s + 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 + +# 使用示例： +# 开发环境：docker-compose up -d +# 生产环境：docker-compose --profile production up -d diff --git a/example_usage.py b/example_usage.py new file mode 100644 index 0000000..b2713e2 --- /dev/null +++ b/example_usage.py @@ -0,0 +1,167 @@ +""" +示例使用脚本 +演示如何使用多智能体系统生成代码和文档 +""" + +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 new file mode 100644 index 0000000..e50a245 --- /dev/null +++ b/generated_output/README.md @@ -0,0 +1,188 @@ +# 生成的代码和文档 + +本目录包含多智能体系统自动生成的所有产出物。 + +## 📁 目录结构 + +每次任务执行后，会在 `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/main.py b/main.py new file mode 100644 index 0000000..c8f644a --- /dev/null +++ b/main.py @@ -0,0 +1,709 @@ +""" +FastAPI 主入口 +提供 RESTful API 和 SSE 流式接口 +""" + +import asyncio +import json +from datetime import datetime +from typing import Dict, Any, Optional +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import StreamingResponse, JSONResponse, HTMLResponse +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel, Field + +from crew_factory import CrewFactory, run_multi_agent_task +from stream_manager import stream_manager, create_sse_generator + + +# ==================== 数据模型 ==================== + +class RunTaskRequest(BaseModel): + """运行任务请求体""" + user_requirement: str = Field( + ..., + description="用户需求描述", + example="开发一个在线商城系统，支持用户注册、商品浏览、购物车和支付功能" + ) + skip_confirmation: bool = Field( + default=True, + description="是否跳过 Coordinator 的人工确认环节" + ) + + +class RunTaskResponse(BaseModel): + """运行任务响应体""" + task_id: str + status: str + message: Optional[str] = None + + +class SSEEvent(BaseModel): + """SSE 事件数据结构""" + agent: str + type: str + content: str + timestamp: str + task_id: str + + +# ==================== 生命周期管理 ==================== + +@asynccontextmanager +async def lifespan(app: FastAPI): + """应用生命周期管理""" + # 启动时执行 + print("🚀 Multi-Agent System 启动中...") + print("📡 SSE 流服务已就绪") + + # 启动后台任务：定期清理旧流 + asyncio.create_task(cleanup_streams_periodically()) + + yield + + # 关闭时清理 + print("👋 正在关闭所有 SSE 流...") + # 可以在这里添加清理逻辑 + + +# ==================== FastAPI 应用 ==================== + +app = FastAPI( + title="Multi-Agent Software Delivery System", + description=""" + 基于 CrewAI + Qwen3.5-flash 的多智能体软件交付系统 + + ## 核心功能 + - **产品需求分析**: ProductManager Agent 分析用户需求，生成 PRD + - **测试计划制定**: QAEngineer Agent 设计测试策略和用例 + - **技术方案设计**: SoftwareDeveloper Agent 输出架构设计和代码框架 + - **质量审核**: Coordinator Agent 审核所有产出物并生成交付报告 + + ## 实时通信 + 支持 SSE (Server-Sent Events) 协议，实时推送每个 Agent 的思考过程和任务状态 + + ## API 端点 + - `POST /api/run_task` - 启动新任务 + - `GET /api/stream/{task_id}` - 订阅任务执行日志（SSE） + - `GET /api/task/{task_id}/status` - 查询任务状态 + - `GET /test-ui` - 测试页面 + """, + version="1.0.0", + lifespan=lifespan, +) + +# CORS 中间件（允许跨域） +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # 生产环境应限制具体域名 + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +# ==================== 后台任务 ==================== + +async def cleanup_streams_periodically(interval: int = 300): + """定期清理旧的流（每 5 分钟）""" + while True: + await asyncio.sleep(interval) + try: + await stream_manager.cleanup_old_streams(max_age_seconds=3600) + except Exception as e: + print(f"清理流失败：{e}") + + +# ==================== API 路由 ==================== + +@app.post( + "/api/run_task", + response_model=RunTaskResponse, + summary="启动多智能体任务", + description="接收用户需求，启动 CrewAI 流程，异步执行并立即返回 task_id" +) +async def run_task(request: RunTaskRequest): + """ + 启动新的多智能体任务 + + - **user_requirement**: 用户需求描述 + - **skip_confirmation**: 是否跳过人工确认（默认 True） + + 返回 task_id 用于后续 SSE 流订阅 + """ + try: + # 生成 task_id 并启动任务 + task_id = await run_multi_agent_task( + user_requirement=request.user_requirement, + skip_confirmation=request.skip_confirmation + ) + + return RunTaskResponse( + task_id=task_id, + status="started", + message="任务已启动，请通过 /api/stream/{task_id} 订阅执行日志" + ) + + except Exception as e: + raise HTTPException( + status_code=500, + detail=f"启动任务失败：{str(e)}" + ) + + +@app.get( + "/api/stream/{task_id}", + summary="订阅任务执行日志 (SSE)", + description="建立 SSE 连接，实时接收任务执行过程中的所有事件" +) +async def stream_task_logs(task_id: str): + """ + SSE 端点 - 实时推送任务执行日志 + + 事件类型包括: + - **start**: 任务开始 + - **agent_start**: Agent 开始执行 + - **thought**: Agent 思考过程 + - **action**: Agent 执行动作 + - **output**: Agent 输出结果 + - **step_end**: 步骤完成 + - **end**: 任务结束 + - **error**: 发生错误 + + 数据格式: + ```json + { + "agent": "ProductManager", + "type": "thought", + "content": "正在分析用户需求...", + "timestamp": "2024-01-01T12:00:00", + "task_id": "uuid" + } + ``` + """ + # 检查任务是否存在 + stream = await stream_manager.get_stream(task_id) + if stream is None: + # 任务不存在，返回错误 + return StreamingResponse( + iter([f"data: {json.dumps({'error': 'Task not found', 'task_id': task_id})}\n\n"]), + media_type="text/event-stream" + ) + + # 创建 SSE 流 + return StreamingResponse( + create_sse_generator(task_id), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # Nginx 禁用缓冲 + } + ) + + +@app.get( + "/api/task/{task_id}/status", + summary="查询任务状态", + description="获取任务的当前状态和基本信息" +) +async def get_task_status(task_id: str): + """查询任务状态""" + stream = await stream_manager.get_stream(task_id) + + if stream is None: + return {"task_id": task_id, "status": "not_found"} + + return { + "task_id": task_id, + "status": "closed" if stream.is_closed else "running", + "queue_size": stream.queue.qsize() if hasattr(stream, 'queue') else 0, + } + + +@app.get( + "/api/streams", + summary="列出所有活跃流", + description="查看当前所有正在执行的任务" +) +async def list_active_streams(): + """列出所有活跃的 SSE 流""" + streams = stream_manager.list_active_streams() + return { + "total": len(streams), + "streams": streams + } + + +@app.get( + "/health", + summary="健康检查", + description="检查服务是否正常运行" +) +async def health_check(): + """健康检查端点""" + return { + "status": "healthy", + "timestamp": datetime.now().isoformat(), + "service": "Multi-Agent Software Delivery System" + } + + +# ==================== 测试页面 ==================== + +@app.get( + "/test-ui", + response_class=HTMLResponse, + summary="测试 UI 页面", + description="一个简单的 HTML 页面，用于测试 SSE 流功能" +) +async def test_ui(): + """返回测试页面""" + return HTMLResponse(content=get_test_page_html()) + + +def get_test_page_html() -> str: + """生成测试页面 HTML""" + return """ + + + + + 多智能体系统 - 测试 UI + + + +
+

🤖 多智能体软件交付系统

+ + +
+
+ + + +
+ 示例需求：
+ + + +
+
+ +
+
+ + +
+
+ + +
+ + +
+
+
+ + 等待中... +
+
+ Task ID: - +
+ +
+
+ + +
+

📊 实时事件日志

+
+
+ 暂无事件，点击上方"启动任务"按钮开始 +
+
+
+
+ + + +""" + + +# ==================== 主程序入口 ==================== + +if __name__ == "__main__": + import uvicorn + + # 加载环境变量 + try: + from dotenv import load_dotenv + load_dotenv() + except ImportError: + pass + + uvicorn.run( + "main:app", + host="0.0.0.0", + port=8000, + reload=True, + log_level="info" + ) diff --git a/nginx.conf b/nginx.conf new file mode 100644 index 0000000..cc043fa --- /dev/null +++ b/nginx.conf @@ -0,0 +1,129 @@ +events { + worker_connections 1024; +} + +http { + # 上游服务器配置 + upstream multi_agent_backend { + server multi-agent-system:8000; + keepalive 32; + } + + # 日志格式 + log_format main '$remote_addr - $remote_user [$time_local] "$request" ' + '$status $body_bytes_sent "$http_referer" ' + '"$http_user_agent" "$http_x_forwarded_for"'; + + access_log /var/log/nginx/access.log main; + error_log /var/log/nginx/error.log warn; + + # 连接超时设置 + proxy_connect_timeout 60s; + proxy_send_timeout 60s; + proxy_read_timeout 60s; + + # SSE 特殊配置：禁用缓冲 + # 这对于 Server-Sent Events 至关重要 + map $http_accept $sse_connection { + default "keep-alive"; + "text/event-stream" "keep-alive"; + } + + server { + listen 80; + server_name localhost; + + # 客户端请求体大小限制 + client_max_body_size 10M; + + # API 代理配置 + location /api/ { + proxy_pass http://multi_agent_backend; + + # 必要的代理头 + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # HTTP/1.1 支持（SSE 必需） + proxy_http_version 1.1; + proxy_set_header Connection ""; + + # 禁用缓冲（SSE 关键配置） + proxy_buffering off; + proxy_cache off; + proxy_request_buffering off; + + # Chunked 传输编码 + chunked_transfer_encoding on; + } + + # SSE 流端点特殊配置 + location /api/stream/ { + proxy_pass http://multi_agent_backend; + + # 代理头 + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + + # HTTP/1.1 和 Connection + proxy_http_version 1.1; + proxy_set_header Connection ""; + + # SSE 关键：完全禁用缓冲 + proxy_buffering off; + proxy_cache off; + proxy_request_buffering off; + + # Nginx 特殊指令：禁用 FastCGI 缓冲 + fastcgi_buffering off; + + # 保持长连接 + proxy_connect_timeout 60s; + proxy_send_timeout 300s; + proxy_read_timeout 300s; + + # SSE 心跳 + proxy_ignore_client_abort off; + } + + # 测试 UI 页面 + location /test-ui { + proxy_pass http://multi_agent_backend; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + } + + # 健康检查端点 + location /health { + proxy_pass http://multi_agent_backend; + access_log off; + } + + # API 文档 + location /docs { + proxy_pass http://multi_agent_backend; + proxy_set_header Host $host; + } + + location /openapi.json { + proxy_pass http://multi_agent_backend; + proxy_set_header Host $host; + } + } + + # HTTPS 配置（可选，取消注释启用） + # server { + # listen 443 ssl http2; + # server_name your-domain.com; + # + # ssl_certificate /etc/nginx/ssl/fullchain.pem; + # ssl_certificate_key /etc/nginx/ssl/privkey.pem; + # ssl_protocols TLSv1.2 TLSv1.3; + # ssl_ciphers HIGH:!aNULL:!MD5; + # + # # 同样的代理配置... + # } +} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..17e4b7b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +crewai==0.51.0 +langchain==0.1.0 +langchain-community==0.0.10 +dashscope==1.14.1 +python-dotenv==1.0.0 +pydantic==2.5.3 +uuid6==2024.1.12 +sse-starlette==2.0.0 diff --git a/stream_manager.py b/stream_manager.py new file mode 100644 index 0000000..f96ff08 --- /dev/null +++ b/stream_manager.py @@ -0,0 +1,283 @@ +""" +SSE 流管理器 +负责管理任务执行过程中的消息队列和 SSE 连接 +确保多用户并发时不同 task_id 的流互不干扰 + +关键技术点： +1. 使用 asyncio.Queue 实现异步非阻塞消息队列 +2. 通过 asyncio.Lock 保证并发安全 +3. 每个 task_id 独立队列，实现任务隔离 +4. 支持从同步线程（CrewAI）安全地发布事件到异步队列 +""" + +import asyncio +from datetime import datetime +from typing import Dict, AsyncGenerator, Optional, Any +from collections import deque +import uuid +import threading +from concurrent.futures import ThreadPoolExecutor + + +class StreamEvent: + """ + SSE 事件数据结构 + + 统一的 JSON 格式设计，便于前端解析： + { + "task_id": "550e8400-e29b...", + "sequence": 1, + "agent_name": "ProductManager", + "event_type": "thought", // 或 action, output, complete + "content": "正在分析用户需求...", + "timestamp": "2023-10-27T10:00:00Z" + } + """ + + # 全局序列号计数器（每个 task_id 独立） + _sequence_counters: Dict[str, int] = {} + + def __init__( + self, + event_type: str, + agent: str, + content: str, + task_id: str, + timestamp: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None + ): + self.event_type = event_type # start, thought, action, output, end, error + self.agent = agent + self.content = content + self.task_id = task_id + self.timestamp = timestamp or datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ') + self.metadata = metadata or {} + + # 为每个 task_id 维护独立的序列号 + if task_id not in StreamEvent._sequence_counters: + StreamEvent._sequence_counters[task_id] = 0 + StreamEvent._sequence_counters[task_id] += 1 + self.sequence = StreamEvent._sequence_counters[task_id] + + def to_dict(self) -> Dict[str, Any]: + """转换为字典格式用于 JSON 序列化""" + return { + "task_id": self.task_id, + "sequence": self.sequence, + "agent_name": self.agent, + "event_type": self.event_type, + "content": self.content, + "timestamp": self.timestamp, + **(self.metadata if self.metadata else {}) + } + + def to_sse_format(self) -> str: + """转换为 SSE 数据格式""" + import json + return f"data: {json.dumps(self.to_dict(), ensure_ascii=False)}\n\n" + + @classmethod + def reset_sequence(cls, task_id: str): + """重置指定 task_id 的序列号（任务重新开始时调用）""" + cls._sequence_counters[task_id] = 0 + + +class TaskStreamQueue: + """ + 单个任务的流式消息队列（线程安全） + + 并发处理逻辑： + - CrewAI 默认是同步运行的，而 FastAPI 和 SSE 需要异步 + - 使用 asyncio.Queue 的 run_coroutine_threadsafe 方法从同步线程安全地发布事件 + - 确保 stream_manager 能安全地在线程间传递消息 + """ + + def __init__(self, task_id: str, max_size: int = 1000): + self.task_id = task_id + self.queue: asyncio.Queue[StreamEvent] = asyncio.Queue(maxsize=max_size) + self.is_closed = False + self.subscribers: int = 0 + self._loop = asyncio.get_event_loop() + self._lock = threading.Lock() # 用于保护同步操作 + + async def put(self, event: StreamEvent) -> bool: + """向队列添加事件（异步调用）""" + if self.is_closed: + return False + try: + await asyncio.wait_for(self.queue.put(event), timeout=5.0) + return True + except asyncio.TimeoutError: + return False + except Exception: + return False + + def put_nowait(self, event: StreamEvent) -> bool: + """ + 从同步线程（如 CrewAI 事件处理器）安全地发布事件 + + 使用 run_coroutine_threadsafe 将协程提交到事件循环执行 + 这是实现 CrewAI（同步）与 SSE（异步）集成的关键 + """ + if self.is_closed: + return False + + try: + # 将协程提交到事件循环线程安全地执行 + future = asyncio.run_coroutine_threadsafe( + self.queue.put(event), + self._loop + ) + # 等待完成（带超时） + future.result(timeout=5.0) + return True + except Exception as e: + # print(f"发布事件失败：{e}") + return False + + async def get(self) -> Optional[StreamEvent]: + """从队列获取事件""" + if self.is_closed and self.queue.empty(): + return None + try: + return await self.queue.get() + except Exception: + return None + + def close(self): + """关闭队列""" + with self._lock: + self.is_closed = True + + async def stream_events(self) -> AsyncGenerator[StreamEvent, None]: + """异步生成器，持续产出事件直到队列关闭""" + while not (self.is_closed and self.queue.empty()): + try: + event = await asyncio.wait_for(self.queue.get(), timeout=30.0) + yield event + except asyncio.TimeoutError: + if self.is_closed and self.queue.empty(): + break + continue + except Exception: + break + + +class StreamManager: + """全局流管理器 - 管理所有任务的 SSE 流""" + + _instance: Optional['StreamManager'] = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + if self._initialized: + return + self._initialized = True + # task_id -> TaskStreamQueue 映射 + self.streams: Dict[str, TaskStreamQueue] = {} + self._lock = asyncio.Lock() + + async def create_stream(self, task_id: str) -> TaskStreamQueue: + """为指定 task_id 创建新的流队列""" + async with self._lock: + if task_id in self.streams: + # 如果已存在，先关闭旧的 + old_stream = self.streams[task_id] + old_stream.close() + + # 重置序列号计数器 + StreamEvent.reset_sequence(task_id) + + queue = TaskStreamQueue(task_id) + self.streams[task_id] = queue + return queue + + async def get_stream(self, task_id: str) -> Optional[TaskStreamQueue]: + """获取指定 task_id 的流队列""" + async with self._lock: + return self.streams.get(task_id) + + async def publish_event( + self, + task_id: str, + event_type: str, + agent: str, + content: str, + metadata: Optional[Dict[str, Any]] = None + ) -> bool: + """发布事件到指定任务的流队列""" + async with self._lock: + stream = self.streams.get(task_id) + if stream is None: + return False + + event = StreamEvent( + event_type=event_type, + agent=agent, + content=content, + task_id=task_id, + metadata=metadata + ) + return await stream.put(event) + + async def close_stream(self, task_id: str): + """关闭指定任务的流队列""" + async with self._lock: + if task_id in self.streams: + self.streams[task_id].close() + # 可以选择删除或保留（如果需要历史记录） + # del self.streams[task_id] + + async def cleanup_old_streams(self, max_age_seconds: int = 3600): + """清理超过指定时间的旧流（定期调用）""" + now = datetime.now() + to_remove = [] + + async with self._lock: + for task_id, stream in self.streams.items(): + if stream.is_closed: + to_remove.append(task_id) + + async with self._lock: + for task_id in to_remove: + del self.streams[task_id] + + def list_active_streams(self) -> list: + """列出所有活跃的流""" + return [ + {"task_id": tid, "closed": s.is_closed} + for tid, s in self.streams.items() + ] + + +# 全局单例 +stream_manager = StreamManager() + + +async def create_sse_generator(task_id: str) -> AsyncGenerator[str, None]: + """ + 创建 SSE 异步生成器 + 供 FastAPI StreamingResponse 使用 + """ + stream = await stream_manager.get_stream(task_id) + if stream is None: + # 创建一个新的流（如果不存在） + stream = await stream_manager.create_stream(task_id) + + try: + async for event in stream.stream_events(): + yield event.to_sse_format() + finally: + # 发送结束标记 + import json + end_event = { + "type": "connection_end", + "task_id": task_id, + "timestamp": datetime.now().isoformat() + } + yield f"data: {json.dumps(end_event)}\n\n" diff --git a/test_import.py b/test_import.py new file mode 100644 index 0000000..0104475 --- /dev/null +++ b/test_import.py @@ -0,0 +1,207 @@ +""" +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()