135 lines
3.6 KiB
Markdown
Executable File
135 lines
3.6 KiB
Markdown
Executable File
# 2026-04-03 项目记录 - Celery集成修订
|
|
|
|
## 今日工作摘要
|
|
|
|
完成bbav20311多Agent系统的Celery异步任务队列集成修订，解决API层与Celery层任务状态不一致问题。
|
|
|
|
---
|
|
|
|
## 1. 问题诊断
|
|
|
|
### 发现的问题
|
|
- `/api/agent/execute` 接口使用 `asyncio.create_task()` 直接执行，未调用Celery
|
|
- 任务状态存储在内存字典 `agent_tasks` 中，与Celery任务状态不同步
|
|
- API返回 completed，但Celery任务状态为 PENDING
|
|
|
|
### 根因
|
|
- 代码实现了Celery任务定义 (`tasks.py`)，但API层未实际调用
|
|
- 异步执行使用了纯asyncio方式，绕过了Celery队列
|
|
|
|
---
|
|
|
|
## 2. 修订内容
|
|
|
|
### 2.1 main.py (apps/web/vue3/main.py)
|
|
|
|
#### `/api/agent/execute` 接口修改
|
|
```python
|
|
# 修改前：使用 asyncio.create_task() 直接执行
|
|
asyncio.create_task(execute_agent_task_async())
|
|
|
|
# 修改后：提交到Celery队列
|
|
if CELERY_AVAILABLE and execute_agent_task is not None:
|
|
celery_task = execute_agent_task.delay(
|
|
task_id=task_id,
|
|
message=request.message,
|
|
session_id=request.session_id or task_id,
|
|
history=request.history if request.history else []
|
|
)
|
|
agent_tasks[task_id]["celery_task_id"] = celery_task.id
|
|
```
|
|
|
|
#### `/api/agent/status/{task_id}` 接口修改
|
|
```python
|
|
# 添加Celery状态查询
|
|
if CELERY_AVAILABLE and "celery_task_id" in task:
|
|
from celery_app import app as celery_app
|
|
celery_task = celery_app.AsyncResult(task["celery_task_id"])
|
|
# 根据celery_task.state更新状态
|
|
```
|
|
|
|
### 2.2 tasks.py
|
|
|
|
#### 修复 `update_state()` 调用
|
|
```python
|
|
# 修改前：self.update_state(state='RUNNING', ...) # task_id为空
|
|
# 修改后：
|
|
celery_task_id = self.request.id if self.request else task_id
|
|
self.update_state(
|
|
task_id=celery_task_id,
|
|
state='RUNNING',
|
|
meta={...}
|
|
)
|
|
```
|
|
|
|
---
|
|
|
|
## 3. 验证结果
|
|
|
|
### 3.1 服务状态
|
|
| 组件 | 状态 | 端口/PID |
|
|
|------|------|----------|
|
|
| FastAPI | 运行中 | :7865 (PID: 170514) |
|
|
| Celery Worker | 运行中 | 3进程 |
|
|
| Redis | 运行中 | :6379 |
|
|
| RAG API | 运行中 | :8001 |
|
|
|
|
### 3.2 任务提交流程验证
|
|
```bash
|
|
# 提交任务
|
|
curl -X POST http://localhost:7865/api/agent/execute \
|
|
-d '{"message": "测试任务", "session_id": "test_001"}'
|
|
|
|
# 响应
|
|
{"success": true, "task_id": "agent_task_...", "status": "pending"}
|
|
|
|
# 查询状态
|
|
curl http://localhost:7865/api/agent/status/{task_id}
|
|
|
|
# 状态流转: pending → running → completed
|
|
```
|
|
|
|
### 3.3 测试任务执行结果
|
|
- 任务ID: `agent_task_1775226937_cd1bce7d`
|
|
- Celery任务ID: `d99ef56b-25ab-4f42-b3c5-aa05bbdadcbb`
|
|
- 执行时间: ~8秒
|
|
- 结果: 成功完成意图识别和分析方案生成
|
|
|
|
---
|
|
|
|
## 4. 已知问题
|
|
|
|
### 4.1 RNAseq QC执行问题
|
|
- 任务提交成功，但执行阶段返回 `分析失败: []`
|
|
- 需要进一步诊断Orchestrator执行逻辑
|
|
- 可能与输入数据路径或工具配置有关
|
|
|
|
### 4.2 服务启动方式
|
|
- 需要使用 `nohup` 或 `systemd` 保持服务后台运行
|
|
- 直接 `&` 后台运行会导致进程退出
|
|
|
|
---
|
|
|
|
## 5. 文件变更
|
|
|
|
| 文件 | 变更类型 | 说明 |
|
|
|------|----------|------|
|
|
| `apps/web/vue3/main.py` | 修改 | Agent执行接口改用Celery |
|
|
| `tasks.py` | 修改 | 修复update_state task_id问题 |
|
|
|
|
---
|
|
|
|
## 6. 后续计划
|
|
|
|
1. **修复RNAseq QC执行问题** - 诊断Orchestrator执行失败原因
|
|
2. **添加任务重试机制** - 失败任务自动重试
|
|
3. **完善日志记录** - 添加更详细的执行日志
|
|
|
|
---
|
|
|
|
## 参考
|
|
|
|
- 项目路径: `/home/xia-lab/wutx/00test/bbav20311/`
|
|
- Celery Worker启动: `celery -A celery_app worker --loglevel=info --concurrency=2`
|
|
- FastAPI启动: `nohup python3 apps/web/vue3/main.py > backend.log 2>&1 &`
|