1237 lines
54 KiB
Python
Executable File
1237 lines
54 KiB
Python
Executable File
"""
|
||
Celery任务定义
|
||
"""
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
# 添加项目路径
|
||
PROJECT_ROOT = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from celery_app import app
|
||
|
||
|
||
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 {})
|
||
try:
|
||
from gfa_core.mas.stage_catalog import clamp_stage_cpu, get_stage_resource_profile
|
||
|
||
requested_cpu = params.get("requested_cpu", params.get("cpu"))
|
||
effective_cpu = clamp_stage_cpu(stage, params.get("cpu"))
|
||
profile = get_stage_resource_profile(stage)
|
||
if stage in {"rnaseq_qc", "rnaseq_align"}:
|
||
from gfa_core.mas.resource_governor import recommend_rnaseq_parallelism, rnaseq_project_cpu_limit
|
||
|
||
explicit_cpu = requested_cpu not in {None, "", 0, "0"}
|
||
project_limit = rnaseq_project_cpu_limit(requested_cpu if explicit_cpu else None)
|
||
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"])
|
||
params["sample_parallel_jobs"] = plan["sample_parallel_jobs"]
|
||
params["threads_per_sample"] = plan["threads_per_sample"]
|
||
params["rnaseq_project_max_cpu"] = plan["project_cpu_limit"]
|
||
params["resource_decision"] = {
|
||
"stage": stage,
|
||
"requested_cpu": requested_cpu,
|
||
"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",
|
||
}
|
||
except Exception as exc:
|
||
requested_cpu = params.get("requested_cpu", params.get("cpu"))
|
||
try:
|
||
effective_cpu = max(1, min(int(requested_cpu), 4))
|
||
except (TypeError, ValueError):
|
||
effective_cpu = 4
|
||
profile = {
|
||
"queue": "medium",
|
||
"resource_class": "medium",
|
||
"default_cpu": effective_cpu,
|
||
"max_cpu": effective_cpu,
|
||
"profile_error": str(exc),
|
||
}
|
||
|
||
params["requested_cpu"] = requested_cpu
|
||
params["cpu"] = effective_cpu
|
||
_sync_member_set_id_with_scope(stage, params)
|
||
return params, profile
|
||
|
||
|
||
def _sync_member_set_id_with_scope(stage: str, params: dict) -> None:
|
||
"""Keep a scope-derived member_set_id consistent with the execute-time scope_id.
|
||
|
||
The execute layer can hand down a comp-polluted scope file together with a
|
||
plan-time (no-comp) member_set_id, producing a self-contradictory manifest
|
||
that postflight rejects as stage_identity_contract_mismatch. Re-derive the
|
||
scope token from the actual scope file so the CLI, the manifest and the
|
||
postflight check all agree. User-defined subsets (subfamily-xxx) are NOT
|
||
scope-derived and are left untouched.
|
||
"""
|
||
member_set_id = str(params.get("memberSetId") or params.get("member_set_id") or "").strip()
|
||
scope_file = str(params.get("scopeFile") or params.get("scope_file") or "").strip()
|
||
if not member_set_id or not scope_file:
|
||
return
|
||
if "subfamily-" in member_set_id:
|
||
return
|
||
parts = member_set_id.split(":")
|
||
if len(parts) != 3 or parts[2] not in {"all_members", "domain_refined"}:
|
||
return
|
||
try:
|
||
from gfa_core.gene_family_scope import read_scope, sanitize_token
|
||
|
||
scope_path = Path(scope_file)
|
||
if not scope_path.is_absolute():
|
||
scope_path = PROJECT_ROOT / scope_path
|
||
if not scope_path.exists():
|
||
return
|
||
scope_id = str(read_scope(scope_path).get("scope_id") or "").strip()
|
||
if not scope_id:
|
||
return
|
||
expected_token = sanitize_token(scope_id)
|
||
analysis_route = str(params.get("analysisRoute") or params.get("analysis_route") or parts[1] or "").strip()
|
||
if analysis_route == "unscoped":
|
||
analysis_route = "single_species" if stage == "identify" else ""
|
||
expected_route = analysis_route or parts[1]
|
||
expected_selector = "domain_refined" if expected_route == "subfamily_evidence" else parts[2]
|
||
if parts[0] == expected_token and parts[1] == expected_route and parts[2] == expected_selector:
|
||
return
|
||
new_member_set_id = f"{expected_token}:{expected_route}:{expected_selector}"
|
||
print(f"[MEMBER_SET SYNC] scope_id={scope_id} member_set_id {member_set_id} -> {new_member_set_id}")
|
||
if params.get("memberSetId"):
|
||
params["memberSetId"] = new_member_set_id
|
||
if params.get("member_set_id"):
|
||
params["member_set_id"] = new_member_set_id
|
||
if not params.get("memberSetId") and not params.get("member_set_id"):
|
||
params["memberSetId"] = new_member_set_id
|
||
except Exception as exc:
|
||
print(f"[MEMBER_SET SYNC] skipped: {exc}")
|
||
|
||
|
||
def _nonempty_prefix(value) -> str:
|
||
text = str(value or "").strip()
|
||
return text if text and text.lower() not in {"none", "null", "undefined", "nan", "-"} else ""
|
||
|
||
|
||
def _merge_manifest_identity_params(params: dict | None, manifest: dict | None) -> dict:
|
||
"""Prefer the identity actually written by the stage for postflight checks."""
|
||
from gfa_core.mas.tools.stage_identity import merge_manifest_identity_params
|
||
|
||
return merge_manifest_identity_params(params, manifest)
|
||
|
||
|
||
def _gene_family_from_config(config_path: Path):
|
||
if not config_path.exists():
|
||
return None
|
||
try:
|
||
import pandas as pd
|
||
sheets = pd.read_excel(config_path, sheet_name=None)
|
||
for sheet_name, df in sheets.items():
|
||
if str(sheet_name).lower() in {"analysis", "meta", "metadata"}:
|
||
for _, row in df.fillna("").iterrows():
|
||
cells = [str(v).strip() for v in row.tolist()]
|
||
for idx, cell in enumerate(cells[:-1]):
|
||
if cell.lower() in {"family", "gene_family", "analysis_family", "project_family"}:
|
||
value = _nonempty_prefix(cells[idx + 1])
|
||
if value:
|
||
return value
|
||
if "Gene_Family" in df.columns:
|
||
for value in df["Gene_Family"].tolist():
|
||
value = _nonempty_prefix(value)
|
||
if value:
|
||
return value
|
||
except Exception as exc:
|
||
print(f"[Celery] Failed to infer gene family from {config_path}: {exc}")
|
||
return None
|
||
|
||
|
||
def _rnaseq_prefix_from_final(final_dir: Path) -> str:
|
||
if not final_dir.exists():
|
||
return ""
|
||
preferred = [
|
||
"*.rnaseq_wgcna.manifest.json",
|
||
"*.rnaseq_fgsea.manifest.json",
|
||
"*.rnaseq_candidate_family.manifest.json",
|
||
"*.rnaseq_sample_qc.manifest.json",
|
||
"*.rnaseq_de.manifest.json",
|
||
"*.rnaseq_quant.manifest.json",
|
||
"*.rnaseq_align.manifest.json",
|
||
"*.rnaseq_qc.manifest.json",
|
||
"*.annotation.manifest.json",
|
||
"*.genome_prepare.manifest.json",
|
||
]
|
||
for pattern in preferred:
|
||
matches = sorted(final_dir.glob(pattern))
|
||
if len(matches) == 1:
|
||
suffix = pattern[1:]
|
||
return matches[0].name[: -len(suffix)]
|
||
return ""
|
||
|
||
|
||
def resolve_effective_prefix(job_id: str, prefix: str = "", stage: str = "") -> str:
|
||
provided = _nonempty_prefix(prefix)
|
||
gene_family_stages = {
|
||
"identify", "phylogeny", "motif", "domain", "structure", "promoter", "tfbs",
|
||
"synteny", "interspecies_synteny", "karyotype", "combo", "ppi",
|
||
"pep", "summary", "reference_classification", "tree_plot",
|
||
}
|
||
|
||
def normalize_gene_family_prefix(value: str) -> str:
|
||
value = _nonempty_prefix(value)
|
||
if stage in gene_family_stages and value.startswith("Multi."):
|
||
family = value.split(".", 1)[1].strip()
|
||
if family:
|
||
return f"Target.{family}"
|
||
return value
|
||
|
||
final_dir = PROJECT_ROOT / "workspace" / job_id / "final"
|
||
if stage in {"rnaseq_enrich", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_sample_qc", "rnaseq_fgsea", "rnaseq_wgcna", "rnaseq_candidate_family"}:
|
||
rnaseq_prefix = _rnaseq_prefix_from_final(final_dir)
|
||
if rnaseq_prefix and rnaseq_prefix != provided:
|
||
print(f"[Celery] Overriding RNA-seq prefix {provided or '<empty>'} -> {rnaseq_prefix}")
|
||
return rnaseq_prefix
|
||
if provided:
|
||
return normalize_gene_family_prefix(provided)
|
||
input_dir = PROJECT_ROOT / "input" / job_id
|
||
species_config = input_dir / "genome_prepare" / "species_config.json"
|
||
if species_config.exists():
|
||
try:
|
||
import json
|
||
data = json.loads(species_config.read_text(encoding="utf-8"))
|
||
target = data.get("target") if isinstance(data.get("target"), dict) else None
|
||
if target:
|
||
value = _nonempty_prefix(target.get("prefix"))
|
||
if value:
|
||
return normalize_gene_family_prefix(value)
|
||
value = _nonempty_prefix(data.get("prefix"))
|
||
if value:
|
||
return normalize_gene_family_prefix(value)
|
||
except Exception as exc:
|
||
print(f"[Celery] Failed to read species_config prefix: {exc}")
|
||
for base in [final_dir, input_dir / "final"]:
|
||
if base.exists():
|
||
for mf in sorted(base.glob("Target.*.genome_prepare.manifest.json")):
|
||
name = mf.name
|
||
return name.split(".genome_prepare.manifest", 1)[0]
|
||
for mf in sorted(base.glob("Multi.*.genome_prepare.manifest.json")):
|
||
name = mf.name
|
||
return normalize_gene_family_prefix(name.split(".genome_prepare.manifest", 1)[0])
|
||
family = _gene_family_from_config(input_dir / "config.xlsx")
|
||
if family:
|
||
return f"Target.{family}"
|
||
for pattern in ("Target.*.final_members.pep.fa", "Multi.*.final_members.pep.fa"):
|
||
for file in sorted(final_dir.glob(pattern)):
|
||
return normalize_gene_family_prefix(file.name[:-len(".final_members.pep.fa")])
|
||
return ""
|
||
|
||
|
||
def _split_prefix(value: str) -> tuple[str, str]:
|
||
text = str(value or "").strip()
|
||
if "." in text:
|
||
left, right = text.split(".", 1)
|
||
return left.strip(), right.strip()
|
||
return "", text
|
||
|
||
|
||
def _list_param(value) -> list[str]:
|
||
if isinstance(value, list):
|
||
return [str(item).strip() for item in value if str(item).strip()]
|
||
import re
|
||
|
||
return [item.strip() for item in re.split(r"[,;，；\s]+", str(value or "")) if item.strip()]
|
||
|
||
|
||
def _resolved_artifacts_from_decisions(decisions: dict) -> dict:
|
||
resolved = {}
|
||
for key, decision in (decisions or {}).items():
|
||
if not isinstance(decision, dict):
|
||
continue
|
||
selected = decision.get("selected") if isinstance(decision.get("selected"), dict) else None
|
||
if selected and selected.get("path"):
|
||
resolved[key] = {
|
||
"path": selected.get("path"),
|
||
"source": selected.get("source", ""),
|
||
"kind": selected.get("kind", ""),
|
||
"role": selected.get("role", ""),
|
||
"species": selected.get("species", ""),
|
||
"family": selected.get("family", ""),
|
||
}
|
||
return resolved
|
||
|
||
|
||
def attach_semantic_execution_context(job_id: str, stage: str, prefix: str, params: dict | None) -> dict:
|
||
"""Attach resolver decisions to params for execution and diagnosis.
|
||
|
||
/api/execute already performs a semantic preflight, but Celery workers can
|
||
also be called directly or run after code reloads. Rebuilding here keeps
|
||
execution, diagnosis, and logs on the same project-context view.
|
||
"""
|
||
params = dict(params or {})
|
||
if isinstance(params.get("semanticPreflight"), dict) and isinstance(params.get("resolvedArtifacts"), dict):
|
||
return params
|
||
try:
|
||
from gfa_core.mas.artifact_resolver import ArtifactResolver
|
||
from gfa_core.mas.context_index import build_project_context
|
||
|
||
context = build_project_context(job_id=job_id, prefix=prefix, project_root=PROJECT_ROOT)
|
||
resolver = ArtifactResolver(context)
|
||
prefix_species, prefix_family = _split_prefix(prefix)
|
||
family = str(params.get("geneFamily") or params.get("family") or prefix_family or context.family or "").strip()
|
||
target_species = _list_param(params.get("targetSpecies") or params.get("target_species") or [])
|
||
if target_species:
|
||
species_list = target_species
|
||
elif prefix_species and prefix_species.lower() not in {"multi", "target"}:
|
||
species_list = [prefix_species]
|
||
elif len(context.target_species) == 1:
|
||
species_list = [context.target_species[0]]
|
||
else:
|
||
species_list = [""]
|
||
|
||
decisions = {}
|
||
for species in species_list:
|
||
for role, decision in resolver.resolve_stage(stage, species=species, family=family).items():
|
||
decisions[f"{species or 'auto'}:{role}"] = decision.to_dict()
|
||
|
||
semantic_preflight = {
|
||
"valid": True,
|
||
"stage": stage,
|
||
"prefix": prefix,
|
||
"decisions": decisions,
|
||
"context": {
|
||
"family": context.family,
|
||
"target_species": context.target_species,
|
||
"reference_species": context.reference_species,
|
||
"comparative_species": context.comparative_species,
|
||
"config_path": context.config_path,
|
||
"input_dir": context.input_dir,
|
||
"final_dir": context.final_dir,
|
||
},
|
||
}
|
||
params["semanticPreflight"] = semantic_preflight
|
||
params["resolvedArtifacts"] = _resolved_artifacts_from_decisions(decisions)
|
||
except Exception as exc:
|
||
params["semanticContextError"] = str(exc)
|
||
return params
|
||
|
||
# 初始化Orchestrator（每个worker进程初始化一次）
|
||
_orchestrator = None
|
||
|
||
def get_orchestrator():
|
||
global _orchestrator
|
||
if _orchestrator is None:
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.memory import MemoryManager
|
||
from gfa_core.mas.core.langgraph_orchestrator import LangGraphOrchestrator
|
||
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_config["openai_api_key"] = mas_config.settings.openai_api_key
|
||
llm_config["openai_base_url"] = mas_config.settings.openai_base_url
|
||
llm_manager = LLMManager(config=llm_config)
|
||
_orchestrator = LangGraphOrchestrator(llm_manager=llm_manager)
|
||
return _orchestrator
|
||
|
||
|
||
def validate_single_stage_artifacts(job_id: str, prefix: str, stage: str, result: dict | None = None) -> dict | None:
|
||
"""Return a failure response if a single-stage task did not produce valid artifacts."""
|
||
from gfa_core.mas.tools.stage_identity import get_stage_status_for_identity, resolve_stage_identity
|
||
from gfa_core.mas.tools.stage_status import get_stage_status
|
||
|
||
actual_stage = stage
|
||
actual_prefix = prefix
|
||
manifest = {}
|
||
if isinstance(result, dict):
|
||
final_output = result.get("final_output", {})
|
||
if isinstance(final_output, dict) and isinstance(final_output.get("manifest"), dict):
|
||
manifest = final_output.get("manifest") or {}
|
||
elif isinstance(result.get("manifest"), dict):
|
||
manifest = result.get("manifest") or {}
|
||
if isinstance(manifest, dict) and manifest:
|
||
actual_stage = _nonempty_prefix(manifest.get("stage")) or actual_stage
|
||
actual_prefix = _nonempty_prefix(manifest.get("prefix")) or actual_prefix
|
||
|
||
stage_context = manifest.get("stage_context") or manifest.get("stageContext") or {} if isinstance(manifest, dict) else {}
|
||
identity_params = {
|
||
"prefix": actual_prefix,
|
||
"scopeFile": manifest.get("scope_file") or stage_context.get("scope_file") or "" if isinstance(manifest, dict) else "",
|
||
"scopeId": manifest.get("scope_id") or stage_context.get("scope_id") or "" if isinstance(manifest, dict) else "",
|
||
"scopeMode": manifest.get("scope_mode") or stage_context.get("scope_mode") or "" if isinstance(manifest, dict) else "",
|
||
"analysisRoute": manifest.get("analysis_route") or stage_context.get("analysis_route") or "" if isinstance(manifest, dict) else "",
|
||
"memberSetId": manifest.get("member_set_id") or stage_context.get("member_set_id") or "" if isinstance(manifest, dict) else "",
|
||
"geneFamily": manifest.get("gene_family") or manifest.get("family") or stage_context.get("family") or "" if isinstance(manifest, dict) else "",
|
||
"targetSpecies": manifest.get("target_species") or stage_context.get("target_species") or [] if isinstance(manifest, dict) else [],
|
||
"referenceSpecies": manifest.get("reference_species") or stage_context.get("reference_species") or [] if isinstance(manifest, dict) else [],
|
||
"comparativeSpecies": manifest.get("comparative_species") or stage_context.get("comparative_species") or [] if isinstance(manifest, dict) else [],
|
||
"wgsSampleSetId": manifest.get("wgs_sample_set_id") if isinstance(manifest, dict) else "",
|
||
"wgsReferenceId": manifest.get("wgs_reference_id") if isinstance(manifest, dict) else "",
|
||
"wgsAnalysisId": manifest.get("wgs_analysis_id") if isinstance(manifest, dict) else "",
|
||
"genotypeScopeId": manifest.get("genotype_scope_id") if isinstance(manifest, dict) else "",
|
||
"phenotypeScopeId": manifest.get("phenotype_scope_id") if isinstance(manifest, dict) else "",
|
||
"covariateScopeId": manifest.get("covariate_scope_id") if isinstance(manifest, dict) else "",
|
||
"modelScopeId": manifest.get("model_scope_id") if isinstance(manifest, dict) else "",
|
||
"sourceModelScopeId": manifest.get("source_model_scope_id") if isinstance(manifest, dict) else "",
|
||
"analysisScopeId": manifest.get("analysis_scope_id") if isinstance(manifest, dict) else "",
|
||
"runId": manifest.get("run_id") if isinstance(manifest, dict) else "",
|
||
}
|
||
try:
|
||
identity = resolve_stage_identity(
|
||
stage=actual_stage,
|
||
prefix=actual_prefix,
|
||
params=identity_params,
|
||
job_id=job_id,
|
||
)
|
||
stage_status = get_stage_status_for_identity(
|
||
job_id=job_id,
|
||
identity=identity,
|
||
strict_identity=bool(
|
||
identity.scope_id
|
||
or identity.analysis_route
|
||
or identity.member_set_id
|
||
or identity.wgs_sample_set_id
|
||
or identity.wgs_reference_id
|
||
or identity.wgs_analysis_id
|
||
or identity.genotype_scope_id
|
||
or identity.phenotype_scope_id
|
||
or identity.covariate_scope_id
|
||
or identity.model_scope_id
|
||
or identity.source_model_scope_id
|
||
or identity.analysis_scope_id
|
||
),
|
||
)
|
||
except Exception as exc:
|
||
scope_id = str(identity_params.get("scopeId") or "")
|
||
analysis_route = str(identity_params.get("analysisRoute") or "")
|
||
member_set_id = str(identity_params.get("memberSetId") or "")
|
||
print(f"[Celery] Identity-aware artifact validation fallback for {stage}: {exc}")
|
||
stage_status = get_stage_status(
|
||
job_id=job_id,
|
||
prefix=actual_prefix,
|
||
stage=actual_stage,
|
||
scope_id=scope_id,
|
||
analysis_route=analysis_route,
|
||
member_set_id=member_set_id,
|
||
strict_identity=bool(scope_id or analysis_route or member_set_id),
|
||
)
|
||
if actual_stage in {"rnaseq_go_enrich", "rnaseq_kegg_enrich"} and stage_status.completed:
|
||
final_dir = PROJECT_ROOT / "workspace" / job_id / "final"
|
||
suffix = ".GO_enrich.txt" if actual_stage == "rnaseq_go_enrich" else ".KEGG_enrich.txt"
|
||
if any(final_dir.glob(f"*{suffix}")):
|
||
return None
|
||
|
||
if stage_status.completed:
|
||
return None
|
||
|
||
business_messages: list[str] = []
|
||
if isinstance(result, dict):
|
||
for error in result.get("errors", []) or []:
|
||
if error:
|
||
business_messages.append(str(error))
|
||
final_output = result.get("final_output", {})
|
||
if isinstance(final_output, dict):
|
||
for key in ("response", "message", "error"):
|
||
value = final_output.get(key)
|
||
if value:
|
||
business_messages.append(str(value))
|
||
|
||
business_error = next(
|
||
(
|
||
msg for msg in business_messages
|
||
if "❌" in msg or "failed" in msg.lower() or "cannot choose a background" in msg.lower()
|
||
),
|
||
"",
|
||
)
|
||
|
||
detail_parts = []
|
||
if stage_status.errors:
|
||
detail_parts.extend(stage_status.errors)
|
||
if stage_status.missing_artifacts:
|
||
detail_parts.append("缺失产物: " + "; ".join(stage_status.missing_artifacts))
|
||
detail_msg = "；".join(detail_parts) or f"{stage} 未生成有效输出"
|
||
error_msg = business_error or f"{stage} 执行失败：{detail_msg}"
|
||
print(f"[Celery] 单阶段产物校验失败: {stage_status.to_dict()}")
|
||
|
||
return {
|
||
'status': 'failed',
|
||
'success': False,
|
||
'current_stage': stage,
|
||
'progress': 0,
|
||
'message': error_msg,
|
||
'errors': [error_msg],
|
||
'stage_status': stage_status.to_dict(),
|
||
'result': (result or {}).get("final_output", {}) if isinstance(result, dict) else {},
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': error_msg}
|
||
]
|
||
}
|
||
|
||
|
||
def _diagnose_single_stage_failure(
|
||
job_id: str,
|
||
stage: str,
|
||
prefix: str = "",
|
||
*,
|
||
prepare_patch_workspace: bool = False,
|
||
) -> dict:
|
||
"""DiagnosisAgent is disabled; keep failures as plain execution failures."""
|
||
return {
|
||
"status": "diagnosis_disabled",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"issues": [],
|
||
}
|
||
try:
|
||
from gfa_core.mas.agents.diagnosis import DiagnosisAgent
|
||
|
||
diagnosis = DiagnosisAgent(None).diagnose_job_stage(
|
||
job_id=job_id,
|
||
stage=stage,
|
||
prefix=prefix or "",
|
||
project_root=PROJECT_ROOT,
|
||
prepare_patch_workspace=prepare_patch_workspace,
|
||
)
|
||
return diagnosis if isinstance(diagnosis, dict) else {}
|
||
except Exception as exc:
|
||
print(f"[Celery] DiagnosisAgent failed for {job_id}/{stage}: {exc}")
|
||
return {
|
||
"status": "diagnosis_failed",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"error": str(exc),
|
||
"policy": {
|
||
"auto_modify_input": False,
|
||
"auto_modify_global_code": False,
|
||
"auto_rerun_stage": False,
|
||
},
|
||
}
|
||
|
||
|
||
def _diagnose_stage_gate(
|
||
job_id: str,
|
||
stage: str,
|
||
prefix: str = "",
|
||
*,
|
||
gate: str,
|
||
params: dict | None = None,
|
||
) -> dict:
|
||
return {
|
||
"status": "diagnosis_disabled",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"gate": gate,
|
||
"issues": [],
|
||
}
|
||
try:
|
||
from gfa_core.mas.agents.diagnosis import DiagnosisAgent
|
||
|
||
diagnosis_prefix = "genome_prepare" if stage in {"genome_prepare", "genome_prepare_v2"} else (prefix or "")
|
||
return DiagnosisAgent(None).diagnose_stage_gate(
|
||
job_id=job_id,
|
||
stage=stage,
|
||
prefix=diagnosis_prefix,
|
||
gate=gate,
|
||
project_root=PROJECT_ROOT,
|
||
params=params or {},
|
||
)
|
||
except Exception as exc:
|
||
print(f"[Celery] Diagnosis gate failed for {job_id}/{stage}/{gate}: {exc}")
|
||
return {
|
||
"status": "diagnosis_failed",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"gate": gate,
|
||
"error": str(exc),
|
||
"issues": [],
|
||
}
|
||
|
||
|
||
def _apply_runtime_repairs(
|
||
job_id: str,
|
||
stage: str,
|
||
prefix: str = "",
|
||
diagnosis: dict | None = None,
|
||
*,
|
||
execute_heavy: bool = False,
|
||
) -> dict:
|
||
return {
|
||
"success": False,
|
||
"status": "diagnosis_disabled",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"results": [],
|
||
}
|
||
try:
|
||
from gfa_core.mas.agents.diagnosis import DiagnosisAgent
|
||
|
||
return DiagnosisAgent(None).apply_runtime_repairs(
|
||
job_id=job_id,
|
||
stage=stage,
|
||
prefix=prefix or "",
|
||
diagnosis=diagnosis,
|
||
project_root=PROJECT_ROOT,
|
||
execute_heavy=execute_heavy,
|
||
)
|
||
except Exception as exc:
|
||
print(f"[Celery] Runtime repair failed for {job_id}/{stage}: {exc}")
|
||
return {
|
||
"success": False,
|
||
"status": "repair_failed",
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"error": str(exc),
|
||
"results": [],
|
||
}
|
||
|
||
|
||
def _attach_diagnosis_to_failure(response: dict, job_id: str, stage: str, prefix: str = "") -> dict:
|
||
"""Keep failed task responses free of DiagnosisAgent side effects."""
|
||
if not isinstance(response, dict) or response.get("success") is not False:
|
||
return response
|
||
response.pop("pending_confirmation", None)
|
||
response.pop("diagnosis", None)
|
||
return response
|
||
|
||
diagnosis = response.get("diagnosis")
|
||
if not isinstance(diagnosis, dict) or not diagnosis:
|
||
diagnosis = _diagnose_single_stage_failure(job_id=job_id, stage=stage, prefix=prefix)
|
||
response["diagnosis"] = diagnosis
|
||
|
||
session_id = str(response.get("session_id") or job_id or "").strip()
|
||
if diagnosis.get("requires_user_confirmation"):
|
||
response["pending_confirmation"] = {
|
||
"status": "waiting",
|
||
"type": "diagnosis_review",
|
||
"session_id": session_id,
|
||
"stage": stage,
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"diagnosis": diagnosis,
|
||
"message": "检测到阶段失败，已生成诊断。请用户确认后再修订输入、执行副本修复或升级全局代码。",
|
||
"allowed_actions": [
|
||
"review_only",
|
||
"prepare_job_patch",
|
||
"apply_confirmed_global_fix_after_backup",
|
||
],
|
||
}
|
||
|
||
return response
|
||
|
||
|
||
@app.task(bind=True, max_retries=3)
|
||
def execute_agent_task(self, task_id: str, message: str, session_id: str, job_id: str = "", history: list = None, prefix: str = ""):
|
||
"""
|
||
执行Agent分析任务
|
||
"""
|
||
try:
|
||
print(f"[Celery] Starting task {task_id}, job_id={job_id}")
|
||
|
||
# 更新任务状态为运行中（使用self.request.id获取Celery任务ID）
|
||
celery_task_id = self.request.id if self.request else task_id
|
||
self.update_state(
|
||
task_id=celery_task_id,
|
||
state='RUNNING',
|
||
meta={
|
||
'current_stage': 'intent_parsing',
|
||
'progress': 0,
|
||
'message': message,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'info', 'message': '开始分析任务...'},
|
||
{'time': datetime.now().isoformat(), 'level': 'info', 'message': f'解析用户意图: {message}'}
|
||
]
|
||
}
|
||
)
|
||
|
||
# 获取Orchestrator
|
||
orchestrator = get_orchestrator()
|
||
|
||
# 查询已完成的阶段
|
||
completed_stages = []
|
||
if job_id:
|
||
from gfa_core.mas.tools.stage_status import list_completed_stages
|
||
|
||
completed_stages = list_completed_stages(job_id=job_id, prefix=prefix or "")
|
||
print(f"[tasks.py] Completed stages: {completed_stages}")
|
||
|
||
# 构建初始状态
|
||
from gfa_core.mas.core.state import GraphState
|
||
initial_state = {
|
||
"query": message,
|
||
"session_id": session_id,
|
||
"job_id": job_id, # 传入 job_id 用于 auto_continue
|
||
"files": [],
|
||
"current_agent": "router",
|
||
"status": "running",
|
||
"errors": [],
|
||
"final_output": {},
|
||
"history": history or [],
|
||
"messages": [], # 添加 messages 字段
|
||
"completed_stages": completed_stages, # 添加已完成阶段
|
||
"retry_count": 0, # 添加重试计数
|
||
"max_retries": 3, # 添加最大重试次数
|
||
"user_input": {"prefix": prefix} # 传递 prefix 参数
|
||
}
|
||
|
||
print(f"[tasks.py] initial_state keys: {list(initial_state.keys())}")
|
||
print(f"[tasks.py] initial_state['completed_stages']: {initial_state.get('completed_stages', 'NOT FOUND')}")
|
||
|
||
# 运行Orchestrator
|
||
import asyncio
|
||
result = asyncio.run(orchestrator.run(initial_state))
|
||
|
||
# 处理结果
|
||
final_output = result.get("final_output", {})
|
||
|
||
if result.get("status") == "completed":
|
||
# ============ 生成累积解读报告（新增） ============
|
||
try:
|
||
import asyncio
|
||
from gfa_core.mas.agents.result_parser import create_result_parser_agent
|
||
from gfa_core.mas.llm import LLMManager
|
||
|
||
# 从 result 中提取 prefix（如果传入的 prefix 为空）
|
||
if not prefix:
|
||
prefix = result.get("final_output", {}).get("user_input", {}).get("prefix", "")
|
||
if not prefix:
|
||
print("[Celery] 未解析到输出标签，跳过累积解读报告生成")
|
||
raise RuntimeError("missing_prefix_for_interpretation")
|
||
|
||
# 获取已完成的阶段列表（统一 manifest/artifacts 校验）
|
||
from gfa_core.mas.tools.stage_status import list_completed_stages
|
||
|
||
completed_stages = list_completed_stages(job_id=job_id, prefix=prefix)
|
||
|
||
# 按顺序排序
|
||
stage_order = ["identify", "phylogeny", "structure", "motif", "domain", "promoter", "synteny", "interspecies_synteny", "karyotype", "expression", "kaks", "ppi"]
|
||
completed_stages = [s for s in stage_order if s in completed_stages] + [s for s in completed_stages if s not in stage_order]
|
||
|
||
print(f"[Celery] 生成累积解读，已完成阶段: {completed_stages}")
|
||
|
||
# 创建 ResultParserAgent（不需要 LLMManager）
|
||
agent = create_result_parser_agent(None)
|
||
|
||
# 生成累积解读报告
|
||
report = asyncio.run(agent.generate_cumulative_report(
|
||
job_id=job_id,
|
||
prefix=prefix,
|
||
completed_stages=completed_stages
|
||
))
|
||
|
||
# 保存解读报告
|
||
save_success = agent.save_interpretation(job_id, prefix, report)
|
||
|
||
if save_success:
|
||
print(f"[Celery] 解读报告已保存到 interpretation.json")
|
||
else:
|
||
print(f"[Celery] 保存解读报告失败")
|
||
|
||
except Exception as e:
|
||
# 解读失败不阻塞主流程
|
||
print(f"[Celery] 生成解读报告失败（非阻塞）: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
return {
|
||
'status': 'completed',
|
||
'current_stage': result.get("current_agent", "unknown"),
|
||
'progress': 100,
|
||
'message': message,
|
||
'result': {
|
||
'response': final_output.get("response", ""),
|
||
'analysis_plan': final_output.get("analysis_plan", {}),
|
||
'execution_result': final_output.get("execution_result", {})
|
||
},
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'success', 'message': '分析任务完成'}
|
||
]
|
||
}
|
||
else:
|
||
return {
|
||
'status': 'failed',
|
||
'current_stage': result.get("current_agent", "unknown"),
|
||
'progress': 0,
|
||
'message': message,
|
||
'result': {},
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': f"分析失败: {result.get('errors', ['Unknown error'])}"}
|
||
]
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
error_msg = f"任务异常: {str(e)}\n{traceback.format_exc()}"
|
||
print(f"[Celery] Task {task_id} failed: {error_msg}")
|
||
celery_task_id = self.request.id if self.request else task_id
|
||
self.update_state(
|
||
task_id=celery_task_id,
|
||
state='FAILURE',
|
||
meta={
|
||
'exc_type': type(e).__name__,
|
||
'exc_message': str(e),
|
||
}
|
||
)
|
||
raise
|
||
|
||
|
||
@app.task(bind=True, max_retries=3)
|
||
def execute_single_stage_task(self, task_id: str, stage: str, job_id: str, prefix: str,
|
||
params: dict = None, session_id: str = None):
|
||
"""
|
||
执行单个分析阶段任务（用于/api/execute统一架构）
|
||
|
||
参数:
|
||
- task_id: 任务ID
|
||
- stage: 分析阶段（如identify, phylogeny等）
|
||
- job_id: 任务目录ID
|
||
- prefix: 分析前缀（例如 At.WRKY；基因家族流程通常由 config/scope 自动生成）
|
||
- params: 执行参数
|
||
- session_id: 会话ID
|
||
"""
|
||
resource_acquired = False
|
||
resource_outcome = "failed"
|
||
resource_lease = None
|
||
try:
|
||
print(f"[Celery] Starting single stage task {task_id}, stage={stage}, job_id={job_id}")
|
||
params, resource_profile = normalize_stage_params(stage, params)
|
||
print(
|
||
"[Celery] Resource profile "
|
||
f"stage={stage}, queue={resource_profile.get('queue')}, "
|
||
f"class={resource_profile.get('resource_class')}, "
|
||
f"requested_cpu={params.get('requested_cpu')}, effective_cpu={params.get('cpu')}"
|
||
)
|
||
|
||
celery_task_id = self.request.id if self.request else task_id
|
||
|
||
try:
|
||
from gfa_core.mas.resource_governor import ResourceLease
|
||
from gfa_core.mas.stage_catalog import get_stage_queue
|
||
|
||
resource_queue = get_stage_queue(stage)
|
||
resource_lease = ResourceLease(
|
||
task_id=task_id,
|
||
stage=stage,
|
||
job_id=job_id,
|
||
prefix=prefix,
|
||
cpu=int(params.get("cpu") or 1),
|
||
queue=resource_queue,
|
||
resource_class=str(resource_profile.get("resource_class") or ""),
|
||
concurrency_group=str(resource_profile.get("concurrency_group") or ""),
|
||
celery_task_id=celery_task_id,
|
||
)
|
||
|
||
def _update_waiting_resource(wait_result: dict):
|
||
usage = wait_result.get("usage", {}) if isinstance(wait_result, dict) else {}
|
||
reason = wait_result.get("reason", "等待资源可用") if isinstance(wait_result, dict) else "等待资源可用"
|
||
self.update_state(
|
||
task_id=celery_task_id,
|
||
state="WAITING_RESOURCE",
|
||
meta={
|
||
"status": "waiting_resource",
|
||
"current_stage": stage,
|
||
"progress": 0,
|
||
"message": reason,
|
||
"resource_usage": usage,
|
||
"resource_profile": resource_profile,
|
||
"logs": [
|
||
{
|
||
"time": datetime.now().isoformat(),
|
||
"level": "info",
|
||
"message": reason,
|
||
}
|
||
],
|
||
},
|
||
)
|
||
|
||
acquire_result = resource_lease.wait(update_state=_update_waiting_resource)
|
||
resource_acquired = True
|
||
print(f"[Celery] Resource lease granted for {task_id}: {acquire_result.get('usage')}")
|
||
except Exception as exc:
|
||
print(f"[Celery] Resource governor unavailable, continuing without lease: {exc}")
|
||
|
||
# 更新任务状态
|
||
self.update_state(
|
||
task_id=celery_task_id,
|
||
state='RUNNING',
|
||
meta={
|
||
'current_stage': stage,
|
||
'progress': 10,
|
||
'message': (
|
||
f"开始执行 {stage}（请求CPU {params.get('requested_cpu') or params.get('cpu')}，"
|
||
f"调度CPU {params.get('cpu')}，{resource_profile.get('resource_class')}队列）..."
|
||
),
|
||
'logs': [
|
||
{
|
||
'time': datetime.now().isoformat(),
|
||
'level': 'info',
|
||
'message': (
|
||
f"开始执行 {stage}; requested_cpu={params.get('requested_cpu')}, "
|
||
f"effective_cpu={params.get('cpu')}, queue_class={resource_profile.get('resource_class')}"
|
||
)
|
||
}
|
||
]
|
||
}
|
||
)
|
||
|
||
# /api/execute 是“用户已明确选择阶段”的入口。这里直接调用
|
||
# StageToolRegistry，避免再走自然语言规划/PreFlight，把
|
||
# rnaseq_enrich 误展开成 rnaseq_qc 并错误要求 FASTQ。
|
||
try:
|
||
import asyncio
|
||
from gfa_core.mas.tools.stage_tools import stage_registry
|
||
|
||
prefix = resolve_effective_prefix(job_id, prefix, stage)
|
||
if params is not None:
|
||
params["prefix"] = prefix
|
||
params = attach_semantic_execution_context(job_id, stage, prefix, params)
|
||
print(f"[Celery] Effective prefix for {stage}: {prefix or '<empty>'}")
|
||
if params.get("resolvedArtifacts"):
|
||
print(f"[Celery] Resolved artifacts for {stage}: {params.get('resolvedArtifacts')}")
|
||
qa_preflight = _diagnose_stage_gate(job_id, stage, prefix, gate="preflight", params=params)
|
||
qa_repair = {}
|
||
if qa_preflight.get("runtime_repair_candidates"):
|
||
execute_heavy = bool((params or {}).get("diagnosisHeavyRepair", True))
|
||
qa_repair = _apply_runtime_repairs(
|
||
job_id,
|
||
stage,
|
||
prefix,
|
||
qa_preflight,
|
||
execute_heavy=execute_heavy,
|
||
)
|
||
qa_preflight = _diagnose_stage_gate(job_id, stage, prefix, gate="preflight", params=params)
|
||
blocking_preflight = [
|
||
issue for issue in qa_preflight.get("issues", [])
|
||
if issue.get("severity") == "blocking" or issue.get("blocks_downstream")
|
||
]
|
||
if blocking_preflight:
|
||
error_msg = "❌ 诊断门控阻断执行：发现输入/上游产物存在会导致阶段失败的问题"
|
||
return _attach_diagnosis_to_failure({
|
||
'status': 'failed',
|
||
'success': False,
|
||
'current_stage': stage,
|
||
'progress': 0,
|
||
'session_id': session_id or job_id,
|
||
'message': error_msg,
|
||
'errors': [error_msg],
|
||
'diagnosis': qa_preflight,
|
||
'qa_preflight': qa_preflight,
|
||
'qa_runtime_repair': qa_repair,
|
||
'semantic_preflight': params.get("semanticPreflight"),
|
||
'resolved_artifacts': params.get("resolvedArtifacts", {}),
|
||
'resource_profile': resource_profile,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': error_msg}
|
||
]
|
||
}, job_id, stage, prefix)
|
||
direct_result = asyncio.run(stage_registry.execute(
|
||
stage_id=stage,
|
||
job_id=job_id,
|
||
prefix=prefix,
|
||
extra_params=params or {},
|
||
))
|
||
print(f"[Celery] Direct single-stage result: {direct_result}")
|
||
|
||
success = direct_result.get("status") == "success"
|
||
if success:
|
||
result_manifest = direct_result.get("manifest") if isinstance(direct_result.get("manifest"), dict) else {}
|
||
result_stage = _nonempty_prefix(result_manifest.get("stage")) or stage
|
||
result_prefix = _nonempty_prefix(result_manifest.get("prefix")) or prefix
|
||
artifact_failure = validate_single_stage_artifacts(job_id, result_prefix, result_stage, {
|
||
"final_output": {
|
||
"message": direct_result.get("message", ""),
|
||
"manifest": result_manifest,
|
||
"manifest_path": direct_result.get("manifest_path"),
|
||
}
|
||
})
|
||
if artifact_failure:
|
||
artifact_failure["session_id"] = session_id or job_id
|
||
return _attach_diagnosis_to_failure(artifact_failure, job_id, result_stage, result_prefix)
|
||
postflight_params = _merge_manifest_identity_params(params, result_manifest)
|
||
qa_postflight = _diagnose_stage_gate(job_id, result_stage, result_prefix, gate="postflight", params=postflight_params)
|
||
qa_post_repair = {}
|
||
if qa_postflight.get("runtime_repair_candidates"):
|
||
qa_post_repair = _apply_runtime_repairs(
|
||
job_id,
|
||
result_stage,
|
||
result_prefix,
|
||
qa_postflight,
|
||
execute_heavy=bool((params or {}).get("diagnosisHeavyRepair", True)),
|
||
)
|
||
qa_postflight = _diagnose_stage_gate(job_id, result_stage, result_prefix, gate="postflight", params=postflight_params)
|
||
blocking_postflight = [
|
||
issue for issue in qa_postflight.get("issues", [])
|
||
if issue.get("severity") == "blocking" or issue.get("blocks_downstream")
|
||
]
|
||
if blocking_postflight:
|
||
error_msg = "❌ 阶段已运行，但诊断 postflight 发现关键产物不可安全下游使用"
|
||
return _attach_diagnosis_to_failure({
|
||
'status': 'failed',
|
||
'success': False,
|
||
'current_stage': result_stage,
|
||
'progress': 100,
|
||
'session_id': session_id or job_id,
|
||
'message': error_msg,
|
||
'errors': [error_msg],
|
||
'result': direct_result,
|
||
'diagnosis': qa_postflight,
|
||
'qa_preflight': qa_preflight,
|
||
'qa_runtime_repair': qa_repair,
|
||
'qa_postflight': qa_postflight,
|
||
'qa_postflight_repair': qa_post_repair,
|
||
'semantic_preflight': params.get("semanticPreflight"),
|
||
'resolved_artifacts': params.get("resolvedArtifacts", {}),
|
||
'resource_profile': resource_profile,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': error_msg}
|
||
]
|
||
}, job_id, result_stage, result_prefix)
|
||
resource_outcome = "completed"
|
||
return {
|
||
'status': 'completed',
|
||
'success': True,
|
||
'current_stage': result_stage,
|
||
'progress': 100,
|
||
'message': f'{result_stage} 执行完成',
|
||
'result': direct_result,
|
||
'qa_preflight': qa_preflight,
|
||
'qa_runtime_repair': qa_repair,
|
||
'qa_postflight': qa_postflight,
|
||
'qa_postflight_repair': qa_post_repair,
|
||
'semantic_preflight': params.get("semanticPreflight"),
|
||
'resolved_artifacts': params.get("resolvedArtifacts", {}),
|
||
'resource_profile': resource_profile,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'success', 'message': direct_result.get("message", f'{stage} 执行成功')}
|
||
]
|
||
}
|
||
|
||
error_msg = direct_result.get("error") or direct_result.get("message") or f"{stage} 执行失败"
|
||
qa_failure = _diagnose_stage_gate(job_id, stage, prefix, gate="failure", params=params)
|
||
return _attach_diagnosis_to_failure({
|
||
'status': 'failed',
|
||
'success': False,
|
||
'current_stage': stage,
|
||
'progress': 0,
|
||
'session_id': session_id or job_id,
|
||
'message': error_msg,
|
||
'errors': [error_msg],
|
||
'result': direct_result,
|
||
'diagnosis': qa_failure,
|
||
'qa_preflight': qa_preflight,
|
||
'qa_runtime_repair': qa_repair,
|
||
'qa_failure': qa_failure,
|
||
'semantic_preflight': params.get("semanticPreflight"),
|
||
'resolved_artifacts': params.get("resolvedArtifacts", {}),
|
||
'resource_profile': resource_profile,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': error_msg}
|
||
]
|
||
}, job_id, stage, prefix)
|
||
except Exception as direct_exc:
|
||
print(f"[Celery] Direct single-stage execution failed, falling back to orchestrator: {direct_exc}")
|
||
|
||
# 获取Orchestrator
|
||
orchestrator = get_orchestrator()
|
||
|
||
# 构建初始状态（单stage模式）
|
||
from gfa_core.mas.core.langgraph_orchestrator import GraphState
|
||
initial_state: GraphState = {
|
||
"query": f"执行{stage}分析",
|
||
"session_id": session_id or task_id,
|
||
"job_id": job_id,
|
||
"files": [],
|
||
"current_agent": "router",
|
||
"status": "running",
|
||
"errors": [],
|
||
"final_output": {},
|
||
"messages": [],
|
||
"retry_count": 0,
|
||
"max_retries": 3,
|
||
"user_input": {
|
||
"prefix": prefix,
|
||
"stage": stage, # 指定要执行的stage
|
||
"params": params or {}
|
||
},
|
||
"single_stage_mode": True, # 标记为单stage模式
|
||
"target_stage": stage # 目标stage
|
||
}
|
||
|
||
print(f"[tasks.py] 单stage模式: target={stage}, prefix={prefix}")
|
||
|
||
# 运行LangGraph
|
||
import asyncio
|
||
result = asyncio.run(orchestrator.run(initial_state))
|
||
|
||
# 处理结果
|
||
if result.get("status") in ["completed", "success"]:
|
||
# 单阶段任务的最终成功标准必须是真实产物，而不是仅依赖
|
||
# LangGraph/Celery 的运行状态。
|
||
artifact_failure = validate_single_stage_artifacts(job_id, prefix, stage, result)
|
||
if artifact_failure:
|
||
artifact_failure["session_id"] = session_id or job_id
|
||
return _attach_diagnosis_to_failure(artifact_failure, job_id, stage, prefix)
|
||
|
||
# ============ 生成累积解读报告（新增） ============
|
||
try:
|
||
import asyncio
|
||
from gfa_core.mas.agents.result_parser import create_result_parser_agent
|
||
|
||
# 获取已完成的阶段列表（统一 manifest/artifacts 校验）
|
||
from gfa_core.mas.tools.stage_status import list_completed_stages
|
||
|
||
completed_stages = list_completed_stages(job_id=job_id, prefix=prefix)
|
||
print(f"[Celery DEBUG] Unified completed_stages: {completed_stages}")
|
||
|
||
# 按顺序排序
|
||
stage_order = ["identify", "phylogeny", "structure", "motif", "domain", "promoter", "synteny", "interspecies_synteny", "karyotype", "expression", "kaks", "ppi"]
|
||
completed_stages = [s for s in stage_order if s in completed_stages] + [s for s in completed_stages if s not in stage_order]
|
||
|
||
print(f"[Celery] 生成累积解读，已完成阶段: {completed_stages}")
|
||
|
||
# 创建 ResultParserAgent（不需要 LLMManager）
|
||
agent = create_result_parser_agent(None)
|
||
|
||
# 生成累积解读报告
|
||
report = asyncio.run(agent.generate_cumulative_report(
|
||
job_id=job_id,
|
||
prefix=prefix,
|
||
completed_stages=completed_stages
|
||
))
|
||
|
||
# 保存解读报告
|
||
save_success = agent.save_interpretation(job_id, prefix, report)
|
||
|
||
if save_success:
|
||
print(f"[Celery] 解读报告已保存到 interpretation.json")
|
||
else:
|
||
print(f"[Celery] 保存解读报告失败")
|
||
|
||
except Exception as e:
|
||
# 解读失败不阻塞主流程
|
||
print(f"[Celery] 生成解读报告失败（非阻塞）: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
resource_outcome = "completed"
|
||
return {
|
||
'status': 'completed',
|
||
'success': True, # 添加 success 字段，供前端判断
|
||
'current_stage': stage,
|
||
'progress': 100,
|
||
'message': f'{stage} 执行完成',
|
||
'result': result.get("final_output", {}),
|
||
'resource_profile': resource_profile,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'success', 'message': f'{stage} 执行成功'}
|
||
]
|
||
}
|
||
else:
|
||
# 检查是否有等待确认的状态（优先检查 result.status）
|
||
if result.get("status") == "waiting_confirmation":
|
||
pending_confirmation = result.get("pending_confirmation", {})
|
||
# 将确认状态存储到Celery任务meta中
|
||
return {
|
||
'status': 'waiting_confirmation',
|
||
'current_stage': stage,
|
||
'progress': 50,
|
||
'message': '等待用户确认修复方案',
|
||
'pending_confirmation': pending_confirmation,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'warning', 'message': '检测到错误，等待用户确认修复方案'}
|
||
]
|
||
}
|
||
|
||
# 旧逻辑：检查 pending_confirmation.status == "waiting"（兼容）
|
||
pending_confirmation = result.get("pending_confirmation")
|
||
if pending_confirmation and pending_confirmation.get("status") == "waiting":
|
||
# 将确认状态存储到Celery任务meta中
|
||
return {
|
||
'status': 'waiting_confirmation',
|
||
'current_stage': stage,
|
||
'progress': 50,
|
||
'message': '等待用户确认修复方案',
|
||
'pending_confirmation': pending_confirmation,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'warning', 'message': '检测到错误，等待用户确认修复方案'}
|
||
]
|
||
}
|
||
|
||
# 构建详细错误消息：优先保留 run_stage/stage_tools 返回的业务错误，
|
||
# 避免被后续 manifest 校验错误覆盖成 "Manifest not found"。
|
||
final_output = result.get("final_output", {}) or {}
|
||
execution_results = result.get("execution_results", []) or []
|
||
business_error = (
|
||
final_output.get("response")
|
||
or final_output.get("message")
|
||
or final_output.get("error")
|
||
or result.get("execution_error")
|
||
)
|
||
if not business_error:
|
||
for item in execution_results:
|
||
stage_result = item.get("result", {}) if isinstance(item, dict) else {}
|
||
candidate = stage_result.get("error") or stage_result.get("message")
|
||
if candidate:
|
||
business_error = candidate
|
||
break
|
||
|
||
if isinstance(business_error, str) and (
|
||
"❌" in business_error
|
||
or "failed" in business_error.lower()
|
||
or "无法" in business_error
|
||
or "缺少" in business_error
|
||
):
|
||
error_msg = business_error
|
||
else:
|
||
error_msg = f'{stage} 执行失败'
|
||
diagnosis = result.get("diagnosis_result", {})
|
||
if diagnosis:
|
||
suggestion = diagnosis.get("suggestion", "")
|
||
error_type = diagnosis.get("error_type", "")
|
||
if suggestion:
|
||
error_msg += f"\n\n诊断: {error_type}\n建议: {suggestion}"
|
||
|
||
return _attach_diagnosis_to_failure({
|
||
'status': 'failed',
|
||
'success': False, # 添加 success 字段，供前端判断
|
||
'current_stage': stage,
|
||
'progress': 0,
|
||
'session_id': session_id or job_id,
|
||
'message': error_msg,
|
||
'errors': result.get("errors", ["Unknown error"]),
|
||
'diagnosis': diagnosis,
|
||
'logs': [
|
||
{'time': datetime.now().isoformat(), 'level': 'error', 'message': error_msg}
|
||
]
|
||
}, job_id, stage, prefix)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
error_msg = f"任务异常: {str(e)}\n{traceback.format_exc()}"
|
||
print(f"[Celery] Single stage task {task_id} failed: {error_msg}")
|
||
celery_task_id = self.request.id if self.request else task_id
|
||
self.update_state(
|
||
task_id=celery_task_id,
|
||
state='FAILURE',
|
||
meta={
|
||
'exc_type': type(e).__name__,
|
||
'exc_message': str(e),
|
||
}
|
||
)
|
||
raise
|
||
finally:
|
||
if resource_acquired and resource_lease is not None:
|
||
try:
|
||
resource_lease.release(status=resource_outcome, message=f"{stage} {resource_outcome}")
|
||
except Exception as release_exc:
|
||
print(f"[Celery] Failed to release resource lease for {task_id}: {release_exc}")
|