Spaces:
Paused
Paused
File size: 7,409 Bytes
6778532 | 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """OpenAI-compatible HTTP surface for a mesh node.
Only the subset ThoxRoute and the ThoxOS apps actually call is implemented:
``/v1/chat/completions``, ``/v1/models`` and health. Implementing the surface
narrowly and honestly is deliberate — a node that advertises streaming or
embeddings it cannot serve would be selected by the router and then fail, which
is worse for the mesh than not advertising them.
Every request is timed and recorded into the ``TelemetryRecorder``, because that
recording *is* how the node earns or loses its place in the router's ranking.
"""
from __future__ import annotations
import logging
import time
import uuid
from typing import Any, Literal
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator
from .backends import Backend
from .errors import BackendError
from .telemetry import TelemetryRecorder
logger = logging.getLogger("thox.server")
class ChatMessage(BaseModel):
"""One message in an OpenAI-style chat request."""
role: Literal["system", "user", "assistant", "tool"]
content: str = ""
@field_validator("content", mode="before")
@classmethod
def _coerce_content(cls, value: Any) -> str:
"""Accept the content-parts array form and flatten it to text.
Newer OpenAI clients send ``[{"type": "text", "text": "..."}]``. A node
that 422s on that shape looks broken to a caller that is, by the spec,
well-behaved.
"""
if value is None:
return ""
if isinstance(value, str):
return value
if isinstance(value, list):
parts = [
str(part.get("text", ""))
for part in value
if isinstance(part, dict) and part.get("type") in (None, "text")
]
return "".join(parts)
return str(value)
class ChatCompletionRequest(BaseModel):
"""Request body for ``/v1/chat/completions``."""
model: str | None = None
messages: list[ChatMessage] = Field(min_length=1)
max_tokens: int | None = Field(default=None, ge=1)
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
stop: list[str] | str | None = None
stream: bool = False
def create_app(
*,
backend: Backend,
recorder: TelemetryRecorder,
model_id: str,
context_window: int,
max_tokens_cap: int,
node_state: dict[str, Any] | None = None,
) -> FastAPI:
"""Build the ASGI app.
``node_state`` is a live dict owned by the agent; the health endpoint reads
it so operators (and the controller's health loop) can see registration
state without a second channel.
"""
app = FastAPI(title="THOX Mesh Node", version="1.0.0")
state = node_state if node_state is not None else {}
@app.get("/health")
def health() -> dict[str, Any]:
"""Liveness plus current mesh registration and telemetry."""
return {
"status": "ok",
"model_id": model_id,
"context_window": context_window,
"mesh": {
"transport": state.get("transport"),
"endpoint_id": state.get("endpoint_id"),
"registered": bool(state.get("endpoint_id")),
"public_url": state.get("public_url"),
"last_heartbeat_at": state.get("last_heartbeat_at"),
"last_error": state.get("last_error"),
},
"telemetry": recorder.snapshot(),
}
@app.get("/v1/models")
def list_models() -> dict[str, Any]:
"""Model list in OpenAI's shape, so generic clients can discover us."""
return {
"object": "list",
"data": [
{
"id": model_id,
"object": "model",
"owned_by": "thox",
"created": int(state.get("started_at", time.time())),
"context_window": context_window,
}
],
}
@app.post("/v1/chat/completions")
def chat_completions(body: ChatCompletionRequest) -> JSONResponse:
"""Generate one completion and record its outcome as telemetry."""
if body.stream:
# Refused explicitly rather than silently returning a non-stream body,
# which would leave a streaming client hanging on a parse it can never finish.
raise HTTPException(
status_code=400,
detail="streaming is not supported by this node; retry with stream=false",
)
requested = body.max_tokens or max_tokens_cap
max_tokens = max(1, min(requested, max_tokens_cap))
stop = [body.stop] if isinstance(body.stop, str) else body.stop
recorder.request_started()
started = time.perf_counter()
try:
completion = backend.generate(
[m.model_dump() for m in body.messages],
max_tokens=max_tokens,
temperature=body.temperature,
stop=stop,
)
except BackendError as exc:
recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False)
logger.error("generation failed: %s", exc)
raise HTTPException(status_code=503, detail=str(exc)) from exc
except Exception as exc: # noqa: BLE001 - never leak a 500 without telemetry
recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False)
logger.exception("unexpected generation error")
raise HTTPException(status_code=500, detail="internal generation error") from exc
latency_ms = (time.perf_counter() - started) * 1000
recorder.request_finished(
latency_ms=latency_ms,
ok=True,
completion_tokens=completion.completion_tokens,
)
return JSONResponse(
{
"id": f"chatcmpl-{uuid.uuid4().hex[:24]}",
"object": "chat.completion",
"created": int(time.time()),
"model": body.model or model_id,
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": completion.text},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": completion.prompt_tokens,
"completion_tokens": completion.completion_tokens,
"total_tokens": completion.total_tokens,
},
"thox_node": {
"latency_ms": round(latency_ms, 1),
"endpoint_id": state.get("endpoint_id"),
"node_kind": state.get("node_kind"),
},
}
)
@app.get("/")
def root(request: Request) -> dict[str, Any]:
"""Human-facing landing payload; also what a Space's health probe hits."""
return {
"service": "THOX Mesh Node",
"model_id": model_id,
"openai_base_url": str(request.base_url).rstrip("/") + "/v1",
"endpoints": ["/v1/chat/completions", "/v1/models", "/health"],
"registered": bool(state.get("endpoint_id")),
}
return app
|