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>
This commit is contained in:
@@ -13,6 +13,19 @@ sys.path.insert(0, str(PROJECT_ROOT))
|
||||
from celery_app import app
|
||||
|
||||
|
||||
@app.task(
|
||||
bind=True,
|
||||
name="tasks.execute_lightweight_extension_operation",
|
||||
time_limit=None,
|
||||
soft_time_limit=None,
|
||||
)
|
||||
def execute_lightweight_extension_operation(self, payload: dict) -> dict:
|
||||
"""Run one lightweight-extension operation without a Celery deadline."""
|
||||
from gfa_core.extension_stage_lightweight.api import execute_queued_operation
|
||||
|
||||
return execute_queued_operation(payload)
|
||||
|
||||
|
||||
def normalize_stage_params(stage: str, params: dict | None) -> tuple[dict, dict]:
|
||||
"""Clamp per-stage CPU at the worker boundary and return resource metadata."""
|
||||
params = dict(params or {})
|
||||
@@ -27,16 +40,19 @@ def normalize_stage_params(stage: str, params: dict | None) -> tuple[dict, dict]
|
||||
|
||||
explicit_cpu = requested_cpu not in {None, "", 0, "0"}
|
||||
project_limit = rnaseq_project_cpu_limit(requested_cpu if explicit_cpu else None)
|
||||
explicit_threads = params.get("threads_per_sample") or params.get("threadsPerSample")
|
||||
plan = recommend_rnaseq_parallelism(
|
||||
total_cpu=project_limit,
|
||||
sample_parallel_jobs=params.get("sample_parallel_jobs") or params.get("sampleParallelJobs"),
|
||||
threads_per_sample=params.get("threads_per_sample") or params.get("threadsPerSample"),
|
||||
sample_count=params.get("sample_count") or params.get("sampleCount"),
|
||||
stage=stage,
|
||||
)
|
||||
effective_cpu = clamp_stage_cpu(stage, plan["total_cpu"])
|
||||
effective_cpu = clamp_stage_cpu(stage, project_limit)
|
||||
params["sample_parallel_jobs"] = plan["sample_parallel_jobs"]
|
||||
params["threads_per_sample"] = plan["threads_per_sample"]
|
||||
if explicit_threads:
|
||||
params["threads_per_sample"] = max(1, int(explicit_threads))
|
||||
else:
|
||||
params.pop("threads_per_sample", None)
|
||||
params.pop("threadsPerSample", None)
|
||||
params["rnaseq_project_max_cpu"] = plan["project_cpu_limit"]
|
||||
params["resource_decision"] = {
|
||||
"stage": stage,
|
||||
@@ -44,8 +60,11 @@ def normalize_stage_params(stage: str, params: dict | None) -> tuple[dict, dict]
|
||||
"effective_cpu": effective_cpu,
|
||||
"project_cpu_limit": plan["project_cpu_limit"],
|
||||
"sample_parallel_jobs": plan["sample_parallel_jobs"],
|
||||
"threads_per_sample": plan["threads_per_sample"],
|
||||
"reason": f"{stage} project CPU budget split across sample-level parallel jobs",
|
||||
"threads_per_sample": max(1, int(explicit_threads)) if explicit_threads else None,
|
||||
"reason": (
|
||||
f"{stage} preserves the selected total CPU budget; per-sample threads "
|
||||
"are derived after the real sample count is known"
|
||||
),
|
||||
}
|
||||
except Exception as exc:
|
||||
requested_cpu = params.get("requested_cpu", params.get("cpu"))
|
||||
|
||||
Reference in New Issue
Block a user