553 lines
24 KiB
Python
Executable File
553 lines
24 KiB
Python
Executable File
#!/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()))
|