Llm Router¶

llm_router — mvp.llm_router

Cluster: Core Infrastructure | Type: component | MCP Tools: 35

Overview¶

LLM routing layer via OpenRouter and Ollama with resource enforcement, model discovery, hardware-aware recommendations, and LiteLLM-based completion with custom callback hooks.

When to use:

• Routing LLM calls through OpenRouter or local Ollama backends
• Local inference with hardware-aware model selection
• Resource-bounded LLM usage with token tracking

Example:

from mvp.llm_router import LLMBlock

block = LLMBlock(name="llm")
result = block.infer("Summarise this text in one sentence.")
# result.ok → True; result.value → str (LLM response)

Works well with: agent_claude, config, goal_engine

Public API¶

CrossCallResult¶

Result of a cross-package call with fallback metadata.

LLMBlock(AIBlock[str, str, None])¶

AIBlock wrapper around an LLM call with configurable backend.

Methods:

discover_models(timeout_seconds: float = 5.0, limit: int = 100) -> Result[list[dict]]¶

Return live OpenRouter model pricing metadata for UI/tooling callers.

estimate_current_model_cost(input_tokens: int, output_tokens: int, timeout_seconds: float = 5.0) -> Result[float]¶

Estimate current model cost using live OpenRouter data when available.

infer(data: str | dict) -> Result[str]¶

Call an LLM with data as the prompt.

complete(prompt: str, model: str | None = None, max_tokens: int = 4096, temperature: float = 0.7, system_prompt: str | None = None, **kwargs: Any) -> Result[str]¶

Satisfy mvp.core.llm_protocol.LLMProtocol.

bias() -> dict¶

ModelPricing¶

Normalized model price metadata.

Methods:

prompt_usd_per_1k() -> float¶

completion_usd_per_1k() -> float¶

estimate(input_tokens: int, output_tokens: int) -> float¶

model_dump() -> dict[str, Any]¶

Functions¶

cross_call(block_cls: type, input_cls: type, op: str, llm: object | None = None, fallback_fn: Callable[..., dict] | None = None, enhance_prompt: str = '', timeout: float = 60.0, **kwargs: Any) -> Result[CrossCallResult, str]¶

Railway-oriented cross-package call.

route_llm_model_task(task_type: str = 'prompt_template_change', risk_level: str = 'medium') -> dict[str, Any]¶

Expose Phase-02b empirical routing context for LLM model selection.

get_openrouter_api_key() -> tuple[bool, str]¶

Resolve the OpenRouter key (in server mode: return it, never install it).

get_ollama_models(base_url: str = 'http://localhost:11434', timeout_seconds: float = OLLAMA_METADATA_TIMEOUT_SECONDS) -> tuple[bool, list[str] | str]¶

Docstring for get_ollama_models

get_openrouter_models(timeout_seconds: float = MODEL_CATALOG_TIMEOUT_SECONDS) -> tuple[bool, list[str] | str]¶

Retrieves a list of all available models from OpenRouter, formatted for LiteLLM.

check_model_availability(litellm_model: str) -> bool¶

Docstring for check_model_availability

estimate_tokens(model: str, messages: List[Dict[str, str]]) -> tuple[bool, Any]¶

simple_safe_completion(litellm_model: str, prompt: List[Dict[str, str]], max_completion_tokens: int = 25000, timeout: float | None = None, temperature: float | None = None) -> tuple[bool, Any]¶

The purpose of this function is to provide a safe wrapper around the litellm completion function, which will catch and

advanced_safe_completion(litellm_model: str, prompt: List[Dict[str, str]], max_completion_tokens: int = 25000, **kwargs) -> tuple[bool, Any]¶

Docstring for advanced_safe_completion

static_pricing_catalog(static_costs_per_1k: dict[str, float]) -> PricingCatalog¶

Build a catalog from legacy same-rate input/output per-1K prices.

fetch_openrouter_model_catalog(timeout_seconds: float = 5.0, http_get: Callable[..., Any] | None = None) -> tuple[bool, PricingCatalog | str]¶

Fetch live OpenRouter model metadata and normalize pricing.

estimate_model_cost_usd(model: str, input_tokens: int, output_tokens: int, catalog: PricingCatalog) -> float | None¶

catalog_to_public_list(catalog: PricingCatalog, limit: int = 100) -> list[dict[str, Any]]¶

Return a de-duplicated, display-safe model list.

MCP Tools¶