The `llm_wire.py` module#

Summary#

Classes

`ModelConfig`	Resolved chat endpoint configuration.
`RetrySpec`	Retry/backoff policy for LLM transport calls.
`LLMProfile`	Resolved provider and transport settings for an LLM call.
`AaliChatModel`	A single resolved chat-model entry from the AALI models config.

Exceptions

LLMTransportError

Raised when an LLM transport call fails.

Functions

`env_flag`	Read a boolean environment flag using common truthy and falsy strings.
`resolve_tls_verify`	Resolve TLS verification settings for outbound LLM HTTP requests.
`first_model_token`	Return the first model identifier from a comma-separated model list.
`normalize_endpoint`	Normalize an LLM endpoint URL by removing trailing slashes.
`resolve_model_config`	Resolve provider, transport, endpoint, and model settings from the environment.
`detect_provider`	Infer the LLM provider from an explicit provider, model name, or endpoint URL.
`resolve_litellm_route`	Resolve the LiteLLM model route and API key for the selected provider.
`native_provider_configured`	Return whether a native provider has enough configuration to run.
`resolve_profile`	Resolve the full LLM wire profile used for model calls.
`resolve_model_quirks`	Resolve request-format quirks required by the selected model.
`max_tokens_param_for`	Return the max-token request parameter name for a provider and model.
`send_temperature_for`	Return whether temperature should be sent for a provider and model.
`build_chat_body`	Build an OpenAI-compatible chat-completions request body.
`build_litellm_kwargs`	Build keyword arguments for a LiteLLM completion call.
`auth_headers`	Build HTTP authorization headers for an LLM provider profile.
`extract_chat_text`	Extract assistant text from a chat-completion provider response.
`call`	Send a synchronous chat-completion request through the configured LLM transport.
`acall`	Send an asynchronous chat-completion request through the configured LLM transport.
`load_aali_chat_model`	Return the first usable chat model from the AALI config, or `None`.

Attributes

logger

Constants

`DEFAULT_MODEL`
`PROVIDER_OPENAI`
`PROVIDER_AZURE`
`PROVIDER_ANTHROPIC`
`PROVIDER_GEMINI`
`PROVIDER_COMPAT`
`TRANSPORT_LITELLM`
`TRANSPORT_COMPAT`

Description#

Wire helpers for LiteLLM/OpenAI-compatible chat endpoints.

Module detail#

llm_wire.env_flag(name: str, *, default: bool, env=os.environ) → bool#

Read a boolean environment flag using common truthy and falsy strings.

Parameters:

namestr: Name of the object, resource, or field to process.
defaultbool: Fallback value used when no explicit setting is present.
envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

bool: Boolean answer for the requested condition.

llm_wire.resolve_tls_verify(env=os.environ) → bool | str#

Resolve TLS verification settings for outbound LLM HTTP requests.

Parameters:

envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

bool | str: Value computed by the helper for the requested CFX workflow.

llm_wire.first_model_token(raw: str | None) → str | None#

Return the first model identifier from a comma-separated model list.

Parameters:

rawstr | None: Raw payload returned by PyCFX or a provider API.

Returns:

str | None: Value computed by the helper for the requested CFX workflow.

llm_wire.normalize_endpoint(url: str) → str#

Normalize an LLM endpoint URL by removing trailing slashes.

Parameters:

urlstr: Endpoint URL to normalize or inspect.

Returns:

str: String value produced for the requested CFX or provider operation.

llm_wire.resolve_model_config(*, endpoint: str | None = None, api_key: str | None = None, model: str | None = None, auth_style: str | None = None, env=os.environ) → ModelConfig#

Resolve provider, transport, endpoint, and model settings from the environment.

Parameters:

endpointstr | None, optional: Provider endpoint used for the LLM request. Default is None.
api_keystr | None, optional: API key used to authenticate the provider request. Default is None.
modelstr | None, optional: Model identifier selected for the request. Default is None.
auth_stylestr | None, optional: Authorization-header style required by the provider. Default is None.
envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

ModelConfig: Value computed by the helper for the requested CFX workflow.

llm_wire.detect_provider(model: str | None, endpoint: str | None, *, env=os.environ) → str#

Infer the LLM provider from an explicit provider, model name, or endpoint URL.

Parameters:

modelstr | None: Model identifier selected for the request.
endpointstr | None: Provider endpoint used for the LLM request.
envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

str: String value produced for the requested CFX or provider operation.

llm_wire.resolve_litellm_route(provider: str, model: str) → str#

Resolve the LiteLLM model route and API key for the selected provider.

Parameters:

providerstr: LLM provider name selected for the request.
modelstr: Model identifier selected for the request.

Returns:

str: String value produced for the requested CFX or provider operation.

llm_wire.native_provider_configured(model: str | None, *, env=os.environ) → bool#

Return whether a native provider has enough configuration to run.

Parameters:

modelstr | None: Model identifier selected for the request.
envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

bool: Boolean answer for the requested condition.

llm_wire.resolve_profile(*, model: str | None = None, endpoint: str | None = None, auth_style: str | None = None, env=os.environ) → LLMProfile#

Resolve the full LLM wire profile used for model calls.

Parameters:

modelstr | None, optional: Model identifier selected for the request. Default is None.
endpointstr | None, optional: Provider endpoint used for the LLM request. Default is None.
auth_stylestr | None, optional: Authorization-header style required by the provider. Default is None.
envAny, optional: Environment mapping used to resolve LLM configuration. Default is os.environ.

Returns:

LLMProfile: Value computed by the helper for the requested CFX workflow.

llm_wire.resolve_model_quirks(model: str) → dict[str, Any]#

Resolve request-format quirks required by the selected model.

Parameters:

modelstr: Model identifier selected for the request.

Returns:

dict[str, Any]: Structured response payload for the requested operation.

llm_wire.max_tokens_param_for(model: str) → str#

Return the max-token request parameter name for a provider and model.

Parameters:

modelstr: Model identifier selected for the request.

Returns:

str: String value produced for the requested CFX or provider operation.

llm_wire.send_temperature_for(model: str) → bool#

Return whether temperature should be sent for a provider and model.

Parameters:

modelstr: Model identifier selected for the request.

Returns:

bool: Boolean answer for the requested condition.

llm_wire.build_chat_body(*, model: str, messages: Sequence[dict[str, Any]], max_tokens: int | None = None, temperature: float | None = None) → dict[str, Any]#

Build an OpenAI-compatible chat-completions request body.

Parameters:

modelstr: Model identifier selected for the request.
messagesSequence[dict[str, Any]]: Chat messages to send to the model.
max_tokensint | None, optional: Maximum number of tokens requested from the model. Default is None.
temperaturefloat | None, optional: Sampling temperature for providers that support it. Default is None.

Returns:

dict[str, Any]: Structured response payload for the requested operation.

llm_wire.build_litellm_kwargs(profile: LLMProfile, messages: Sequence[dict[str, Any]], *, max_tokens: int | None = None, temperature: float | None = None, api_key: str | None = None, api_base: str | None = None, api_version: str | None = None) → dict[str, Any]#

Build keyword arguments for a LiteLLM completion call.

Parameters:

profileLLMProfile: Resolved LLM provider profile used to build requests.
messagesSequence[dict[str, Any]]: Chat messages to send to the model.
max_tokensint | None, optional: Maximum number of tokens requested from the model. Default is None.
temperaturefloat | None, optional: Sampling temperature for providers that support it. Default is None.
api_keystr | None, optional: API key used to authenticate the provider request. Default is None.
api_basestr | None, optional: Provider API base URL override. Default is None.
api_versionstr | None, optional: Provider API version override, when required. Default is None.

Returns:

dict[str, Any]: Structured response payload for the requested operation.

llm_wire.auth_headers(api_key: str | None, *, auth_style: str = 'bearer', base: dict[str, str] | None = None) → dict[str, str]#

Build HTTP authorization headers for an LLM provider profile.

Parameters:

api_keystr | None: API key used to authenticate the provider request.
auth_stylestr, optional: Authorization-header style required by the provider. Default is 'bearer'.
basedict[str, str] | None, optional: Existing headers to extend. Default is None.

Returns:

dict[str, str]: Value computed by the helper for the requested CFX workflow.

llm_wire.extract_chat_text(payload: dict[str, Any]) → str#

Extract assistant text from a chat-completion provider response.

Parameters:

payloaddict[str, Any]: Structured payload to process.

Returns:

str: String value produced for the requested CFX or provider operation.

llm_wire.call(profile: LLMProfile, messages: Sequence[dict[str, Any]], *, max_tokens: int | None = None, temperature: float | None = None, api_key: str | None = None, api_base: str | None = None, api_version: str | None = None) → dict[str, Any]#

Send a synchronous chat-completion request through the configured LLM transport.

Parameters:

profileLLMProfile: Resolved LLM provider profile used to build requests.
messagesSequence[dict[str, Any]]: Chat messages to send to the model.
max_tokensint | None, optional: Maximum number of tokens requested from the model. Default is None.
temperaturefloat | None, optional: Sampling temperature for providers that support it. Default is None.
api_keystr | None, optional: API key used to authenticate the provider request. Default is None.
api_basestr | None, optional: Provider API base URL override. Default is None.
api_versionstr | None, optional: Provider API version override, when required. Default is None.

Returns:

dict[str, Any]: Structured response payload for the requested operation.

async llm_wire.acall(profile: LLMProfile, messages: Sequence[dict[str, Any]], *, max_tokens: int | None = None, temperature: float | None = None, api_key: str | None = None, api_base: str | None = None, api_version: str | None = None) → dict[str, Any]#

Send an asynchronous chat-completion request through the configured LLM transport.

Parameters:

profileLLMProfile: Resolved LLM provider profile used to build requests.
messagesSequence[dict[str, Any]]: Chat messages to send to the model.
max_tokensint | None, optional: Maximum number of tokens requested from the model. Default is None.
temperaturefloat | None, optional: Sampling temperature for providers that support it. Default is None.
api_keystr | None, optional: API key used to authenticate the provider request. Default is None.
api_basestr | None, optional: Provider API base URL override. Default is None.
api_versionstr | None, optional: Provider API version override, when required. Default is None.

Returns:

dict[str, Any]: Structured response payload for the requested operation.

llm_wire.load_aali_chat_model() → AaliChatModel | None#

Return the first usable chat model from the AALI config, or None.

Never raises: any I/O or parse error is logged and yields None so the caller can fall back to its own defaults.

Returns:

AaliChatModel | None: Result produced by the function.

llm_wire.DEFAULT_MODEL = 'gpt-4o-mini'#

llm_wire.PROVIDER_OPENAI = 'openai'#

llm_wire.PROVIDER_AZURE = 'azure'#

llm_wire.PROVIDER_ANTHROPIC = 'anthropic'#

llm_wire.PROVIDER_GEMINI = 'gemini'#

llm_wire.PROVIDER_COMPAT = 'compat'#

llm_wire.TRANSPORT_LITELLM = 'litellm'#

llm_wire.TRANSPORT_COMPAT = 'openai_compat'#

llm_wire.logger#

The llm_wire.py module#

Summary#

Description#

Module detail#

The `llm_wire.py` module#