File size: 2,941 Bytes
48d895c | 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 | """Reverse pipeline planner — build_plan() from model spec and request metadata.
Determines the endpoint URL, transport kind, pool/mode IDs, and timeout
for a given operation. Does NOT execute anything — pure data transform.
"""
from typing import Any
from app.control.model.spec import ModelSpec
from app.dataplane.reverse.runtime.endpoint_table import (
CHAT, MEDIA_POST, WS_IMAGINE,
)
from .types import ReversePlan, TransportKind
# ---------------------------------------------------------------------------
# Profile defaults (timeout / content-type per transport)
# ---------------------------------------------------------------------------
_DEFAULTS: dict[TransportKind, dict[str, Any]] = {
TransportKind.HTTP_SSE: {"timeout_s": 120.0, "content_type": "application/json"},
TransportKind.HTTP_JSON: {"timeout_s": 30.0, "content_type": "application/json"},
TransportKind.WEBSOCKET: {"timeout_s": 300.0, "content_type": "application/json"},
TransportKind.GRPC_WEB: {"timeout_s": 15.0, "content_type": "application/grpc-web+proto"},
}
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def build_plan(spec: ModelSpec, request: dict[str, Any] | None = None) -> ReversePlan:
"""Produce a ReversePlan for the given model spec.
``request`` is the raw API request body — used to refine the plan
(e.g. detect image-edit vs image-gen) but may be ``None``.
"""
endpoint, tkind = _resolve_endpoint(spec, request or {})
defaults = _DEFAULTS.get(tkind, _DEFAULTS[TransportKind.HTTP_JSON])
return ReversePlan(
endpoint = endpoint,
transport_kind = tkind,
pool_candidates = spec.pool_candidates(),
mode_id = int(spec.mode_id),
timeout_s = defaults["timeout_s"],
content_type = defaults["content_type"],
)
# ---------------------------------------------------------------------------
# Internal routing logic
# ---------------------------------------------------------------------------
def _resolve_endpoint(
spec: ModelSpec,
request: dict[str, Any],
) -> tuple[str, TransportKind]:
"""Determine (endpoint_url, transport_kind) for the given capability."""
if spec.is_chat():
return CHAT, TransportKind.HTTP_SSE
if spec.is_image():
return WS_IMAGINE, TransportKind.WEBSOCKET
if spec.is_image_edit():
return CHAT, TransportKind.HTTP_SSE
if spec.is_video():
return MEDIA_POST, TransportKind.HTTP_JSON
if spec.is_voice():
# LiveKit negotiation is HTTP JSON, actual voice is WebSocket.
return CHAT, TransportKind.HTTP_SSE
# Fallback: treat as chat.
return CHAT, TransportKind.HTTP_SSE
__all__ = ["build_plan"]
|