File size: 3,072 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 | """A minimal OpenAI-compatible server standing in for the Cortex.0 runtime.
Cortex.0 itself needs ~45 GiB of RAM and a GPU, neither of which this machine
has. This server speaks the same protocol so the fusion code can be validated
for real: real HTTP, real SDK calls, real agent loop.
"""
from __future__ import annotations
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
calls: list[dict] = []
def log_message(self, *args): # silence the default stderr noise
pass
def _send(self, code: int, payload: dict) -> None:
body = json.dumps(payload).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
if self.path.rstrip("/").endswith("/models"):
self._send(200, {"object": "list", "data": [{"id": "cortex0", "object": "model"}]})
else:
self._send(404, {"error": "not found"})
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
request = json.loads(self.rfile.read(length) or b"{}")
Handler.calls.append(request)
messages = request.get("messages", [])
tools = request.get("tools") or []
# First turn: request a tool. Later turns: answer in plain text.
already_acted = any(m.get("role") == "tool" for m in messages)
if tools and not already_acted:
self._send(200, {
"id": "cmpl-1", "object": "chat.completion", "model": "cortex0",
"choices": [{
"index": 0, "finish_reason": "tool_calls",
"message": {
"role": "assistant", "content": None,
"tool_calls": [{
"id": "call_1", "type": "function",
"function": {"name": tools[0]["function"]["name"],
"arguments": json.dumps({"command": "echo cortex0-fused"})},
}],
},
}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
})
else:
self._send(200, {
"id": "cmpl-2", "object": "chat.completion", "model": "cortex0",
"choices": [{
"index": 0, "finish_reason": "stop",
"message": {"role": "assistant", "content": "Task complete."},
}],
"usage": {"prompt_tokens": 20, "completion_tokens": 4, "total_tokens": 24},
})
def start(port: int = 0) -> tuple[ThreadingHTTPServer, str]:
server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
return server, f"http://127.0.0.1:{server.server_address[1]}/v1"
|