File size: 3,526 Bytes
c429240 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """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,
}
|