"""Unified Gateway Runner — single entry point for all LLM interactions.
The ``Gateway`` class ties together prompt compilation, adapter selection,
tool injection, and response standardisation into one high-level interface.
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ValidationError
from ractogateway.adapters.base import BaseLLMAdapter, LLMResponse
from ractogateway.exceptions import ResponseModelValidationError
from ractogateway.prompts.engine import RactoPrompt
from ractogateway.tools.registry import ToolRegistry
[docs]
class Gateway:
"""Unified entry point that wraps any ``BaseLLMAdapter``.
Parameters
----------
adapter:
A concrete adapter instance (``OpenAILLMKit``, ``GoogleLLMKit``,
``AnthropicLLMKit``).
tools:
An optional ``ToolRegistry`` containing registered tools that the
LLM is allowed to call.
default_prompt:
An optional ``RactoPrompt`` to use when ``run()`` is called
without an explicit prompt.
Usage::
from ractogateway import RactoPrompt, Gateway
from ractogateway.adapters import OpenAILLMKit
adapter = OpenAILLMKit(model="gpt-4o", api_key="sk-...")
prompt = RactoPrompt(...)
gw = Gateway(adapter=adapter)
response = gw.run(prompt, "Analyse this code for bugs.")
print(response.parsed) # auto-parsed JSON dict
"""
def __init__(
self,
adapter: BaseLLMAdapter,
*,
tools: ToolRegistry | None = None,
default_prompt: RactoPrompt | None = None,
) -> None:
self.adapter = adapter
self.tools = tools
self.default_prompt = default_prompt
# ------------------------------------------------------------------
# Execution
# ------------------------------------------------------------------
[docs]
def run(
self,
prompt: RactoPrompt | None = None,
user_message: str = "",
*,
tools: ToolRegistry | None = None,
temperature: float = 0.0,
max_tokens: int = 4096,
response_model: type[BaseModel] | None = None,
**kwargs: Any,
) -> LLMResponse:
"""Execute a request and return a standardised ``LLMResponse``.
Parameters
----------
prompt:
The RACTO prompt. Falls back to ``default_prompt``.
user_message:
The end-user's query.
tools:
Override the gateway-level tool registry for this call.
temperature:
Sampling temperature.
max_tokens:
Maximum tokens in the response.
response_model:
Optional Pydantic model to validate ``parsed`` output against.
If provided and the LLM returns valid JSON, it is validated
through this model and attached to ``response.parsed``.
**kwargs:
Passed through to the adapter.
"""
effective_prompt = prompt or self.default_prompt
if effective_prompt is None:
raise ValueError("No prompt provided and no default_prompt configured on the Gateway.")
effective_tools = tools or self.tools
response = self.adapter.run(
effective_prompt,
user_message,
tools=effective_tools,
temperature=temperature,
max_tokens=max_tokens,
**kwargs,
)
if response_model is not None and response.parsed is not None:
response = self._validate_response(response, response_model)
return response
[docs]
async def arun(
self,
prompt: RactoPrompt | None = None,
user_message: str = "",
*,
tools: ToolRegistry | None = None,
temperature: float = 0.0,
max_tokens: int = 4096,
response_model: type[BaseModel] | None = None,
**kwargs: Any,
) -> LLMResponse:
"""Async variant of ``run()``."""
effective_prompt = prompt or self.default_prompt
if effective_prompt is None:
raise ValueError("No prompt provided and no default_prompt configured on the Gateway.")
effective_tools = tools or self.tools
response = await self.adapter.arun(
effective_prompt,
user_message,
tools=effective_tools,
temperature=temperature,
max_tokens=max_tokens,
**kwargs,
)
if response_model is not None and response.parsed is not None:
response = self._validate_response(response, response_model)
return response
# ------------------------------------------------------------------
# Response validation
# ------------------------------------------------------------------
@staticmethod
def _validate_response(
response: LLMResponse,
model: type[BaseModel],
) -> LLMResponse:
"""Validate and re-parse the response through a Pydantic model.
On success, ``response.parsed`` is replaced with the validated
model's ``.model_dump()``. On failure a
:class:`~ractogateway.exceptions.ResponseModelValidationError` is
raised (the Gateway does not retry — callers control retry logic).
"""
if not isinstance(response.parsed, dict):
return response
try:
validated = model.model_validate(response.parsed)
response.parsed = validated.model_dump()
except ValidationError as exc:
raise ResponseModelValidationError(
f"response_model validation failed. Last error: {exc}",
attempts=1,
last_error=exc,
raw_response=response.content,
) from exc
return response