Biomni-ReAct/biomni_react/cli.py

from __future__ import annotations

import argparse
import json
from pathlib import Path

from .agent import BiomniReActAgent
from .config import AgentConfig
from .schema import TaskSpec


def load_task(path: Path, workspace_override: Path | None = None) -> TaskSpec:
    payload = json.loads(path.read_text(encoding="utf-8"))
    workspace = workspace_override or Path(payload.get("workspace", "runs/default"))
    outputs = []
    for raw in payload.get("expected_outputs", []):
        output = Path(raw)
        outputs.append(output if output.is_absolute() else workspace / output)
    return TaskSpec(
        name=payload["name"],
        objective=payload["objective"],
        workspace=workspace,
        expected_outputs=outputs,
        constraints=list(payload.get("constraints", [])),
        metadata={str(key): str(value) for key, value in payload.get("metadata", {}).items()},
    )


def main() -> None:
    parser = argparse.ArgumentParser(description="Run a Biomni-ReAct task.")
    parser.add_argument("--task", required=True, type=Path, help="Path to a JSON task specification.")
    parser.add_argument("--workspace", type=Path, help="Override the task workspace.")
    parser.add_argument("--model", help="Override BIOMNI_REACT_MODEL.")
    parser.add_argument("--top-k", type=int, help="Number of resources to retrieve.")
    args = parser.parse_args()

    config = AgentConfig()
    if args.model:
        config.model = args.model
    if args.top_k:
        config.retrieval_top_k = args.top_k

    task = load_task(args.task, args.workspace)
    result = BiomniReActAgent(config=config).run(task)
    print(json.dumps({"success": result.success, "error": result.error, "summary": str(result.artifact_paths["summary"])}, indent=2))


if __name__ == "__main__":
    main()