Spaces:
Running
Running
File size: 3,904 Bytes
f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e faaafef f049d1e | 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 | from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from registry import normalize_tool
from context_adapter import DecisionContext
class RouteRequest(BaseModel):
decision_context: DecisionContext = Field(default_factory=DecisionContext)
input: str = Field(..., min_length=1)
tools: list[dict[str, Any]] = Field(..., min_length=1, max_length=5)
system: str | None = None
@field_validator("tools")
@classmethod
def _normalize_tools(cls, value: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [normalize_tool(item) for item in value]
class RouteResponse(BaseModel):
outcome: Literal["selected", "confirm", "abstain", "error"]
function_calls: list[dict[str, Any]] = Field(default_factory=list)
suppressed_calls: list[dict[str, Any]] = Field(default_factory=list)
confidence: float | None = None
reasoning: str | None = None
validation: dict[str, Any] | None = None
model: str
class AutoRouteRequest(BaseModel):
decision_context: DecisionContext = Field(default_factory=DecisionContext)
query: str = Field(..., min_length=1)
tools: list[dict[str, Any]] = Field(..., min_length=1)
system: str | None = None
max_candidates: int = Field(5, ge=1, le=5)
@field_validator("tools")
@classmethod
def _normalize_tools(cls, value: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [normalize_tool(item) for item in value]
class AutoRouteResponse(BaseModel):
outcome: Literal["selected", "confirm", "abstain", "error"]
candidates: list[dict[str, Any]]
retrieval_scores: list[float]
function_calls: list[dict[str, Any]]
suppressed_calls: list[dict[str, Any]]
confidence: float | None
reasoning: str | None
validation: dict[str, Any] | None
model: str
diagnostics: dict[str, Any] = Field(default_factory=dict)
class ToolRegistration(BaseModel):
namespace: str
tools: list[dict[str, Any]]
class ToolRegistrationResponse(BaseModel):
registered_count: int
namespace: str
class RetrieveRequest(BaseModel):
query: str
namespace: str | None = None
max_candidates: int = Field(5, ge=1, le=5)
workflow_context: dict[str, Any] | None = None
class EvaluateRequest(BaseModel):
tools: list[dict[str, Any]]
class EvaluateResponse(BaseModel):
evaluations: list[dict[str, Any]]
class Message(BaseModel):
role: str
content: str | list[dict[str, Any]] | None = None
tool_call_id: str | None = None
name: str | None = None
tool_calls: list[dict[str, Any]] | None = None
class RegistryReference(BaseModel):
model_config = ConfigDict(extra="forbid")
version: int = Field(1, ge=1, le=1)
namespace: str = Field(..., min_length=1, max_length=256)
sha256: str = Field(..., pattern="^[a-f0-9]{64}$")
class ChatCompletionRequest(BaseModel):
registry_reference: RegistryReference | None = None
decision_context: DecisionContext = Field(default_factory=DecisionContext)
model: str
messages: list[Message]
tools: list[dict[str, Any]] | None = None
stream: bool | None = False
@model_validator(mode="after")
def _one_tool_source(self):
if self.registry_reference is not None and self.tools is not None:
raise ValueError("provide tools or registry_reference, not both")
return self
@field_validator("tools")
@classmethod
def _normalize_chat_tools(cls, value):
return [normalize_tool(t) for t in value] if value is not None else None
class LayaEvaluationRequest(BaseModel):
workflow_state: str
candidates: list[dict[str, Any]]
scores: list[float]
function_calls: list[dict[str, Any]]
confidence: float | None
class LayaEvaluationResponse(BaseModel):
decision: Literal["proceed", "escalate", "abstain"]
reasoning: str | None = None
|