| 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 = [ |
| |
| self.project_root / "toolbase" / "register_mcp_servers_to_biomni.py", |
| |
| self.project_root / "agent_system" / "toolbase" / "register_mcp_servers_to_biomni.py", |
| |
| 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 |
| |
| 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 |
|
|
| 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): |
| |
| 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 |
|
|
| 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 |
|
|
| 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) |
| |
| 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 |
|
|
