#!/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 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)) # 导入Celery任务 try: from tasks import execute_agent_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 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" app = FastAPI(title="BotanicalBaseAgent API") # 挂载静态文件目录 WEB_DIR = Path(__file__).parent app.mount("/static", StaticFiles(directory=str(WEB_DIR)), name="static") # 添加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: {llm_config.get('openai_api_key', 'NOT SET')[:20]}...") 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", # 其他分析 "pep", "ppi", "kaks", "expression", # 启动子分析（外部服务） "promoter", # 基因组准备 "genome_prepare", # RNA-seq全流程 "rnaseq_qc", "rnaseq_align", "rnaseq_quant", "rnaseq_de", "rnaseq_enrich" } 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" geoId: str = "" # RNA-seq参数 rnaseqMode: str = "upload" speciesName: str = "" # RNA-seq QC 参数 rnaseqDataMode: str = "upload" rnaseqDataPath: str = "" # RNA-seq Enrich参数 topPathways: int = 20 # 富集分析热图通路数量 # 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 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分钟，请耐心等待..." # 执行 result_msg, manifest = run_stage( stage=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), # 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"), ) # 恢复工作目录 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"], "rnaseq_qc": ["rnaseq_qc"], "rnaseq_align": ["rnaseq_align"], "rnaseq_quant": ["rnaseq_quant"], "rnaseq_de": ["rnaseq_de"], "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.replace("Ft.SPL.", "") 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.replace("Ft.SPL.", "") 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) # 整理 final 目录结构 (figures/reports/tables) try: sys.path.insert(0, str(Path(__file__).parent.parent.parent / "scripts")) from reorganize_workspace import reorganize_workspace as reorganize reorganize(job_id) except Exception as e: print(f"[WARN] Failed to reorganize workspace: {e}") # 严格验证：检查输出 manifest 是否存在且状态为 success expected_manifest = final_dir / f"{params['prefix']}.{stage}.manifest.json" manifest_valid = False manifest_status = None if expected_manifest.exists(): try: with open(expected_manifest, 'r', encoding='utf-8') as f: manifest_data = json.load(f) manifest_status = manifest_data.get("status") manifest_valid = ( manifest_status == "success" and # manifest 状态必须是 success "❌" not in result_msg # 结果消息不包含失败标记 ) except Exception as e: print(f"[WARN] Failed to read manifest {expected_manifest}: {e}") if not manifest_valid: # manifest 验证失败 tasks[task_id]["status"] = "failed" tasks[task_id]["message"] = f"❌ {stage} 失败：manifest 状态为 {manifest_status}。{result_msg}" tasks[task_id]["end_time"] = datetime.now().isoformat() tasks[task_id]["result"] = { "success": False, "error": f"Manifest validation failed: expected {expected_manifest}, status={manifest_status}", "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 } 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 } @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 current_job_id = f"job_{int(time.time()*1000)}" 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.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 # ===== 新增：检查是否已完成 ===== try: import json final_dir = WORKSPACE_ROOT / job_id / "final" if final_dir.exists(): manifest_files = list(final_dir.glob("*.manifest.json")) completed_stages = [] for mf in manifest_files: try: with open(mf) as f: data = json.load(f) if data.get("status") in ["success", "partial_success"]: completed_stages.append(data.get("stage")) except: pass print(f"[API DEBUG] Completed stages from manifest: {completed_stages}") if req.stage in completed_stages: 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 } except Exception as e: print(f"[API DEBUG] Error checking completed stages: {e}") # 继续执行，不阻断 # 对于耗时长的阶段，使用异步模式 if req.stage in ASYNC_STAGES: task_id = str(uuid.uuid4()) # 创建任务记录 tasks[task_id] = { "id": task_id, "stage": req.stage, "job_id": job_id, "status": "pending", "message": f"任务 {req.stage} 已提交", "created_at": datetime.now().isoformat(), "result": None } # 提交后台任务 params = { "prefix": req.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), # 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'), } 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 } # 其他阶段保持同步执行 job_dir = INPUT_ROOT / job_id workspace_dir = WORKSPACE_ROOT / job_id job_dir.mkdir(parents=True, exist_ok=True) workspace_dir.mkdir(parents=True, exist_ok=True) # RNA-seq QC: 处理软链接模式 if req.stage == "rnaseq_qc" and req.rnaseqDataMode == "symlink" and req.rnaseqDataPath: data_paths = [p.strip() for p in req.rnaseqDataPath.split(",")] symlink_count = 0 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, "async": False, "message": f"❌ 数据路径不存在或不是目录: {data_path_str}", "job_id": job_id } # 只创建 FASTQ 文件的软链接 fastq_patterns = ['*.fastq', '*.fastq.gz', '*.fq', '*.fq.gz'] for pattern in fastq_patterns: for fastq_file in data_path.glob(pattern): if fastq_file.is_file(): link_path = job_dir / fastq_file.name if not link_path.exists(): try: os.symlink(str(fastq_file), str(link_path)) symlink_count += 1 print(f"[SYMLINK] Created: {link_path} -> {fastq_file}") except Exception as e: print(f"[SYMLINK ERROR] {fastq_file}: {e}") # 复制 samples.yaml（从第一个包含该文件的路径） samples_yaml = data_path / "samples.yaml" if samples_yaml.exists() and not (job_dir / "samples.yaml").exists(): import shutil shutil.copy2(str(samples_yaml), str(job_dir / "samples.yaml")) print(f"[COPY] samples.yaml from {data_path} -> {job_dir}") print(f"[SYMLINK] Total {symlink_count} files linked from {len(data_paths)} directories") from apps.web.pipeline_backend import run_stage, JobInfo 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) # DEBUG: 打印传入的RNA-seq参数 print(f"[STAGE CALL DEBUG] rnaseq_mode={getattr(req, 'rnaseqMode', 'NOT_SET')}, species_name={getattr(req, 'speciesName', 'NOT_SET')}") try: result_msg, manifest = run_stage( stage=req.stage, job=job, prefix=req.prefix, cpu=req.cpu, required_pfams=req.requiredPfams or None, ref_species=req.refSpecies or None, id_evalue=req.idEvalue, id_coverage=req.idCoverage, blast_evalue=req.blastEvalue, phy_mafft=req.mafftParams, phy_iqtree=req.iqtreeParams, motif_nmotifs=req.nmotifs, motif_minw=req.minw, motif_maxw=req.maxw, dom_evalue=req.domEvalue, dom_coverage=req.domCoverage, id_transform=req.idTransform, geo_id=getattr(req, 'geoId', ''), # RNA-seq参数 rnaseq_mode=getattr(req, 'rnaseqMode', 'upload'), species_name=getattr(req, 'speciesName', ''), # RNA-seq Enrich参数 top_pathways=getattr(req, 'topPathways', 20), # PPI参数 ppi_species=getattr(req, 'ppiSpecies', 'ath'), ppi_score=getattr(req, 'ppiScore', 700), # KaKs参数 kaks_neutral_rate=getattr(req, 'kaksNeutralRate', 7.08e-9), # Genome Prepare参数 genome_prepare_type=getattr(req, 'genomePrepareType', 'auto'), ) os.chdir(original_cwd) # 读取日志 log_file = Path(workspace_dir) / "logs" log_files = sorted(log_file.glob("*.log"), key=lambda p: p.stat().st_mtime, reverse=True) log_content = "" if log_files: log_content = log_files[0].read_text()[-5000:] # 收集结果 images = [] files = [] final_dir = workspace_dir / "final" if final_dir.exists(): for ext in ['.png', '.jpg', '.jpeg', '.svg']: for f in final_dir.glob(f"*{ext}"): images.append(f"/workspace/{job_id}/final/{f.name}") for f in final_dir.iterdir(): if f.is_file(): files.append({ "name": f.name, "size": f.stat().st_size, "path": f"/workspace/{job_id}/final/{f.name}" }) return { "success": True, "async": False, "message": result_msg, "log": log_content, "images": images, "files": files, "job_id": job_id } except Exception as e: os.chdir(original_cwd) return { "success": False, "error": str(e) } @app.get("/api/task/{task_id}") async def get_task_status(task_id: str): """查询异步任务状态""" if task_id not in tasks: raise HTTPException(status_code=404, detail="Task not found") task = tasks[task_id] 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") } @app.get("/api/stage-status") async def get_stage_status(job_id: str, stage: str): """检查阶段是否已完成（通过manifest文件）""" from pathlib import Path import json workspace_dir = WORKSPACE_ROOT / job_id final_dir = workspace_dir / "final" # 查找manifest文件 manifest_files = list(final_dir.glob(f"*.{stage}.manifest.json")) if not manifest_files: return {"completed": False, "stage": stage} try: with open(manifest_files[0], 'r') as f: manifest = json.load(f) # manifest 中的 status 是 "success"，但前端使用 "completed" is_completed = manifest.get("status") == "success" return { "completed": is_completed, "stage": stage, "status": "completed" if is_completed else manifest.get("status", "unknown"), "manifest": manifest } except Exception: return {"completed": False, "stage": stage} @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(): """获取当前任务结果""" global current_job_id if not current_job_id: return {"identify": {"images": [], "files": []}} workspace_dir = WORKSPACE_ROOT / current_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}, "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"], "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", "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分析"], "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 "Ft.SPL" # 收集所有匹配的阶段（不再用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) # 处理 "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 prefix: str = "" # 前端设置的 prefix，用于解析物种和家族 history: list = [] @app.post("/api/agent/chat") async def agent_chat(request: AgentChatRequest): """Agent智能对话接口 - 只做意图识别，不执行分析 修改说明： - 原版本调用 langgraph_orchestrator.run() 会执行完整分析流程 - 新版本只做 Router + Intent 检测，返回意图结果 - 前端根据意图结果决定下一步： - FAQ: 显示回答 - Analysis: 调用 /api/agent/plan + 弹窗确认 + /api/agent/execute """ try: # 优先使用 LangGraph Orchestrator（新版本） if langgraph_orchestrator is not None and LANGGRAPH_AVAILABLE: 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 # 创建初始状态 initial_state = langgraph_orchestrator.create_initial_state( session_id=session_id, query=request.message, job_id=request.job_id, files=[], prefix=request.prefix # 传递前端设置的 prefix ) # 添加 completed_stages 到初始状态 initial_state["completed_stages"] = completed_stages # 【修改】只调用 Router 进行意图识别，不执行后续流程 print(f"[Agent Chat] 只做意图识别，不执行分析: {request.message[:50]}...") router_result = await langgraph_orchestrator._router_node(initial_state) # 获取意图结果 intent_result = router_result.get("intent_result", {}) mode = router_result.get("mode", "faq") print(f"[Agent Chat] 意图识别结果: mode={mode}, gene_family={intent_result.get('gene_family', '未知')}") # 返回意图识别结果（不执行分析） if mode == "analysis": # 分析任务：返回意图信息，前端需要调用 /api/agent/plan 获取详细计划 return { "success": True, "intent": { "stage": intent_result.get("stage", ""), "taskType": "analysis", "gene_family": intent_result.get("gene_family", ""), "species": intent_result.get("species", ""), "params": intent_result, "confidence": intent_result.get("confidence", 0.8), "message": f"检测到分析任务: {intent_result.get('gene_family', '未知')} 基因家族分析" }, "model_used": request.model, "session_id": session_id, "mode": "intent_only", # 标记：只做了意图识别 "current_agent": "router" } else: # FAQ 模式：返回意图信息，前端调用 RAG return { "success": True, "intent": { "stage": None, "taskType": "faq", "params": {}, "confidence": intent_result.get("confidence", 0.8), "message": intent_result.get("message", "检测到FAQ查询") }, "model_used": request.model, "session_id": session_id, "mode": "faq", "current_agent": "router" } # 降级：使用原 Orchestrator（保留兼容）- 也只做意图识别 elif orchestrator is not None and MAS_AVAILABLE: # 使用意图解析器进行意图识别（不执行分析） print(f"[Agent Chat - Fallback] 使用意图解析器只做意图识别...") session_id = request.session_id or f"session_{int(time.time())}" # 使用 intent_parser 的正则匹配（已修复） try: from gfa_core.mas.utils.intent_parser import parse_gene_family, parse_intent_from_query gene_family = parse_gene_family(request.message) intent_result = parse_intent_from_query(request.message) mode = "analysis" if gene_family else "faq" print(f"[Agent Chat - Fallback] 意图识别结果: mode={mode}, gene_family={gene_family}") if mode == "analysis": return { "success": True, "intent": { "stage": intent_result.get("stage", "identify"), "taskType": "analysis", "gene_family": gene_family, "species": intent_result.get("species", "Ft"), "params": intent_result, "confidence": 0.85, "message": f"检测到分析任务: {gene_family} 基因家族分析" }, "model_used": request.model, "session_id": session_id, "mode": "intent_only", "current_agent": "router" } else: return { "success": True, "intent": { "stage": None, "taskType": "faq", "params": {}, "confidence": 0.85, "message": "检测到FAQ查询" }, "model_used": request.model, "session_id": session_id, "mode": "faq", "current_agent": "router" } except Exception as intent_err: print(f"[Agent Chat - Fallback] 意图解析错误: {intent_err}") return { "success": False, "message": f"意图识别失败: {str(intent_err)}", "errors": [str(intent_err)] } else: # 降级：使用简单意图解析 print("⚠️ Orchestrator不可用，使用降级方案") result = parse_natural_language(request.message) if result["success"] and result["intent"]: intent = result["intent"] intent["model"] = request.model intent["taskType"] = "complex_analysis" intent["confidence"] = 0.85 return { "success": True, "intent": intent, "model_used": request.model, "session_id": request.session_id or f"session_{int(time.time())}", "mode": "simple" # 标记使用简单方案 } else: return { "success": False, "message": result.get("message", "未能理解您的意图"), "model_used": request.model, "mode": "simple" } except Exception as e: import traceback error_detail = traceback.format_exc() print(f"❌ Agent处理失败: {str(e)}\n{error_detail}") return { "success": False, "message": f"Agent处理失败: {str(e)}", "model_used": request.model, "error_detail": error_detail } # RNA-seq分析相关 import subprocess # RNA-seq任务存储 rnaseq_tasks: Dict[str, dict] = {} class RnaseqRequest(BaseModel): stage: str # align, quant, de, enrich sample: str # 样本名 params: dict = {} class RnaseqJobRequest(BaseModel): sample_names: List[str] group_config: dict = {} # {sample: group} @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") # 脚本路径 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") 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") 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 model: str = "kimi" history: list = [] auto_execute: bool = True # 是否自动执行分析 prefix: str = "" # 前端传递的 prefix，格式如 "Ft.SPL" 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, "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 # 传递 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" # 更新全局 current_job_id，确保 /api/results 能返回正确结果 global current_job_id if request.job_id: current_job_id = request.job_id logger.info(f"[Agent] 更新 current_job_id: {current_job_id}") if orchestrator is not None and MAS_AVAILABLE: from gfa_core.mas.core.state import GraphState # 设置进度为运行中状态 agent_tasks[task_id]["progress"] = 50 agent_tasks[task_id]["current_stage"] = "执行中" 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 [], "user_input": {"prefix": request.prefix} # 传递 prefix } 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 状态 sync_celery_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"], "start_time": task.get("start_time"), "end_time": task.get("end_time"), "celery_task_id": task.get("celery_task_id", "") } } def sync_celery_status(task: dict) -> dict: """同步 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': task["status"] = "completed" task["progress"] = 100 result = celery_task.result if result: task["result"] = result.get('result', {}) if 'logs' in result: task["logs"] = result['logs'] 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}") return task @app.get("/api/agent/tasks") async def list_agent_tasks(): """列出所有Agent任务""" tasks = [] for task_id, task in agent_tasks.items(): # 同步 Celery 状态 sync_celery_status(task) 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 = "" history: list = [] @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, "files": [], "current_agent": "planning", "status": "running", "errors": [], "final_output": {}, "history": request.history if request.history else [], "results": {} # 用于存储已完成阶段 } # 查询已完成的阶段（包括 success 和 partial_success） if request.session_id: workspace_dir = WORKSPACE_ROOT / request.session_id / "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) status = manifest_data.get("status", "") # 接受 success 和 partial_success 作为已完成状态 if 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 print(f"[Agent Plan] Completed stages: {completed_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", {}) # 构建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", "Ft.SPL"), "description": plan.get("description", ""), "estimated_time": plan.get("estimated_time", "未知"), "execution_strategy": plan.get("execution_strategy", "parallel"), "max_parallel": plan.get("max_parallel", 4), "total_stages": plan.get("total_stages", 0), "stages": plan.get("stages", []), "stages_detail": plan.get("stages_detail", []), "parallel_groups": plan.get("parallel_groups", []), "params": plan.get("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) }, "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", "Ft.SPL"), "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", "Ft.SPL"), "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() } # 从路径加载文件（创建软链接） 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_v2 import KnowledgeAgentV2, 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已索引 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"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7868)