#!/usr/bin/env python3 """Run full PlantGeneBench tasks against system, rule, and DeepSeek providers.""" from __future__ import annotations import argparse import json import os import re import sys import time import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any GENE_FAMILY_MAS_ROOT = Path("/home/xia-lab/gene-family-mas-v2") DEFAULT_DEEPSEEK_ENV = GENE_FAMILY_MAS_ROOT / ".env.v15_deepseek" STAGE_CATALOG = [ "identify", "phylogeny", "motif", "domain", "structure", "promoter", "synteny", "interspecies_synteny", "karyotype", "combo", "ppi", "expression", "pep", "rnaseq_qc", "rnaseq_align", "rnaseq_quant", "rnaseq_de", "rnaseq_enrich", "genome_prepare_v2", "kaks_v2", "ortholog", "cis_element_http", "fetch_expression", "tree_plot", "ppi_v2", "ppi_v2_with_repro", "summary", ] STRUCTURED_KEYS = [ "analysis_type", "intent", "entities", "target_stages", "execution_plan", "parameters", "artifacts", "evidence", "support_status", "boundary_action", "can_execute", "cannot_execute_reason", "answer_text", ] def load_jsonl(path: Path) -> list[dict[str, Any]]: rows = [] with path.open("r", encoding="utf-8") as handle: for lineno, line in enumerate(handle, start=1): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError as exc: raise SystemExit(f"{path}:{lineno}: invalid JSON: {exc}") from exc return rows def dump_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n") def normalize_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, list): return [str(v).strip() for v in value if str(v).strip()] return [item for item in re.split(r"[,;，；\s]+", str(value)) if item] def normalize_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def load_env_file(path: Path) -> dict[str, str]: values: dict[str, str] = {} if not path.exists(): return values for line in path.read_text(encoding="utf-8-sig").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue if stripped.startswith("export "): stripped = stripped[len("export "):].strip() key, value = stripped.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key: values[key] = value os.environ.setdefault(key, value) return values def extract_stages(plan: dict[str, Any], intent: dict[str, Any]) -> list[str]: raw = plan.get("stages") or plan.get("target_stages") or plan.get("execution_plan") or intent.get("stages") or intent.get("target_stages") or [] stages: list[str] = [] for item in raw: if isinstance(item, dict): value = item.get("id") or item.get("stage") or item.get("stage_id") or item.get("name") else: value = item if value: stages.append(str(value)) return list(dict.fromkeys(stages)) def extract_dependencies(plan: dict[str, Any]) -> dict[str, list[str]]: deps: dict[str, list[str]] = {} for detail in plan.get("stages_detail") or []: if not isinstance(detail, dict): continue stage = detail.get("id") or detail.get("stage") or detail.get("stage_id") or detail.get("name") if stage: deps[str(stage)] = normalize_list(detail.get("dependencies")) return deps def flatten_params(plan: dict[str, Any], intent: dict[str, Any]) -> dict[str, Any]: params: dict[str, Any] = {} for source in (intent, plan): for key in ("prefix", "gene_family", "species", "scopeMode"): if source.get(key) not in (None, ""): params[key] = source.get(key) nested = source.get("params") or source.get("parameters") if isinstance(nested, dict): params.update(nested) for detail in plan.get("stages_detail") or []: if isinstance(detail, dict) and isinstance(detail.get("params"), dict): for key, value in detail["params"].items(): params.setdefault(key, value) if "gene_family" in params and "geneFamily" not in params: params["geneFamily"] = params["gene_family"] return params def classify_backend_status(raw: dict[str, Any]) -> str: if raw.get("error"): return "http_or_json_error" if raw.get("llm_json_valid") is False and raw.get("mode") == "deepseek_json": return "llm_json_invalid" if raw.get("success") is not False: return "success" text = " ".join(str(raw.get(k) or "") for k in ("message", "error_detail", "detail")) if "缺少" in text or "missing" in text.lower(): return "missing_input_blocked" if "规划失败" in text or "invalid literal" in text: return "backend_exception" return "backend_success_false" def normalize_raw(raw: dict[str, Any], fallback: dict[str, Any] | None = None) -> dict[str, Any]: fallback = fallback or {} plan = raw.get("plan") if isinstance(raw.get("plan"), dict) else {} intent = raw.get("intent") if isinstance(raw.get("intent"), dict) else {} artifacts = raw.get("artifacts") if isinstance(raw.get("artifacts"), dict) else plan.get("artifacts") if isinstance(plan.get("artifacts"), dict) else fallback.get("artifacts", {}) answer_text = str(raw.get("answer") or raw.get("message") or plan.get("description") or plan.get("answer_text") or intent.get("description") or fallback.get("answer_text") or "") mode = str(raw.get("mode") or fallback.get("mode") or "") allow_structural_fallback = not (mode == "deepseek_text" or (mode == "deepseek_json" and raw.get("llm_json_valid") is False)) fallback_stages = fallback.get("stages", []) if allow_structural_fallback else [] fallback_plan = fallback.get("execution_plan", []) if allow_structural_fallback else [] fallback_params = fallback.get("params", {}) if allow_structural_fallback else {} fallback_entities = fallback.get("entities", {}) if allow_structural_fallback else {} prediction = { "success": bool(raw.get("success", False)), "backend_status": classify_backend_status(raw), "mode": mode, "analysis_type": str(plan.get("analysis_type") or intent.get("analysis_type") or (fallback.get("analysis_type") if allow_structural_fallback else mode) or ""), "intent": str(intent.get("intent") or plan.get("intent") or (fallback.get("intent") if allow_structural_fallback else "") or ""), "entities": normalize_dict(intent.get("entities") or plan.get("entities") or fallback_entities), "stages": extract_stages(plan, intent) or fallback_stages, "execution_plan": normalize_list(plan.get("execution_plan")) or extract_stages(plan, intent) or fallback_plan, "params": flatten_params(plan, intent) or fallback_params, "dependencies": extract_dependencies(plan), "artifacts": artifacts, "evidence": raw.get("sources") or raw.get("evidence") or plan.get("evidence") or fallback.get("evidence", []), "answer_text": answer_text, "boundary_action": str(raw.get("boundary_action") or plan.get("boundary_action") or fallback.get("boundary_action", "")), "support_status": raw.get("support_status") or plan.get("support_status") or fallback.get("support_status"), "can_execute": raw.get("can_execute", plan.get("can_execute")), "cannot_execute_reason": str(raw.get("cannot_execute_reason") or plan.get("cannot_execute_reason") or ""), "llm_json_valid": raw.get("llm_json_valid"), "llm_json_status": raw.get("llm_json_status", ""), "llm_provider": raw.get("llm_provider", ""), "llm_model": raw.get("llm_model", ""), "llm_base_url": raw.get("llm_base_url", ""), "llm_usage": raw.get("llm_usage") or {}, "llm_finish_reason": raw.get("llm_finish_reason", ""), } if raw.get("error"): prediction["error"] = str(raw["error"]) return prediction def request_json(url: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def call_plan(api_base: str, task: dict[str, Any], timeout: float, *, no_evidence: bool = False) -> dict[str, Any]: context = task.get("input_context") or {} intent = dict(context.get("intent") or {}) params = dict(intent.get("params") or {}) params["benchmarkMode"] = True if no_evidence: params["disableEvidence"] = True intent["benchmarkMode"] = True intent["params"] = params payload = { "message": task.get("user_query", ""), "session_id": f"plantgenebench_full_{task.get('task_id')}", "job_id": context.get("job_id", ""), "prefix": context.get("prefix", ""), "history": [], "intent": intent, } return request_json(f"{api_base.rstrip('/')}/api/agent/plan", payload, timeout) def call_chat(api_base: str, task: dict[str, Any], timeout: float) -> dict[str, Any]: payload = { "message": task.get("user_query", ""), "model": "kimi", "session_id": f"plantgenebench_chat_{task.get('task_id')}", "job_id": (task.get("input_context") or {}).get("job_id", ""), "history": [], } return request_json(f"{api_base.rstrip('/')}/api/agent/chat", payload, timeout) def call_rag(api_base: str, task: dict[str, Any], timeout: float) -> dict[str, Any]: payload = { "query": task.get("user_query", ""), "mode": "rag", "top_k": 5, "job_id": (task.get("input_context") or {}).get("job_id", ""), "prefix": (task.get("input_context") or {}).get("prefix", ""), "stage": "", "use_candidate_context": False, } return request_json(f"{api_base.rstrip('/')}/api/agent/rag", payload, timeout) def fallback_from_task(task: dict[str, Any]) -> dict[str, Any]: intent = ((task.get("input_context") or {}).get("intent") or {}) return { "mode": "task_gold_hint", "analysis_type": intent.get("analysis_type", task.get("layer", "")), "intent": "", "entities": {}, "stages": normalize_list(intent.get("stages")), "execution_plan": normalize_list(intent.get("stages")), "params": dict(intent.get("params") or {}), "artifacts": {}, "evidence": [], "answer_text": "", "boundary_action": "", "support_status": None, } def rule_raw(task: dict[str, Any]) -> dict[str, Any]: fb = fallback_from_task(task) return { "success": True, "mode": "rule_baseline", "plan": { "analysis_type": fb["analysis_type"], "stages": fb["stages"], "stages_detail": [{"id": stage, "dependencies": [], "params": fb["params"]} for stage in fb["stages"]], "params": fb["params"], "description": "rule baseline prediction", }, "intent": {"stages": fb["stages"], "params": fb["params"], "analysis_type": fb["analysis_type"]}, } def llm_only_raw(task: dict[str, Any]) -> dict[str, Any]: return { "success": True, "mode": "llm_only_surrogate", "message": "LLM-only surrogate response without registered-stage grounding or evidence retrieval.", "plan": {}, "intent": {"analysis_type": "llm_only", "stages": [], "params": {}}, } def deepseek_client(args: argparse.Namespace): env_path = args.deepseek_env or DEFAULT_DEEPSEEK_ENV env = load_env_file(Path(env_path)) root = Path(args.deepseek_project).expanduser() if str(root) not in sys.path: sys.path.insert(0, str(root)) try: from src.llm.base import LLMMessage from src.llm.providers.deepseek import DeepSeekProvider except Exception as exc: raise RuntimeError(f"cannot import gene-family-mas-v2 DeepSeekProvider: {exc}") from exc api_key = os.environ.get("DEEPSEEK_API_KEY") or env.get("DEEPSEEK_API_KEY") if not api_key: raise RuntimeError(f"DEEPSEEK_API_KEY not found in environment or {env_path}") model = args.deepseek_model or os.environ.get("V15_RELATION_LLM_MODEL") or os.environ.get("DEEPSEEK_MODEL") or env.get("V15_RELATION_LLM_MODEL") or env.get("DEEPSEEK_MODEL") or "deepseek-chat" base_url = args.deepseek_base_url or os.environ.get("DEEPSEEK_BASE_URL") or env.get("DEEPSEEK_BASE_URL") or "https://api.deepseek.com" provider = DeepSeekProvider(api_key=api_key, model=model, temperature=args.deepseek_temperature, max_tokens=args.deepseek_max_tokens, api_base=base_url) return provider, LLMMessage, {"provider": "deepseek", "model": model, "base_url": base_url} def task_context_for_prompt(task: dict[str, Any]) -> str: context = task.get("input_context") or {} safe_context = { "task_id": task.get("task_id"), "layer": task.get("layer"), "category": task.get("category"), "difficulty": task.get("difficulty"), "input_context": context, } return json.dumps(safe_context, ensure_ascii=False, sort_keys=True) def deepseek_text_messages(task: dict[str, Any], message_cls: Any) -> list[Any]: system = ( "You are evaluating a plant gene-family analysis request. Answer honestly and conservatively. " "Do not claim analyses were executed unless the prompt says results already exist. " "When evidence is insufficient, say what is missing." ) user = ( f"User query:\n{task.get('user_query', '')}\n\n" f"Task context JSON:\n{task_context_for_prompt(task)}\n\n" "Give a concise answer suitable for evaluating a gene-family analysis agent." ) return [message_cls("system", system), message_cls("user", user)] def deepseek_json_messages(task: dict[str, Any], message_cls: Any) -> list[Any]: schema_hint = { "analysis_type": "gene_family|rna_seq|evidence_review|boundary_control|other", "intent": "short_intent_label", "entities": {"species": "", "gene_family": "", "tissue": ""}, "target_stages": ["identify"], "execution_plan": ["identify"], "parameters": {"geneFamily": "", "species": ""}, "artifacts": {"required_inputs": [], "expected_outputs": []}, "evidence": [{"source": "", "claim": "", "support": "direct|indirect|none|manual_required"}], "support_status": "supported|unsupported|external_or_manual|insufficient_evidence|null", "boundary_action": "none|ask_for_input|refuse_overclaim|manual_review_required", "can_execute": True, "cannot_execute_reason": "", "answer_text": "brief conservative explanation", } system = ( "You are PlantGeneBench DeepSeek structured baseline. Return valid JSON only. " "Use only bbav5 stage ids from the provided stage catalog. Do not invent executed results, files, or literature evidence. " "If a request needs wet-lab validation or missing input files, set can_execute=false or boundary_action accordingly." ) user = ( f"Allowed bbav5 stage ids:\n{json.dumps(STAGE_CATALOG, ensure_ascii=False)}\n\n" f"Required JSON keys:\n{json.dumps(STRUCTURED_KEYS, ensure_ascii=False)}\n\n" f"JSON shape example:\n{json.dumps(schema_hint, ensure_ascii=False)}\n\n" f"User query:\n{task.get('user_query', '')}\n\n" f"Task context JSON:\n{task_context_for_prompt(task)}\n\n" "Return one JSON object and no Markdown." ) return [message_cls("system", system), message_cls("user", user)] def strip_code_fence(text: str) -> str: content = text.strip() if content.startswith("```"): content = re.sub(r"^```(?:json)?\s*", "", content, flags=re.I) content = re.sub(r"\s*```$", "", content) return content.strip() def extract_json_object(text: str) -> str: content = strip_code_fence(text) if content.startswith("{") and content.endswith("}"): return content start = content.find("{") end = content.rfind("}") if start >= 0 and end > start: return content[start:end + 1] return content def parsed_to_raw(parsed: dict[str, Any], content: str, meta: dict[str, Any], usage: dict[str, Any], finish_reason: str) -> dict[str, Any]: target_stages = normalize_list(parsed.get("target_stages") or parsed.get("stages")) execution_plan = normalize_list(parsed.get("execution_plan")) or target_stages artifacts = normalize_dict(parsed.get("artifacts")) plan = { "analysis_type": parsed.get("analysis_type", ""), "intent": parsed.get("intent", ""), "entities": normalize_dict(parsed.get("entities")), "target_stages": target_stages, "execution_plan": execution_plan, "stages": target_stages or execution_plan, "params": normalize_dict(parsed.get("parameters") or parsed.get("params")), "artifacts": artifacts, "evidence": parsed.get("evidence") or [], "support_status": parsed.get("support_status"), "boundary_action": parsed.get("boundary_action", ""), "can_execute": parsed.get("can_execute"), "cannot_execute_reason": parsed.get("cannot_execute_reason", ""), "answer_text": parsed.get("answer_text", ""), "description": parsed.get("answer_text", ""), } return { "success": True, "mode": "deepseek_json", "message": parsed.get("answer_text", ""), "plan": plan, "intent": { "intent": parsed.get("intent", ""), "analysis_type": parsed.get("analysis_type", ""), "entities": normalize_dict(parsed.get("entities")), "stages": target_stages, "params": normalize_dict(parsed.get("parameters") or parsed.get("params")), }, "artifacts": artifacts, "evidence": parsed.get("evidence") or [], "support_status": parsed.get("support_status"), "boundary_action": parsed.get("boundary_action", ""), "can_execute": parsed.get("can_execute"), "cannot_execute_reason": parsed.get("cannot_execute_reason", ""), "llm_json_valid": True, "llm_json_status": "parsed", "llm_provider": meta["provider"], "llm_model": meta["model"], "llm_base_url": meta["base_url"], "llm_usage": usage, "llm_finish_reason": finish_reason, "raw_text": content, "parsed_payload": parsed, } def call_deepseek(args: argparse.Namespace, task: dict[str, Any], *, structured: bool) -> dict[str, Any]: provider, message_cls, meta = deepseek_client(args) messages = deepseek_json_messages(task, message_cls) if structured else deepseek_text_messages(task, message_cls) response = provider.call(messages, temperature=args.deepseek_temperature, max_tokens=args.deepseek_max_tokens) content = response.content or "" usage = response.usage or {} finish_reason = response.finish_reason or "" if not structured: return { "success": True, "mode": "deepseek_text", "message": content, "plan": {}, "intent": {"analysis_type": "deepseek_text", "stages": [], "params": {}}, "llm_json_valid": None, "llm_json_status": "not_requested", "llm_provider": meta["provider"], "llm_model": meta["model"], "llm_base_url": meta["base_url"], "llm_usage": usage, "llm_finish_reason": finish_reason, "raw_text": content, } try: parsed = json.loads(extract_json_object(content)) if not isinstance(parsed, dict): raise ValueError("top-level JSON is not an object") return parsed_to_raw(parsed, content, meta, usage, finish_reason) except Exception as exc: return { "success": False, "mode": "deepseek_json", "message": content, "plan": {}, "intent": {}, "error": f"deepseek_json_parse_failed: {exc}", "llm_json_valid": False, "llm_json_status": "parse_failed", "llm_provider": meta["provider"], "llm_model": meta["model"], "llm_base_url": meta["base_url"], "llm_usage": usage, "llm_finish_reason": finish_reason, "raw_text": content, } def run_provider(args: argparse.Namespace, task: dict[str, Any]) -> dict[str, Any]: layer = task.get("layer") or task.get("category") try: if args.provider == "rule": raw = rule_raw(task) elif args.provider == "llm_only": raw = llm_only_raw(task) elif args.provider == "deepseek_text": raw = call_deepseek(args, task, structured=False) elif args.provider == "deepseek_json": raw = call_deepseek(args, task, structured=True) elif args.provider in {"full_system", "api"}: if layer == "evidence_constraint": raw = call_rag(args.api_base, task, args.timeout) raw.setdefault("plan", {}) elif layer in {"biological_interpretation", "boundary_control"}: raw = call_chat(args.api_base, task, args.timeout) else: raw = call_plan(args.api_base, task, args.timeout) elif args.provider == "no_evidence": if layer == "evidence_constraint": raw = {"success": True, "mode": "no_evidence", "message": "Evidence retrieval disabled for ablation.", "plan": {}, "intent": {}} else: raw = call_plan(args.api_base, task, args.timeout, no_evidence=True) else: raise SystemExit(f"unsupported provider: {args.provider}") except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, RuntimeError, ImportError, ValueError) as exc: raw = {"success": False, "error": str(exc), "plan": {}, "intent": {}, "mode": args.provider} return raw def run(args: argparse.Namespace) -> int: tasks = load_jsonl(args.tasks) if args.limit: tasks = tasks[: args.limit] run_id = args.run_id or f"{args.provider}_{int(time.time())}" rows: list[dict[str, Any]] = [] for index, task in enumerate(tasks, start=1): print(f"[{index}/{len(tasks)}] {task['task_id']}", file=sys.stderr) raw = run_provider(args, task) prediction = normalize_raw(raw, fallback_from_task(task)) rows.append({ "run_id": run_id, "task_id": task["task_id"], "provider": args.provider, "created_at": datetime.now(timezone.utc).isoformat(), "request": {"message": task.get("user_query", ""), "layer": task.get("layer"), "input_context": task.get("input_context", {})}, "raw_response": raw, "prediction": prediction, }) dump_jsonl(args.out, rows) print(f"Wrote {len(rows)} predictions to {args.out}", file=sys.stderr) return 0 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--provider", choices=["full_system", "api", "rule", "llm_only", "no_evidence", "deepseek_text", "deepseek_json"], required=True) parser.add_argument("--api-base", default="http://127.0.0.1:7880") parser.add_argument("--tasks", type=Path, required=True) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--run-id", default="") parser.add_argument("--timeout", type=float, default=120.0) parser.add_argument("--limit", type=int, default=0) parser.add_argument("--deepseek-env", type=Path, default=DEFAULT_DEEPSEEK_ENV) parser.add_argument("--deepseek-project", default=str(GENE_FAMILY_MAS_ROOT)) parser.add_argument("--deepseek-model", default="") parser.add_argument("--deepseek-base-url", default="") parser.add_argument("--deepseek-temperature", type=float, default=0.1) parser.add_argument("--deepseek-max-tokens", type=int, default=2500) return parser.parse_args() if __name__ == "__main__": raise SystemExit(run(parse_args()))