from __future__ import annotations import csv import json import os import subprocess import sys import importlib.util import statistics import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any from ..schemas import Hypothesis, Insight, ValidationReport from ..shared_memory import SharedKnowledgeSpace class BioinfoE1Validator: """ E1 validator: layered and cost-aware hypothesis validation. First version: - L1: logical consistency / rule checks - L2: light tooling readiness check using Biomni + registered MCP servers - L3: lightweight data-backed validation on biomni_data - L4: extended data-backed validation with fuller passes and numeric checks """ LEVEL_CONFIDENCE = { "L1": 0.30, "L2": 0.60, "L3": 0.85, "L4": 0.95, } def __init__( self, memory: SharedKnowledgeSpace, biomni_root: str | Path, mcp_config_path: str | Path, project_root: str | Path, ): self.memory = memory self.biomni_root = Path(biomni_root).resolve() self.mcp_config_path = Path(mcp_config_path).resolve() self.project_root = Path(project_root).resolve() self._registration_state: dict[str, Any] | None = None self._biomni_probe_cache: dict[str, Any] | None = None self._biomni_agent_cache: dict[str, Any] | None = None def _resolve_register_script(self) -> Path | None: candidates = [ # when project_root=/.../BioScientist/agent_system self.project_root / "toolbase" / "register_mcp_servers_to_biomni.py", # when project_root=/.../BioScientist self.project_root / "agent_system" / "toolbase" / "register_mcp_servers_to_biomni.py", # fallback based on current file location Path(__file__).resolve().parents[1] / "toolbase" / "register_mcp_servers_to_biomni.py", ] for c in candidates: if c.exists(): return c return None def ensure_mcp_registered(self, dry_run: bool = False) -> dict[str, Any]: script = self._resolve_register_script() if script is None: return { "ok": False, "status": "missing_script", "error": ( "register script not found under either " f"{self.project_root}/toolbase or {self.project_root}/agent_system/toolbase" ), } cmd = [ sys.executable, str(script), "--biomni-root", str(self.biomni_root), "--config-out", str(self.mcp_config_path), ] if dry_run: cmd.append("--dry-run") try: completed = subprocess.run(cmd, capture_output=True, text=True, check=False) state = { "ok": completed.returncode == 0, "status": "registered" if completed.returncode == 0 else "failed", "return_code": completed.returncode, "stdout_tail": "\n".join((completed.stdout or "").splitlines()[-20:]), "stderr_tail": "\n".join((completed.stderr or "").splitlines()[-20:]), "config_path": str(self.mcp_config_path), } self._registration_state = state return state except Exception as exc: state = {"ok": False, "status": "failed", "error": str(exc), "config_path": str(self.mcp_config_path)} self._registration_state = state return state def _coerce_hypothesis(self, hypothesis: Hypothesis | dict[str, Any]) -> dict[str, Any]: return hypothesis.to_dict() if isinstance(hypothesis, Hypothesis) else hypothesis def _prepare_biomni_import(self) -> dict[str, Any]: """ Ensure `import biomni` works without requiring pip install -e. """ info = { "biomni_root": str(self.biomni_root), "path_injected": False, "package_dir_exists": (self.biomni_root / "biomni").exists(), } root_str = str(self.biomni_root) if root_str not in sys.path: sys.path.insert(0, root_str) info["path_injected"] = True return info def _resolve_biomni_data_root(self) -> Path: env = os.getenv("BIOCLAW_BIOMNI_DATA_ROOT", "").strip() if env: p = Path(env).expanduser().resolve() if p.exists(): return p candidates = [ self.project_root / "agent_system" / "toolbase" / "data" / "biomni_data", self.project_root / "toolbase" / "data" / "biomni_data", Path("/225040511/project/BioScientist/agent_system/toolbase/data/biomni_data"), ] for c in candidates: if c.exists(): return c return candidates[0] def _resolve_biomni_llm_source(self) -> tuple[str, str]: """ Resolve runtime LLM/source for Biomni execution. Priority: 1) BIOCLAW_BIOMNI_LLM / BIOCLAW_BIOMNI_SOURCE 2) GEMINI_MODEL / BIOMNI_SOURCE 3) safe Gemini defaults """ llm = ( os.getenv("BIOCLAW_BIOMNI_LLM", "").strip() or os.getenv("GEMINI_MODEL", "").strip() or "gemini-2.5-flash-lite" ) source = ( os.getenv("BIOCLAW_BIOMNI_SOURCE", "").strip() or os.getenv("BIOMNI_SOURCE", "").strip() or "Gemini" ) return llm, source def _quick_biomni_probe(self, hypothesis_text: str = "", force: bool = False) -> dict[str, Any]: """ Minimal-cost Biomni runtime probe. Reuses cached probe; otherwise builds/uses a cached Biomni agent context. """ if self._biomni_probe_cache is not None and not force: cached = dict(self._biomni_probe_cache) cached["cache_hit"] = True return cached t0 = datetime.now(timezone.utc) ctx = self._get_biomni_agent_context(hypothesis_text, force_rebuild=force) dt = datetime.now(timezone.utc) - t0 if not ctx.get("ok", False): result = { "ok": False, "status": "biomni_probe_failed", "error": str(ctx.get("error", "unknown")), "mcp_registration_ok": bool(ctx.get("trace", {}).get("mcp_registration_ok", False)), "duration_ms": int(dt.total_seconds() * 1000), "cache_hit": False, "mcp_config_used": str(ctx.get("trace", {}).get("mcp_config_used", self.mcp_config_path)), **(ctx.get("trace", {}) if isinstance(ctx.get("trace", {}), dict) else {}), } self._biomni_probe_cache = result return result trace = ctx.get("trace", {}) if isinstance(ctx.get("trace", {}), dict) else {} result = { "ok": True, "status": "ready", "mcp_registration_ok": True, "custom_tool_count": int(trace.get("custom_tool_count", 0)), "related_tool_hits": 0, "duration_ms": int(dt.total_seconds() * 1000), "cache_hit": False, "mcp_config_used": str(trace.get("mcp_config_used", self.mcp_config_path)), "loaded_server_count": int(trace.get("loaded_server_count", -1)), "minimal_config_selected_count": int(trace.get("minimal_config_selected_count", 0)), "full_config_server_count": int(trace.get("full_config_server_count", 0)), "path_injected": bool(trace.get("path_injected", False)), "package_dir_exists": bool(trace.get("package_dir_exists", False)), "biomni_root": str(trace.get("biomni_root", self.biomni_root)), } self._biomni_probe_cache = result return result def _get_biomni_agent_context(self, hypothesis_text: str = "", force_rebuild: bool = False) -> dict[str, Any]: """ Build/reuse a Biomni A1 agent with MCP loaded. Returns a context dict: { ok: bool, agent: A1 | None, trace: dict[str, Any], error: str } """ if self._biomni_agent_cache is not None and not force_rebuild: return self._biomni_agent_cache registration = self._registration_state or self.ensure_mcp_registered(dry_run=False) if not registration.get("ok", False): ctx = { "ok": False, "agent": None, "trace": {"mcp_registration_ok": False}, "error": "mcp_registration_failed", } self._biomni_agent_cache = ctx return ctx deps_ok, missing_modules, deps_evidence = self._check_l2_dependencies() if not deps_ok: ctx = { "ok": False, "agent": None, "trace": { "mcp_registration_ok": True, "missing_modules": missing_modules, }, "error": deps_evidence, } self._biomni_agent_cache = ctx return ctx try: import_info = self._prepare_biomni_import() from biomni.agent import A1 llm_name, llm_source = self._resolve_biomni_llm_source() min_cfg = self._build_minimal_mcp_config(hypothesis_text) used_config_path = str(self.mcp_config_path) loaded_server_count = -1 minimal_server_count = 0 full_server_count = 0 if min_cfg is not None: minimal_server_count = int(min_cfg.get("selected_count", 0)) full_server_count = int(min_cfg.get("full_count", 0)) if minimal_server_count > 0: used_config_path = str(min_cfg["path"]) agent = A1(llm=llm_name, source=llm_source) agent.add_mcp(config_path=used_config_path) tools = agent.list_custom_tools() or [] if min_cfg is not None and minimal_server_count > 0: loaded_server_count = minimal_server_count elif min_cfg is not None: loaded_server_count = full_server_count if not tools and used_config_path != str(self.mcp_config_path): agent = A1(llm=llm_name, source=llm_source) agent.add_mcp(config_path=str(self.mcp_config_path)) tools = agent.list_custom_tools() or [] used_config_path = str(self.mcp_config_path) loaded_server_count = full_server_count if full_server_count > 0 else -1 ctx = { "ok": True, "agent": agent, "trace": { "custom_tool_count": len(tools), "mcp_config_used": used_config_path, "loaded_server_count": loaded_server_count, "minimal_config_selected_count": minimal_server_count, "full_config_server_count": full_server_count, "biomni_llm": llm_name, "biomni_source": llm_source, **import_info, }, "error": "", } self._biomni_agent_cache = ctx return ctx except Exception as exc: ctx = { "ok": False, "agent": None, "trace": {"mcp_registration_ok": True}, "error": str(exc), } self._biomni_agent_cache = ctx return ctx @staticmethod def _extract_json_like_block(text: str) -> dict[str, Any]: s = (text or "").strip() if not s: return {} try: return json.loads(s) if s.startswith("{") else {} except Exception: pass # naive fallback: find first {...} start = s.find("{") end = s.rfind("}") if start >= 0 and end > start: block = s[start : end + 1] try: return json.loads(block) except Exception: return {} return {} def _execute_hypothesis_with_biomni(self, h: dict[str, Any]) -> dict[str, Any]: """ Execute hypothesis using Biomni directly (A1.go) with lightweight prompt. Returns execution summary and artifact path. """ text = f"{h.get('title', '')} {h.get('hypothesis', '')}".strip() ctx = self._get_biomni_agent_context(text) if not ctx.get("ok", False): return { "ok": False, "error": f"biomni_agent_unavailable: {ctx.get('error', 'unknown')}", "trace": ctx.get("trace", {}), } agent = ctx.get("agent") data_sources = [str(x) for x in h.get("data_sources", [])][:5] ops = [x for x in h.get("data_operations", []) if isinstance(x, dict)] prompt = ( "You are validating a bioinformatics hypothesis in lightweight mode.\n" "Use available MCP tools where useful and provide concise execution output.\n" "Return JSON with keys: status, summary, key_metrics, recommendation.\n\n" f"Domain: {h.get('domain', '')}\n" f"Hypothesis title: {h.get('title', '')}\n" f"Hypothesis: {h.get('hypothesis', '')}\n" f"Expected improvement: {h.get('expected_improvement', '')}\n" f"Theoretical basis: {h.get('theoretical_basis', '')}\n" f"Data sources: {json.dumps(data_sources, ensure_ascii=True)}\n" f"Requested operations: {json.dumps(ops, ensure_ascii=True)}\n" "Please execute a minimal-cost validation and return machine-readable JSON." ) try: logs, final_message = agent.go(prompt) parsed = self._extract_json_like_block(str(final_message)) status_raw = str(parsed.get("status", "")).lower() mapped_status = "success" if status_raw in {"success", "ok", "passed"} else ("failed" if status_raw in {"failed", "error"} else "inconclusive") summary = str(parsed.get("summary", "")).strip() or str(final_message)[:1000] key_metrics = parsed.get("key_metrics", {}) if isinstance(parsed.get("key_metrics", {}), dict) else {} recommendation = str(parsed.get("recommendation", "")).strip() out_dir = self.project_root / "agent_system" / "results" / "biomni_exec" out_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") out_path = out_dir / f"biomni_exec_{h.get('hypothesis_id', 'unknown')}_{ts}.json" out_payload = { "hypothesis_id": h.get("hypothesis_id", ""), "domain": h.get("domain", ""), "prompt": prompt, "parsed": parsed, "status": mapped_status, "summary": summary, "key_metrics": key_metrics, "recommendation": recommendation, "final_message": str(final_message), "log_tail": logs[-20:] if isinstance(logs, list) else [], "trace": ctx.get("trace", {}), "saved_at": datetime.now(timezone.utc).isoformat(), } out_path.write_text(json.dumps(out_payload, indent=2, ensure_ascii=True), encoding="utf-8") return { "ok": True, "status": mapped_status, "summary": summary, "key_metrics": key_metrics, "recommendation": recommendation, "artifact_file": str(out_path), "trace": ctx.get("trace", {}), } except Exception as exc: return { "ok": False, "error": f"biomni_execution_failed: {exc}", "trace": ctx.get("trace", {}), } def _build_minimal_mcp_config(self, hypothesis_text: str) -> dict[str, Any] | None: yaml_spec = importlib.util.find_spec("yaml") if yaml_spec is None: return None import yaml # type: ignore if not self.mcp_config_path.exists(): return None try: raw = self.mcp_config_path.read_text(encoding="utf-8") doc = yaml.safe_load(raw) or {} except Exception: return None servers = doc.get("mcp_servers", {}) if not isinstance(servers, dict) or not servers: return None full_count = len(servers) max_servers = int(os.getenv("BIOCLAW_MIN_MCP_SERVERS", "40")) max_servers = max(5, min(200, max_servers)) q = (hypothesis_text or "").lower() tokens = [t for t in q.replace("-", " ").replace("_", " ").split() if len(t) >= 3][:30] scored: list[tuple[int, str, dict[str, Any]]] = [] for name, cfg in servers.items(): if not isinstance(cfg, dict): continue cmd = " ".join([str(x).lower() for x in cfg.get("command", [])]) if isinstance(cfg.get("command"), list) else str(cfg.get("command", "")).lower() desc = str(cfg.get("description", "")).lower() hay = f"{name.lower()} {desc} {cmd}" s = 0 for tok in tokens: if tok in hay: s += 2 if "single" in hay or "cell" in hay or "rna" in hay: s += 1 scored.append((s, name, cfg)) scored.sort(key=lambda x: x[0], reverse=True) chosen = [x for x in scored if x[0] > 0][:max_servers] if len(chosen) < min(10, max_servers): # Backfill with top entries to keep probe meaningful. fallback_pool = [x for x in scored if x[1] not in {c[1] for c in chosen}] need = min(10, max_servers) - len(chosen) chosen.extend(fallback_pool[:need]) if not chosen: chosen = scored[: min(10, max_servers)] if not chosen: return None selected_servers = {name: cfg for _, name, cfg in chosen} out_dir = self.project_root / "agent_system" / "results" / "e1_runtime" / "min_configs" out_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") out_path = out_dir / f"mcp_min_{ts}.yaml" out_path.write_text( yaml.safe_dump({"mcp_servers": selected_servers}, sort_keys=False, allow_unicode=False), encoding="utf-8", ) return { "path": out_path, "selected_count": len(selected_servers), "full_count": full_count, } def _persist_runtime_artifact(self, report_dict: dict[str, Any], hypothesis: dict[str, Any]) -> str: runtime_dir = self.project_root / "agent_system" / "results" / "e1_runtime" runtime_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") out_path = runtime_dir / f"{report_dict.get('level', 'Lx').lower()}_{report_dict.get('validation_id', 'unknown')}_{ts}.json" payload = { "validation": report_dict, "hypothesis": { "hypothesis_id": hypothesis.get("hypothesis_id", ""), "domain": hypothesis.get("domain", ""), "title": hypothesis.get("title", ""), }, "saved_at": datetime.now(timezone.utc).isoformat(), } out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=True), encoding="utf-8") summary_path = runtime_dir / "summary.json" if summary_path.exists(): try: summary = json.loads(summary_path.read_text(encoding="utf-8")) except Exception: summary = {} else: summary = {} by_level = summary.setdefault("by_level", {}) lvl = report_dict.get("level", "UNKNOWN") s = by_level.setdefault(lvl, {"count": 0, "success": 0, "inconclusive": 0, "failed": 0, "avg_score": 0.0}) old_count = int(s.get("count", 0)) old_avg = float(s.get("avg_score", 0.0)) new_score = float(report_dict.get("score", 0.0)) new_count = old_count + 1 s["count"] = new_count s["avg_score"] = (old_avg * old_count + new_score) / max(1, new_count) st = str(report_dict.get("status", "inconclusive")) s[st] = int(s.get(st, 0)) + 1 summary["total_reports"] = int(summary.get("total_reports", 0)) + 1 summary["last_report_file"] = str(out_path) summary["last_updated"] = datetime.now(timezone.utc).isoformat() summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=True), encoding="utf-8") return str(out_path) def _index_data_files(self, data_root: Path, limit: int = 1000) -> list[Path]: if not data_root.exists(): return [] files: list[Path] = [] for p in sorted(data_root.rglob("*")): if p.is_file(): files.append(p) if len(files) >= limit: break return files def _download_online_example_data(self, domain: str, max_files: int = 2) -> list[Path]: """ Best-effort fallback downloader when no usable local dataset is available. Downloads small public tabular files so L3/L4 can still produce bound validation reports. """ out_dir = self.project_root / "agent_system" / "results" / "downloaded_example_data" out_dir.mkdir(parents=True, exist_ok=True) domain_l = str(domain).lower() single_cell_candidates = [ ( "pbmc3k_marker_genes.csv", "https://raw.githubusercontent.com/scverse/scanpy-tutorials/main/pbmc3k/marker_genes.csv", ), ( "pbmc3k_obs_metadata.csv", "https://raw.githubusercontent.com/scverse/scanpy-tutorials/main/pbmc3k/obs_metadata.csv", ), ] generic_candidates = [ ( "iris.csv", "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/iris.csv", ), ( "tips.csv", "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/tips.csv", ), ] candidates = single_cell_candidates + generic_candidates if ("single" in domain_l or "cell" in domain_l) else generic_candidates + single_cell_candidates downloaded: list[Path] = [] for filename, url in candidates: target = out_dir / filename if target.exists() and target.stat().st_size > 0: downloaded.append(target) if len(downloaded) >= max_files: break continue try: urllib.request.urlretrieve(url, str(target)) if target.exists() and target.stat().st_size > 0: downloaded.append(target) if len(downloaded) >= max_files: break except Exception: continue return downloaded def _safe_read_tabular_preview(self, file_path: Path, sample_rows: int = 500) -> dict[str, Any]: suffix = file_path.suffix.lower() result: dict[str, Any] = { "file": str(file_path), "suffix": suffix, "status": "skipped", "rows_scanned": 0, "columns": [], "missing_cells": 0, "total_cells": 0, "numeric_columns": [], } if suffix in {".csv", ".tsv"}: delim = "," if suffix == ".csv" else "\t" with file_path.open("r", encoding="utf-8", errors="ignore", newline="") as fh: reader = csv.DictReader(fh, delimiter=delim) cols = reader.fieldnames or [] result["columns"] = cols rows = 0 missing = 0 total = 0 numeric_candidates: dict[str, int] = {c: 0 for c in cols} for row in reader: rows += 1 for c in cols: v = (row.get(c) or "").strip() total += 1 if v == "": missing += 1 else: try: float(v) numeric_candidates[c] += 1 except Exception: pass if rows >= sample_rows: break result["rows_scanned"] = rows result["missing_cells"] = missing result["total_cells"] = total result["numeric_columns"] = [c for c, n in numeric_candidates.items() if n > 0] result["status"] = "loaded" return result if suffix == ".parquet": pd_spec = importlib.util.find_spec("pandas") if pd_spec is None: result["status"] = "skipped" result["error"] = "pandas_not_installed_for_parquet" return result import pandas as pd # type: ignore df = pd.read_parquet(file_path) if len(df) > sample_rows: df = df.head(sample_rows) result["rows_scanned"] = int(len(df)) cols = [str(c) for c in df.columns.tolist()] result["columns"] = cols total = int(df.shape[0] * df.shape[1]) if df.shape[1] > 0 else 0 missing = int(df.isna().sum().sum()) if total > 0 else 0 numeric_cols = [str(c) for c in df.select_dtypes(include=["number"]).columns.tolist()] result["missing_cells"] = missing result["total_cells"] = total result["numeric_columns"] = numeric_cols result["status"] = "loaded" return result if suffix == ".json": with file_path.open("r", encoding="utf-8", errors="ignore") as fh: payload = json.load(fh) if isinstance(payload, list): rows = min(sample_rows, len(payload)) cols = sorted({k for item in payload[:rows] if isinstance(item, dict) for k in item.keys()}) result["rows_scanned"] = rows result["columns"] = cols result["status"] = "loaded" elif isinstance(payload, dict): result["rows_scanned"] = 1 result["columns"] = sorted(payload.keys()) result["status"] = "loaded" else: result["status"] = "loaded" return result result["status"] = "skipped" result["error"] = f"unsupported_suffix:{suffix or 'none'}" return result def _safe_read_tabular_full(self, file_path: Path, max_rows: int = 200000) -> dict[str, Any]: """ L4-oriented data pass. Attempts a fuller scan (or up to max_rows for safety) and includes basic numeric summaries. """ suffix = file_path.suffix.lower() result: dict[str, Any] = { "file": str(file_path), "suffix": suffix, "status": "skipped", "rows_scanned": 0, "columns": [], "missing_cells": 0, "total_cells": 0, "numeric_columns": [], "truncated": False, "numeric_summary": {}, } if suffix in {".csv", ".tsv"}: delim = "," if suffix == ".csv" else "\t" with file_path.open("r", encoding="utf-8", errors="ignore", newline="") as fh: reader = csv.DictReader(fh, delimiter=delim) cols = reader.fieldnames or [] result["columns"] = cols rows = 0 missing = 0 total = 0 numeric_samples: dict[str, list[float]] = {c: [] for c in cols} for row in reader: rows += 1 for c in cols: v = (row.get(c) or "").strip() total += 1 if v == "": missing += 1 else: try: fv = float(v) if len(numeric_samples[c]) < 5000: numeric_samples[c].append(fv) except Exception: pass if rows >= max_rows: result["truncated"] = True break result["rows_scanned"] = rows result["missing_cells"] = missing result["total_cells"] = total numeric_cols = [c for c, arr in numeric_samples.items() if arr] result["numeric_columns"] = numeric_cols summary = {} for c in numeric_cols[:20]: arr = numeric_samples[c] summary[c] = { "count": len(arr), "mean": float(statistics.fmean(arr)), "min": float(min(arr)), "max": float(max(arr)), } result["numeric_summary"] = summary result["status"] = "loaded" return result if suffix == ".parquet": pd_spec = importlib.util.find_spec("pandas") if pd_spec is None: result["status"] = "skipped" result["error"] = "pandas_not_installed_for_parquet" return result import pandas as pd # type: ignore df = pd.read_parquet(file_path) if len(df) > max_rows: df = df.head(max_rows) result["truncated"] = True result["rows_scanned"] = int(len(df)) cols = [str(c) for c in df.columns.tolist()] result["columns"] = cols total = int(df.shape[0] * df.shape[1]) if df.shape[1] > 0 else 0 missing = int(df.isna().sum().sum()) if total > 0 else 0 numeric_cols = [str(c) for c in df.select_dtypes(include=["number"]).columns.tolist()] result["missing_cells"] = missing result["total_cells"] = total result["numeric_columns"] = numeric_cols summary = {} for c in numeric_cols[:20]: s = df[c].dropna() if len(s) == 0: continue summary[str(c)] = { "count": int(len(s)), "mean": float(s.mean()), "min": float(s.min()), "max": float(s.max()), } result["numeric_summary"] = summary result["status"] = "loaded" return result if suffix == ".json": with file_path.open("r", encoding="utf-8", errors="ignore") as fh: payload = json.load(fh) if isinstance(payload, list): rows = len(payload) if rows > max_rows: rows = max_rows result["truncated"] = True cols = sorted({k for item in payload[:rows] if isinstance(item, dict) for k in item.keys()}) result["rows_scanned"] = rows result["columns"] = cols result["status"] = "loaded" elif isinstance(payload, dict): result["rows_scanned"] = 1 result["columns"] = sorted(payload.keys()) result["status"] = "loaded" else: result["status"] = "loaded" return result result["status"] = "skipped" result["error"] = f"unsupported_suffix:{suffix or 'none'}" return result @staticmethod def _domain_relevance_score(domain: str, selected_files: list[Path], per_file_results: list[dict[str, Any]]) -> tuple[float, list[str]]: domain_l = str(domain).lower() if not selected_files: return 0.0, ["no_selected_files"] hints = [] if "single_cell" in domain_l or "single" in domain_l: hints = ["single", "cell", "rna", "census", "marker", "celltype", "expression"] else: hints = [x for x in domain_l.split("_") if x] matched = 0 explanations: list[str] = [] for p in selected_files: p_low = str(p).lower() if any(h in p_low for h in hints): matched += 1 explanations.append(f"file_match:{p.name}") for r in per_file_results: cols = " ".join([str(c).lower() for c in r.get("columns", [])]) if any(h in cols for h in hints): matched += 1 explanations.append(f"column_match:{Path(str(r.get('file', 'unknown'))).name}") denom = max(1, len(selected_files) + len(per_file_results)) return min(1.0, matched / denom), explanations[:10] @staticmethod def _build_interpretability_summary( *, level: str, loaded_count: int, selected_count: int, rows_scanned: int, missing_rate: float, numeric_cols_total: int, relevance_score: float, ) -> dict[str, Any]: quality = "high" if missing_rate < 0.01 else ("medium" if missing_rate < 0.1 else "low") coverage = loaded_count / max(1, selected_count) coverage_label = "high" if coverage >= 0.8 else ("medium" if coverage >= 0.5 else "low") relevance_label = "high" if relevance_score >= 0.6 else ("medium" if relevance_score >= 0.3 else "low") return { "level": level, "data_quality": quality, "coverage": coverage_label, "domain_relevance": relevance_label, "rows_scanned": rows_scanned, "numeric_signal_strength": "high" if numeric_cols_total >= 10 else ("medium" if numeric_cols_total >= 3 else "low"), "recommendations": [ "Prefer domain-relevant files if domain_relevance is low.", "Increase file coverage when coverage is medium/low.", "Escalate to L4 only after L3 relevance is acceptable.", ], } @staticmethod def _check_l2_dependencies() -> tuple[bool, list[str], str]: required_modules = ["nest_asyncio", "mcp", "yaml"] missing = [m for m in required_modules if importlib.util.find_spec(m) is None] if not missing: return True, [], "" module_to_pip = { "nest_asyncio": "nest_asyncio", "mcp": "mcp", "yaml": "PyYAML", } pip_pkgs = [module_to_pip[m] for m in missing] install_cmd = f"{sys.executable} -m pip install " + " ".join(pip_pkgs) evidence = ( "L2 dependency check failed. Missing modules: " f"{', '.join(missing)}. " f"Install with: {install_cmd}. " f"(module->pip: {', '.join(f'{m}->{module_to_pip[m]}' for m in missing)})" ) return False, missing, evidence def _validate_l1(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]: text = f"{h.get('title', '')} {h.get('hypothesis', '')}".lower() violations: list[str] = [] if "敲除" in text and "增殖加快" in text: violations.append("possible essential-gene-growth contradiction") if "不存在的工具" in text: violations.append("depends on non-existing tool") if "必须" in text and "不需要数据" in text: violations.append("self-contradictory requirement") if violations: return ( "failed", 0.1, {"rule_violations": violations, "violation_count": len(violations)}, "biological_or_tooling_inconsistency", "L1 rules detected contradictions.", ) return ( "success", 0.6, {"rule_violations": [], "violation_count": 0}, "", "L1 rules found no obvious contradiction.", ) def _validate_l2(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]: registration = self._registration_state or self.ensure_mcp_registered(dry_run=False) if not registration.get("ok", False): return ( "inconclusive", 0.3, {"mcp_registration_ok": False}, "mcp_registration_failed", "L2 could not ensure MCP registration.", ) deps_ok, missing_modules, deps_evidence = self._check_l2_dependencies() if not deps_ok: return ( "inconclusive", 0.2, { "mcp_registration_ok": True, "l2_dependency_check_ok": False, "missing_modules": missing_modules, }, "biomni_dependency_missing", deps_evidence, ) try: import_info = self._prepare_biomni_import() from biomni.agent import A1 llm_name, llm_source = self._resolve_biomni_llm_source() agent = A1(llm=llm_name, source=llm_source) agent.add_mcp(config_path=str(self.mcp_config_path)) tools = agent.list_custom_tools() or [] if not tools: return ( "inconclusive", 0.35, { "mcp_registration_ok": True, "custom_tool_count": 0, "biomni_llm": llm_name, "biomni_source": llm_source, **import_info, }, "no_mcp_tools_discovered", "Biomni loaded config but no tools were discovered.", ) text = f"{h.get('title', '')} {h.get('hypothesis', '')}".lower() related = [t for t in tools if any(tok in t.lower() for tok in text.split()[:20])] score = 0.5 + min(0.4, len(related) * 0.05) return ( "success", score, { "mcp_registration_ok": True, "custom_tool_count": len(tools), "related_tool_hits": len(related), "biomni_llm": llm_name, "biomni_source": llm_source, **import_info, }, "", "L2 passed using Biomni MCP tool readiness check.", ) except Exception as exc: import_info = self._prepare_biomni_import() return ( "inconclusive", 0.25, {"mcp_registration_ok": True, **import_info}, "biomni_probe_failed", f"Biomni probe failed: {exc}", ) def _validate_l3(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]: probe = self._quick_biomni_probe(f"{h.get('title', '')} {h.get('hypothesis', '')}") if not probe.get("ok", False): return ( "inconclusive", 0.25, { "mcp_registration_ok": bool(probe.get("mcp_registration_ok", False)), "biomni_probe_status": probe.get("status", ""), "biomni_probe_duration_ms": probe.get("duration_ms", 0), "biomni_call_trace": { "registration_attempted": True, "biomni_runtime_called": False, "cache_hit": bool(probe.get("cache_hit", False)), }, }, str(probe.get("status", "biomni_probe_failed")), f"L3 aborted: Biomni quick probe failed ({probe.get('error', 'unknown_error')}).", ) data_root = self._resolve_biomni_data_root() data_files = self._index_data_files(data_root, limit=1500) if not data_files: return ( "inconclusive", 0.3, {"data_root": str(data_root), "indexed_files": 0}, "l3_data_unavailable", "L3 could not locate biomni_data files.", ) requested = [Path(str(x)) for x in h.get("data_sources", []) if str(x).strip()] if requested: selected = [p for p in requested if p.exists()][:5] else: selected = data_files[:5] downloaded_examples: list[Path] = [] if not selected: downloaded_examples = self._download_online_example_data(h.get("domain", ""), max_files=2) selected = downloaded_examples[:] operations = [x for x in h.get("data_operations", []) if isinstance(x, dict)] if not operations: operations = [ {"action": "load_table", "target": "all_selected"}, {"action": "sample_rows", "rows": 500}, {"action": "profile_missingness"}, {"action": "numeric_summary"}, ] per_file_results: list[dict[str, Any]] = [] loaded_count = 0 for p in selected: try: r = self._safe_read_tabular_preview(p, sample_rows=500) per_file_results.append(r) if r.get("status") == "loaded": loaded_count += 1 except Exception as exc: per_file_results.append( { "file": str(p), "status": "failed", "error": str(exc), } ) op_names = [str(op.get("action", "")) for op in operations] rows_scanned = sum(int(r.get("rows_scanned", 0) or 0) for r in per_file_results) total_cells = sum(int(r.get("total_cells", 0) or 0) for r in per_file_results) missing_cells = sum(int(r.get("missing_cells", 0) or 0) for r in per_file_results) missing_rate = (missing_cells / total_cells) if total_cells > 0 else 0.0 numeric_cols_total = sum(len(r.get("numeric_columns", []) or []) for r in per_file_results) relevance_score, relevance_evidence = self._domain_relevance_score(h.get("domain", ""), selected, per_file_results) results_dir = self.project_root / "agent_system" / "results" / "l3_reports" results_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") artifact_path = results_dir / f"l3_{h.get('hypothesis_id', 'unknown')}_{ts}.json" artifact_payload = { "hypothesis_id": h.get("hypothesis_id", ""), "domain": h.get("domain", ""), "data_root": str(data_root), "selected_files": [str(p) for p in selected], "operations": operations, "per_file_results": per_file_results, "rows_scanned": rows_scanned, "missing_rate": missing_rate, } artifact_path.write_text(json.dumps(artifact_payload, indent=2, ensure_ascii=True), encoding="utf-8") if loaded_count == 0: return ( "inconclusive", 0.35, { "data_root": str(data_root), "indexed_files": len(data_files), "selected_files": [str(p) for p in selected], "loaded_files": 0, "operations_requested": op_names, "l3_report_file": str(artifact_path), }, "l3_no_loadable_data", "L3 selected files but none were loadable in lightweight mode.", ) score_components = { "base": 0.45, "loaded_files_component": 0.08 * loaded_count, "operation_component": 0.03 * min(8, len(op_names)), "domain_relevance_component": 0.08 * relevance_score, } score = min(0.92, sum(score_components.values())) status = "success" if loaded_count >= 1 else "inconclusive" biomni_exec = self._execute_hypothesis_with_biomni(h) if biomni_exec.get("ok", False): b_status = biomni_exec.get("status", "inconclusive") score = min(1.0, score + (0.08 if b_status == "success" else 0.0)) if b_status == "failed": status = "failed" elif status != "failed" and b_status == "inconclusive": status = "inconclusive" else: score = max(0.0, score - 0.08) if status == "success": status = "inconclusive" evidence = ( "L3 executed lightweight full-data checks with sampled previews; " f"loaded {loaded_count}/{len(selected)} files, scanned_rows={rows_scanned}, " f"missing_rate={missing_rate:.4f}, domain_relevance={relevance_score:.2f}. Report: {artifact_path}" ) return ( status, score, { "data_root": str(data_root), "indexed_files": len(data_files), "selected_files": [str(p) for p in selected], "loaded_files": loaded_count, "rows_scanned": rows_scanned, "missing_rate": missing_rate, "numeric_columns_total": numeric_cols_total, "operations_requested": op_names, "domain_relevance_score": relevance_score, "domain_relevance_evidence": relevance_evidence, "score_breakdown": score_components, "interpretability_summary": self._build_interpretability_summary( level="L3", loaded_count=loaded_count, selected_count=len(selected), rows_scanned=rows_scanned, missing_rate=missing_rate, numeric_cols_total=numeric_cols_total, relevance_score=relevance_score, ), "biomni_call_trace": { "registration_attempted": bool(self._registration_state), "biomni_runtime_called": True, "cache_hit": bool(probe.get("cache_hit", False)), "biomni_probe_duration_ms": probe.get("duration_ms", 0), "custom_tool_count": probe.get("custom_tool_count", 0), "related_tool_hits": probe.get("related_tool_hits", 0), "mcp_config_used": probe.get("mcp_config_used", str(self.mcp_config_path)), "loaded_server_count": probe.get("loaded_server_count", -1), "minimal_config_selected_count": probe.get("minimal_config_selected_count", 0), "full_config_server_count": probe.get("full_config_server_count", 0), "note": "L3 first performs a minimal Biomni runtime probe, then runs lightweight data checks.", }, "downloaded_example_files": [str(p) for p in downloaded_examples], "l3_report_file": str(artifact_path), "biomni_execution": biomni_exec, }, "" if biomni_exec.get("ok", False) else "biomni_execution_failed", evidence, ) def _validate_l4(self, h: dict[str, Any]) -> tuple[str, float, dict[str, Any], str, str]: probe = self._quick_biomni_probe(f"{h.get('title', '')} {h.get('hypothesis', '')}") if not probe.get("ok", False): return ( "inconclusive", 0.3, { "mcp_registration_ok": bool(probe.get("mcp_registration_ok", False)), "biomni_probe_status": probe.get("status", ""), "biomni_probe_duration_ms": probe.get("duration_ms", 0), "biomni_call_trace": { "registration_attempted": True, "biomni_runtime_called": False, "cache_hit": bool(probe.get("cache_hit", False)), }, }, str(probe.get("status", "biomni_probe_failed")), f"L4 aborted: Biomni quick probe failed ({probe.get('error', 'unknown_error')}).", ) data_root = self._resolve_biomni_data_root() data_files = self._index_data_files(data_root, limit=3000) if not data_files: return ( "inconclusive", 0.3, {"data_root": str(data_root), "indexed_files": 0}, "l4_data_unavailable", "L4 could not locate biomni_data files.", ) requested = [Path(str(x)) for x in h.get("data_sources", []) if str(x).strip()] if requested: selected = [p for p in requested if p.exists()][:8] else: selected = data_files[:5] downloaded_examples: list[Path] = [] if not selected: downloaded_examples = self._download_online_example_data(h.get("domain", ""), max_files=3) selected = downloaded_examples[:] operations = [x for x in h.get("data_operations", []) if isinstance(x, dict)] if not operations: operations = [ {"action": "load_table", "target": "all_selected"}, {"action": "sample_rows", "rows": 500}, {"action": "profile_missingness"}, {"action": "numeric_summary"}, ] max_rows = int(os.getenv("BIOCLAW_L4_MAX_ROWS", "200000")) max_rows = max(1000, min(1000000, max_rows)) per_file_results: list[dict[str, Any]] = [] loaded_count = 0 truncated_count = 0 for p in selected: try: r = self._safe_read_tabular_full(p, max_rows=max_rows) per_file_results.append(r) if r.get("status") == "loaded": loaded_count += 1 if r.get("truncated"): truncated_count += 1 except Exception as exc: per_file_results.append( { "file": str(p), "status": "failed", "error": str(exc), } ) rows_scanned = sum(int(r.get("rows_scanned", 0) or 0) for r in per_file_results) total_cells = sum(int(r.get("total_cells", 0) or 0) for r in per_file_results) missing_cells = sum(int(r.get("missing_cells", 0) or 0) for r in per_file_results) missing_rate = (missing_cells / total_cells) if total_cells > 0 else 0.0 numeric_cols_total = sum(len(r.get("numeric_columns", []) or []) for r in per_file_results) relevance_score, relevance_evidence = self._domain_relevance_score(h.get("domain", ""), selected, per_file_results) op_names = [str(op.get("action", "")) for op in operations] results_dir = self.project_root / "agent_system" / "results" / "l4_reports" results_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") artifact_path = results_dir / f"l4_{h.get('hypothesis_id', 'unknown')}_{ts}.json" artifact_payload = { "hypothesis_id": h.get("hypothesis_id", ""), "domain": h.get("domain", ""), "data_root": str(data_root), "selected_files": [str(p) for p in selected], "operations": operations, "max_rows_per_file": max_rows, "per_file_results": per_file_results, "rows_scanned": rows_scanned, "missing_rate": missing_rate, "numeric_columns_total": numeric_cols_total, "truncated_file_count": truncated_count, } artifact_path.write_text(json.dumps(artifact_payload, indent=2, ensure_ascii=True), encoding="utf-8") if loaded_count == 0: return ( "inconclusive", 0.35, { "data_root": str(data_root), "indexed_files": len(data_files), "selected_files": [str(p) for p in selected], "loaded_files": 0, "l4_report_file": str(artifact_path), }, "l4_no_loadable_data", "L4 selected files but none were loadable.", ) completeness = 1.0 - min(1.0, missing_rate * 10.0) coverage = min(1.0, loaded_count / max(1, len(selected))) richness = min(1.0, numeric_cols_total / 20.0) trunc_penalty = 0.08 if truncated_count > 0 else 0.0 score_components = { "base": 0.45, "coverage_component": 0.25 * coverage, "completeness_component": 0.2 * completeness, "richness_component": 0.15 * richness, "domain_relevance_component": 0.1 * relevance_score, "truncation_penalty": -trunc_penalty, } score = max(0.0, min(1.0, sum(score_components.values()))) status = "success" if (loaded_count >= max(1, min(2, len(selected))) and completeness >= 0.5) else "inconclusive" biomni_exec = self._execute_hypothesis_with_biomni(h) if biomni_exec.get("ok", False): b_status = biomni_exec.get("status", "inconclusive") score = min(1.0, score + (0.1 if b_status == "success" else 0.0)) if b_status == "failed": status = "failed" elif status != "failed" and b_status == "inconclusive": status = "inconclusive" else: score = max(0.0, score - 0.1) if status == "success": status = "inconclusive" evidence = ( "L4 executed extended data checks; " f"loaded {loaded_count}/{len(selected)} files, rows_scanned={rows_scanned}, " f"missing_rate={missing_rate:.4f}, numeric_columns_total={numeric_cols_total}, domain_relevance={relevance_score:.2f}, " f"truncated_files={truncated_count}. Report: {artifact_path}" ) return ( status, score, { "data_root": str(data_root), "indexed_files": len(data_files), "selected_files": [str(p) for p in selected], "loaded_files": loaded_count, "rows_scanned": rows_scanned, "missing_rate": missing_rate, "numeric_columns_total": numeric_cols_total, "truncated_file_count": truncated_count, "operations_requested": op_names, "domain_relevance_score": relevance_score, "domain_relevance_evidence": relevance_evidence, "score_breakdown": score_components, "interpretability_summary": self._build_interpretability_summary( level="L4", loaded_count=loaded_count, selected_count=len(selected), rows_scanned=rows_scanned, missing_rate=missing_rate, numeric_cols_total=numeric_cols_total, relevance_score=relevance_score, ), "biomni_call_trace": { "registration_attempted": bool(self._registration_state), "biomni_runtime_called": True, "cache_hit": bool(probe.get("cache_hit", False)), "biomni_probe_duration_ms": probe.get("duration_ms", 0), "custom_tool_count": probe.get("custom_tool_count", 0), "related_tool_hits": probe.get("related_tool_hits", 0), "mcp_config_used": probe.get("mcp_config_used", str(self.mcp_config_path)), "loaded_server_count": probe.get("loaded_server_count", -1), "minimal_config_selected_count": probe.get("minimal_config_selected_count", 0), "full_config_server_count": probe.get("full_config_server_count", 0), "note": "L4 first performs a minimal Biomni runtime probe, then runs extended data checks.", }, "downloaded_example_files": [str(p) for p in downloaded_examples], "l4_report_file": str(artifact_path), "biomni_execution": biomni_exec, }, "" if biomni_exec.get("ok", False) else "biomni_execution_failed", evidence, ) def validate_hypothesis(self, hypothesis: Hypothesis | dict[str, Any], level: str = "L1") -> dict[str, Any]: h = self._coerce_hypothesis(hypothesis) level = level.upper().strip() if level not in self.LEVEL_CONFIDENCE: level = "L1" if level == "L1": status, score, metrics, failure_reason, evidence = self._validate_l1(h) elif level == "L2": status, score, metrics, failure_reason, evidence = self._validate_l2(h) elif level == "L3": status, score, metrics, failure_reason, evidence = self._validate_l3(h) elif level == "L4": status, score, metrics, failure_reason, evidence = self._validate_l4(h) else: status, score, metrics, failure_reason, evidence = ( "inconclusive", 0.4, {"implemented": False}, f"{level}_not_implemented_in_v1", f"{level} validation is scaffolded in first version.", ) report = ValidationReport.build( hypothesis_id=h.get("hypothesis_id", ""), domain=h.get("domain", ""), level=level, status=status, score=score, confidence=self.LEVEL_CONFIDENCE[level], key_metrics=metrics, failure_reason=failure_reason, evidence=evidence, ) self.memory.save_validation_report(report) # Save an insight so T1 can leverage execution outcomes as experience. try: title = f"{h.get('domain', 'general')} {level} {status}" recommendation = report.evidence biomni_exec = report.key_metrics.get("biomni_execution", {}) if isinstance(biomni_exec, dict) and biomni_exec.get("recommendation"): recommendation = str(biomni_exec.get("recommendation")) ins = Insight.build( title=title, hypothesis=str(h.get("hypothesis", ""))[:1000], recommendation=recommendation[:2000], confidence=float(report.confidence), evidence_run_ids=[report.validation_id], tags=[str(h.get("domain", "general")), f"validation-{level.lower()}", status], ) self.memory.save_insight(ins) except Exception: pass self.memory.update_summary_statistics() report_dict = report.to_dict() report_dict["runtime_report_file"] = self._persist_runtime_artifact(report_dict, h) return report_dict def validate_top_hypotheses( self, ranked_hypotheses: list[dict[str, Any]], top_m: int = 3, level: str = "L1", ) -> list[dict[str, Any]]: reports: list[dict[str, Any]] = [] selected = ranked_hypotheses[: max(0, top_m)] total = len(selected) for idx, h in enumerate(selected, start=1): hid = h.get("hypothesis_id", f"idx_{idx}") print(f"[STAGE:VALIDATION] running {idx}/{total} level={level.upper()} hypothesis_id={hid}", flush=True) report = self.validate_hypothesis(h, level=level) print( f"[STAGE:VALIDATION] finished {idx}/{total} hypothesis_id={hid} status={report.get('status', 'unknown')} score={report.get('score', 0)}", flush=True, ) reports.append(report) return reports