Files
PlantAI-dev/celery_app.py
T
potaserandClaude Opus 5 e641c8b9ab feat: 轻量拓展分析 + WGS2GWAS 阶段支持 + 结果区域重构 + 物种角色解析增强 + 会话 Job 隔离
【轻量拓展分析】
- 新增 gfa_core/extension_stage_lightweight/ 核心模块,支持用户上传代码/数据/原始命令
- 前端卡片式交互流程:候选核对 → 验证 → 结果审查 → 授权发布 → 运行
- Celery 不受限任务 + 轮询机制 + sessionStorage 操作恢复
- 启动脚本新增 BBAV5_LIGHTWEIGHT_* 环境变量

【WGS2GWAS 支持】
- 新增 13 个 WGS 阶段定义(QC → MAF/Missing Filter)
- email_haplotype_region.py 新增 LTR-RT 分析功能

【结果区域重构】
- 单层折叠 → 三组双层折叠:RNA-seq / 基因家族 / WGS2GWAS
- 结果按执行时间排序,保留最新尝试

【物种角色解析增强】
- 新增 _resolve_subfamily_request_species_roles()
- 支持 reference 角色的识别
- 直接 stage 请求保留 intent 中的物种角色

【会话 Job 隔离】
- rebind_job() 跨 job 时创建新会话,返回新 conversation_id

【前端交互增强】
- RAG 文献问答模式
- NL 触发 config/samples 自动填写、亚家族确认表、基因家族成员确认表
- 执行状态从后端 resource_tasks 恢复

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-05 09:40:45 +08:00

74 lines
2.0 KiB
Python
Executable File

import os
import sys
from pathlib import Path
from celery import Celery
from kombu import Exchange, Queue
LOCAL_PROJECT_ROOT = Path(__file__).resolve().parent
if str(LOCAL_PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(LOCAL_PROJECT_ROOT))
from gfa_core.project_paths import PROJECT_ROOT
os.environ.setdefault("PROJECT_ROOT", str(PROJECT_ROOT))
os.environ.setdefault("BBAV5_PROJECT_ROOT", str(PROJECT_ROOT))
QUEUE_PREFIX = os.environ.get("BBAV5_QUEUE_PREFIX", "plantai_dev_9000").strip() or "plantai_dev_9000"
REDIS_URL = os.environ.get("BBAV5_REDIS_URL", "redis://localhost:6379/10")
def q(name: str) -> str:
return f"{QUEUE_PREFIX}.{name}"
TASK_EXCHANGE = Exchange(q("tasks"), type="direct")
def queue(name: str) -> Queue:
queue_name = q(name)
return Queue(queue_name, TASK_EXCHANGE, routing_key=queue_name)
app = Celery(
"plantai_dev_9000",
broker=REDIS_URL,
backend=REDIS_URL,
include=["tasks"],
)
app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="Asia/Shanghai",
enable_utc=True,
task_track_started=True,
task_default_queue=q("default"),
task_default_exchange=q("tasks"),
task_default_exchange_type="direct",
task_default_routing_key=q("default"),
task_queues=(
queue("default"),
queue("heavy"),
queue("medium"),
queue("light"),
queue("rnaseq"),
),
task_routes={
"tasks.execute_agent_task": {"queue": q("default"), "routing_key": q("default")},
},
task_time_limit=172800,
task_soft_time_limit=169200,
task_annotations={
"tasks.execute_agent_task": {"time_limit": 3600, "soft_time_limit": 3300},
"tasks.execute_single_stage_task": {"time_limit": 172800, "soft_time_limit": 169200},
"tasks.execute_lightweight_extension_operation": {
"time_limit": None,
"soft_time_limit": None,
},
},
worker_prefetch_multiplier=1,
)
if __name__ == "__main__":
app.start()