mirror of
https://github.com/mims-harvard/ToolUniverse.git
synced 2026-09-19 07:31:47 +08:00
Shanghua Gao
83a6085b8b
* Add examples/case_studies: reproduce the three published case studies Runnable versions of the BLM, OXTR and BCG-CORONA case studies from https://aiscientist.tools/posts/tooluniverse-case-studies Each script makes the same tool calls in the same order as the published run and prints the published value beside the retrieved one, so a run reports either that the case reproduces or which number moved. Values not stated in the post come from the supplementary note it condenses (arXiv:2509.23426) and are marked as such. - case1_blm_target_assessment.py: 15 tools across Open Targets, ClinVar, GeneBe, AlphaFold, PDBe, Europe PMC, PubChem, ADMET-AI and ChEMBL - case2_oxtr_druggability.py: druggability dossier ending in the agonist/antagonist mismatch that decides the case - case3_bcg_ae_severity.py: clinical_trial_ae_severity_test in prepare, chi-square and ordinal modes, with download_bcg_data.py fetching the CC0 trial tables from Zenodo Case 3's tool loads its implementation from the repository's skills/ directory, so it only resolves in a git clone. scripts/prepare_ae_cohort.py bundles the equivalent implementation; the case study uses the tool when it is available and falls back to the bundle otherwise, reporting which path it took. Both produce the published numbers, on pandas 2 and 3 alike. Verified 2026-08-19: 44 published-value checks across the three cases, all matching except the OXTR disease-association count, which Open Targets has re-scored from 393 to 465 since the run. The README records which values are expected to drift and which are not, and notes that the two ChEMBL-dependent steps could not be re-observed because that API is returning HTTP 500 upstream. * Drop the companion-posts pointer from the case-studies README
96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""Shared helpers for the three case-study reproduction scripts.
|
|
|
|
Keeps the case-study scripts readable: they should read as the ordered list of
|
|
tool calls the AI scientist made, not as plumbing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Iterable
|
|
|
|
from tooluniverse import ToolUniverse
|
|
|
|
# Loading all ~2,700 tools takes a while and none of the case studies need more
|
|
# than these categories. Together they supply every tool the three cases call.
|
|
CASE_STUDY_CATEGORIES = [
|
|
"opentarget",
|
|
"clinvar",
|
|
"genebe",
|
|
"alphafold",
|
|
"pdbe_graph",
|
|
"pubchem",
|
|
"admetai",
|
|
"ChEMBL",
|
|
"compose",
|
|
"EuropePMC",
|
|
"clinical_trial_stats",
|
|
]
|
|
|
|
|
|
def load_universe(categories: Iterable[str] | None = None) -> ToolUniverse:
|
|
"""Return a ToolUniverse with just the categories the case studies need."""
|
|
tu = ToolUniverse()
|
|
tu.load_tools(list(categories or CASE_STUDY_CATEGORIES))
|
|
return tu
|
|
|
|
|
|
def call(tu: ToolUniverse, tool_name: str, /, **arguments: Any) -> Any:
|
|
"""Run one tool and return its payload, unwrapping the status envelope.
|
|
|
|
Tools return either ``{"status": "success", "data": ...}`` or a bare value.
|
|
Errors are returned rather than raised so a single unavailable upstream
|
|
service does not abort the whole case study.
|
|
|
|
``tool_name`` is positional-only so that tools taking a ``name`` argument
|
|
(e.g. ``PubChem_get_CID_by_compound_name``) can still be called as
|
|
``call(tu, "...", name="ML216")``.
|
|
"""
|
|
result = tu.run({"name": tool_name, "arguments": arguments})
|
|
if isinstance(result, dict):
|
|
if result.get("status") == "error":
|
|
return {"__error__": result.get("error", "unknown error")}
|
|
if "data" in result and result.get("status") == "success":
|
|
return result["data"]
|
|
return result
|
|
|
|
|
|
def is_error(payload: Any) -> bool:
|
|
return isinstance(payload, dict) and "__error__" in payload
|
|
|
|
|
|
def step(number: int | str, title: str) -> None:
|
|
print(f"\n[{number}] {title}")
|
|
print("-" * 72)
|
|
|
|
|
|
def report(label: str, observed: Any, published: Any = None) -> None:
|
|
"""Print one observed value, and the published value when there is one."""
|
|
if published is None:
|
|
print(f" {label}: {observed}")
|
|
else:
|
|
print(f" {label}: {observed} (published: {published})")
|
|
|
|
|
|
def note_unavailable(tool: str, payload: Any) -> None:
|
|
"""Report a step that could not run, without dumping an HTML error page."""
|
|
message = " ".join(str(payload["__error__"]).split())
|
|
if len(message) > 160:
|
|
message = message[:160] + " ..."
|
|
print(f" [unavailable] {tool}: {message}")
|
|
print(" This step depends on a live external service; see README.md.")
|
|
|
|
|
|
def header(title: str, question: str) -> None:
|
|
print("=" * 72)
|
|
print(title)
|
|
print("=" * 72)
|
|
print(f"Question: {question}")
|
|
|
|
|
|
def footer(lines: Dict[str, str]) -> None:
|
|
print("\n" + "=" * 72)
|
|
print("Result")
|
|
print("=" * 72)
|
|
for key, value in lines.items():
|
|
print(f" {key}: {value}")
|