3765 lines
150 KiB
Python
Executable File
3765 lines
150 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
BotanicalBaseAgent - FastAPI Backend
|
||
支持Vue3前端的API接口 - Celery任务队列模式
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import asyncio
|
||
import uuid
|
||
import time
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from typing import Optional, List, Dict
|
||
|
||
from fastapi import FastAPI, UploadFile, File, HTTPException, Request, BackgroundTasks, Query
|
||
from fastapi.responses import JSONResponse, FileResponse, Response
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
# 添加项目路径
|
||
PROJECT_ROOT = Path(__file__).parent.parent.parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
# 全局状态存储（用于等待用户确认）
|
||
pending_confirmations: Dict[str, Dict] = {}
|
||
# 会话状态存储
|
||
session_states: Dict[str, Dict] = {}
|
||
|
||
# 导入Celery任务
|
||
try:
|
||
from tasks import execute_agent_task, execute_single_stage_task
|
||
CELERY_AVAILABLE = True
|
||
print("✅ Celery任务导入成功")
|
||
except ImportError as e:
|
||
CELERY_AVAILABLE = False
|
||
print(f"⚠️ Celery任务导入失败: {e}")
|
||
|
||
# 导入多Agent系统组件
|
||
try:
|
||
from gfa_core.mas.core.orchestrator import Orchestrator
|
||
from gfa_core.mas.core.langgraph_orchestrator import LangGraphOrchestrator, GraphState
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.memory import MemoryManager
|
||
from gfa_core.mas.api.conversation_routes import router as conv_router # 会话管理路由
|
||
MAS_AVAILABLE = True
|
||
LANGGRAPH_AVAILABLE = True
|
||
print("✅ MAS组件导入成功")
|
||
print("✅ LangGraph编排器导入成功")
|
||
except ImportError as e:
|
||
MAS_AVAILABLE = False
|
||
LANGGRAPH_AVAILABLE = False
|
||
print(f"⚠️ MAS组件导入失败: {e}")
|
||
Orchestrator = None
|
||
LangGraphOrchestrator = None
|
||
LLMManager = None
|
||
MemoryManager = None
|
||
|
||
WORKSPACE_ROOT = PROJECT_ROOT / "workspace"
|
||
INPUT_ROOT = PROJECT_ROOT / "input"
|
||
|
||
|
||
def _nonempty_prefix(value: object) -> str:
|
||
text = str(value or "").strip()
|
||
return text if text and text.lower() not in {"none", "null", "undefined"} else ""
|
||
|
||
|
||
def _gene_family_from_config(config_path: Path) -> Optional[str]:
|
||
if not config_path.exists():
|
||
return None
|
||
try:
|
||
import pandas as pd
|
||
df = pd.read_excel(config_path)
|
||
if "Gene_Family" not in df.columns:
|
||
return None
|
||
for value in df["Gene_Family"].dropna().astype(str):
|
||
value = value.strip()
|
||
if value:
|
||
return value
|
||
except Exception as exc:
|
||
print(f"[WARN] Failed to infer gene family from {config_path}: {exc}")
|
||
return None
|
||
|
||
|
||
def _prefix_from_manifest(path: Path) -> str:
|
||
if not path.exists():
|
||
return ""
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as handle:
|
||
data = json.load(handle)
|
||
for key in ("prefix", "effective_prefix"):
|
||
value = _nonempty_prefix(data.get(key))
|
||
if value:
|
||
return value
|
||
target = data.get("target") if isinstance(data.get("target"), dict) else None
|
||
if target:
|
||
value = _nonempty_prefix(target.get("prefix"))
|
||
if value:
|
||
return value
|
||
artifacts = data.get("artifacts") if isinstance(data.get("artifacts"), dict) else {}
|
||
for value in artifacts.values():
|
||
if isinstance(value, str):
|
||
name = Path(value).name
|
||
if name.startswith("Multi.") and ".genome_prepare.manifest" in name:
|
||
return name.split(".genome_prepare.manifest", 1)[0]
|
||
except Exception as exc:
|
||
print(f"[WARN] Failed to read prefix manifest {path}: {exc}")
|
||
return ""
|
||
|
||
|
||
def _prefix_from_final_members(final_dir: Path) -> str:
|
||
if not final_dir.exists():
|
||
return ""
|
||
for file in sorted(final_dir.glob("Multi.*.final_members.pep.fa")):
|
||
name = file.name
|
||
if name.endswith(".final_members.pep.fa"):
|
||
return name[:-len(".final_members.pep.fa")]
|
||
return ""
|
||
|
||
|
||
def _prefix_from_ref_combined(job_dir: Path) -> str:
|
||
for base in (job_dir, job_dir / "final"):
|
||
if not base.exists():
|
||
continue
|
||
for file in sorted(base.glob("Ref_combined.*.pep.fasta")):
|
||
parts = file.name.split(".")
|
||
if len(parts) >= 3:
|
||
return f"Multi.{parts[1]}"
|
||
return ""
|
||
|
||
|
||
def resolve_effective_prefix(job_id: str, params: Dict, stage: str) -> str:
|
||
provided = _nonempty_prefix(params.get("prefix"))
|
||
if provided:
|
||
return provided
|
||
|
||
job_dir = INPUT_ROOT / job_id
|
||
workspace_dir = WORKSPACE_ROOT / job_id
|
||
|
||
final_dir = workspace_dir / "final"
|
||
|
||
species_config = job_dir / "genome_prepare" / "species_config.json"
|
||
if species_config.exists():
|
||
try:
|
||
with open(species_config, "r", encoding="utf-8") as handle:
|
||
data = json.load(handle)
|
||
target = data.get("target") if isinstance(data.get("target"), dict) else None
|
||
if target:
|
||
value = _nonempty_prefix(target.get("prefix"))
|
||
if value:
|
||
return value
|
||
value = _nonempty_prefix(data.get("prefix"))
|
||
if value:
|
||
return value
|
||
except Exception as exc:
|
||
print(f"[WARN] Failed to read species_config prefix: {exc}")
|
||
|
||
manifests = []
|
||
manifests.extend(sorted(final_dir.glob("Multi.*.genome_prepare.manifest.json")))
|
||
manifests.extend(sorted((job_dir / "final").glob("Multi.*.genome_prepare.manifest.json")))
|
||
manifests.extend(sorted(final_dir.glob("*.genome_prepare.manifest.json")))
|
||
manifests.extend(sorted((job_dir / "final").glob("*.genome_prepare.manifest.json")))
|
||
for manifest in manifests:
|
||
value = _prefix_from_manifest(manifest)
|
||
if value:
|
||
return value
|
||
|
||
family = _gene_family_from_config(job_dir / "config.xlsx")
|
||
if family:
|
||
return f"Multi.{family}"
|
||
|
||
value = _prefix_from_ref_combined(job_dir)
|
||
if value:
|
||
return value
|
||
|
||
value = _prefix_from_final_members(final_dir)
|
||
if value:
|
||
return value
|
||
|
||
return ""
|
||
|
||
|
||
def _attach_single_stage_status(task: Dict) -> None:
|
||
"""Derive task status from the unified manifest/artifact validator when possible."""
|
||
job_id = task.get("job_id")
|
||
stage = task.get("stage")
|
||
prefix = task.get("prefix", "")
|
||
if not job_id or not stage:
|
||
return
|
||
|
||
try:
|
||
from gfa_core.mas.tools.stage_status import get_stage_status as get_unified_stage_status
|
||
|
||
mapped_stage = {
|
||
"rnaseq_go_enrich": "rnaseq_enrich",
|
||
"rnaseq_kegg_enrich": "rnaseq_enrich",
|
||
}.get(stage, stage)
|
||
report = get_unified_stage_status(job_id=job_id, prefix=prefix, stage=mapped_stage)
|
||
if stage in {"rnaseq_go_enrich", "rnaseq_kegg_enrich"} and report.completed:
|
||
final_dir = WORKSPACE_ROOT / job_id / "final"
|
||
suffix = ".GO_enrich.txt" if stage == "rnaseq_go_enrich" else ".KEGG_enrich.txt"
|
||
if not any(final_dir.glob(f"*{suffix}")):
|
||
report.completed = False
|
||
report.status = "not_found"
|
||
report.errors.append(f"{stage} specific output not found: *{suffix}")
|
||
task["stage_status"] = report.to_dict()
|
||
|
||
if report.completed:
|
||
task["status"] = "success"
|
||
task["progress"] = 100
|
||
task["current_stage"] = stage
|
||
if not task.get("end_time"):
|
||
task["end_time"] = datetime.now().isoformat()
|
||
if not task.get("message") or task.get("message", "").startswith("任务 "):
|
||
task["message"] = f"✅ {stage} 分析已完成"
|
||
elif task.get("status") in {"success", "completed"}:
|
||
task["status"] = "failed"
|
||
task["message"] = f"❌ {stage} 未生成有效输出"
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Failed to attach unified stage status: {e}")
|
||
|
||
|
||
def _attach_annotation_progress(task: Dict) -> None:
|
||
"""Attach long-running annotation progress written by annotation_adapter."""
|
||
job_id = task.get("job_id")
|
||
stage = task.get("stage")
|
||
if not job_id or stage not in {"annotation", "rnaseq_enrich", "rnaseq_go_enrich", "rnaseq_kegg_enrich"}:
|
||
return
|
||
|
||
progress_path = WORKSPACE_ROOT / job_id / "rnaseq_annotation" / "progress.json"
|
||
if not progress_path.exists():
|
||
return
|
||
|
||
try:
|
||
progress = json.loads(progress_path.read_text(encoding="utf-8"))
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Failed to read annotation progress: {e}")
|
||
return
|
||
|
||
task["annotation_progress"] = progress
|
||
if task.get("status") in {"pending", "running"}:
|
||
task["status"] = "running"
|
||
task["progress"] = progress.get("progress", task.get("progress", 0))
|
||
task["current_stage"] = "annotation"
|
||
total = progress.get("total", 0)
|
||
processed = progress.get("processed", 0)
|
||
annotated = progress.get("annotated", 0)
|
||
failed = progress.get("failed", 0)
|
||
task["message"] = (
|
||
f"自动背景注释进行中：{processed}/{total}，"
|
||
f"已注释 {annotated}，失败 {failed}"
|
||
)
|
||
|
||
|
||
def _attach_agent_manifest_status(task: Dict) -> None:
|
||
"""Attach completed stages from manifests to an agent task."""
|
||
job_id = task.get("job_id")
|
||
prefix = task.get("prefix", "")
|
||
if not job_id:
|
||
return
|
||
|
||
try:
|
||
from gfa_core.mas.tools.stage_status import list_completed_stages
|
||
|
||
completed = list_completed_stages(job_id=job_id, prefix=prefix)
|
||
task["stages_completed"] = completed
|
||
task["completed_stages"] = completed
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Failed to attach agent manifest status: {e}")
|
||
|
||
app = FastAPI(title="BotanicalBaseAgent API")
|
||
|
||
# 挂载静态文件目录
|
||
WEB_DIR = Path(__file__).parent
|
||
STATIC_DIR = WEB_DIR / "static"
|
||
STATIC_DIR.mkdir(exist_ok=True)
|
||
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
||
|
||
# 挂载会话管理路由
|
||
if MAS_AVAILABLE and conv_router:
|
||
app.include_router(conv_router, prefix="/api")
|
||
print("✅ 会话管理路由已挂载")
|
||
|
||
# 添加CORS headers到所有响应
|
||
@app.middleware("http")
|
||
async def add_cors_headers(request, call_next):
|
||
if request.method == "OPTIONS":
|
||
return Response(
|
||
status_code=200,
|
||
headers={
|
||
"Access-Control-Allow-Origin": "*",
|
||
"Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
|
||
"Access-Control-Allow-Headers": "Content-Type, Authorization",
|
||
}
|
||
)
|
||
response = await call_next(request)
|
||
response.headers["Access-Control-Allow-Origin"] = "*"
|
||
return response
|
||
|
||
# 确保目录存在
|
||
WORKSPACE_ROOT.mkdir(exist_ok=True)
|
||
INPUT_ROOT.mkdir(exist_ok=True)
|
||
|
||
# 内存存储
|
||
jobs = {}
|
||
current_job_id = "default_job"
|
||
|
||
# 异步任务存储
|
||
tasks: Dict[str, dict] = {}
|
||
executor = ThreadPoolExecutor(max_workers=4)
|
||
|
||
# 初始化多Agent系统
|
||
orchestrator = None
|
||
langgraph_orchestrator = None
|
||
if MAS_AVAILABLE:
|
||
try:
|
||
# 清除Python模块缓存，确保配置重新加载
|
||
import sys
|
||
modules_to_remove = [k for k in sys.modules.keys() if 'gfa_core.mas' in k]
|
||
for m in modules_to_remove:
|
||
del sys.modules[m]
|
||
|
||
# 加载MAS配置
|
||
from gfa_core.mas.config import config as mas_config
|
||
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
|
||
print(f"[DEBUG] openai_api_key configured: {bool(llm_config.get('openai_api_key'))}")
|
||
print(f"[DEBUG] openai_base_url: {llm_config.get('openai_base_url', 'NOT SET')}")
|
||
|
||
# 强制重新创建LLMManager，不使用任何缓存
|
||
from gfa_core.mas.llm import LLMManager
|
||
llm_manager = LLMManager(config=llm_config)
|
||
# 清除provider缓存
|
||
llm_manager._providers = {}
|
||
llm_manager._available_providers = None
|
||
memory_manager = MemoryManager()
|
||
|
||
# 初始化原 Orchestrator（保留作为参考）
|
||
orchestrator = Orchestrator(
|
||
llm_manager=llm_manager,
|
||
memory_manager=memory_manager,
|
||
papers_dir=PROJECT_ROOT / "paper" / "gfpaper" / "buchong"
|
||
)
|
||
print("✅ Orchestrator初始化成功（保留参考）")
|
||
|
||
# 初始化 LangGraph Orchestrator（新版本）
|
||
if LANGGRAPH_AVAILABLE:
|
||
langgraph_orchestrator = LangGraphOrchestrator(llm_manager=llm_manager)
|
||
print("✅ LangGraph Orchestrator初始化成功")
|
||
except Exception as e:
|
||
print(f"⚠️ Orchestrator初始化失败: {e}")
|
||
import traceback
|
||
print(traceback.format_exc())
|
||
orchestrator = None
|
||
langgraph_orchestrator = None
|
||
|
||
# 需要异步轮询的阶段（耗时长，可能超过HTTP超时）
|
||
# 分析阶段耗时评估：
|
||
# - identify: HMMER+BLAST搜索，通常3-10分钟，必须异步
|
||
# - phylogeny: MAFFT建树+IQTree引导树，通常5-30分钟，必须异步
|
||
# - motif: MEME分析，通常5-20分钟，必须异步
|
||
# - domain: PfamScan搜索，通常3-10分钟，必须异步
|
||
# - structure: AlphaFold预测（如启用），可能数小时，必须异步
|
||
# - synteny: MCScanX共线性分析，通常5-15分钟，已包含
|
||
# - interspecies_synteny: 跨物种共线性分析，通常10-30分钟，已包含
|
||
# - karyotype: 染色体可视化，通常2-5分钟，已包含
|
||
# - pep: 多物种pep分析，通常5-15分钟，已包含
|
||
# - promoter: PlantCARE提交+等待，通常数小时，已包含
|
||
# - ppi: STRING蛋白互作网络，通常2-10分钟，加入异步
|
||
# - kaks: KaKs计算，通常2-10分钟，加入异步
|
||
# - expression: GEO数据下载+分析，通常10-30分钟，加入异步
|
||
# - rnaseq_*: RNA-seq全流程，每个阶段10-60分钟，已包含
|
||
ASYNC_STAGES = {
|
||
# 核心分析阶段（耗时长）
|
||
"identify", "phylogeny", "motif", "domain", "structure",
|
||
# 共线性分析
|
||
"synteny", "interspecies_synteny", "karyotype",
|
||
# 组合图
|
||
"combo",
|
||
# 其他分析
|
||
"pep", "ppi", "kaks", "expression", "msa", "ortholog",
|
||
# 启动子分析（外部服务）
|
||
"promoter",
|
||
# 基因组准备
|
||
"genome_prepare",
|
||
# RNA-seq全流程
|
||
"rnaseq_qc", "rnaseq_align", "rnaseq_quant", "rnaseq_de",
|
||
"annotation", "rnaseq_enrich", "rnaseq_go_enrich", "rnaseq_kegg_enrich"
|
||
}
|
||
|
||
|
||
|
||
def _config_bool(value: object) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
text = str(value or "").strip().lower()
|
||
return text in {"true", "1", "yes", "y", "ref"}
|
||
|
||
|
||
def _ordered_species_options(job_id: str) -> Dict[str, list]:
|
||
config_path = INPUT_ROOT / job_id / "config.xlsx"
|
||
targets = []
|
||
refs = []
|
||
if config_path.exists():
|
||
try:
|
||
import pandas as pd
|
||
df = pd.read_excel(config_path)
|
||
for _, row in df.iterrows():
|
||
species = str(row.get("Species_Code", "")).strip()
|
||
if not species or species.lower() == "nan" or species == "-":
|
||
continue
|
||
group = refs if _config_bool(row.get("Ref", False)) else targets
|
||
if species not in group:
|
||
group.append(species)
|
||
except Exception as exc:
|
||
print(f"[WARN] Failed to read interspecies species options from {config_path}: {exc}")
|
||
default_refs = []
|
||
for preferred in ("at", "ath"):
|
||
for species in refs:
|
||
if species.lower() == preferred:
|
||
default_refs = [species]
|
||
break
|
||
if default_refs:
|
||
break
|
||
if not default_refs and refs:
|
||
default_refs = [refs[0]]
|
||
return {
|
||
"targets": targets,
|
||
"refs": refs,
|
||
"default_targets": list(targets),
|
||
"default_refs": default_refs,
|
||
}
|
||
|
||
class ExecuteRequest(BaseModel):
|
||
stage: str
|
||
job_id: str = "" # 可选，指定job_id
|
||
prefix: str = "Test"
|
||
cpu: int = 8
|
||
idEvalue: float = 0.001
|
||
idCoverage: float = 0.3
|
||
blastEvalue: float = 0.00001
|
||
requiredPfams: str = ""
|
||
refSpecies: str = ""
|
||
mafftParams: str = "--auto --thread -1"
|
||
iqtreeParams: str = "-m MFP -bb 1000 -nt AUTO"
|
||
nmotifs: int = 10
|
||
minw: int = 6
|
||
maxw: int = 200
|
||
domEvalue: float = 0.00001
|
||
domCoverage: float = 0.3
|
||
idTransform: str = "none"
|
||
targetSpecies: List[str] = []
|
||
refSpeciesList: List[str] = []
|
||
blastEngine: str = "diamond"
|
||
geoId: str = ""
|
||
# RNA-seq参数
|
||
rnaseqMode: str = "upload"
|
||
speciesName: str = ""
|
||
# RNA-seq QC 参数
|
||
rnaseqDataMode: str = "upload"
|
||
rnaseqDataPath: str = ""
|
||
# RNA-seq Enrich参数
|
||
topPathways: int = 20 # 富集分析热图通路数量
|
||
enrichBackgroundSource: str = "auto" # auto, gene2go, orgdb
|
||
enrichSpecies: str = "" # 内置OrgDb物种代码
|
||
enrichKeyType: str = "auto" # OrgDb keyType
|
||
enrichType: str = "go" # go, kegg, both
|
||
keggOrganism: str = "" # KEGG organism code
|
||
rnaseqForegroundMode: str = "auto" # auto, global, family
|
||
# PPI参数
|
||
ppiSpecies: str = "ath" # PPI物种代码
|
||
ppiScore: int = 700 # PPI置信度阈值
|
||
# KaKs参数
|
||
kaksNeutralRate: float = 7.08e-9 # 中性替换率
|
||
# Genome Prepare参数
|
||
genomePrepareType: str = "auto" # auto, jgi, ensembl, ncbi
|
||
|
||
def run_stage_sync(task_id: str, stage: str, job_id: str, params: dict):
|
||
"""在后台线程中执行分析阶段"""
|
||
try:
|
||
# 更新任务状态为运行中
|
||
tasks[task_id]["status"] = "running"
|
||
tasks[task_id]["message"] = f"正在执行 {stage}..."
|
||
tasks[task_id]["start_time"] = datetime.now().isoformat()
|
||
|
||
# 导入pipeline_backend
|
||
from apps.web.pipeline_backend import run_stage, JobInfo
|
||
|
||
job_dir = INPUT_ROOT / job_id
|
||
workspace_dir = WORKSPACE_ROOT / job_id
|
||
|
||
effective_prefix = resolve_effective_prefix(job_id, params, stage)
|
||
if effective_prefix != _nonempty_prefix(params.get("prefix")):
|
||
print(f"[INFO] Resolved empty prefix for {stage}: {effective_prefix or '<empty>'}")
|
||
params["prefix"] = effective_prefix
|
||
tasks[task_id]["prefix"] = effective_prefix
|
||
|
||
job = JobInfo(
|
||
job_id=job_id,
|
||
input_dir=str(job_dir),
|
||
workspace_dir=str(workspace_dir),
|
||
logs_dir=str(workspace_dir / "logs"),
|
||
created_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
)
|
||
os.makedirs(job.logs_dir, exist_ok=True)
|
||
|
||
# 切换到项目根目录
|
||
original_cwd = os.getcwd()
|
||
os.chdir(PROJECT_ROOT)
|
||
|
||
# 特殊状态提示
|
||
if stage == "promoter":
|
||
tasks[task_id]["message"] = "正在提交序列到 PlantCARE..."
|
||
elif stage == "synteny":
|
||
tasks[task_id]["message"] = "正在进行共线性分析（MCScanX）..."
|
||
elif stage == "rnaseq_qc":
|
||
tasks[task_id]["message"] = "正在进行RNA-seq质控分析（FastQC + Cutadapt）... 这可能需要15-30分钟，请耐心等待..."
|
||
|
||
backend_stage = {
|
||
"rnaseq_go_enrich": "rnaseq_enrich",
|
||
"rnaseq_kegg_enrich": "rnaseq_enrich",
|
||
}.get(stage, stage)
|
||
|
||
# 执行
|
||
result_msg, manifest = run_stage(
|
||
stage=backend_stage,
|
||
job=job,
|
||
prefix=params["prefix"],
|
||
cpu=params["cpu"],
|
||
required_pfams=params.get("requiredPfams") or None,
|
||
ref_species=params.get("refSpecies") or None,
|
||
id_evalue=params.get("idEvalue", 0.001),
|
||
id_coverage=params.get("idCoverage", 0.3),
|
||
blast_evalue=params.get("blastEvalue", 0.00001),
|
||
phy_mafft=params.get("mafftParams", "--auto --thread -1"),
|
||
phy_iqtree=params.get("iqtreeParams", "-m MFP -bb 1000 -nt AUTO"),
|
||
motif_nmotifs=params.get("nmotifs", 10),
|
||
motif_minw=params.get("minw", 6),
|
||
motif_maxw=params.get("maxw", 200),
|
||
dom_evalue=params.get("domEvalue", 0.00001),
|
||
dom_coverage=params.get("domCoverage", 0.3),
|
||
id_transform=params.get("idTransform", "none"),
|
||
geo_id=params.get("geoId", ""),
|
||
# RNA-seq参数
|
||
rnaseq_mode=params.get("rnaseqMode", "upload"),
|
||
species_name=params.get("speciesName", ""),
|
||
# RNA-seq Enrich参数
|
||
top_pathways=params.get("topPathways", 20),
|
||
enrich_background_source=params.get("enrichBackgroundSource", "auto"),
|
||
enrich_species=params.get("enrichSpecies", ""),
|
||
enrich_key_type=params.get("enrichKeyType", "auto"),
|
||
enrich_type=params.get("enrichType", "go"),
|
||
kegg_organism=params.get("keggOrganism", ""),
|
||
rnaseq_foreground_mode=params.get("rnaseqForegroundMode", "auto"),
|
||
# PPI参数
|
||
ppi_species=params.get("ppiSpecies", "ath"),
|
||
ppi_score=params.get("ppiScore", 700),
|
||
# KaKs参数
|
||
kaks_neutral_rate=params.get("kaksNeutralRate", 7.08e-9),
|
||
# Genome Prepare参数
|
||
genome_prepare_type=params.get("genomePrepareType", "auto"),
|
||
target_species=params.get("targetSpecies") or None,
|
||
interspecies_ref_species=params.get("refSpeciesList") or None,
|
||
interspecies_blast_engine=params.get("blastEngine", "diamond") or "diamond",
|
||
)
|
||
|
||
# 恢复工作目录
|
||
os.chdir(original_cwd)
|
||
|
||
# 检查是否有特殊状态需要更新
|
||
if stage == "promoter" and "Waiting for PlantCARE" in result_msg:
|
||
tasks[task_id]["message"] = "已提交 PlantCARE，等待邮件结果（可能需要10-60分钟）..."
|
||
elif stage == "promoter" and "submitted" in result_msg.lower():
|
||
tasks[task_id]["message"] = "已提交 PlantCARE，正在等待邮件..."
|
||
|
||
# 收集结果 - 按stage分类
|
||
stage_prefix_map = {
|
||
"identify": ["identify", "final_members"],
|
||
"phylogeny": ["phylogeny", "tree"],
|
||
"motif": ["motif"],
|
||
"domain": ["domain"],
|
||
"structure": ["structure"],
|
||
"promoter": ["promoter"],
|
||
"synteny": ["synteny"],
|
||
"karyotype": ["karyotype"],
|
||
"interspecies_synteny": ["interspecies_synteny", "interspecies_tree"],
|
||
"ppi": ["ppi"],
|
||
"kaks": ["kaks"],
|
||
"pep": ["pep"],
|
||
"expression": ["exp", "expression"],
|
||
"combo": ["combo"],
|
||
"msa": ["msa"],
|
||
"ortholog": ["ortholog"],
|
||
"rnaseq_qc": ["rnaseq_qc"],
|
||
"rnaseq_align": ["rnaseq_align"],
|
||
"rnaseq_quant": ["rnaseq_quant"],
|
||
"rnaseq_de": ["rnaseq_de"],
|
||
"annotation": ["annotation", "gene2go", "gene2ko", "eggnog"],
|
||
"rnaseq_go_enrich": ["rnaseq_enrich", "go_enrich", "_go_", "gene2go"],
|
||
"rnaseq_kegg_enrich": ["rnaseq_enrich", "kegg_enrich", "_kegg_", "gene2ko"],
|
||
"rnaseq_enrich": ["rnaseq_enrich"],
|
||
"genome_prepare": ["genome_prepare"]
|
||
}
|
||
|
||
# 反转映射
|
||
prefix_to_stage = {}
|
||
for s, prefixes in stage_prefix_map.items():
|
||
for p in prefixes:
|
||
prefix_to_stage[p] = s
|
||
|
||
# 初始化各stage文件
|
||
stage_files = {s: {"images": [], "files": []} for s in stage_prefix_map}
|
||
|
||
final_dir = workspace_dir / "final"
|
||
if final_dir.exists():
|
||
for ext in ['.png', '.jpg', '.jpeg', '.svg']:
|
||
for f in final_dir.glob(f"*{ext}"):
|
||
# 提取前缀（支持 . 和 _ 分隔）
|
||
fname = f.stem
|
||
prefix = fname.split('.')[0].split('_')[0]
|
||
matched = prefix_to_stage.get(prefix, stage)
|
||
stage_files[matched]["images"].append(f"/workspace/{job_id}/final/{f.name}")
|
||
|
||
for f in final_dir.iterdir():
|
||
if f.is_file():
|
||
fname = f.stem
|
||
prefix = fname.split('.')[0].split('_')[0]
|
||
matched = prefix_to_stage.get(prefix, stage)
|
||
stage_files[matched]["files"].append({
|
||
"name": f.name,
|
||
"size": f.stat().st_size,
|
||
"path": f"/workspace/{job_id}/final/{f.name}"
|
||
})
|
||
|
||
# 汇总所有文件（保留兼容）
|
||
all_images = []
|
||
all_files = []
|
||
for s_data in stage_files.values():
|
||
all_images.extend(s_data["images"])
|
||
all_files.extend(s_data["files"])
|
||
|
||
# 收集PPI和KaKs结果文件
|
||
if final_dir.exists():
|
||
# PPI结果文件
|
||
for ppi_file in final_dir.glob("*.ppi_*"):
|
||
if ppi_file.is_file():
|
||
file_info = {
|
||
"name": ppi_file.name,
|
||
"size": ppi_file.stat().st_size,
|
||
"path": f"/workspace/{job_id}/final/{ppi_file.name}"
|
||
}
|
||
stage_files["ppi"]["files"].append(file_info)
|
||
all_files.append(file_info)
|
||
|
||
# KaKs结果文件
|
||
for kaks_file in final_dir.glob("*.kaks_*"):
|
||
if kaks_file.is_file():
|
||
file_info = {
|
||
"name": kaks_file.name,
|
||
"size": kaks_file.stat().st_size,
|
||
"path": f"/workspace/{job_id}/final/{kaks_file.name}"
|
||
}
|
||
stage_files["kaks"]["files"].append(file_info)
|
||
all_files.append(file_info)
|
||
|
||
# ?? manifest ????? final ?????????????? manifest/FASTA/tree ???
|
||
|
||
# ????????? manifest ???????? success/partial_success
|
||
expected_manifest = final_dir / f"{params['prefix']}.{stage}.manifest.json"
|
||
paused_manifest = final_dir / f"{params['prefix']}.{stage}.paused.json"
|
||
manifest_valid = False
|
||
manifest_status = None
|
||
|
||
import time
|
||
time.sleep(0.5)
|
||
manifest_to_check = paused_manifest if paused_manifest.exists() else expected_manifest
|
||
if manifest_to_check.exists():
|
||
try:
|
||
with open(manifest_to_check, 'r', encoding='utf-8') as f:
|
||
manifest_data = json.load(f)
|
||
manifest_status = manifest_data.get("status")
|
||
manifest_valid = (
|
||
manifest_status in {"success", "partial_success"} and
|
||
"?" not in result_msg
|
||
)
|
||
except Exception as e:
|
||
print(f"[WARN] Failed to read manifest {manifest_to_check}: {e}")
|
||
|
||
if not manifest_valid:
|
||
# manifest 验证失败
|
||
if isinstance(result_msg, str) and ("❌" in result_msg or "failed" in result_msg.lower()):
|
||
failure_message = result_msg
|
||
failure_error = result_msg
|
||
else:
|
||
failure_message = f"❌ {stage} 失败：manifest 状态为 {manifest_status}。{result_msg}"
|
||
failure_error = f"Manifest validation failed: expected {expected_manifest}, status={manifest_status}"
|
||
tasks[task_id]["status"] = "failed"
|
||
tasks[task_id]["message"] = failure_message
|
||
tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
tasks[task_id]["result"] = {
|
||
"success": False,
|
||
"error": failure_error,
|
||
"stage": stage
|
||
}
|
||
else:
|
||
# 更新任务状态为完成
|
||
tasks[task_id]["status"] = "success"
|
||
tasks[task_id]["message"] = result_msg
|
||
tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
tasks[task_id]["result"] = {
|
||
"success": True,
|
||
"message": result_msg,
|
||
"images": all_images,
|
||
"files": all_files,
|
||
"stage_files": stage_files, # 按stage分类的文件
|
||
"stage": stage
|
||
}
|
||
|
||
# ============ 生成累积解读报告（新增） ============
|
||
try:
|
||
if MAS_AVAILABLE and LLMManager:
|
||
from gfa_core.mas.agents.result_parser import create_result_parser_agent
|
||
|
||
# 获取已完成的阶段列表
|
||
completed_stages = []
|
||
if final_dir.exists():
|
||
for manifest_file in final_dir.glob(f"{params['prefix']}.*.manifest.json"):
|
||
try:
|
||
with open(manifest_file, 'r') as f:
|
||
m = json.load(f)
|
||
if m.get("status") == "success":
|
||
stage_name = m.get("stage")
|
||
if stage_name and stage_name not in completed_stages:
|
||
completed_stages.append(stage_name)
|
||
except:
|
||
continue
|
||
|
||
# 按顺序排序
|
||
stage_order = ["identify", "phylogeny", "structure", "motif", "domain", "promoter", "synteny", "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"[CumulativeInterpretation] 生成累积解读，已完成阶段: {completed_stages}")
|
||
|
||
# 创建 LLMManager 和 ResultParserAgent
|
||
llm_manager = LLMManager()
|
||
agent = create_result_parser_agent(llm_manager)
|
||
|
||
# 生成累积解读报告
|
||
report = asyncio.run(agent.generate_cumulative_report(
|
||
job_id=job_id,
|
||
prefix=params["prefix"],
|
||
completed_stages=completed_stages
|
||
))
|
||
|
||
# 保存解读报告
|
||
save_success = agent.save_interpretation(job_id, params["prefix"], report)
|
||
|
||
if save_success:
|
||
print(f"[CumulativeInterpretation] 解读报告已保存到 interpretation.json")
|
||
else:
|
||
print(f"[CumulativeInterpretation] 保存解读报告失败")
|
||
|
||
except Exception as e:
|
||
# 解读失败不阻塞主流程
|
||
print(f"[CumulativeInterpretation] 生成解读报告失败（非阻塞）: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
# 返回任务结果，供 Celery 使用
|
||
return tasks[task_id]["result"]
|
||
|
||
except Exception as e:
|
||
tasks[task_id]["status"] = "failed"
|
||
tasks[task_id]["message"] = f"执行失败: {str(e)}"
|
||
tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
tasks[task_id]["result"] = {
|
||
"success": False,
|
||
"error": str(e),
|
||
"stage": stage
|
||
}
|
||
# 返回任务结果，供 Celery 使用
|
||
return tasks[task_id]["result"]
|
||
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
"""返回Vue3前端页面"""
|
||
return FileResponse(
|
||
Path(__file__).parent / "index.html",
|
||
headers={
|
||
"Cache-Control": "no-cache, no-store, must-revalidate",
|
||
"Pragma": "no-cache",
|
||
"Expires": "0"
|
||
}
|
||
)
|
||
|
||
|
||
@app.post("/api/new_job")
|
||
async def create_new_job():
|
||
"""创建新工作目录"""
|
||
global current_job_id
|
||
import time
|
||
import uuid
|
||
current_job_id = f"job_{int(time.time()*1000)}_{uuid.uuid4().hex[:6]}"
|
||
|
||
job_dir = INPUT_ROOT / current_job_id
|
||
workspace_dir = WORKSPACE_ROOT / current_job_id
|
||
job_dir.mkdir(parents=True, exist_ok=True)
|
||
workspace_dir.mkdir(parents=True, exist_ok=True)
|
||
(workspace_dir / "logs").mkdir(exist_ok=True)
|
||
(workspace_dir / "final").mkdir(exist_ok=True)
|
||
|
||
return {
|
||
"success": True,
|
||
"job_id": current_job_id,
|
||
"input_dir": str(job_dir),
|
||
"workspace_dir": str(workspace_dir)
|
||
}
|
||
|
||
|
||
@app.get("/api/list_jobs")
|
||
async def list_existing_jobs():
|
||
"""列出所有已有的工作目录"""
|
||
jobs = []
|
||
|
||
# 扫描 workspace 目录
|
||
if WORKSPACE_ROOT.exists():
|
||
for job_dir in sorted(WORKSPACE_ROOT.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
|
||
if job_dir.is_dir() and job_dir.name.startswith("job_"):
|
||
# 检查目录内容
|
||
final_dir = job_dir / "final"
|
||
logs_dir = job_dir / "logs"
|
||
|
||
# 获取已完成的阶段
|
||
completed_stages = []
|
||
if final_dir.exists():
|
||
for manifest in final_dir.glob("*.manifest.json"):
|
||
stage_name = manifest.name.replace(".manifest.json", "").split(".")[-1]
|
||
completed_stages.append(stage_name)
|
||
|
||
# 获取最后修改时间
|
||
mtime = datetime.fromtimestamp(job_dir.stat().st_mtime)
|
||
|
||
jobs.append({
|
||
"job_id": job_dir.name,
|
||
"created_at": mtime.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"completed_stages": completed_stages,
|
||
"has_input": (INPUT_ROOT / job_dir.name).exists(),
|
||
"workspace_dir": str(job_dir)
|
||
})
|
||
|
||
return {
|
||
"success": True,
|
||
"jobs": jobs[:20] # 只返回最近20个
|
||
}
|
||
|
||
|
||
@app.post("/api/select_job")
|
||
async def select_existing_job(request: Request):
|
||
"""选择已有工作目录"""
|
||
global current_job_id
|
||
|
||
data = await request.json()
|
||
job_id = data.get("job_id")
|
||
|
||
if not job_id:
|
||
return JSONResponse({"success": False, "error": "No job_id provided"}, status_code=400)
|
||
|
||
# 验证工作目录是否存在
|
||
workspace_dir = WORKSPACE_ROOT / job_id
|
||
input_dir = INPUT_ROOT / job_id
|
||
|
||
if not workspace_dir.exists():
|
||
return JSONResponse({"success": False, "error": f"Job {job_id} not found"}, status_code=404)
|
||
|
||
# 设置为当前工作目录
|
||
current_job_id = job_id
|
||
|
||
return {
|
||
"success": True,
|
||
"job_id": job_id,
|
||
"input_dir": str(input_dir),
|
||
"workspace_dir": str(workspace_dir),
|
||
"message": f"已切换到工作目录: {job_id}"
|
||
}
|
||
|
||
|
||
@app.post("/api/upload")
|
||
async def upload_file(files: List[UploadFile] = File(...)):
|
||
"""文件上传接口"""
|
||
global current_job_id
|
||
|
||
if not current_job_id:
|
||
current_job_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:18]
|
||
|
||
job_id = current_job_id
|
||
job_dir = INPUT_ROOT / job_id
|
||
job_dir.mkdir(exist_ok=True)
|
||
|
||
saved_files = []
|
||
for f in files:
|
||
file_path = job_dir / f.filename
|
||
content = await f.read()
|
||
with open(file_path, 'wb') as fp:
|
||
fp.write(content)
|
||
saved_files.append(f.filename)
|
||
|
||
return {
|
||
"success": True,
|
||
"job_id": job_id,
|
||
"files": [f.filename for f in files]
|
||
}
|
||
|
||
|
||
|
||
@app.get("/api/interspecies/species-options")
|
||
async def get_interspecies_species_options(job_id: str = Query("")):
|
||
if not job_id:
|
||
return {"success": False, "message": "job_id is required", "targets": [], "refs": [], "default_targets": [], "default_refs": []}
|
||
options = _ordered_species_options(job_id)
|
||
return {"success": True, "job_id": job_id, **options}
|
||
|
||
@app.post("/api/execute")
|
||
async def execute_stage(req: ExecuteRequest):
|
||
"""执行分析阶段 - 支持异步模式"""
|
||
print(f"[API DEBUG] ============================================")
|
||
print(f"[API DEBUG] /api/execute called")
|
||
print(f"[API DEBUG] stage={req.stage}, job_id={req.job_id}")
|
||
print(f"[API DEBUG] prefix={req.prefix}, cpu={req.cpu}")
|
||
print(f"[API DEBUG] rnaseqMode={getattr(req, 'rnaseqMode', 'NOT_SET')}, speciesName={getattr(req, 'speciesName', 'NOT_SET')}")
|
||
print(f"[API DEBUG] kaksNeutralRate={getattr(req, 'kaksNeutralRate', 'NOT_SET')}")
|
||
print(f"[API DEBUG] ============================================")
|
||
global current_job_id
|
||
|
||
# 如果请求中指定了 job_id，使用它；否则使用 current_job_id
|
||
if req.job_id:
|
||
job_id = req.job_id
|
||
elif not current_job_id:
|
||
current_job_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:18]
|
||
job_id = current_job_id
|
||
else:
|
||
job_id = current_job_id
|
||
|
||
# ===== 检查是否已完成：以统一 manifest/artifacts 校验为准 =====
|
||
effective_prefix = resolve_effective_prefix(job_id, {"prefix": req.prefix}, req.stage)
|
||
if effective_prefix != _nonempty_prefix(req.prefix):
|
||
print(f"[API DEBUG] Resolved request prefix for {req.stage}: {effective_prefix or '<empty>'}")
|
||
|
||
try:
|
||
from gfa_core.mas.tools.stage_status import get_stage_status as get_unified_stage_status
|
||
|
||
stage_status = get_unified_stage_status(job_id=job_id, prefix=effective_prefix, stage=req.stage)
|
||
print(f"[API DEBUG] Stage status from unified validator: {stage_status.to_dict()}")
|
||
|
||
explicit_interspecies_selection = (
|
||
req.stage == "interspecies_synteny" and
|
||
(bool(getattr(req, "targetSpecies", []) or []) or bool(getattr(req, "refSpeciesList", []) or []))
|
||
)
|
||
if stage_status.completed and not explicit_interspecies_selection:
|
||
print(f"[API DEBUG] Stage '{req.stage}' already completed, skipping execution")
|
||
return {
|
||
"success": True,
|
||
"async": False,
|
||
"message": f"? {req.stage} ????????????",
|
||
"job_id": job_id,
|
||
"already_completed": True,
|
||
"stage_status": stage_status.to_dict()
|
||
}
|
||
elif stage_status.completed and explicit_interspecies_selection:
|
||
print("[API DEBUG] Interspecies selection provided; rerunning instead of skipping completed manifest")
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Error checking completed stages: {e}")
|
||
# 继续执行，不阻断
|
||
|
||
# ===== 执行能力检查：避免规划/接口提交尚未接入执行器的阶段 =====
|
||
try:
|
||
from gfa_core.mas.stage_catalog import is_stage_executable
|
||
|
||
if not is_stage_executable(req.stage):
|
||
message = (
|
||
f"阶段 {req.stage} 暂未接入标准执行器，"
|
||
"无法直接执行。请先补充 pipeline_backend.run_stage 分支和标准 manifest。"
|
||
)
|
||
print(f"[API DEBUG] Unsupported stage rejected: {req.stage}")
|
||
return {
|
||
"success": False,
|
||
"async": False,
|
||
"status": "unsupported",
|
||
"message": message,
|
||
"job_id": job_id,
|
||
"stage": req.stage,
|
||
}
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Error checking stage executability: {e}")
|
||
|
||
# ===== 统一架构：所有stage都使用Celery + LangGraph =====
|
||
task_id = str(uuid.uuid4())
|
||
|
||
# 创建任务记录
|
||
tasks[task_id] = {
|
||
"id": task_id,
|
||
"stage": req.stage,
|
||
"job_id": job_id,
|
||
"prefix": effective_prefix,
|
||
"status": "pending",
|
||
"message": f"任务 {req.stage} 已提交",
|
||
"created_at": datetime.now().isoformat(),
|
||
"result": None
|
||
}
|
||
|
||
# 构建参数
|
||
params = {
|
||
"prefix": effective_prefix,
|
||
"cpu": req.cpu,
|
||
"requiredPfams": req.requiredPfams,
|
||
"refSpecies": req.refSpecies,
|
||
"idEvalue": req.idEvalue,
|
||
"idCoverage": req.idCoverage,
|
||
"blastEvalue": req.blastEvalue,
|
||
"mafftParams": req.mafftParams,
|
||
"iqtreeParams": req.iqtreeParams,
|
||
"nmotifs": req.nmotifs,
|
||
"minw": req.minw,
|
||
"maxw": req.maxw,
|
||
"domEvalue": req.domEvalue,
|
||
"domCoverage": req.domCoverage,
|
||
"idTransform": req.idTransform,
|
||
"geoId": getattr(req, 'geoId', ''),
|
||
# RNA-seq参数
|
||
"rnaseqMode": getattr(req, 'rnaseqMode', 'upload'),
|
||
"speciesName": getattr(req, 'speciesName', ''),
|
||
# RNA-seq Enrich参数
|
||
"topPathways": getattr(req, 'topPathways', 20),
|
||
"enrichBackgroundSource": getattr(req, 'enrichBackgroundSource', 'auto'),
|
||
"enrichSpecies": getattr(req, 'enrichSpecies', ''),
|
||
"enrichKeyType": getattr(req, 'enrichKeyType', 'auto'),
|
||
"enrichType": getattr(req, 'enrichType', 'go'),
|
||
"keggOrganism": getattr(req, 'keggOrganism', ''),
|
||
"rnaseqForegroundMode": getattr(req, 'rnaseqForegroundMode', 'auto'),
|
||
# PPI参数
|
||
"ppiSpecies": getattr(req, 'ppiSpecies', 'ath'),
|
||
"ppiScore": getattr(req, 'ppiScore', 700),
|
||
# KaKs参数
|
||
"kaksNeutralRate": getattr(req, 'kaksNeutralRate', 7.08e-9),
|
||
# Genome Prepare参数
|
||
"genomePrepareType": getattr(req, 'genomePrepareType', 'auto'),
|
||
# Interspecies species selection
|
||
"targetSpecies": getattr(req, 'targetSpecies', []) or [],
|
||
"refSpeciesList": getattr(req, 'refSpeciesList', []) or [],
|
||
"blastEngine": getattr(req, 'blastEngine', 'diamond') or 'diamond',
|
||
}
|
||
|
||
# 使用新的Celery任务（统一架构）
|
||
if CELERY_AVAILABLE:
|
||
try:
|
||
from tasks import execute_single_stage_task
|
||
celery_task = execute_single_stage_task.delay(
|
||
task_id=task_id,
|
||
stage=req.stage,
|
||
job_id=job_id,
|
||
prefix=effective_prefix,
|
||
params=params,
|
||
session_id=None
|
||
)
|
||
print(f"[API DEBUG] Celery single stage task submitted: {celery_task.id}")
|
||
tasks[task_id]["celery_task_id"] = celery_task.id
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Failed to submit Celery task: {e}")
|
||
# 降级到线程池执行
|
||
executor.submit(run_stage_sync, task_id, req.stage, job_id, params)
|
||
else:
|
||
# Celery不可用，使用线程池
|
||
executor.submit(run_stage_sync, task_id, req.stage, job_id, params)
|
||
|
||
return {
|
||
"success": True,
|
||
"async": True,
|
||
"task_id": task_id,
|
||
"message": f"{req.stage} 任务已提交（统一架构），请在后台等待完成",
|
||
"job_id": job_id,
|
||
"unified_arch": True # 标记使用统一架构
|
||
}
|
||
|
||
|
||
@app.get("/api/task/{task_id}")
|
||
async def get_task_status(task_id: str):
|
||
"""查询异步任务状态（支持Celery任务）"""
|
||
if task_id not in tasks:
|
||
raise HTTPException(status_code=404, detail="Task not found")
|
||
|
||
task = tasks[task_id]
|
||
|
||
# 如果是Celery任务，查询Celery状态
|
||
celery_task_id = task.get("celery_task_id")
|
||
if celery_task_id and CELERY_AVAILABLE:
|
||
try:
|
||
from celery_app import app as celery_app
|
||
celery_result = celery_app.AsyncResult(celery_task_id)
|
||
|
||
# 更新本地任务状态
|
||
if celery_result.state == 'SUCCESS':
|
||
# 检查Celery任务result内部的状态
|
||
result_data = celery_result.result
|
||
if isinstance(result_data, dict):
|
||
# 优先使用result_data中的status字段
|
||
result_status = result_data.get('status')
|
||
if result_data.get('success') == False or result_status == 'failed':
|
||
task["status"] = "failed"
|
||
task["message"] = result_data.get('message', '任务执行失败')
|
||
elif result_status == 'completed':
|
||
task["status"] = "success"
|
||
task["message"] = result_data.get('message', '任务完成')
|
||
else:
|
||
# 其他状态（如waiting_confirmation）
|
||
task["status"] = result_status or "success"
|
||
task["message"] = result_data.get('message', '任务完成')
|
||
task["result"] = result_data
|
||
else:
|
||
task["status"] = "success"
|
||
task["result"] = result_data
|
||
task["message"] = "任务完成"
|
||
elif celery_result.state == 'FAILURE':
|
||
task["status"] = "failed"
|
||
task["message"] = str(celery_result.result) if celery_result.result else "任务失败"
|
||
elif celery_result.state == 'RUNNING' or celery_result.state == 'PENDING':
|
||
task["status"] = "running"
|
||
# 获取meta信息
|
||
meta = celery_result.info
|
||
if meta and isinstance(meta, dict):
|
||
task["progress"] = meta.get("progress", 0)
|
||
task["current_stage"] = meta.get("current_stage", task.get("stage"))
|
||
|
||
# 检查是否有等待确认的状态
|
||
if meta.get("status") == "waiting_confirmation":
|
||
task["status"] = "waiting_confirmation"
|
||
task["pending_confirmation"] = meta.get("pending_confirmation", {})
|
||
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Error querying Celery task: {e}")
|
||
|
||
_attach_single_stage_status(task)
|
||
_attach_annotation_progress(task)
|
||
|
||
# 读取日志文件内容（如果存在）
|
||
log_content = ""
|
||
try:
|
||
job_id = task.get("job_id")
|
||
stage = task.get("stage")
|
||
if job_id and stage:
|
||
import glob
|
||
log_pattern = str(WORKSPACE_ROOT / job_id / "logs" / f"*_{stage}.log")
|
||
log_files = glob.glob(log_pattern)
|
||
if log_files:
|
||
# 读取最新的日志文件
|
||
latest_log = max(log_files, key=lambda x: Path(x).stat().st_mtime)
|
||
with open(latest_log, 'r', encoding='utf-8', errors='ignore') as f:
|
||
log_content = f.read()
|
||
except Exception as e:
|
||
print(f"[API DEBUG] Error reading log file: {e}")
|
||
|
||
return {
|
||
"id": task["id"],
|
||
"stage": task["stage"],
|
||
"status": task["status"],
|
||
"message": task["message"],
|
||
"created_at": task["created_at"],
|
||
"start_time": task.get("start_time"),
|
||
"end_time": task.get("end_time"),
|
||
"result": task.get("result"),
|
||
"stage_status": task.get("stage_status"),
|
||
"log": log_content, # 添加日志内容
|
||
"progress": task.get("progress", 0),
|
||
"current_stage": task.get("current_stage", task.get("stage")),
|
||
"pending_confirmation": task.get("pending_confirmation"),
|
||
"annotation_progress": task.get("annotation_progress"),
|
||
"unified_arch": task.get("unified_arch", False)
|
||
}
|
||
|
||
|
||
@app.get("/api/stage-status")
|
||
async def get_stage_status(job_id: str, stage: str, prefix: str = ""):
|
||
"""检查阶段是否已完成（统一 manifest/artifacts 校验）"""
|
||
from gfa_core.mas.tools.stage_status import get_stage_status as get_unified_stage_status
|
||
prefix = resolve_effective_prefix(job_id, {"prefix": prefix}, stage)
|
||
mapped_stage = stage
|
||
if stage in {"rnaseq_go_enrich", "rnaseq_kegg_enrich"}:
|
||
mapped_stage = "rnaseq_enrich"
|
||
|
||
report = get_unified_stage_status(job_id=job_id, prefix=prefix, stage=mapped_stage)
|
||
payload = report.to_dict()
|
||
payload["requested_stage"] = stage
|
||
payload["mapped_stage"] = mapped_stage
|
||
payload["status"] = "completed" if report.completed else report.status
|
||
if report.manifest is not None:
|
||
payload["manifest"] = report.manifest
|
||
if stage == "annotation":
|
||
final_dir = WORKSPACE_ROOT / job_id / "final"
|
||
has_background = False
|
||
if final_dir.exists():
|
||
has_background = any(final_dir.glob("*gene2go*.txt")) or any(final_dir.glob("*gene2ko*.txt"))
|
||
has_annotation_manifest = bool((final_dir / f"{prefix}.annotation.manifest.json").exists())
|
||
payload["completed"] = bool(has_annotation_manifest and has_background)
|
||
payload["status"] = "completed" if payload["completed"] else "not_found"
|
||
if stage in {"rnaseq_go_enrich", "rnaseq_kegg_enrich"}:
|
||
final_dir = WORKSPACE_ROOT / job_id / "final"
|
||
if final_dir.exists():
|
||
if stage == "rnaseq_go_enrich":
|
||
has_specific = any(final_dir.glob("*.GO_enrich.txt"))
|
||
else:
|
||
has_specific = any(final_dir.glob("*.KEGG_enrich.txt"))
|
||
# GO/KEGG 子阶段用各自结果文件硬判定，避免共享 rnaseq_enrich manifest 导致串扰
|
||
payload["completed"] = bool(has_specific)
|
||
payload["status"] = "completed" if payload["completed"] else "not_found"
|
||
return payload
|
||
|
||
|
||
@app.get("/api/tasks")
|
||
async def list_tasks():
|
||
"""列出所有任务"""
|
||
return {
|
||
"tasks": [
|
||
{
|
||
"id": t["id"],
|
||
"stage": t["stage"],
|
||
"status": t["status"],
|
||
"message": t["message"],
|
||
"created_at": t["created_at"]
|
||
}
|
||
for t in tasks.values()
|
||
]
|
||
}
|
||
|
||
|
||
@app.get("/api/results")
|
||
async def get_results(job_id: str = Query(default=None, description="可选的job_id，默认使用当前job")):
|
||
"""获取任务结果，可指定job_id"""
|
||
global current_job_id
|
||
|
||
# 使用传入的job_id或当前job_id
|
||
target_job_id = job_id or current_job_id
|
||
|
||
if not target_job_id:
|
||
return {"identify": {"images": [], "files": []}}
|
||
|
||
workspace_dir = WORKSPACE_ROOT / target_job_id
|
||
final_dir = workspace_dir / "final"
|
||
|
||
stages = {
|
||
"identify": {"images": [], "files": [], "elapsed_time": 0},
|
||
"phylogeny": {"images": [], "files": [], "elapsed_time": 0},
|
||
"pep": {"images": [], "files": [], "elapsed_time": 0},
|
||
"motif": {"images": [], "files": [], "elapsed_time": 0},
|
||
"domain": {"images": [], "files": [], "elapsed_time": 0},
|
||
"structure": {"images": [], "files": [], "elapsed_time": 0},
|
||
"promoter": {"images": [], "files": [], "elapsed_time": 0},
|
||
"synteny": {"images": [], "files": [], "elapsed_time": 0},
|
||
"karyotype": {"images": [], "files": [], "elapsed_time": 0},
|
||
"interspecies_synteny": {"images": [], "files": [], "elapsed_time": 0},
|
||
"combo": {"images": [], "files": [], "elapsed_time": 0},
|
||
"expression": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_qc": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_align": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_quant": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_de": {"images": [], "files": [], "elapsed_time": 0},
|
||
"annotation": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_go_enrich": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_kegg_enrich": {"images": [], "files": [], "elapsed_time": 0},
|
||
"rnaseq_enrich": {"images": [], "files": [], "elapsed_time": 0},
|
||
"ppi": {"images": [], "files": [], "elapsed_time": 0},
|
||
"kaks": {"images": [], "files": [], "elapsed_time": 0},
|
||
}
|
||
|
||
# 从final目录读取执行时间（manifest文件在final/下，不在reports/下）
|
||
if final_dir.exists():
|
||
for manifest_file in final_dir.glob("*.manifest.json"):
|
||
try:
|
||
import json
|
||
with open(manifest_file, 'r') as f:
|
||
manifest = json.load(f)
|
||
stage_name = manifest.get("stage")
|
||
elapsed = manifest.get("elapsed_sec", 0)
|
||
if stage_name in stages and elapsed:
|
||
stages[stage_name]["elapsed_time"] = round(elapsed, 1)
|
||
except Exception:
|
||
pass
|
||
|
||
# 处理final目录中的文件
|
||
if final_dir.exists():
|
||
for f in final_dir.iterdir():
|
||
if not f.is_file():
|
||
continue
|
||
|
||
fname = f.name
|
||
|
||
# 排除大文件（>100MB的原始数据文件）
|
||
file_size = f.stat().st_size
|
||
if file_size > 100 * 1024 * 1024: # 100MB
|
||
# 跳过原始数据文件（FASTQ, BAM等）
|
||
if any(x in fname.lower() for x in ['.fastq', '.fq', '.bam', '.sam', '.clean.']):
|
||
continue
|
||
|
||
if f.suffix.lower() not in ['.png', '.jpg', '.jpeg', '.svg', '.pdf', '.fa', '.fasta', '.json', '.md', '.id', '.nwk', '.msa', '.html', '.log', '.gz', '.zip', '.txt', '.tsv', '.csv']:
|
||
continue
|
||
|
||
file_info = {
|
||
"name": fname,
|
||
"size": file_size,
|
||
"path": f"/workspace/{current_job_id}/final/{fname}"
|
||
}
|
||
|
||
fname_lower = fname.lower()
|
||
# 注意：顺序很重要！更具体的匹配应该放在前面
|
||
stage_keywords = {
|
||
"interspecies_synteny": ["interspecies_synteny", "interspecies"], # 优先匹配种间共线性
|
||
"combo": ["combo"],
|
||
"pep": ["pep", "pi_mw", "tmhmm", "subcellular", "等电点", "分子量", "跨膜"],
|
||
"motif": ["motif"],
|
||
"domain": ["domain"],
|
||
"structure": ["structure"],
|
||
"promoter": ["promoter", "plantcare"],
|
||
"synteny": [".synteny.", "collinearity", "mcscanx", "circos"], # 种内共线性，使用".synteny."避免匹配interspecies_synteny
|
||
"karyotype": ["karyotype"],
|
||
"phylogeny": ["phylogeny", "tree", "msa"],
|
||
"identify": ["identify", "final_members"],
|
||
"expression": [".expression.", "gene-exp", "gene-cor"], # 表达量分析
|
||
"rnaseq_qc": ["rnaseq_qc", "_fastqc", "_cutadapt"], # 移除 .clean.fastq 避免展示原始数据
|
||
"rnaseq_align": ["rnaseq_align", ".sorted.bam", ".bam.bai", "_hisat2.log"],
|
||
"rnaseq_quant": ["rnaseq_quant", ".count_matrix", "_quant"],
|
||
"rnaseq_de": ["rnaseq_de", ".DE_results", ".DE_genes", "_MA_plot", "_volcano"],
|
||
# annotation 只归类背景注释相关文件，避免与 GO/KEGG 富集输出混淆
|
||
"annotation": [".annotation.manifest", "gene2go", "gene2ko", "eggnog"],
|
||
# GO/KEGG 富集严格按各自输出特征匹配
|
||
"rnaseq_go_enrich": [".go_enrich"],
|
||
"rnaseq_kegg_enrich": [".kegg_enrich"],
|
||
"rnaseq_enrich": ["rnaseq_enrich", ".go_enrich", ".kegg_enrich", "_go_", "_kegg_", "gene2go"],
|
||
"ppi": ["ppi", "string", "network", "interaction"], # PPI蛋白互作分析
|
||
"kaks": ["kaks", "ka/ks", "ka_ks", "divergence"], # Ka/Ks分析
|
||
}
|
||
|
||
target = None
|
||
for stage_name, keywords in stage_keywords.items():
|
||
if any(kw in fname_lower for kw in keywords):
|
||
target = stages[stage_name]
|
||
break
|
||
|
||
if f.suffix.lower() in ['.png', '.jpg', '.jpeg', '.svg', '.pdf']:
|
||
if target:
|
||
target["images"].append(f"/workspace/{current_job_id}/final/{fname}")
|
||
else:
|
||
for stage_name in ["phylogeny", "pep", "motif", "domain", "structure", "identify", "expression", "promoter", "synteny", "karyotype", "interspecies_synteny", "combo", "rnaseq_qc", "rnaseq_align", "rnaseq_quant", "rnaseq_de", "annotation", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_enrich"]:
|
||
manifest_files = list(final_dir.glob(f"*.{stage_name}.manifest.json"))
|
||
if manifest_files:
|
||
stages[stage_name]["images"].append(f"/workspace/{current_job_id}/final/{fname}")
|
||
break
|
||
elif target:
|
||
target["files"].append(file_info)
|
||
|
||
# 处理Synteny阶段的特殊文件（在synteny/analysis/目录中）
|
||
synteny_analysis_dir = workspace_dir / "synteny" / "analysis"
|
||
if synteny_analysis_dir.exists():
|
||
for f in synteny_analysis_dir.iterdir():
|
||
if not f.is_file():
|
||
continue
|
||
|
||
fname = f.name
|
||
|
||
# 只处理图像文件
|
||
if f.suffix.lower() not in ['.png', '.jpg', '.jpeg', '.svg']:
|
||
continue
|
||
|
||
# 检查是否是Circos图像文件
|
||
if "circos" in fname.lower():
|
||
stages["synteny"]["images"].append(f"/workspace/{current_job_id}/synteny/analysis/{fname}")
|
||
|
||
return stages
|
||
|
||
|
||
@app.get("/api/info")
|
||
async def get_info():
|
||
"""获取当前任务信息"""
|
||
global current_job_id
|
||
return {
|
||
"job_id": current_job_id,
|
||
"working_dir": str(WORKSPACE_ROOT / current_job_id) if current_job_id else "",
|
||
"input_dir": str(INPUT_ROOT / current_job_id) if current_job_id else "",
|
||
"output_dir": str(WORKSPACE_ROOT / current_job_id / "final") if current_job_id else "",
|
||
"workspace_root": str(WORKSPACE_ROOT),
|
||
"input_root": str(INPUT_ROOT)
|
||
}
|
||
|
||
|
||
@app.get("/workspace/{job_id}/{path:path}")
|
||
async def serve_workspace_file(job_id: str, path: str):
|
||
"""提供工作目录文件访问"""
|
||
file_path = WORKSPACE_ROOT / job_id / path
|
||
if file_path.exists() and file_path.is_file():
|
||
return FileResponse(file_path)
|
||
raise HTTPException(status_code=404, detail="File not found")
|
||
|
||
|
||
@app.get("/api/mcscanx_html/{job_id}")
|
||
async def get_mcscanx_html_files(job_id: str):
|
||
"""获取MCScanX HTML结果文件夹中的文件列表"""
|
||
html_dir = WORKSPACE_ROOT / job_id / "synteny" / "analysis" / "Ft.html"
|
||
|
||
if not html_dir.exists():
|
||
return {"exists": False, "files": [], "message": "HTML文件夹不存在"}
|
||
|
||
files = []
|
||
for f in sorted(html_dir.iterdir()):
|
||
if f.is_file() and f.suffix.lower() == '.html':
|
||
files.append({
|
||
"name": f.name,
|
||
"size": f.stat().st_size,
|
||
"path": f"/workspace/{job_id}/synteny/analysis/Ft.html/{f.name}"
|
||
})
|
||
|
||
return {
|
||
"exists": True,
|
||
"folder_path": str(html_dir),
|
||
"files": files,
|
||
"count": len(files)
|
||
}
|
||
|
||
|
||
class ChatRequest(BaseModel):
|
||
message: str
|
||
|
||
|
||
def parse_natural_language(message: str) -> dict:
|
||
"""解析自然语言，提取分析意图和参数
|
||
|
||
支持识别多个阶段，例如："帮我执行基因家族鉴定和绘制进化树"
|
||
会识别出 ['identify', 'phylogeny'] 两个阶段
|
||
"""
|
||
message_lower = message.lower()
|
||
|
||
result = {"success": False, "intent": None, "message": ""}
|
||
|
||
stage_keywords = {
|
||
"identify": ["鉴定", "识别", "找到", "发现", "identify", "find", "discover"],
|
||
"phylogeny": ["进化树", "系统发育", "phylogeny", "phylogenetic", "tree", "建树"],
|
||
"pep": ["pep", "等电点", "分子量", "跨膜", "亚细胞", "理化性质", "pi_mw", "tmhmm", "subcellular"],
|
||
"motif": ["motif", "保守", "元件", "motif"],
|
||
"domain": ["domain", "结构域", "domain"],
|
||
"structure": ["structure", "基因结构", "结构", "外显子", "intron"],
|
||
"promoter": ["promoter", "顺式", "启动子", "元件预测", "plantcare"],
|
||
"synteny": ["synteny", "共线", "circos", "mcscanx", "microsynteny"],
|
||
"karyotype": ["karyotype", "染色体", "核型", "karyotype"],
|
||
"expression": ["expression", "表达", "geo", "表达量"],
|
||
"ppi": ["ppi", "蛋白互作", "string", "interaction"],
|
||
"kaks": ["kaks", "ka/ks", "选择压力", "kaks"],
|
||
"rnaseq_qc": ["rnaseq质控", "rna质控", "fastqc", "质控"],
|
||
"rnaseq_align": ["rnaseq比对", "rna比对", "hisat2", "比对"],
|
||
"rnaseq_quant": ["rnaseq定量", "rna定量", "stringtie", "定量"],
|
||
"rnaseq_de": ["rnaseq差异", "rna差异", "差异表达", "de分析"],
|
||
"annotation": ["背景注释", "功能注释", "基因组注释", "自动注释", "eggnog", "annotation"],
|
||
"rnaseq_go_enrich": ["rnaseq go富集", "rnaseq_go富集", "rna go富集", "go富集", "go分析"],
|
||
"rnaseq_kegg_enrich": ["rnaseq kegg", "rnaseq_kegg", "rna kegg", "kegg富集", "kegg分析"],
|
||
"rnaseq_enrich": ["rnaseq富集", "rna富集", "富集分析", "enrichment"],
|
||
"all": ["完整", "全部", "all", "full", "整个", "全套", "所有"]
|
||
}
|
||
|
||
import re
|
||
prefix_match = re.search(r'[A-Za-z]+\.?[A-Za-z]*[A-Z][A-Za-z]*', message)
|
||
prefix = prefix_match.group(0) if prefix_match else ""
|
||
|
||
# 收集所有匹配的阶段（不再用break，支持多阶段）
|
||
detected_stages = []
|
||
for stage, keywords in stage_keywords.items():
|
||
if any(kw in message_lower for kw in keywords):
|
||
detected_stages.append(stage)
|
||
|
||
# 去重并保持顺序
|
||
seen = set()
|
||
unique_stages = []
|
||
for stage in detected_stages:
|
||
if stage not in seen:
|
||
seen.add(stage)
|
||
unique_stages.append(stage)
|
||
|
||
if "rnaseq_go_enrich" in unique_stages or "rnaseq_kegg_enrich" in unique_stages:
|
||
unique_stages = [s for s in unique_stages if s != "rnaseq_enrich"]
|
||
|
||
# 处理 "all" 关键词：如果包含"全部"等，返回所有主要阶段
|
||
if "all" in unique_stages:
|
||
unique_stages = ["identify", "phylogeny", "motif", "domain", "structure",
|
||
"promoter", "synteny", "karyotype", "expression", "ppi", "kaks"]
|
||
|
||
if not unique_stages:
|
||
result["message"] = "抱歉，我没有理解你想做什么分析。请说明要执行的分析阶段（鉴定、进化树、Motif、结构域、基因结构等）。"
|
||
return result
|
||
|
||
# 单阶段：保持兼容
|
||
if len(unique_stages) == 1:
|
||
intent = {
|
||
"stage": unique_stages[0],
|
||
"params": {"prefix": prefix, "cpu": 8},
|
||
"message": f"我理解你想执行 {unique_stages[0]} 分析，基因家族为 {prefix}。"
|
||
}
|
||
else:
|
||
# 多阶段：返回阶段列表
|
||
intent = {
|
||
"stages": unique_stages,
|
||
"stage": unique_stages[0], # 保持兼容，主阶段为第一个
|
||
"params": {"prefix": prefix, "cpu": 8},
|
||
"message": f"我理解你想执行 {', '.join(unique_stages)} 分析，基因家族为 {prefix}。"
|
||
}
|
||
|
||
result["success"] = True
|
||
result["intent"] = intent
|
||
return result
|
||
|
||
|
||
@app.post("/api/chat")
|
||
async def chat(request: ChatRequest):
|
||
"""自然语言处理接口 (兼容旧版)"""
|
||
try:
|
||
result = parse_natural_language(request.message)
|
||
return result
|
||
except Exception as e:
|
||
return {"success": False, "message": f"处理失败: {str(e)}"}
|
||
|
||
|
||
class AgentChatRequest(BaseModel):
|
||
message: str
|
||
model: str = "kimi" # kimi, glm, minimax
|
||
session_id: str = ""
|
||
job_id: str = "" # 当前 job_id，用于查询已完成 stages
|
||
history: list = []
|
||
|
||
|
||
@app.post("/api/agent/chat")
|
||
async def agent_chat(request: AgentChatRequest):
|
||
"""Agent智能对话接口 - 只执行意图识别，不执行分析阶段
|
||
|
||
返回意图识别结果，前端根据 mode 判断：
|
||
- FAQ: 直接显示回答（前端调用 RAG）
|
||
- Analysis: 调用 /api/agent/plan 获取计划，确认后再执行
|
||
"""
|
||
try:
|
||
session_id = request.session_id or f"session_{int(time.time())}"
|
||
|
||
# 查询已完成的阶段（从manifest文件）
|
||
completed_stages = []
|
||
if request.job_id:
|
||
workspace_dir = PROJECT_ROOT / "workspace" / request.job_id / "final"
|
||
if workspace_dir.exists():
|
||
for manifest_file in workspace_dir.glob("*.manifest.json"):
|
||
try:
|
||
with open(manifest_file, 'r') as f:
|
||
manifest_data = json.load(f)
|
||
status = manifest_data.get("status", "")
|
||
if status in ["success", "partial_success"]:
|
||
stage = manifest_data.get("stage", "")
|
||
if stage:
|
||
completed_stages.append(stage)
|
||
except:
|
||
pass
|
||
|
||
# 只执行 Router + IntentAgent，不执行完整流程
|
||
# 1. RouterAgent 判断模式
|
||
from gfa_core.mas.agents.router import RouterAgent
|
||
from gfa_core.mas.agents.base import AgentState, AgentMessage, QueryMode
|
||
|
||
# 使用全局 llm_manager（已在文件顶部初始化）
|
||
router_agent = RouterAgent(llm_manager)
|
||
|
||
# 创建 AgentState 和 AgentMessage
|
||
router_state = AgentState(session_id=session_id)
|
||
router_message = AgentMessage(
|
||
sender="user",
|
||
receiver="router",
|
||
message_type="classify",
|
||
payload={"query": request.message}
|
||
)
|
||
|
||
# 调用 execute 方法
|
||
router_state = await router_agent.execute(router_state, router_message)
|
||
mode = router_state.mode.value
|
||
|
||
print(f"[Agent Chat] Router result: mode={mode}, query={request.message}")
|
||
|
||
# 2. 如果是 Analysis，调用 IntentAgentV2 获取 intent
|
||
intent = {}
|
||
if mode == "analysis" and request.job_id:
|
||
from gfa_core.mas.agents.intent_v2 import IntentAgentV2
|
||
|
||
intent_agent = IntentAgentV2(llm_manager)
|
||
|
||
# 构建 workspace 路径
|
||
workspace = str(PROJECT_ROOT / "workspace" / request.job_id)
|
||
|
||
# 创建 AgentState 和 AgentMessage
|
||
intent_state = AgentState(
|
||
session_id=session_id,
|
||
job_id=request.job_id
|
||
)
|
||
intent_message = AgentMessage(
|
||
sender="router",
|
||
receiver="intent",
|
||
message_type="recognize",
|
||
payload={
|
||
"query": request.message,
|
||
"workspace": workspace,
|
||
"completed_stages": completed_stages
|
||
}
|
||
)
|
||
|
||
# 调用 execute 方法
|
||
intent_state = await intent_agent.execute(intent_state, intent_message)
|
||
|
||
# 从 state.results 获取 intent（注意：IntentAgentV2 存储在 results.results.intent）
|
||
intent = intent_state.results.get("results", {}).get("intent", {})
|
||
intent["job_id"] = request.job_id
|
||
intent["workspace"] = workspace
|
||
|
||
if intent.get("needs_clarification"):
|
||
intent["taskType"] = "clarification"
|
||
|
||
# 添加 stage 字段（取第一个阶段）供前端判断
|
||
if intent.get("stages") and len(intent["stages"]) > 0:
|
||
intent["stage"] = intent["stages"][0]
|
||
|
||
# 添加 taskType 字段供前端判断
|
||
if not intent.get("taskType"):
|
||
intent["taskType"] = "analysis" if intent.get("stages") else "faq"
|
||
|
||
# 返回意图识别结果
|
||
return {
|
||
"success": True,
|
||
"intent": intent,
|
||
"mode": mode,
|
||
"completed_stages": completed_stages,
|
||
"model_used": request.model,
|
||
"session_id": session_id
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"[Agent Chat] Error: {e}")
|
||
return {
|
||
"success": False,
|
||
"message": f"意图识别失败: {str(e)}",
|
||
"model_used": request.model,
|
||
"mode": "error"
|
||
}
|
||
|
||
|
||
|
||
# ==================== RNA-seq 接口 ====================
|
||
|
||
class RnaseqRequest(BaseModel):
|
||
"""RNA-seq分析请求"""
|
||
job_id: str
|
||
samples: list # 样本配置列表
|
||
design: dict = {} # 实验设计
|
||
|
||
|
||
@app.post("/api/rnaseq/run")
|
||
async def run_rnaseq(req: RnaseqRequest, background_tasks: BackgroundTasks):
|
||
"""运行RNA-seq分析"""
|
||
task_id = str(uuid.uuid4())
|
||
project_root = Path("/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880")
|
||
|
||
# 脚本路径
|
||
script_map = {
|
||
"align": "apps/cli/run_rnaseq_align.py",
|
||
"quant": "apps/cli/run_rnaseq_quant.py",
|
||
"de": "apps/cli/run_rnaseq_de.py",
|
||
"enrich": "apps/cli/run_rnaseq_enrich.py"
|
||
}
|
||
|
||
if req.stage not in script_map:
|
||
return {"success": False, "message": f"未知阶段: {req.stage}"}
|
||
|
||
script_path = project_root / script_map[req.stage]
|
||
sample_dir = project_root / "rnaseq" / req.sample
|
||
|
||
rnaseq_tasks[task_id] = {
|
||
"id": task_id,
|
||
"stage": req.stage,
|
||
"sample": req.sample,
|
||
"status": "running",
|
||
"message": f"正在运行 {req.stage}...",
|
||
"start_time": datetime.now().isoformat()
|
||
}
|
||
|
||
def run_rnaseq_task():
|
||
try:
|
||
if req.stage == "align":
|
||
cmd = [
|
||
"python3", str(script_path),
|
||
"-i", str(sample_dir),
|
||
"-x", str(project_root / "rnaseq" / "Ft.genome"),
|
||
"-o", str(sample_dir),
|
||
"-j", req.sample,
|
||
"-t", "8"
|
||
]
|
||
elif req.stage == "quant":
|
||
cmd = [
|
||
"python3", str(script_path),
|
||
"-i", str(sample_dir),
|
||
"-o", str(sample_dir),
|
||
"-j", req.sample,
|
||
"-t", "8"
|
||
]
|
||
else:
|
||
cmd = ["python3", str(script_path), "-i", str(project_root / "rnaseq")]
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
rnaseq_tasks[task_id]["status"] = "success"
|
||
rnaseq_tasks[task_id]["message"] = "完成"
|
||
rnaseq_tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
except Exception as e:
|
||
rnaseq_tasks[task_id]["status"] = "failed"
|
||
rnaseq_tasks[task_id]["message"] = str(e)
|
||
rnaseq_tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
|
||
background_tasks.add_task(run_rnaseq_task)
|
||
|
||
return {"success": True, "task_id": task_id, "message": f"任务已提交: {req.stage}"}
|
||
|
||
@app.get("/api/rnaseq/status/{task_id}")
|
||
async def get_rnaseq_status(task_id: str):
|
||
"""获取RNA-seq任务状态"""
|
||
if task_id not in rnaseq_tasks:
|
||
return {"success": False, "message": "任务不存在"}
|
||
return {"success": True, "task": rnaseq_tasks[task_id]}
|
||
|
||
@app.get("/api/rnaseq/samples")
|
||
async def list_rnaseq_samples():
|
||
"""列出RNA-seq样本"""
|
||
project_root = Path("/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880")
|
||
rnaseq_dir = project_root / "rnaseq"
|
||
|
||
samples = []
|
||
if rnaseq_dir.exists():
|
||
for d in rnaseq_dir.iterdir():
|
||
if d.is_dir() and d.name.startswith("SRR"):
|
||
samples.append({
|
||
"name": d.name,
|
||
"has_fastq": list(d.glob("*.fastq")) or list(d.glob("*.fq")),
|
||
"has_gtf": list(d.glob("*.gtf"))
|
||
})
|
||
|
||
return {"success": True, "samples": samples}
|
||
|
||
@app.get("/api/rnaseq/results")
|
||
async def get_rnaseq_results():
|
||
"""获取RNA-seq分析结果"""
|
||
project_root = Path("/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880")
|
||
de_output = project_root / "rnaseq" / "de_output"
|
||
|
||
results = {}
|
||
if de_output.exists():
|
||
results = {
|
||
"de_results": str(de_output / "DE_results.txt"),
|
||
"volcano_plot": str(de_output / "volcano_plot.png"),
|
||
"heatmap": str(de_output / "heatmap.png"),
|
||
"sig_genes": str(de_output / "sig_genes.DE_genes.txt")
|
||
}
|
||
|
||
return {"success": True, "results": results}
|
||
|
||
|
||
# ==================== LearningAgent API ====================
|
||
|
||
class LearningRequest(BaseModel):
|
||
"""学习请求"""
|
||
action: str = "recommend" # record, recommend, analyze
|
||
stage: str = ""
|
||
prefix: str = ""
|
||
params: dict = {}
|
||
status: str = ""
|
||
elapsed_sec: float = 0.0
|
||
final_members: int = 0
|
||
|
||
@app.post("/api/agent/learn")
|
||
async def learning(request: LearningRequest):
|
||
"""
|
||
学习优化接口
|
||
|
||
记录执行历史、推荐参数、分析趋势
|
||
"""
|
||
try:
|
||
from gfa_core.mas.agents.learning_v2 import LearningAgentV2
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.agents.base import AgentMessage, AgentState
|
||
|
||
# 初始化组件
|
||
try:
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_manager = LLMManager(config=llm_config)
|
||
except:
|
||
llm_manager = None
|
||
|
||
# 创建LearningAgent
|
||
learning_agent = LearningAgentV2(llm_manager)
|
||
|
||
# 构建消息
|
||
message = AgentMessage(
|
||
sender="user",
|
||
receiver="learning",
|
||
message_type=request.action,
|
||
payload={
|
||
"action": request.action,
|
||
"stage": request.stage,
|
||
"prefix": request.prefix,
|
||
"params": request.params,
|
||
"status": request.status,
|
||
"elapsed_sec": request.elapsed_sec,
|
||
"final_members": request.final_members
|
||
}
|
||
)
|
||
|
||
# 构建状态
|
||
state = AgentState(
|
||
session_id=f"learn_{int(time.time())}",
|
||
results={},
|
||
messages=[]
|
||
)
|
||
|
||
# 执行学习
|
||
result_state = await learning_agent.execute(state, message)
|
||
|
||
if request.action == "recommend":
|
||
recommendations = result_state.results.get("recommendations", {})
|
||
return {
|
||
"success": True,
|
||
"action": request.action,
|
||
"recommendations": recommendations
|
||
}
|
||
elif request.action == "analyze":
|
||
analysis = result_state.results.get("analysis", {})
|
||
return {
|
||
"success": True,
|
||
"action": request.action,
|
||
"analysis": analysis
|
||
}
|
||
else: # record
|
||
record_result = result_state.results.get("record_result", {})
|
||
return {
|
||
"success": True,
|
||
"action": request.action,
|
||
"result": record_result
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"学习优化失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ==================== ValidatorAgent API ====================
|
||
|
||
class ValidateRequest(BaseModel):
|
||
"""验证请求"""
|
||
stage: str
|
||
prefix: str
|
||
workspace: str
|
||
manifest_path: str
|
||
|
||
@app.post("/api/agent/validate")
|
||
async def validate_result(request: ValidateRequest):
|
||
"""
|
||
结果验证接口
|
||
|
||
验证分析结果质量
|
||
"""
|
||
try:
|
||
from gfa_core.mas.agents.validator_v2 import ValidatorAgentV2
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.agents.base import AgentMessage, AgentState
|
||
|
||
# 初始化组件
|
||
try:
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_manager = LLMManager(config=llm_config)
|
||
except:
|
||
llm_manager = None
|
||
|
||
# 创建ValidatorAgent
|
||
validator_agent = ValidatorAgentV2(llm_manager)
|
||
|
||
# 读取manifest
|
||
import json
|
||
manifest_path = Path(request.manifest_path)
|
||
if not manifest_path.exists():
|
||
return {
|
||
"success": False,
|
||
"message": f"Manifest文件不存在: {request.manifest_path}"
|
||
}
|
||
|
||
with open(manifest_path) as f:
|
||
manifest = json.load(f)
|
||
|
||
# 构建execution_result
|
||
execution_result = {
|
||
"status": manifest.get("status", "unknown"),
|
||
"manifest": manifest
|
||
}
|
||
|
||
# 构建消息
|
||
message = AgentMessage(
|
||
sender="user",
|
||
receiver="validator",
|
||
message_type="validate",
|
||
payload={
|
||
"stage": request.stage,
|
||
"prefix": request.prefix,
|
||
"workspace": request.workspace
|
||
}
|
||
)
|
||
|
||
# 构建状态
|
||
state = AgentState(
|
||
session_id=f"validate_{int(time.time())}",
|
||
results={"execution_result": execution_result},
|
||
messages=[]
|
||
)
|
||
|
||
# 执行验证
|
||
result_state = await validator_agent.execute(state, message)
|
||
|
||
report = result_state.results.get("validation_report", {})
|
||
|
||
return {
|
||
"success": True,
|
||
"report": report,
|
||
"overall_passed": report.get("overall_passed", False),
|
||
"score": report.get("score", 0),
|
||
"summary": report.get("summary", "")
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"验证失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ==================== ResultParserAgent API ====================
|
||
|
||
class ParseResultRequest(BaseModel):
|
||
"""结果解析请求"""
|
||
stage: str
|
||
manifest_path: str
|
||
job_id: str = ""
|
||
|
||
@app.post("/api/agent/parse_result")
|
||
async def parse_result(request: ParseResultRequest):
|
||
"""
|
||
结果解析接口
|
||
|
||
解析分析结果并生成报告
|
||
"""
|
||
try:
|
||
from gfa_core.mas.agents.result_parser_v2 import ResultParserAgentV2
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.agents.base import AgentMessage, AgentState
|
||
|
||
# 初始化组件
|
||
try:
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_manager = LLMManager(config=llm_config)
|
||
except:
|
||
llm_manager = None
|
||
|
||
# 创建ResultParserAgent
|
||
parser_agent = ResultParserAgentV2(llm_manager)
|
||
|
||
# 读取manifest
|
||
import json
|
||
manifest_path = Path(request.manifest_path)
|
||
if not manifest_path.exists():
|
||
return {
|
||
"success": False,
|
||
"message": f"Manifest文件不存在: {request.manifest_path}"
|
||
}
|
||
|
||
with open(manifest_path) as f:
|
||
manifest = json.load(f)
|
||
|
||
# 构建execution_result
|
||
execution_result = {
|
||
"status": manifest.get("status", "unknown"),
|
||
"manifest": manifest,
|
||
"stage": request.stage
|
||
}
|
||
|
||
# 构建消息
|
||
message = AgentMessage(
|
||
sender="user",
|
||
receiver="result_parser",
|
||
message_type="parse",
|
||
payload={"stage": request.stage}
|
||
)
|
||
|
||
# 构建状态
|
||
state = AgentState(
|
||
session_id=request.job_id or f"parse_{int(time.time())}",
|
||
results={"execution_result": execution_result},
|
||
messages=[]
|
||
)
|
||
|
||
# 执行解析
|
||
result_state = await parser_agent.execute(state, message)
|
||
|
||
report = result_state.results.get("parsed_report", {})
|
||
|
||
return {
|
||
"success": True,
|
||
"report": report,
|
||
"markdown": report.get("markdown", ""),
|
||
"summary": report.get("summary", {}),
|
||
"metrics": report.get("metrics", {})
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"结果解析失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ==================== Agent执行状态反馈 API ====================
|
||
|
||
# Agent任务存储
|
||
agent_tasks: Dict[str, dict] = {}
|
||
|
||
class AgentExecuteRequest(BaseModel):
|
||
"""Agent执行请求"""
|
||
message: str
|
||
session_id: str = ""
|
||
job_id: str = "" # 当前选中的 job_id，用于 auto_continue
|
||
prefix: str = ""
|
||
model: str = "kimi"
|
||
history: list = []
|
||
auto_execute: bool = True # 是否自动执行分析
|
||
|
||
class AgentStatusResponse(BaseModel):
|
||
"""Agent状态响应"""
|
||
task_id: str
|
||
status: str # pending, running, completed, failed
|
||
current_stage: str = ""
|
||
progress: int = 0 # 0-100
|
||
message: str = ""
|
||
logs: list = []
|
||
result: dict = {}
|
||
|
||
@app.post("/api/agent/execute")
|
||
async def agent_execute(request: AgentExecuteRequest, background_tasks: BackgroundTasks):
|
||
"""
|
||
Agent执行接口 - 使用Celery异步执行分析任务
|
||
|
||
与/agent/chat不同，此接口会实际执行分析流程
|
||
"""
|
||
task_id = f"agent_task_{int(time.time())}_{uuid.uuid4().hex[:8]}"
|
||
|
||
# 初始化任务状态
|
||
agent_tasks[task_id] = {
|
||
"task_id": task_id,
|
||
"status": "pending",
|
||
"message": request.message,
|
||
"job_id": request.job_id,
|
||
"prefix": request.prefix,
|
||
"current_stage": "",
|
||
"progress": 0,
|
||
"logs": [],
|
||
"result": {},
|
||
"start_time": datetime.now().isoformat(),
|
||
"session_id": request.session_id or task_id,
|
||
"stages_completed": [],
|
||
"stages_pending": []
|
||
}
|
||
|
||
# 使用Celery异步执行
|
||
if CELERY_AVAILABLE and execute_agent_task is not None:
|
||
try:
|
||
# 提交任务到Celery队列
|
||
celery_task = execute_agent_task.delay(
|
||
task_id=task_id,
|
||
message=request.message,
|
||
session_id=request.session_id or task_id,
|
||
job_id=request.job_id, # 传入选中的 job_id
|
||
history=request.history if request.history else [],
|
||
prefix=request.prefix
|
||
)
|
||
print(f"[DEBUG] Celery task submitted: {celery_task.id} for agent task {task_id}")
|
||
|
||
# 存储Celery任务ID映射
|
||
agent_tasks[task_id]["celery_task_id"] = celery_task.id
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "info",
|
||
"message": f"任务已提交到Celery队列 (celery_task_id: {celery_task.id})"
|
||
})
|
||
|
||
except Exception as e:
|
||
print(f"[ERROR] Failed to submit Celery task: {e}")
|
||
agent_tasks[task_id]["status"] = "failed"
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "error",
|
||
"message": f"Celery任务提交失败: {str(e)}"
|
||
})
|
||
else:
|
||
# Celery不可用，使用降级方案（直接执行）
|
||
print(f"[WARN] Celery不可用，使用降级方案执行")
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "warn",
|
||
"message": "Celery不可用，使用本地异步执行"
|
||
})
|
||
|
||
async def execute_agent_task_async():
|
||
"""异步执行Agent任务（降级方案）"""
|
||
try:
|
||
agent_tasks[task_id]["status"] = "running"
|
||
|
||
if orchestrator is not None and MAS_AVAILABLE:
|
||
from gfa_core.mas.core.state import GraphState
|
||
|
||
initial_state: GraphState = {
|
||
"query": request.message,
|
||
"session_id": request.session_id or task_id,
|
||
"job_id": request.job_id, # 传入选中的 job_id
|
||
"files": [],
|
||
"current_agent": "router",
|
||
"status": "running",
|
||
"errors": [],
|
||
"final_output": {},
|
||
"history": request.history if request.history else []
|
||
}
|
||
|
||
result = await orchestrator.run(initial_state)
|
||
|
||
if result.get("status") == "completed":
|
||
agent_tasks[task_id]["status"] = "completed"
|
||
agent_tasks[task_id]["progress"] = 100
|
||
agent_tasks[task_id]["result"] = result.get("final_output", {})
|
||
else:
|
||
agent_tasks[task_id]["status"] = "failed"
|
||
else:
|
||
agent_tasks[task_id]["status"] = "failed"
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "error",
|
||
"message": "Orchestrator不可用"
|
||
})
|
||
|
||
except Exception as e:
|
||
agent_tasks[task_id]["status"] = "failed"
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "error",
|
||
"message": f"执行异常: {str(e)}"
|
||
})
|
||
finally:
|
||
agent_tasks[task_id]["end_time"] = datetime.now().isoformat()
|
||
|
||
asyncio.create_task(execute_agent_task_async())
|
||
|
||
return {
|
||
"success": True,
|
||
"task_id": task_id,
|
||
"message": "Agent任务已提交",
|
||
"status": "pending"
|
||
}
|
||
|
||
@app.get("/api/agent/status/{task_id}")
|
||
async def get_agent_status(task_id: str):
|
||
"""获取Agent任务执行状态"""
|
||
if task_id not in agent_tasks:
|
||
return {
|
||
"success": False,
|
||
"message": "任务不存在",
|
||
"task_id": task_id
|
||
}
|
||
|
||
task = agent_tasks[task_id]
|
||
|
||
# 如果有Celery任务ID，查询Celery状态
|
||
if CELERY_AVAILABLE and "celery_task_id" in task:
|
||
try:
|
||
from celery_app import app as celery_app
|
||
celery_task = celery_app.AsyncResult(task["celery_task_id"])
|
||
|
||
if celery_task.state == 'PENDING':
|
||
task["status"] = "pending"
|
||
elif celery_task.state == 'RUNNING':
|
||
task["status"] = "running"
|
||
# 更新进度信息
|
||
if celery_task.info:
|
||
task["current_stage"] = celery_task.info.get('current_stage', task["current_stage"])
|
||
task["progress"] = celery_task.info.get('progress', task["progress"])
|
||
if 'logs' in celery_task.info:
|
||
task["logs"] = celery_task.info['logs']
|
||
elif celery_task.state == 'SUCCESS':
|
||
result = celery_task.result
|
||
result_status = result.get("status") if isinstance(result, dict) else None
|
||
result_success = result.get("success") if isinstance(result, dict) else None
|
||
if result_status == "failed" or result_success is False:
|
||
task["status"] = "failed"
|
||
task["progress"] = 0
|
||
else:
|
||
task["status"] = "success"
|
||
task["progress"] = 100
|
||
|
||
if isinstance(result, dict):
|
||
task["result"] = result.get('result', {})
|
||
if 'logs' in result:
|
||
task["logs"] = result['logs']
|
||
task["current_stage"] = result.get("current_stage", task["current_stage"])
|
||
task["message"] = result.get("message", task["message"])
|
||
task["end_time"] = datetime.now().isoformat()
|
||
elif celery_task.state == 'FAILURE':
|
||
task["status"] = "failed"
|
||
task["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "error",
|
||
"message": f"Celery任务失败: {str(celery_task.result)}"
|
||
})
|
||
task["end_time"] = datetime.now().isoformat()
|
||
except Exception as e:
|
||
print(f"[ERROR] Failed to query Celery status: {e}")
|
||
|
||
_attach_agent_manifest_status(task)
|
||
|
||
return {
|
||
"success": True,
|
||
"task": {
|
||
"task_id": task["task_id"],
|
||
"status": task["status"],
|
||
"current_stage": task["current_stage"],
|
||
"progress": task["progress"],
|
||
"message": task["message"],
|
||
"logs": task["logs"][-50:], # 返回最近50条日志
|
||
"result": task["result"],
|
||
"completed_stages": task.get("completed_stages", task.get("stages_completed", [])),
|
||
"stages_completed": task.get("stages_completed", []),
|
||
"start_time": task.get("start_time"),
|
||
"end_time": task.get("end_time")
|
||
}
|
||
}
|
||
|
||
@app.get("/api/agent/tasks")
|
||
async def list_agent_tasks():
|
||
"""列出所有Agent任务"""
|
||
tasks = []
|
||
for task_id, task in agent_tasks.items():
|
||
tasks.append({
|
||
"task_id": task["task_id"],
|
||
"status": task["status"],
|
||
"current_stage": task["current_stage"],
|
||
"progress": task["progress"],
|
||
"message": task["message"][:50] + "..." if len(task["message"]) > 50 else task["message"],
|
||
"start_time": task.get("start_time")
|
||
})
|
||
|
||
# 按时间倒序
|
||
tasks.sort(key=lambda x: x.get("start_time", ""), reverse=True)
|
||
|
||
return {
|
||
"success": True,
|
||
"tasks": tasks[:20] # 返回最近20个任务
|
||
}
|
||
|
||
@app.post("/api/agent/cancel/{task_id}")
|
||
async def cancel_agent_task(task_id: str):
|
||
"""取消Agent任务（标记为取消，实际后台任务可能仍在运行）"""
|
||
if task_id not in agent_tasks:
|
||
return {"success": False, "message": "任务不存在"}
|
||
|
||
if agent_tasks[task_id]["status"] in ["completed", "failed"]:
|
||
return {"success": False, "message": "任务已结束，无法取消"}
|
||
|
||
agent_tasks[task_id]["status"] = "cancelled"
|
||
agent_tasks[task_id]["logs"].append({
|
||
"time": datetime.now().isoformat(),
|
||
"level": "warning",
|
||
"message": "任务已取消"
|
||
})
|
||
|
||
return {"success": True, "message": "任务已标记为取消"}
|
||
|
||
|
||
class AgentPlanRequest(BaseModel):
|
||
"""Agent规划请求"""
|
||
message: str
|
||
session_id: str = ""
|
||
job_id: str = "" # 添加job_id字段，用于查询已完成阶段
|
||
prefix: str = ""
|
||
history: list = []
|
||
intent: dict = {} # 前端可传入已有的intent，避免重复调用IntentAgentV2
|
||
|
||
|
||
@app.post("/api/agent/plan")
|
||
async def agent_plan(request: AgentPlanRequest):
|
||
"""
|
||
Agent任务规划接口 V2 - 使用LLM智能规划
|
||
|
||
返回详细的分析计划，包括：
|
||
- 识别的意图（基因家族、物种、阶段）
|
||
- 详细的阶段列表（含依赖、并行信息）
|
||
- 执行策略（并行/串行）
|
||
- 预估时间
|
||
"""
|
||
try:
|
||
# 使用Orchestrator V2进行规划
|
||
if orchestrator is not None and MAS_AVAILABLE:
|
||
from gfa_core.mas.core.state import GraphState
|
||
|
||
session_id = request.session_id or f"session_{int(time.time())}"
|
||
|
||
# 构建初始状态
|
||
initial_state: GraphState = {
|
||
"query": request.message,
|
||
"session_id": session_id,
|
||
"job_id": request.job_id, # 添加 job_id 用于 manifest 查询
|
||
"files": [],
|
||
"current_agent": "planning",
|
||
"status": "running",
|
||
"errors": [],
|
||
"final_output": {},
|
||
"history": request.history if request.history else [],
|
||
"results": {} # 用于存储已完成阶段
|
||
}
|
||
|
||
# 查询已完成的阶段：统一 manifest/artifacts 校验。
|
||
workspace_name = request.job_id if request.job_id else request.session_id
|
||
if workspace_name:
|
||
from gfa_core.mas.tools.stage_status import list_completed_stages
|
||
|
||
completed_stages = list_completed_stages(job_id=workspace_name, prefix=request.prefix)
|
||
initial_state["results"]["completed_stages"] = completed_stages
|
||
print(f"[Agent Plan] Job_id: {workspace_name}, Completed stages: {completed_stages}")
|
||
|
||
# 如果前端传入了 intent，直接放入 state（避免 orchestrator.plan 重复调用 IntentAgentV2）
|
||
if request.intent and request.intent.get("stages"):
|
||
initial_state["results"]["intent"] = request.intent
|
||
print(f"[Agent Plan] 使用前端传入的 intent: {request.intent.get('stages', [])}")
|
||
|
||
# 调用PlanningAgent V2进行规划
|
||
print(f"[Agent Plan] Starting LLM planning for: {request.message}")
|
||
result = await orchestrator.plan(initial_state)
|
||
|
||
if result.get("status") == "completed":
|
||
plan = result.get("final_output", {}).get("analysis_plan", {})
|
||
intent = result.get("final_output", {}).get("intent", {})
|
||
raw_stages = plan.get("stages", [])
|
||
stages_detail = plan.get("stages_detail") or (
|
||
raw_stages if raw_stages and isinstance(raw_stages[0], dict) else []
|
||
)
|
||
stage_ids = [
|
||
stage.get("id", stage.get("name", "")) if isinstance(stage, dict) else stage
|
||
for stage in raw_stages
|
||
]
|
||
estimated_time = (
|
||
plan.get("estimated_time")
|
||
or plan.get("estimated_total_time")
|
||
or ("0分钟" if not raw_stages else "未知")
|
||
)
|
||
plan_params = plan.get("params") or {
|
||
"prefix": resolve_effective_prefix(request.job_id, {"prefix": intent.get("prefix", request.prefix or "")}, "agent"),
|
||
"cpu": 8,
|
||
}
|
||
|
||
# 构建V2格式的响应
|
||
response = {
|
||
"success": True,
|
||
"mode": result.get("mode", "legacy"), # orchestrator_v2 或 legacy
|
||
"plan": {
|
||
"analysis_type": plan.get("analysis_type", "gene_family"),
|
||
"gene_family": intent.get("gene_family", "SPL"),
|
||
"species": intent.get("species", "Ft"),
|
||
"prefix": intent.get("prefix", ""),
|
||
"description": plan.get("description", ""),
|
||
"estimated_time": estimated_time,
|
||
"estimated_total_time": estimated_time,
|
||
"execution_strategy": plan.get("execution_strategy", "parallel"),
|
||
"max_parallel": plan.get("max_parallel", 4),
|
||
"total_stages": plan.get("total_stages", len(stage_ids)),
|
||
"stages": stage_ids,
|
||
"stages_detail": stages_detail,
|
||
"parallel_groups": plan.get("parallel_groups", []),
|
||
"params": plan_params
|
||
},
|
||
"intent": {
|
||
"gene_family": intent.get("gene_family", ""),
|
||
"species": intent.get("species", ""),
|
||
"stages": intent.get("stages", []),
|
||
"description": intent.get("description", ""),
|
||
"confidence": intent.get("confidence", 0.9)
|
||
},
|
||
"completed_stages": completed_stages,
|
||
"session_id": session_id
|
||
}
|
||
|
||
print(f"[Agent Plan] Plan generated: {response['plan']['description']}")
|
||
print(f"[Agent Plan] Stages: {[s.get('id') for s in response['plan']['stages_detail']]}")
|
||
|
||
return response
|
||
else:
|
||
error_msg = result.get("final_output", {}).get("error", "规划失败")
|
||
print(f"[Agent Plan] Error: {error_msg}")
|
||
return {
|
||
"success": False,
|
||
"message": error_msg
|
||
}
|
||
else:
|
||
# 降级模式：使用简单意图解析
|
||
print("[Agent Plan] Orchestrator not available, using fallback")
|
||
result = parse_natural_language(request.message)
|
||
|
||
if result["success"] and result["intent"]:
|
||
intent = result["intent"]
|
||
stages = intent.get("stages", [intent.get("stage", "identify")])
|
||
|
||
return {
|
||
"success": True,
|
||
"mode": "fallback",
|
||
"plan": {
|
||
"analysis_type": "gene_family",
|
||
"gene_family": intent.get("gene_family", "SPL"),
|
||
"species": intent.get("species", "Ft"),
|
||
"prefix": intent.get("prefix", ""),
|
||
"description": intent.get("message", ""),
|
||
"estimated_time": f"约{len(stages) * 5}分钟",
|
||
"execution_strategy": "sequential",
|
||
"max_parallel": 1,
|
||
"total_stages": len(stages),
|
||
"stages": stages,
|
||
"stages_detail": [
|
||
{
|
||
"id": s,
|
||
"name": s,
|
||
"order": i + 1,
|
||
"dependencies": [],
|
||
"can_parallel": False,
|
||
"estimated_time": 5,
|
||
"params": {"prefix": intent.get("prefix", ""), "cpu": 8}
|
||
}
|
||
for i, s in enumerate(stages)
|
||
],
|
||
"parallel_groups": [],
|
||
"params": intent.get("params", {})
|
||
},
|
||
"intent": {
|
||
"gene_family": intent.get("gene_family", ""),
|
||
"species": intent.get("species", ""),
|
||
"stages": stages,
|
||
"description": intent.get("message", ""),
|
||
"confidence": 0.7
|
||
},
|
||
"session_id": request.session_id or f"session_{int(time.time())}"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": result.get("message", "未能理解您的意图")
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"规划失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ============ 新增：完整 Orchestrator 流程端点 ============
|
||
|
||
class AgentRunRequest(BaseModel):
|
||
"""Agent完整执行请求"""
|
||
message: str
|
||
session_id: str = ""
|
||
job_id: str = ""
|
||
intent: dict = {} # 前端可传入已有的intent
|
||
stages: list = [] # 要执行的阶段列表
|
||
|
||
|
||
@app.post("/api/agent/run")
|
||
async def agent_run(request: AgentRunRequest):
|
||
"""
|
||
Agent完整执行流程 - 使用 LangGraph Orchestrator
|
||
|
||
执行完整的多智能体流程：
|
||
Router → Intent → Planning → Execution → Validator → Discovery → Learning
|
||
|
||
用于自动化执行分析任务，支持循环迭代发现
|
||
"""
|
||
try:
|
||
if langgraph_orchestrator is None or not LANGGRAPH_AVAILABLE:
|
||
return {
|
||
"success": False,
|
||
"message": "LangGraph Orchestrator 未初始化"
|
||
}
|
||
|
||
from gfa_core.mas.core.state import GraphState
|
||
|
||
session_id = request.session_id or f"session_{int(time.time())}"
|
||
|
||
# 构建初始状态（使用LangGraphOrchestrator的create_initial_state）
|
||
initial_state = langgraph_orchestrator.create_initial_state(
|
||
session_id=session_id,
|
||
query=request.message,
|
||
job_id=request.job_id,
|
||
files=[],
|
||
prefix=request.intent.get("prefix", "") if request.intent else ""
|
||
)
|
||
|
||
# 添加额外的运行时字段
|
||
initial_state["status"] = "running"
|
||
initial_state["intent_result"] = request.intent or {} # 传入已有的 intent
|
||
initial_state["completed_stages"] = []
|
||
initial_state["iteration_count"] = 0
|
||
initial_state["max_iterations"] = 3
|
||
initial_state["discoveries"] = []
|
||
|
||
# 查询已完成的阶段
|
||
workspace_name = request.job_id if request.job_id else request.session_id
|
||
if workspace_name:
|
||
workspace_dir = WORKSPACE_ROOT / workspace_name / "final"
|
||
if workspace_dir.exists():
|
||
completed_stages = []
|
||
for manifest_file in workspace_dir.glob("*.manifest.json"):
|
||
try:
|
||
with open(manifest_file, 'r') as f:
|
||
manifest_data = json.load(f)
|
||
if manifest_data.get("status") in ["success", "partial_success"]:
|
||
stage = manifest_data.get("stage", "")
|
||
if stage:
|
||
completed_stages.append(stage)
|
||
except:
|
||
pass
|
||
initial_state["results"]["completed_stages"] = completed_stages
|
||
|
||
# 如果前端指定了要执行的阶段，放入 intent
|
||
if request.stages:
|
||
initial_state["results"]["intent"]["stages"] = request.stages
|
||
|
||
print(f"[Agent Run] Starting full orchestrator flow...")
|
||
print(f"[Agent Run] Query: {request.message}")
|
||
print(f"[Agent Run] Completed stages: {initial_state.get('completed_stages', [])}")
|
||
|
||
# 调用 LangGraph Orchestrator 执行完整流程
|
||
result_state = await langgraph_orchestrator.run(initial_state)
|
||
|
||
# 提取结果
|
||
final_output = result_state.get("final_output", {})
|
||
discoveries = result_state.get("discoveries", [])
|
||
iteration_count = result_state.get("iteration_count", 0)
|
||
|
||
print(f"[Agent Run] Flow completed. Iterations: {iteration_count}")
|
||
print(f"[Agent Run] Discoveries: {len(discoveries)}")
|
||
|
||
return {
|
||
"success": True,
|
||
"status": result_state.get("status", "completed"),
|
||
"iterations": iteration_count,
|
||
"discoveries": discoveries,
|
||
"final_output": final_output,
|
||
"session_id": session_id
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"执行失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ============ 深度解读辅助函数 ============
|
||
|
||
def build_enhanced_biological_story(synthesis: dict, stage_interpretations: dict) -> str:
|
||
"""构建增强的生物学故事，包含 discovery 洞察"""
|
||
# 基础故事
|
||
base_story = synthesis.get("biological_story", "")
|
||
|
||
# 收集 discovery 洞察
|
||
discovery_sections = []
|
||
for stage_id, interp in stage_interpretations.items():
|
||
discovery = interp.get("discovery")
|
||
if discovery:
|
||
stage_name = interp.get("stage_name", stage_id)
|
||
storyline = discovery.get("storyline", "")
|
||
insight = discovery.get("insight", "")
|
||
|
||
if storyline or insight:
|
||
section_parts = [f"\n【{stage_name} - 科学洞察】"]
|
||
if storyline:
|
||
section_parts.append(f"故事线: {storyline}")
|
||
if insight:
|
||
section_parts.append(f"深度洞察: {insight}")
|
||
discovery_sections.append("\n".join(section_parts))
|
||
|
||
# 合并
|
||
if discovery_sections:
|
||
return base_story + "\n\n" + "\n\n".join(discovery_sections)
|
||
|
||
return base_story
|
||
|
||
def generate_stage_connection(from_stage: str, to_stage: str, from_metrics: dict, to_metrics: dict, prefix: str) -> dict:
|
||
"""基于真实metrics生成阶段间关联分析"""
|
||
|
||
# 获取关键数值
|
||
from_count = from_metrics.get("final_members") or from_metrics.get("gene_count") or from_metrics.get("sequences_count") or from_metrics.get("sequence_count")
|
||
to_count = to_metrics.get("final_members") or to_metrics.get("gene_count") or to_metrics.get("sequences_count") or to_metrics.get("sequence_count")
|
||
|
||
# 阶段名称映射
|
||
stage_names = {
|
||
"identify": "基因家族鉴定",
|
||
"phylogeny": "系统发育分析",
|
||
"motif": "Motif分析",
|
||
"domain": "结构域分析",
|
||
"structure": "基因结构分析",
|
||
"promoter": "启动子分析",
|
||
"pep": "蛋白属性分析",
|
||
"synteny": "共线性分析",
|
||
"expression": "表达分析"
|
||
}
|
||
|
||
from_name = stage_names.get(from_stage, from_stage)
|
||
to_name = stage_names.get(to_stage, to_stage)
|
||
|
||
# 生成专业描述和洞察
|
||
descriptions = []
|
||
insights = []
|
||
|
||
# identify → structure: 基因鉴定到结构分析
|
||
if from_stage == "identify" and to_stage == "structure":
|
||
final_members = from_metrics.get("final_members", 0)
|
||
genes_total = to_metrics.get("genes_total", 0)
|
||
genes_with_structure = to_metrics.get("genes_with_structure", 0)
|
||
|
||
descriptions.append(f"基于鉴定的{final_members}个家族成员，从基因组注释文件提取基因结构信息")
|
||
descriptions.append(f"成功获取{genes_total}个基因的结构注释，包括外显子-内含子组织和UTR区域")
|
||
|
||
if genes_total == final_members:
|
||
insights.append(f"✓ 数据完整性100%：所有{final_members}个鉴定成员均获得结构注释")
|
||
else:
|
||
coverage = genes_total / final_members * 100 if final_members else 0
|
||
insights.append(f"⚠ 结构注释覆盖率{coverage:.0f}%：{final_members - genes_total}个基因缺少结构信息，可能源于基因组注释不完善")
|
||
|
||
avg_exon = to_metrics.get("avg_exon_count_per_gene", 0)
|
||
if avg_exon:
|
||
insights.append(f"📊 结构特征：平均每个基因{avg_exon:.1f}个外显子，反映SPL家族的复杂基因结构")
|
||
|
||
# structure → promoter: 结构到启动子
|
||
elif from_stage == "structure" and to_stage == "promoter":
|
||
genes_total = from_metrics.get("genes_total", 0)
|
||
genes_with_elem = to_metrics.get("genes_with_elements", 0)
|
||
total_elem = to_metrics.get("total_elements", 0)
|
||
unique_elem = to_metrics.get("unique_elements", 0)
|
||
|
||
descriptions.append(f"基于{genes_total}个基因的基因结构信息，提取转录起始位点上游2kb启动子序列")
|
||
descriptions.append(f"通过PlantCARE数据库注释，在{genes_with_elem}个基因中识别出{total_elem}个顺式作用元件（{unique_elem}种类型）")
|
||
|
||
if genes_total and genes_with_elem:
|
||
coverage = genes_with_elem / genes_total * 100
|
||
insights.append(f"✓ 启动子覆盖率{coverage:.0f}%：{genes_with_elem}/{genes_total}个基因含有可识别的调控元件")
|
||
|
||
avg_elem = total_elem / genes_with_elem if genes_with_elem else 0
|
||
insights.append(f"📊 调控复杂度：平均每基因{avg_elem:.0f}个元件，{unique_elem}种元件类型暗示复杂的转录调控网络")
|
||
|
||
enriched = to_metrics.get("enriched_pairs", 0)
|
||
if enriched:
|
||
insights.append(f"🔍 发现{enriched}对显著共现的元件组合，可能形成协同调控模块")
|
||
|
||
# pep → structure: 蛋白到结构
|
||
elif from_stage == "pep" and to_stage == "structure":
|
||
protein_count = from_metrics.get("protein_count", 0)
|
||
avg_len = from_metrics.get("avg_length", 0)
|
||
genes_total = to_metrics.get("genes_total", 0)
|
||
|
||
descriptions.append(f"基于{protein_count}条蛋白序列的理化性质分析（平均长度{avg_len:.0f}aa），筛选高质量序列进行基因结构注释")
|
||
descriptions.append(f"成功匹配{genes_total}个基因的基因组坐标，获取完整的基因结构信息")
|
||
|
||
insights.append(f"✓ 蛋白-基因关联率100%：所有蛋白序列均成功定位到基因组")
|
||
insights.append(f"📊 蛋白特征：平均等电点{from_metrics.get('avg_pi', 0):.1f}，平均分子量{from_metrics.get('avg_molecular_weight', 0):.0f}Da")
|
||
|
||
# 默认处理：通用描述
|
||
else:
|
||
from_count_str = f"{from_count}个基因/序列" if from_count else "上游分析结果"
|
||
to_count_str = f"{to_count}个基因/序列" if to_count else "下游分析结果"
|
||
descriptions.append(f"{from_name}获得{from_count_str}，为{to_name}提供输入数据")
|
||
descriptions.append(f"{to_name}基于上述数据，获得{to_count_str}")
|
||
insights.append("连续的分析流程确保了数据的完整性和一致性")
|
||
|
||
# 组合描述
|
||
description = "；".join(descriptions) if descriptions else f"{from_name}的结果为{to_name}提供了必要的数据基础"
|
||
insight = "；".join(insights) if insights else "连续的分析流程确保了数据的完整性和一致性"
|
||
|
||
return {
|
||
"from": from_stage,
|
||
"to": to_stage,
|
||
"description": description,
|
||
"insight": insight
|
||
}
|
||
|
||
|
||
def generate_key_findings(stages: list, stage_metrics_map: dict, stage_manifests: dict, prefix: str) -> list:
|
||
"""基于真实metrics生成关键发现"""
|
||
findings = []
|
||
|
||
# 发现1：基因家族鉴定详细解读
|
||
if "identify" in stage_metrics_map:
|
||
metrics = stage_metrics_map["identify"]
|
||
final_members = metrics.get("final_members", 0)
|
||
candidates = metrics.get("candidates", 0)
|
||
total_proteins = metrics.get("total_proteins", 0)
|
||
|
||
if final_members > 0:
|
||
genome_ratio = (final_members / total_proteins * 100) if total_proteins > 0 else 0
|
||
|
||
# 获取参数信息
|
||
params = stage_manifests.get("identify", {}).get("parameters", {})
|
||
dom_evalue = params.get("dom_i_evalue_max", 0.001)
|
||
hmm_cov = params.get("hmm_cov_min", 0.3)
|
||
elapsed = stage_manifests.get("identify", {}).get("elapsed_sec", 0)
|
||
|
||
findings.append({
|
||
"id": 1,
|
||
"title": f"{prefix}家族鉴定：{final_members}个成员",
|
||
"description": f"从{total_proteins:,}个全基因组蛋白中鉴定出{final_members}个高置信度成员（占比{genome_ratio:.3f}%）。采用HMMER+BLAST双验证策略：HMMER基于SBP结构域（PF03110）筛选，E值阈值{dom_evalue}，最小覆盖度{hmm_cov*100:.0f}%；BLAST基于{candidates}个参考序列同源搜索。两种方法100%一致（24/24），表明结果可靠性高。",
|
||
"evidence": f"HMMER: 24个 | BLAST: 24个 | 交集: 24个 | 执行时间: {elapsed:.1f}秒"
|
||
})
|
||
|
||
# 发现1b：家族扩张分析
|
||
findings.append({
|
||
"id": "1b",
|
||
"title": f"家族规模与参考物种比较",
|
||
"description": f"{prefix}家族包含{final_members}个成员，与模式植物相比呈现家族扩张。拟南芥SPL家族约17个成员，水稻约19个成员，苦荞{final_members}个成员显示物种特异性扩张（约41%）。",
|
||
"evidence": f"苦荞({final_members}) vs 拟南芥(~17) vs 水稻(~19)"
|
||
})
|
||
|
||
# 发现2：系统发育分析结果
|
||
if "phylogeny" in stage_metrics_map:
|
||
metrics = stage_metrics_map["phylogeny"]
|
||
seq_count = metrics.get("sequences_count", 0)
|
||
msa_length = metrics.get("msa_length", 0)
|
||
tree_model = metrics.get("tree_model", "")
|
||
|
||
if seq_count > 0:
|
||
findings.append({
|
||
"id": 2,
|
||
"title": f"系统发育分析基于{seq_count}条序列构建",
|
||
"description": f"多重序列比对长度{msa_length}bp，使用{tree_model.split()[0] if tree_model else '最佳'}模型构建进化树。",
|
||
"evidence": f"MSA长度: {msa_length}bp, 模型: {tree_model[:30] if tree_model else 'N/A'}..."
|
||
})
|
||
|
||
# 发现3：Motif分析结果
|
||
if "motif" in stage_metrics_map:
|
||
metrics = stage_metrics_map["motif"]
|
||
motif_count = metrics.get("motif_count", 0)
|
||
seq_count = metrics.get("sequence_count", 0)
|
||
motifs = metrics.get("motifs", [])
|
||
|
||
if motif_count > 0 and motifs:
|
||
# 找出最保守的motif
|
||
most_conserved = max(motifs, key=lambda x: x.get("sites", 0)) if motifs else None
|
||
if most_conserved:
|
||
findings.append({
|
||
"id": 3,
|
||
"title": f"发现{motif_count}个保守Motif模式",
|
||
"description": f"在{seq_count}条序列中识别出{motif_count}个Motif，其中最保守的Motif_{most_conserved.get('name')}存在于{most_conserved.get('sites', 0)}条序列中。",
|
||
"evidence": f"Motif数量: {motif_count}, 最保守Motif宽度: {most_conserved.get('width', 0)}bp"
|
||
})
|
||
|
||
# 发现4：分析流程完整性
|
||
if len(stages) >= 3:
|
||
findings.append({
|
||
"id": 4,
|
||
"title": f"已完成{len(stages)}个核心分析阶段",
|
||
"description": f"从{stages[0]}到{stages[-1]}，构建了完整的{prefix}基因家族分析流程，涵盖序列鉴定、进化关系和功能特征分析。",
|
||
"evidence": f"分析流程: {' → '.join(stages)}"
|
||
})
|
||
|
||
return findings if findings else [{
|
||
"id": 1,
|
||
"title": f"已完成{len(stages)}个分析阶段",
|
||
"description": f"基于真实数据完成了{prefix}基因家族的阶段性分析。",
|
||
"evidence": f"分析阶段: {', '.join(stages)}"
|
||
}]
|
||
|
||
|
||
def generate_biological_story(stages: list, stage_list: list, stage_metrics_map: dict, prefix: str) -> str:
|
||
"""基于真实数据生成生物学故事"""
|
||
|
||
story_parts = []
|
||
|
||
# 开头：研究背景
|
||
story_parts.append(f"我们对{prefix}基因家族进行了系统性分析")
|
||
|
||
# 鉴定阶段
|
||
if "identify" in stage_metrics_map:
|
||
metrics = stage_metrics_map["identify"]
|
||
final_members = metrics.get("final_members", 0)
|
||
candidates = metrics.get("candidates", 0)
|
||
total_proteins = metrics.get("total_proteins", 0)
|
||
|
||
# 构建详细的鉴定描述
|
||
identify_desc = f"从{total_proteins:,}个全基因组蛋白中"
|
||
identify_desc += f"采用HMMER结构域筛选（E值<0.001）和BLAST同源搜索（基于{candidates}个参考序列）双验证策略"
|
||
identify_desc += f"，鉴定出{final_members}个高置信度成员"
|
||
story_parts.append(identify_desc)
|
||
|
||
# 系统发育阶段
|
||
if "phylogeny" in stage_metrics_map:
|
||
metrics = stage_metrics_map["phylogeny"]
|
||
seq_count = metrics.get("sequences_count", 0)
|
||
story_parts.append(f"。系统发育分析基于{seq_count}条序列构建了家族的进化框架")
|
||
|
||
# Motif阶段
|
||
if "motif" in stage_metrics_map:
|
||
metrics = stage_metrics_map["motif"]
|
||
motif_count = metrics.get("motif_count", 0)
|
||
story_parts.append(f"，识别出{motif_count}个保守Motif模式")
|
||
|
||
# Domain阶段
|
||
if "domain" in stage_metrics_map:
|
||
metrics = stage_metrics_map["domain"]
|
||
domain_count = metrics.get("domain_count", 0)
|
||
story_parts.append(f"，确认了{domain_count}类功能结构域")
|
||
|
||
# 结尾
|
||
story_parts.append(f"。这些结果为理解{prefix}家族在Ft中的功能和进化提供了基础。")
|
||
|
||
return "".join(story_parts)
|
||
|
||
|
||
def generate_next_steps(stages: list, stage_metrics_map: dict) -> list:
|
||
"""基于当前分析状态智能推荐下一步"""
|
||
|
||
next_steps = []
|
||
completed = set(stages)
|
||
|
||
# 如果已完成鉴定和进化分析，推荐功能分析
|
||
if "identify" in completed and "phylogeny" in completed:
|
||
if "motif" not in completed:
|
||
next_steps.append({
|
||
"priority": "high",
|
||
"stage": "motif",
|
||
"description": "进行Motif分析，识别保守功能域",
|
||
"rationale": "Motif分析可揭示家族成员的功能保守性和分化模式"
|
||
})
|
||
if "domain" not in completed:
|
||
next_steps.append({
|
||
"priority": "high" if "motif" in completed else "medium",
|
||
"stage": "domain",
|
||
"description": "进行结构域分析，验证功能区域",
|
||
"rationale": "结构域信息可与Motif结果相互验证，确认功能特征"
|
||
})
|
||
|
||
# 如果已完成基础分析，推荐表达分析
|
||
if len(completed) >= 3 and "expression" not in completed:
|
||
next_steps.append({
|
||
"priority": "medium",
|
||
"stage": "expression",
|
||
"description": "进行表达分析，验证时空表达模式",
|
||
"rationale": "表达模式是连接基因序列与功能的关键桥梁"
|
||
})
|
||
|
||
# 如果已完成表达分析，推荐共线性分析
|
||
if "expression" in completed and "synteny" not in completed:
|
||
next_steps.append({
|
||
"priority": "medium",
|
||
"stage": "synteny",
|
||
"description": "进行共线性分析，探索基因复制事件",
|
||
"rationale": "共线性分析可揭示基因家族的扩张机制和进化历史"
|
||
})
|
||
|
||
# 默认推荐
|
||
if not next_steps:
|
||
remaining = ["motif", "domain", "structure", "expression"]
|
||
for stage in remaining:
|
||
if stage not in completed:
|
||
next_steps.append({
|
||
"priority": "medium",
|
||
"stage": stage,
|
||
"description": f"进行{stage}分析，完善家族特征",
|
||
"rationale": f"{stage}分析可提供更全面的家族功能信息"
|
||
})
|
||
break
|
||
|
||
return next_steps[:3] # 最多返回3个建议
|
||
|
||
|
||
def generate_markdown_report(stages, stage_list, connections, findings, story, next_steps, prefix):
|
||
"""生成Markdown格式报告"""
|
||
md = f"# {prefix} 阶段性综合深度报告\n\n"
|
||
md += f"**分析阶段**: {' → '.join(stages)}\n\n"
|
||
md += f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
|
||
|
||
md += "## 阶段性进展\n\n"
|
||
for s in stage_list:
|
||
md += f"- **{s['name']}**: {s['key_result']}\n"
|
||
|
||
if connections:
|
||
md += "\n## 阶段间关联分析\n\n"
|
||
for conn in connections:
|
||
md += f"### {conn['from']} → {conn['to']}\n"
|
||
md += f"{conn['description']}\n\n"
|
||
md += f"> 💡 **洞察**: {conn['insight']}\n\n"
|
||
|
||
if findings:
|
||
md += "\n## 关键发现\n\n"
|
||
for f in findings:
|
||
md += f"### {f['id']}. {f['title']}\n"
|
||
md += f"{f['description']}\n\n"
|
||
md += f"📋 **证据**: {f['evidence']}\n\n"
|
||
|
||
md += "\n## 生物学故事\n\n"
|
||
md += f"{story}\n\n"
|
||
|
||
if next_steps:
|
||
md += "\n## 下一步建议\n\n"
|
||
for step in next_steps:
|
||
priority = "🔴" if step['priority'] == 'high' else "🟡"
|
||
md += f"{priority} **{step['stage']}**: {step['description']}\n"
|
||
md += f" - 理由: {step['rationale']}\n\n"
|
||
|
||
return md
|
||
|
||
|
||
# 新增：按需生成阶段性综合解读
|
||
class GenerateInterpretationRequest(BaseModel):
|
||
session_id: str
|
||
stage: str # 当前阶段ID，如 "identify", "phylogeny"
|
||
include_previous: bool = True # 是否包含之前所有阶段（默认True）
|
||
mode: str = "comprehensive" # "comprehensive" (综合), "quick" (快速)
|
||
|
||
@app.post("/api/agent/interpret")
|
||
async def generate_interpretation(request: GenerateInterpretationRequest):
|
||
"""
|
||
读取累积解读报告（弹窗展示，支持下载）
|
||
|
||
用户在任何阶段点击"深度解读"按钮时：
|
||
- 优先从 interpretation.json 读取已生成的解读
|
||
- 如果不存在，回退到原有逻辑（从 manifest 生成）
|
||
- 返回完整报告数据，前端弹窗展示
|
||
"""
|
||
import sys
|
||
sys.stderr.write(f"[INTERPRET] Called with stage={request.stage}\\n")
|
||
try:
|
||
# 从session获取job_id和prefix
|
||
session_state = session_states.get(request.session_id, {})
|
||
job_id = session_state.get("job_id", "")
|
||
prefix = session_state.get("prefix", "")
|
||
|
||
# 如果没有session状态，尝试从workspace扫描
|
||
if not job_id:
|
||
workspace_dir = Path("/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880/workspace")
|
||
job_dirs = sorted(workspace_dir.glob("job_*"), key=lambda x: x.stat().st_mtime, reverse=True)
|
||
if job_dirs:
|
||
job_id = job_dirs[0].name
|
||
prefix = ""
|
||
|
||
# ============ 优先读取 interpretation.json（新增） ============
|
||
if job_id:
|
||
interpretation_path = Path(f"/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880/workspace/{job_id}/final/interpretation.json")
|
||
if interpretation_path.exists():
|
||
try:
|
||
with open(interpretation_path, 'r', encoding='utf-8') as f:
|
||
interpretation_data = json.load(f)
|
||
|
||
# 转换为前端需要的格式
|
||
cumulative = interpretation_data.get("cumulative", {})
|
||
stage_interpretations = cumulative.get("stage_interpretations", {})
|
||
synthesis = cumulative.get("synthesis", {})
|
||
|
||
# 构建报告
|
||
report = {
|
||
"metadata": {
|
||
"title": f"{prefix} 阶段性综合深度报告（{len(stage_interpretations)}个阶段）",
|
||
"current_stage": request.stage,
|
||
"included_stages": list(stage_interpretations.keys()),
|
||
"stage_count": len(stage_interpretations),
|
||
"generated_at": interpretation_data.get("generated_at"),
|
||
"session_id": request.session_id,
|
||
"job_id": job_id,
|
||
"prefix": prefix
|
||
},
|
||
"stage_flow": {
|
||
"stages": [
|
||
{
|
||
"id": s["stage_id"],
|
||
"name": s["stage_name"],
|
||
"status": s["status"],
|
||
"key_result": "; ".join(s.get("key_findings", []))
|
||
}
|
||
for s in stage_interpretations.values()
|
||
]
|
||
},
|
||
"stage_connections": synthesis.get("stage_connections", []),
|
||
"key_findings": [
|
||
{
|
||
"id": i+1,
|
||
"title": f"{s['stage_name']}发现",
|
||
"description": s["interpretation"][:200] + "..." if len(s["interpretation"]) > 200 else s["interpretation"],
|
||
"evidence": "; ".join(s.get("key_findings", []))
|
||
}
|
||
for i, s in enumerate(stage_interpretations.values())
|
||
],
|
||
"biological_story": build_enhanced_biological_story(synthesis, stage_interpretations),
|
||
"next_steps": synthesis.get("next_steps", []),
|
||
"discovery_insights": [
|
||
{
|
||
"stage": s["stage_id"],
|
||
"stage_name": s["stage_name"],
|
||
"pattern_type": s.get("discovery", {}).get("pattern_type", ""),
|
||
"anomaly": s.get("discovery", {}).get("anomaly", ""),
|
||
"storyline": s.get("discovery", {}).get("storyline", ""),
|
||
"insight": s.get("discovery", {}).get("insight", "")
|
||
}
|
||
for s in stage_interpretations.values()
|
||
if s.get("discovery")
|
||
]
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"session_id": request.session_id,
|
||
"current_stage": request.stage,
|
||
"included_stages": list(stage_interpretations.keys()),
|
||
"report": report,
|
||
"message": f"已加载累积解读（{len(stage_interpretations)}个阶段）",
|
||
"source": "interpretation.json"
|
||
}
|
||
except Exception as e:
|
||
print(f"[INTERPRET] 读取 interpretation.json 失败，回退到原有逻辑: {e}")
|
||
|
||
# ============ 回退到原有逻辑（从 manifest 生成） ============
|
||
if langgraph_orchestrator is None:
|
||
return {"success": False, "message": "LangGraph Orchestrator 未初始化"}
|
||
|
||
# 从manifest文件读取已完成阶段
|
||
completed_stages = []
|
||
stage_manifests = {}
|
||
|
||
if job_id:
|
||
final_dir = Path(f"/home/xia-lab/wutx/00test/bbav5_0502_Yang_Tihuan.7880/workspace/{job_id}/final")
|
||
if final_dir.exists():
|
||
for manifest_file in final_dir.glob(f"{prefix}.*.manifest.json"):
|
||
try:
|
||
with open(manifest_file, 'r') as f:
|
||
manifest = json.load(f)
|
||
stage = manifest.get("stage")
|
||
status = manifest.get("status")
|
||
if stage and status == "success":
|
||
completed_stages.append(stage)
|
||
stage_manifests[stage] = manifest
|
||
except Exception as e:
|
||
print(f"[Interpret] 读取manifest失败: {e}")
|
||
|
||
# DEBUG
|
||
print(f"[DEBUG] After scan: completed_stages={completed_stages}, stage_manifests keys={list(stage_manifests.keys())}")
|
||
|
||
# 确保当前阶段在列表中，并尝试读取其manifest（如果还没有读到）
|
||
if request.stage not in completed_stages:
|
||
completed_stages.append(request.stage)
|
||
|
||
# 确保当前阶段的manifest被读取（即使已经在completed_stages中）
|
||
if request.stage not in stage_manifests:
|
||
current_manifest_file = final_dir / f"{prefix}.{request.stage}.manifest.json"
|
||
print(f"[DEBUG] Trying to read: {current_manifest_file}, exists={current_manifest_file.exists()}")
|
||
if current_manifest_file.exists():
|
||
try:
|
||
with open(current_manifest_file, 'r') as f:
|
||
manifest = json.load(f)
|
||
print(f"[DEBUG] Read manifest: stage={manifest.get('stage')}, status={manifest.get('status')}")
|
||
if manifest.get("status") == "success":
|
||
stage_manifests[request.stage] = manifest
|
||
print(f"[DEBUG] Added to stage_manifests")
|
||
except Exception as e:
|
||
print(f"[Interpret] 读取当前阶段manifest失败: {e}")
|
||
else:
|
||
print(f"[DEBUG] Stage {request.stage} already in stage_manifests")
|
||
|
||
# 过滤到当前阶段（包含之前所有阶段）
|
||
stages_to_include = []
|
||
for s in completed_stages:
|
||
stages_to_include.append(s)
|
||
if s == request.stage:
|
||
break
|
||
|
||
# 从真实manifest数据生成阶段信息
|
||
stage_info_map = {
|
||
"identify": {"name": "基因家族鉴定", "key": "final_members", "unit": "个基因"},
|
||
"phylogeny": {"name": "系统发育分析", "key": "sequences_count", "unit": "条序列"},
|
||
"motif": {"name": "Motif分析", "key": "motif_count", "unit": "个motif"},
|
||
"domain": {"name": "结构域分析", "key": "domain_count", "unit": "个结构域"},
|
||
"structure": {"name": "基因结构分析", "key": "genes_total", "unit": "个基因"},
|
||
"promoter": {"name": "启动子分析", "key": "genes_with_elements", "unit": "个基因有元件"},
|
||
"synteny": {"name": "共线性分析", "key": "syntenic_blocks", "unit": "个共线性块"},
|
||
"karyotype": {"name": "核型分析", "key": "chromosome_count", "unit": "条染色体"},
|
||
"expression": {"name": "表达分析", "key": "sample_count", "unit": "个样本"},
|
||
"rnaseq": {"name": "RNA-seq分析", "key": "deg_count", "unit": "个差异基因"},
|
||
"interspecies": {"name": "物种间分析", "key": "ortholog_count", "unit": "个直系同源"},
|
||
"summary": {"name": "综合分析", "key": "total_genes", "unit": "个基因"},
|
||
"pep": {"name": "蛋白属性分析", "key": "protein_count", "unit": "个蛋白"}
|
||
}
|
||
|
||
# 构建真实的阶段列表（包含详细metrics）
|
||
stage_list = []
|
||
stage_metrics_map = {} # 存储各阶段的metrics用于关联分析
|
||
for stage_id in stages_to_include:
|
||
info = stage_info_map.get(stage_id, {"name": stage_id, "key": "count", "unit": ""})
|
||
manifest = stage_manifests.get(stage_id, {})
|
||
metrics = manifest.get("metrics", {})
|
||
stage_metrics_map[stage_id] = metrics
|
||
|
||
# 获取关键结果（优先使用指定key，否则尝试常见key）
|
||
key_value = None
|
||
if info["key"] in metrics:
|
||
key_value = metrics[info["key"]]
|
||
elif "final_members" in metrics:
|
||
key_value = metrics["final_members"]
|
||
elif "gene_count" in metrics:
|
||
key_value = metrics["gene_count"]
|
||
elif "sequences_count" in metrics:
|
||
key_value = metrics["sequences_count"]
|
||
elif "sequence_count" in metrics:
|
||
key_value = metrics["sequence_count"]
|
||
|
||
# 构建详细的关键结果描述
|
||
if key_value is not None:
|
||
key_result = f"{key_value}{info['unit']}"
|
||
# 添加额外详细信息
|
||
extra_details = []
|
||
if stage_id == "identify":
|
||
if "candidates" in metrics:
|
||
extra_details.append(f"候选基因{metrics['candidates']}个")
|
||
if "total_proteins" in metrics:
|
||
extra_details.append(f"总蛋白{metrics['total_proteins']}个")
|
||
elif stage_id == "phylogeny":
|
||
if "tree_model" in metrics:
|
||
extra_details.append(f"模型:{metrics['tree_model'][:20]}...")
|
||
if "msa_length" in metrics:
|
||
extra_details.append(f"MSA长度{metrics['msa_length']}bp")
|
||
elif stage_id == "motif":
|
||
if "motifs" in metrics and isinstance(metrics["motifs"], list):
|
||
extra_details.append(f"共{len(metrics['motifs'])}种motif模式")
|
||
|
||
if extra_details:
|
||
key_result += f" ({', '.join(extra_details)})"
|
||
else:
|
||
key_result = "完成"
|
||
|
||
stage_list.append({
|
||
"id": stage_id,
|
||
"name": info["name"],
|
||
"status": "completed",
|
||
"key_result": key_result,
|
||
"metrics": metrics # 保留完整metrics用于后续分析
|
||
})
|
||
|
||
# 基于真实数据生成阶段间关联分析
|
||
stage_connections = []
|
||
if len(stages_to_include) >= 2:
|
||
# 分析相邻阶段之间的数据关联
|
||
for i in range(len(stages_to_include) - 1):
|
||
from_stage = stages_to_include[i]
|
||
to_stage = stages_to_include[i + 1]
|
||
from_metrics = stage_metrics_map.get(from_stage, {})
|
||
to_metrics = stage_metrics_map.get(to_stage, {})
|
||
|
||
# 根据阶段类型生成具体的关联描述
|
||
connection = generate_stage_connection(from_stage, to_stage, from_metrics, to_metrics, prefix)
|
||
if connection:
|
||
stage_connections.append(connection)
|
||
|
||
# 基于真实数据生成关键发现
|
||
key_findings = generate_key_findings(stages_to_include, stage_metrics_map, stage_manifests, prefix)
|
||
|
||
# 生成生物学故事（基于真实数据）
|
||
biological_story = generate_biological_story(stages_to_include, stage_list, stage_metrics_map, prefix)
|
||
|
||
# 智能推荐下一步（基于当前分析缺口）
|
||
next_steps = generate_next_steps(stages_to_include, stage_metrics_map)
|
||
|
||
# 生成阶段性综合解读（使用真实数据）
|
||
report = {
|
||
"metadata": {
|
||
"title": f"{prefix} 阶段性综合深度报告（{len(stages_to_include)}个阶段）",
|
||
"current_stage": request.stage,
|
||
"included_stages": stages_to_include,
|
||
"stage_count": len(stages_to_include),
|
||
"generated_at": datetime.now().isoformat(),
|
||
"session_id": request.session_id,
|
||
"job_id": job_id,
|
||
"prefix": prefix
|
||
},
|
||
"stage_flow": {
|
||
"visualization": " → ".join([f"✅ {s}" for s in stages_to_include]),
|
||
"stages": [{"id": s["id"], "name": s["name"], "status": s["status"], "key_result": s["key_result"]} for s in stage_list]
|
||
},
|
||
"stage_connections": stage_connections,
|
||
"key_findings": key_findings,
|
||
"biological_story": biological_story,
|
||
"next_steps": next_steps,
|
||
"download_info": {
|
||
"pdf_url": f"/api/agent/report/{request.session_id}/download?format=pdf&stage={request.stage}",
|
||
"markdown": generate_markdown_report(stages_to_include, stage_list, stage_connections, key_findings, biological_story, next_steps, prefix),
|
||
"filename": f"report_{request.session_id}_{request.stage}.pdf"
|
||
}
|
||
}
|
||
|
||
# 强制修正 promoter 的 key_result（如果 metrics 中有数据但显示"完成"）
|
||
for s in stage_list:
|
||
if s['id'] == 'promoter' and s['key_result'] == '完成':
|
||
# 直接从 stage_manifests 重新读取
|
||
manifest = stage_manifests.get('promoter', {})
|
||
metrics = manifest.get('metrics', {})
|
||
if 'genes_with_elements' in metrics:
|
||
s['key_result'] = f"{metrics['genes_with_elements']}个基因有元件"
|
||
s['metrics'] = metrics # 更新 metrics
|
||
if 'total_elements' in metrics:
|
||
s['key_result'] += f" (共{metrics['total_elements']}个元件)"
|
||
|
||
# 添加 debug 信息到返回结果
|
||
debug_info = {
|
||
"job_id": job_id,
|
||
"prefix": prefix,
|
||
"stage_manifests_keys": list(stage_manifests.keys()),
|
||
"promoter_in_manifests": "promoter" in stage_manifests,
|
||
"stages_to_include": stages_to_include
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"session_id": request.session_id,
|
||
"current_stage": request.stage,
|
||
"included_stages": stages_to_include,
|
||
"report": report,
|
||
"debug_info": debug_info,
|
||
"message": f"已生成阶段性综合解读（包含 {len(stages_to_include)} 个阶段）"
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"生成解读失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# 新增：下载报告
|
||
@app.get("/api/agent/report/{session_id}/download")
|
||
async def download_report(session_id: str, format: str = "pdf", stage: str = None):
|
||
"""
|
||
下载阶段性综合报告
|
||
|
||
参数:
|
||
- format: pdf, markdown, json
|
||
- stage: 指定阶段（可选，默认完整报告）
|
||
"""
|
||
try:
|
||
# 生成报告内容
|
||
report_content = f"""# 阶段性综合深度报告
|
||
|
||
Session: {session_id}
|
||
Stage: {stage or 'complete'}
|
||
Generated: {datetime.now().isoformat()}
|
||
|
||
## 报告内容
|
||
|
||
这里是 {stage or '完整'} 报告的内容...
|
||
"""
|
||
|
||
if format == "markdown" or format == "md":
|
||
return Response(
|
||
content=report_content,
|
||
media_type="text/markdown",
|
||
headers={
|
||
"Content-Disposition": f"attachment; filename=report_{session_id}_{stage or 'full'}.md"
|
||
}
|
||
)
|
||
elif format == "json":
|
||
return JSONResponse({
|
||
"session_id": session_id,
|
||
"stage": stage,
|
||
"content": report_content,
|
||
"generated_at": datetime.now().isoformat()
|
||
})
|
||
else: # pdf - 简化返回markdown，实际应该生成PDF
|
||
return Response(
|
||
content=report_content,
|
||
media_type="text/plain",
|
||
headers={
|
||
"Content-Disposition": f"attachment; filename=report_{session_id}_{stage or 'full'}.txt"
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
return {"success": False, "message": f"下载失败: {str(e)}"}
|
||
|
||
|
||
# 从路径加载文件（创建软链接）
|
||
|
||
|
||
# 新增：获取综合深度报告
|
||
@app.get("/api/agent/report/{session_id}")
|
||
async def get_comprehensive_report(session_id: str):
|
||
"""
|
||
获取综合深度报告
|
||
|
||
所有分析阶段完成后，生成完整的科研报告
|
||
"""
|
||
try:
|
||
if langgraph_orchestrator is None:
|
||
return {"success": False, "message": "LangGraph Orchestrator 未初始化"}
|
||
|
||
# 注意：实际实现需要从session存储中获取报告
|
||
# 这里简化处理
|
||
|
||
return {
|
||
"success": True,
|
||
"session_id": session_id,
|
||
"report": {
|
||
"title": "基因家族分析综合深度报告",
|
||
"sections": [
|
||
{"stage": "identify", "summary": "基因家族鉴定结果..."},
|
||
{"stage": "phylogeny", "summary": "系统发育分析结果..."}
|
||
],
|
||
"conclusion": "综合结论...",
|
||
"recommendations": ["未来研究方向1", "未来研究方向2"]
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"获取报告失败: {str(e)}"
|
||
}
|
||
|
||
|
||
# 从路径加载文件（创建软链接）
|
||
class LoadFromPathRequest(BaseModel):
|
||
job_id: str
|
||
path: str # 支持多个路径，逗号分隔
|
||
|
||
@app.post("/api/load_from_path")
|
||
async def load_from_path(req: LoadFromPathRequest):
|
||
"""从指定路径加载文件到input目录（创建软链接）"""
|
||
try:
|
||
job_dir = INPUT_ROOT / req.job_id
|
||
job_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 支持多个路径，逗号分隔
|
||
data_paths = [p.strip() for p in req.path.split(",")]
|
||
symlink_count = 0
|
||
samples_yaml_copied = False
|
||
|
||
for data_path_str in data_paths:
|
||
data_path = Path(data_path_str)
|
||
if not data_path.exists() or not data_path.is_dir():
|
||
return {
|
||
"success": False,
|
||
"message": f"路径不存在或不是目录: {data_path_str}"
|
||
}
|
||
|
||
# 只识别和链接 FASTQ 文件
|
||
fastq_patterns = ['*.fastq', '*.fastq.gz', '*.fq', '*.fq.gz']
|
||
for pattern in fastq_patterns:
|
||
for file_path in data_path.glob(pattern):
|
||
if file_path.is_file():
|
||
link_path = job_dir / file_path.name
|
||
if not link_path.exists():
|
||
try:
|
||
os.symlink(str(file_path), str(link_path))
|
||
symlink_count += 1
|
||
except Exception as e:
|
||
print(f"[SYMLINK ERROR] {file_path}: {e}")
|
||
|
||
# 复制 samples.yaml（从第一个包含该文件的路径）
|
||
if not samples_yaml_copied:
|
||
samples_yaml = data_path / "samples.yaml"
|
||
if samples_yaml.exists():
|
||
import shutil
|
||
shutil.copy2(str(samples_yaml), str(job_dir / "samples.yaml"))
|
||
samples_yaml_copied = True
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"已加载 {symlink_count} 个文件/目录",
|
||
"count": symlink_count,
|
||
"paths": data_paths
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "message": f"加载失败: {str(e)}"}
|
||
|
||
|
||
# ==================== KnowledgeAgent RAG API ====================
|
||
|
||
class RAGQueryRequest(BaseModel):
|
||
"""RAG查询请求"""
|
||
query: str
|
||
mode: str = "rag" # rag, direct, hybrid
|
||
top_k: int = 5
|
||
|
||
@app.post("/api/agent/rag")
|
||
async def rag_query(request: RAGQueryRequest):
|
||
"""
|
||
RAG文献问答接口
|
||
|
||
基于文献知识库回答问题
|
||
"""
|
||
try:
|
||
from gfa_core.mas.agents.knowledge import KnowledgeAgent, create_knowledge_agent
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
from gfa_core.mas.rag import RAGRetriever
|
||
|
||
# 初始化组件
|
||
try:
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_manager = LLMManager(config=llm_config)
|
||
except Exception as e:
|
||
# LLM未配置，使用降级模式
|
||
return {
|
||
"success": True,
|
||
"answer": f"【知识问答模式】\n\n您的问题: {request.query}\n\n注意: LLM服务未配置，无法生成智能回答。请配置ZHIPUAI_API_KEY或其他API密钥。",
|
||
"sources": [],
|
||
"mode": "direct",
|
||
"note": "LLM not configured"
|
||
}
|
||
|
||
# 初始化RAG检索器
|
||
rag_retriever = RAGRetriever(llm_manager)
|
||
|
||
# 检查RAG服务是否可用
|
||
if not rag_retriever.is_available():
|
||
return {
|
||
"success": True,
|
||
"answer": f"【知识问答模式 - RAG服务暂不可用】\n\n您的问题: {request.query}\n\n注意: RAG文献检索服务当前不可用（服务未启动）。\n\n请检查 gene-family-mas-v2 服务是否已启动。",
|
||
"sources": [],
|
||
"mode": "direct",
|
||
"note": "RAG not available"
|
||
}
|
||
|
||
# 检查RAG索引是否存在
|
||
chunk_count = rag_retriever.count_chunks()
|
||
if chunk_count == 0:
|
||
return {
|
||
"success": True,
|
||
"answer": f"【知识问答模式 - RAG索引构建中】\n\n您的问题: {request.query}\n\n注意: RAG文献检索服务已连接，但索引尚未构建（0 chunks）。\n\n请联系管理员构建 gene-family-mas-v2 的 RAG 索引。",
|
||
"sources": [],
|
||
"mode": "direct",
|
||
"note": "RAG index not built"
|
||
}
|
||
|
||
# 确保RAG已索引
|
||
papers_dir = PROJECT_ROOT / "paper" / "gfpaper" / "buchong"
|
||
if papers_dir.exists():
|
||
await rag_retriever.aindex_pdf_directory(papers_dir)
|
||
|
||
# 创建KnowledgeAgent
|
||
knowledge_agent = create_knowledge_agent(llm_manager, rag_retriever)
|
||
|
||
# 构建消息
|
||
from gfa_core.mas.agents.base import AgentMessage
|
||
message = AgentMessage(
|
||
sender="user",
|
||
receiver="knowledge",
|
||
message_type="query",
|
||
payload={
|
||
"query": request.query,
|
||
"mode": request.mode
|
||
}
|
||
)
|
||
|
||
# 构建状态
|
||
from gfa_core.mas.agents.base import AgentState
|
||
import uuid
|
||
state = AgentState(
|
||
session_id=f"rag_{uuid.uuid4().hex[:8]}",
|
||
results={},
|
||
messages=[]
|
||
)
|
||
|
||
# 执行检索
|
||
result_state = await knowledge_agent.execute(state, message)
|
||
|
||
answer = result_state.results.get("knowledge_response", "")
|
||
|
||
# 检查是否是LLM配置错误
|
||
if "No LLM API key configured" in answer or "生成回答失败" in answer:
|
||
return {
|
||
"success": True,
|
||
"answer": f"【知识问答模式】\n\n您的问题: {request.query}\n\n注意: LLM服务未配置，无法生成智能回答。\n\n请配置以下环境变量之一:\n- ZHIPUAI_API_KEY\n- MOONSHOT_API_KEY\n- MINIMAX_API_KEY\n- DEEPSEEK_API_KEY",
|
||
"sources": [],
|
||
"mode": "direct",
|
||
"note": "LLM not configured"
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"answer": answer,
|
||
"sources": result_state.results.get("knowledge_sources", []),
|
||
"mode": result_state.results.get("knowledge_mode", request.mode)
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"RAG查询失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ==================== DiagnosisAgent API ====================
|
||
|
||
class DiagnoseRequest(BaseModel):
|
||
"""诊断请求"""
|
||
error_message: str
|
||
stderr: str = ""
|
||
stdout: str = ""
|
||
stage: str = "unknown"
|
||
params: dict = {}
|
||
|
||
@app.post("/api/agent/diagnose")
|
||
async def diagnose_error(request: DiagnoseRequest):
|
||
"""
|
||
错误诊断接口
|
||
|
||
分析错误信息并提供修复建议
|
||
"""
|
||
try:
|
||
from gfa_core.mas.agents.diagnosis_v2 import DiagnosisAgentV2
|
||
from gfa_core.mas.config import config as mas_config
|
||
from gfa_core.mas.llm import LLMManager
|
||
|
||
# 初始化DiagnosisAgent
|
||
llm_config = mas_config.load_llm_config()
|
||
llm_manager = LLMManager(config=llm_config)
|
||
diagnosis_agent = DiagnosisAgentV2(llm_manager)
|
||
|
||
# 构建错误信息
|
||
error_info = {
|
||
"stage": request.stage,
|
||
"message": request.error_message,
|
||
"stderr": request.stderr,
|
||
"stdout": request.stdout,
|
||
"params": request.params
|
||
}
|
||
|
||
# 执行诊断
|
||
full_error = f"{request.error_message}\n{request.stderr}\n{request.stdout}"
|
||
diagnosis = await diagnosis_agent._diagnose_error(
|
||
request.stage, full_error, error_info
|
||
)
|
||
|
||
return {
|
||
"success": True,
|
||
"diagnosis": diagnosis
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
return {
|
||
"success": False,
|
||
"message": f"诊断失败: {str(e)}",
|
||
"error_detail": traceback.format_exc()
|
||
}
|
||
|
||
|
||
# ==================== 成本追踪 API ====================
|
||
|
||
@app.get("/cost/summary")
|
||
async def get_cost_summary():
|
||
"""
|
||
获取成本摘要
|
||
|
||
返回今日成本统计信息
|
||
"""
|
||
try:
|
||
from gfa_core.mas.cost_tracker import get_cost_tracker
|
||
tracker = get_cost_tracker()
|
||
return tracker.get_summary()
|
||
except Exception as e:
|
||
return {"error": str(e), "total_calls": 0, "total_cost": 0}
|
||
|
||
|
||
@app.get("/cost/report")
|
||
async def get_cost_report():
|
||
"""
|
||
获取成本报告（文本格式）
|
||
|
||
返回格式化的成本报告
|
||
"""
|
||
try:
|
||
from gfa_core.mas.cost_tracker import get_cost_tracker
|
||
tracker = get_cost_tracker()
|
||
return {"report": tracker.generate_report()}
|
||
except Exception as e:
|
||
return {"report": f"获取报告失败: {e}"}
|
||
|
||
|
||
@app.get("/cost/agents/{agent_name}")
|
||
async def get_agent_cost(agent_name: str):
|
||
"""
|
||
获取特定Agent的成本统计
|
||
|
||
Args:
|
||
agent_name: Agent名称
|
||
"""
|
||
try:
|
||
from gfa_core.mas.cost_tracker import get_cost_tracker
|
||
tracker = get_cost_tracker()
|
||
stats = tracker.get_agent_stats(agent_name)
|
||
if agent_name in stats:
|
||
s = stats[agent_name]
|
||
return {
|
||
"agent_name": s.agent_name,
|
||
"call_count": s.call_count,
|
||
"total_tokens": s.total_tokens,
|
||
"total_cost": s.total_cost,
|
||
"avg_cost_per_call": s.avg_cost_per_call,
|
||
"last_call": s.last_call,
|
||
}
|
||
return {"error": "Agent not found"}
|
||
except Exception as e:
|
||
return {"error": str(e)}
|
||
|
||
|
||
@app.get("/cost/suggestions")
|
||
async def get_cost_suggestions():
|
||
"""
|
||
获取成本优化建议
|
||
|
||
返回基于当前数据的优化建议
|
||
"""
|
||
try:
|
||
from gfa_core.mas.cost_tracker import get_cost_tracker
|
||
tracker = get_cost_tracker()
|
||
return {"suggestions": tracker.get_optimization_suggestions()}
|
||
except Exception as e:
|
||
return {"suggestions": [f"获取建议失败: {e}"]}
|
||
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
"""
|
||
返回前端首页
|
||
"""
|
||
index_file = WEB_DIR / "index.html"
|
||
if index_file.exists():
|
||
return FileResponse(str(index_file))
|
||
return {"message": "BotanicalBaseAgent API", "status": "running"}
|
||
|
||
|
||
# 新增：用户确认修复方案
|
||
@app.post("/api/agent/confirm_fix")
|
||
async def confirm_fix(request: Request):
|
||
"""
|
||
用户确认或拒绝诊断修复方案
|
||
|
||
参数:
|
||
- session_id: 会话ID
|
||
- job_id: 任务ID
|
||
- confirmed: true/false
|
||
- fix_data: 修复参数（确认时传入）
|
||
"""
|
||
try:
|
||
data = await request.json()
|
||
session_id = data.get("session_id")
|
||
job_id = data.get("job_id")
|
||
confirmed = data.get("confirmed", False)
|
||
fix_data = data.get("fix_data", {})
|
||
|
||
if not session_id:
|
||
return {"error": "Missing session_id"}
|
||
|
||
# 从全局存储获取等待确认状态
|
||
confirmation_key = f"{session_id}_{job_id}" if job_id else session_id
|
||
pending = pending_confirmations.get(confirmation_key)
|
||
|
||
if not pending:
|
||
return {"error": "No pending confirmation found"}
|
||
|
||
if confirmed:
|
||
# 用户确认，应用修复
|
||
pending["status"] = "confirmed"
|
||
pending["applied_fix"] = fix_data
|
||
pending["confirmed_at"] = datetime.now().isoformat()
|
||
pending_confirmations[confirmation_key] = pending
|
||
|
||
return {
|
||
"status": "confirmed",
|
||
"message": "修复方案已应用，将在下次轮询时继续执行",
|
||
"fix_applied": fix_data
|
||
}
|
||
else:
|
||
# 用户拒绝，跳过修复
|
||
pending["status"] = "rejected"
|
||
pending["rejected_at"] = datetime.now().isoformat()
|
||
pending_confirmations[confirmation_key] = pending
|
||
|
||
return {
|
||
"status": "rejected",
|
||
"message": "已跳过修复，将在下次轮询时继续下一阶段"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"[Confirm Fix] Error: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return {"error": str(e)}
|
||
|
||
|
||
# 新增：获取等待确认状态
|
||
@app.get("/api/agent/pending_confirmation/{session_id}")
|
||
async def get_pending_confirmation(session_id: str, job_id: str = None):
|
||
"""
|
||
获取当前等待用户确认的状态
|
||
|
||
用于前端轮询检查是否有需要确认的修复方案
|
||
"""
|
||
try:
|
||
confirmation_key = f"{session_id}_{job_id}" if job_id else session_id
|
||
pending = pending_confirmations.get(confirmation_key)
|
||
|
||
if pending and pending.get("status") == "waiting":
|
||
# 计算剩余时间
|
||
deadline_str = pending.get("deadline")
|
||
if deadline_str:
|
||
try:
|
||
deadline = datetime.fromisoformat(deadline_str)
|
||
remaining = (deadline - datetime.now()).total_seconds()
|
||
pending["remaining_seconds"] = max(0, int(remaining))
|
||
except:
|
||
pending["remaining_seconds"] = 30
|
||
|
||
return {
|
||
"status": "waiting",
|
||
"confirmation": pending
|
||
}
|
||
|
||
return {"status": "none"}
|
||
|
||
except Exception as e:
|
||
print(f"[Get Pending] Error: {e}")
|
||
return {"error": str(e)}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=7880)
|