| """The fusion itself: Cortex.0 driving the OpenHands action-perception loop.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from pathlib import Path |
| from typing import Any |
|
|
| from openhands.sdk import Conversation |
|
|
| from cortex0.agent import build_agent |
| from cortex0.health import ServerProbe, probe_server |
| from cortex0.settings import Cortex0Settings |
|
|
| logger = logging.getLogger("cortex0.fusion") |
|
|
|
|
| class Cortex0Agent: |
| """A single unified system: Cortex.0 as the brain, the SDK as the body. |
| |
| The model is not invoked as an external tool. It is the reasoning engine |
| the SDK calls on every turn, and the SDK's tools (terminal, file editor, |
| browser) are exposed to it as its own action space. |
| """ |
|
|
| def __init__( |
| self, |
| settings: Cortex0Settings | None = None, |
| verify: bool = True, |
| ) -> None: |
| self.settings = settings or Cortex0Settings.load() |
| self.probe: ServerProbe | None = None |
|
|
| if verify: |
| self.probe = probe_server( |
| base_url=self.settings.base_url, |
| api_key=self.settings.api_key, |
| model=self.settings.served_model_name, |
| timeout=min(self.settings.timeout, 120.0), |
| ) |
| if not self.probe.reachable: |
| raise RuntimeError( |
| "Cortex.0 inference server is unreachable at " |
| f"{self.settings.base_url}. {self.probe.detail}\n" |
| "Start the server (see scripts/serve_llamacpp.sh) or point " |
| "CORTEX0_BASE_URL at a running endpoint." |
| ) |
| if not self.probe.chat_ok: |
| raise RuntimeError( |
| "Cortex.0 answered /models but not chat/completions. " |
| f"{self.probe.detail}" |
| ) |
| if not self.probe.tool_calling_ok: |
| logger.warning( |
| "Cortex.0 did not emit tool_calls during the probe (%s). " |
| "The agent will still run, but autonomous tool use may be weak.", |
| self.probe.detail or "no tool call returned", |
| ) |
|
|
| self.agent = build_agent(self.settings) |
|
|
| def start_conversation(self, workspace: str | Path | None = None) -> Conversation: |
| """Create a conversation whose workspace is the agent's sandbox.""" |
| target = Path(workspace or self.settings.workspace) |
| target.mkdir(parents=True, exist_ok=True) |
| return Conversation(agent=self.agent, workspace=str(target)) |
|
|
| def run(self, task: str, workspace: str | Path | None = None) -> Any: |
| """Run one task end to end and return the conversation.""" |
| conversation = self.start_conversation(workspace) |
| conversation.send_message(task) |
| conversation.run() |
| return conversation |
|
|
| def summary(self) -> dict: |
| """Report the state of the fusion, for logs and the CLI.""" |
| return { |
| "model": self.settings.model, |
| "base_url": self.settings.base_url, |
| "context_window": self.settings.context_window, |
| "tools": list(self.settings.tools), |
| "server_reachable": bool(self.probe and self.probe.reachable), |
| "chat_ok": bool(self.probe and self.probe.chat_ok), |
| "tool_calling_ok": bool(self.probe and self.probe.tool_calling_ok), |
| "probe_detail": self.probe.detail if self.probe else "", |
| "probe_latency_s": self.probe.latency_s if self.probe else None, |
| } |
|
|
