LLM layer¶
The native, type-safe LLM abstraction: the client and its config/result types, message and content-part types, provider adapters, structured-output planning, routing helpers, pricing, telemetry, tracing, retry, and the typed error hierarchy.
Native LLM layer for penguiflow.
This module provides a native, type-safe LLM abstraction layer with: - Typed request/response models - Provider-specific adapters (OpenAI, Anthropic, Google, Bedrock, Databricks, OpenRouter) - Automatic output mode selection (native, tools, prompted) - Schema transformation for provider compatibility - Retry with LLM feedback - Cost tracking and telemetry - Streaming support
Quick Start
from penguiflow.llm import LLMClient, LLMMessage, TextPart from pydantic import BaseModel
class Answer(BaseModel): text: str confidence: float
client = LLMClient("gpt-4o") result = await client.generate( messages=[LLMMessage(role="user", parts=[TextPart(text="What is 2+2?")])], response_model=Answer, ) print(result.data.text)
For backward compatibility with existing planner code
from penguiflow.llm.protocol import NativeLLMAdapter
Drop-in replacement for _LiteLLMJSONClient¶
client = NativeLLMAdapter("gpt-4o") content, cost = await client.complete( messages=[{"role": "user", "content": "Hello"}], response_format={"type": "json_object"}, )
module-attribute
¶
ContentPart = TextPart | ToolCallPart | ToolResultPart | ImagePart | AudioPart
module-attribute
¶
ProviderType = Literal['openai', 'nim', 'anthropic', 'google', 'bedrock', 'databricks', 'openrouter', 'unknown']
LLMClient(model: str, *, api_key: str | None = None, base_url: str | None = None, fallback: ModelFallbackConfig | None = None, cooldown_store: CooldownStore | None = None, config: LLMClientConfig | None = None, provider: Provider | None = None, profile: ModelProfile | None = None, **provider_kwargs: Any)
High-level client for structured LLM interactions.
Provides automatic mode selection, retry with feedback, and cost tracking.
Example
from penguiflow.llm import LLMClient from pydantic import BaseModel
class Answer(BaseModel): text: str confidence: float
client = LLMClient("openai/gpt-4o") result = await client.generate( messages=[LLMMessage(role="user", parts=[TextPart(text="What is 2+2?")])], response_model=Answer, ) print(result.data.text) # "4"
Initialize the LLM client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier (e.g., "gpt-4o", "claude-3-5-sonnet"). |
required |
api_key
|
str | None
|
API key (uses environment variable if not provided). |
None
|
base_url
|
str | None
|
Base URL override for OpenAI-compatible endpoints. |
None
|
fallback
|
ModelFallbackConfig | None
|
Optional model fallback chain applied on 429s. |
None
|
cooldown_store
|
CooldownStore | None
|
Optional shared cooldown store used with fallback. |
None
|
config
|
LLMClientConfig | None
|
Client configuration. |
None
|
provider
|
Provider | None
|
Pre-configured provider instance (overrides model-based creation). |
None
|
profile
|
ModelProfile | None
|
Pre-configured model profile (overrides automatic lookup). |
None
|
**provider_kwargs
|
Any
|
Additional provider-specific configuration. |
{}
|
async
¶
generate(messages: list[LLMMessage], response_model: type[T], *, timeout_s: float | None = None, cancel: CancelToken | None = None, stream: bool = False, on_stream_event: StreamCallback | None = None, max_retries: int | None = None, temperature: float | None = None, force_mode: OutputMode | None = None) -> LLMResult
Generate structured output from an LLM.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
list[LLMMessage]
|
Conversation messages. |
required |
response_model
|
type[T]
|
Pydantic model for structured output. |
required |
timeout_s
|
float | None
|
Request timeout (uses config default if not specified). |
None
|
cancel
|
CancelToken | None
|
Cancellation token. |
None
|
stream
|
bool
|
Enable streaming. |
False
|
on_stream_event
|
StreamCallback | None
|
Streaming callback. |
None
|
max_retries
|
int | None
|
Override max retries. |
None
|
temperature
|
float | None
|
Override temperature. |
None
|
force_mode
|
OutputMode | None
|
Force specific output mode. |
None
|
Returns:
| Type | Description |
|---|---|
LLMResult
|
LLMResult with parsed data, usage, and cost. |
Raises:
| Type | Description |
|---|---|
LLMError
|
If the request fails after all retries. |
ValidationError
|
If the response cannot be validated. |
async
¶
complete_raw(request: LLMRequest, *, timeout_s: float | None = None, cancel: CancelToken | None = None, stream: bool = False, on_stream_event: StreamCallback | None = None) -> CompletionResponse
Execute a raw completion request without structured output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
LLMRequest
|
The LLM request. |
required |
timeout_s
|
float | None
|
Request timeout. |
None
|
cancel
|
CancelToken | None
|
Cancellation token. |
None
|
stream
|
bool
|
Enable streaming. |
False
|
on_stream_event
|
StreamCallback | None
|
Streaming callback. |
None
|
Returns:
| Type | Description |
|---|---|
CompletionResponse
|
Raw completion response. |
dataclass
¶
LLMClientConfig(max_retries: int = 3, retry_on_validation: bool = True, retry_on_parse: bool = True, retry_on_provider_errors: bool = True, timeout_s: float = 120.0, temperature: float | None = None, force_mode: OutputMode | None = None, enable_telemetry: bool = True, enable_cost_tracking: bool = True)
Configuration for LLMClient.
dataclass
¶
LLMResult(data: BaseModel, usage: Usage, cost: Cost, mode_used: OutputMode, attempts: int, raw_response: Any = None)
Result from an LLM structured output call.
dataclass
¶
LLMMessage(role: Role, parts: tuple[ContentPart, ...] | list[ContentPart])
A single message in a conversation.
Providers adapt SDK-specific payload shapes (Anthropic content blocks, Google parts, Bedrock converse formats) to/from this type.
dataclass
¶
LLMRequest(model: str, messages: tuple[LLMMessage, ...] | list[LLMMessage], tools: tuple[ToolSpec, ...] | list[ToolSpec] | None = None, tool_choice: str | None = None, structured_output: StructuredOutputSpec | None = None, temperature: float | None = None, max_tokens: int | None = None, extra: dict[str, Any] | None = None)
A typed request to an LLM provider.
This is the normalized request format that all output strategies produce and all providers consume.
dataclass
¶
ImagePart(data: bytes, media_type: str, detail: Literal['auto', 'low', 'high'] = 'auto')
dataclass
¶
AudioPart(data: bytes, media_type: str)
dataclass
¶
ToolCallPart(name: str, arguments_json: str, call_id: str | None = None)
dataclass
¶
ToolResultPart(name: str, result_json: str, call_id: str | None = None, is_error: bool = False)
dataclass
¶
ToolSpec(name: str, description: str, json_schema: dict[str, Any])
dataclass
¶
StructuredOutputSpec(name: str, json_schema: dict[str, Any], strict: bool = True)
dataclass
¶
CompletionResponse(message: LLMMessage, usage: Usage, raw_response: Any = None, reasoning_content: str | None = None, finish_reason: str | None = None)
Normalized response from a completion call.
Providers adapt SDK responses into this portable shape so that the rest of the system never needs to interpret provider-specific payload formats.
dataclass
¶
StreamEvent(delta_text: str | None = None, delta_reasoning: str | None = None, delta_tool_call: ToolCallPart | None = None, usage: Usage | None = None, done: bool = False, finish_reason: str | None = None)
A streaming event emitted during completion.
This is a minimal common denominator; providers can emit richer events internally.
CancelToken()
dataclass
¶
Cost(input_cost: float, output_cost: float, total_cost: float, currency: str = 'USD')
Cost information for an LLM call.
dataclass
¶
Usage(input_tokens: int, output_tokens: int, total_tokens: int)
Bases: ABC
Abstract base class for LLM providers.
Each provider implementation: - Uses native SDK directly (openai, anthropic, google-genai, boto3) - Handles SDK-specific payload shapes (content blocks, parts, etc.) - Normalizes responses into typed CompletionResponse - Implements streaming via StreamCallback - Respects cancellation tokens and timeouts
abstractmethod
property
¶
provider_name: str
Get the provider name (e.g., 'openai', 'anthropic').
abstractmethod
property
¶
profile: ModelProfile
Get the model profile for capability information.
property
¶
temperature_unsupported: bool
Whether temperature has been disabled for this model at runtime.
Set by :meth:mark_temperature_unsupported after a provider returns a
temperature-related 400, so subsequent requests omit the parameter.
abstractmethod
async
¶
complete(request: LLMRequest, *, timeout_s: float | None = None, cancel: CancelToken | None = None, stream: bool = False, on_stream_event: StreamCallback | None = None) -> CompletionResponse
Execute a completion request.
Requirements:
- Respect cancellation (raise asyncio.CancelledError or LLMCancelledError)
- Enforce timeout_s (raise LLMTimeoutError)
- Emit StreamEvent via on_stream_event if streaming is enabled
- Normalize provider-specific responses into CompletionResponse.message
- Map SDK exceptions to LLMError subclasses
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
LLMRequest
|
The typed request to execute. |
required |
timeout_s
|
float | None
|
Optional timeout in seconds. |
None
|
cancel
|
CancelToken | None
|
Optional cancellation token. |
None
|
stream
|
bool
|
Whether to enable streaming. |
False
|
on_stream_event
|
StreamCallback | None
|
Callback for streaming events. |
None
|
Returns:
| Type | Description |
|---|---|
CompletionResponse
|
Normalized completion response. |
Raises:
| Type | Description |
|---|---|
LLMError
|
For any provider errors (mapped to appropriate subclass). |
CancelledError
|
If cancelled via cancel token. |
validate_request(request: LLMRequest) -> None
Validate a request before sending.
Default implementation does nothing. Providers can override to add provider-specific validation (e.g., tool count limits, schema complexity).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
LLMRequest
|
The request to validate. |
required |
Raises:
| Type | Description |
|---|---|
LLMInvalidRequestError
|
If validation fails. |
mark_temperature_unsupported() -> None
Record that this model rejects the temperature parameter.
Subsequent _build_params calls drop temperature so the same
request can be retried successfully without a static profile change.
dataclass
¶
SchemaPlan(requested_schema: dict[str, Any], transformed_schema: dict[str, Any], strict_requested: bool, strict_applied: bool, compatible_with_native: bool, compatible_with_tools: bool, reasons: tuple[str, ...] = (), estimated_total_keys: int | None = None, has_recursive_refs: bool = False, has_composition: bool = False)
Result of schema planning for a specific provider.
Contains the transformed schema along with compatibility information that guides mode selection and degradation.
NativeLLMAdapter(model: str, *, api_key: str | None = None, base_url: str | None = None, temperature: float | None = None, max_retries: int = 3, timeout_s: float = 120.0, json_schema_mode: bool = True, streaming_enabled: bool = True, use_native_reasoning: bool = True, reasoning_effort: str | None = None, reasoning_display: ReasoningDisplay = None, retry_rate_limit_errors: bool = True, trace_sink: LLMTraceSink | None = None, transport: str | None = None, multimodal_inline_data_limit_bytes: int = INLINE_MULTIMODAL_DATA_LIMIT_BYTES, **provider_kwargs: Any)
Adapter that implements JSONLLMClient protocol using the native LLM layer.
This class provides backward compatibility with the existing planner infrastructure while using the new native provider implementations.
Example
from penguiflow.llm.protocol import NativeLLMAdapter
Create adapter that implements JSONLLMClient protocol¶
client = NativeLLMAdapter("openai/gpt-4o")
Use with existing planner code¶
result = await client.complete( messages=[{"role": "user", "content": "Hello"}], response_format={"type": "json_object"}, )
Initialize the adapter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier (e.g., "gpt-4o", "claude-3-5-sonnet"). |
required |
api_key
|
str | None
|
API key (uses environment variable if not provided). |
None
|
base_url
|
str | None
|
Base URL override. |
None
|
temperature
|
float | None
|
Default temperature. |
None
|
max_retries
|
int
|
Maximum retry attempts. |
3
|
timeout_s
|
float
|
Request timeout in seconds. |
120.0
|
json_schema_mode
|
bool
|
Enable JSON schema mode for structured output. |
True
|
streaming_enabled
|
bool
|
Enable streaming support. |
True
|
use_native_reasoning
|
bool
|
Enable native reasoning for supported models. |
True
|
reasoning_effort
|
str | None
|
Reasoning effort level (e.g., "low", "medium", "high"). |
None
|
reasoning_display
|
ReasoningDisplay
|
Reasoning display mode ("summarized" or "omitted"). |
None
|
retry_rate_limit_errors
|
bool
|
When False, a 429 is raised immediately
instead of being retried in-place. Used by |
True
|
trace_sink
|
LLMTraceSink | None
|
Optional |
None
|
transport
|
str | None
|
|
None
|
multimodal_inline_data_limit_bytes
|
int
|
Maximum inline byte size for each image/audio content part before provider serialization. Defaults to 32 KiB. |
INLINE_MULTIMODAL_DATA_LIMIT_BYTES
|
**provider_kwargs
|
Any
|
Additional provider-specific configuration. |
{}
|
async
¶
complete(*, messages: Sequence[Mapping[str, Any]], response_format: Mapping[str, Any] | None = None, stream: bool = False, on_stream_chunk: Callable[[str, bool], None] | None = None, on_reasoning_chunk: Callable[[str, bool], None] | None = None) -> tuple[str, float]
Execute a completion following the JSONLLMClient protocol.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
Sequence[Mapping[str, Any]]
|
Conversation messages in dict format. |
required |
response_format
|
Mapping[str, Any] | None
|
Optional response format specification. |
None
|
stream
|
bool
|
Enable streaming. |
False
|
on_stream_chunk
|
Callable[[str, bool], None] | None
|
Callback for streaming text chunks. |
None
|
on_reasoning_chunk
|
Callable[[str, bool], None] | None
|
Callback for reasoning content chunks. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[str, float]
|
Tuple of (content, cost) matching JSONLLMClient protocol. |
async
¶
stream_events(*, messages: Sequence[Mapping[str, Any]], response_format: Mapping[str, Any] | None = None, timeout_s: float | None = None) -> AsyncIterator[StreamEvent]
Stream events as an async iterator.
This provides a LiteLLM-style "pull" streaming API (async iteration) on top of the provider callback-based streaming contract.
async
¶
complete_with_tools(*, messages: Sequence[Mapping[str, Any]], tools: Sequence[ToolSpec], stream: bool = False, on_stream_chunk: Callable[[str, bool], None] | None = None, on_reasoning_chunk: Callable[[str, bool], None] | None = None) -> NativeToolCallResult
Execute a completion with provider-native tool declarations.
Additive surface for the planner's native tool-calling mode --
JSONLLMClient.complete() is untouched. Content streams via
on_stream_chunk (the planner decides which channel it belongs
to); tool-call fragment assembly is the provider's job and the
assembled calls are returned, never streamed.
FallbackLLMClient(primary_model: str, config: ModelFallbackConfig, *, cooldown_store: CooldownStore | None = None, adapter_factory: AdapterFactory | None = None, default_api_key: str | None = None, **adapter_kwargs: Any)
A JSONLLMClient that fails over across models on rate limits.
The effective chain leads with primary_model followed by the configured
fallback models, so selection always prefers the primary and reverts to it
once its cooldown expires.
async
¶
complete(*, messages: Sequence[Mapping[str, Any]], response_format: Mapping[str, Any] | None = None, stream: bool = False, on_stream_chunk: Callable[[str, bool], None] | None = None, on_reasoning_chunk: Callable[[str, bool], None] | None = None) -> tuple[str, float]
Run a completion, failing over to fallback models on 429.
async
¶
complete_with_tools(*, messages: Sequence[Mapping[str, Any]], tools: Sequence[Any], stream: bool = False, on_stream_chunk: Callable[[str, bool], None] | None = None, on_reasoning_chunk: Callable[[str, bool], None] | None = None) -> Any
Native tool-calling sibling of complete with the same failover.
dataclass
¶
ModelFallbackConfig(models: list[str], api_keys: list[str] | None = None, cooldown_s: float = 45.0, max_wait_s: float = 30.0)
Developer-facing configuration for rate-limit model fallback.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
models
|
list[str]
|
Ordered fallback chain, highest priority first. |
required |
api_keys
|
list[str] | None
|
Optional key pool. |
None
|
cooldown_s
|
float
|
Cooldown applied to a |
45.0
|
max_wait_s
|
float
|
When every entry is cooling down, the longest a call will block waiting for the soonest cooldown to expire before failing. |
30.0
|
classmethod
¶
from_env(model_vars: Sequence[str], *, api_key_vars: Sequence[str] | None = None, cooldown_s: float = 45.0, max_wait_s: float = 30.0) -> ModelFallbackConfig
Build a config from named environment variables.
model_vars are read in order; missing/empty variables are skipped.
CooldownStore()
Tracks cooldown expiry per (model, key_index) pair.
The default implementation keeps state in memory for the lifetime of the
instance (v1: one store shared by all clients of a single planner run).
Subclass and override the methods to widen the scope (e.g. a process-wide
singleton) without changing :class:FallbackLLMClient.
set_cooldown(entry: Entry, seconds: float) -> float
Place entry in cooldown for seconds; returns the expiry time.
seconds_until_earliest(entries: Sequence[Entry]) -> float | None
Seconds until the soonest cooldown among entries expires.
Returns None when at least one entry is already available.
dataclass
¶
ParsedModel(provider: ProviderType, model_id: str, original: str, sub_provider: str | None = None)
dataclass
¶
ModelProfile(supports_schema_guided_output: bool = False, supports_json_only_output: bool = True, supports_tools: bool = True, supports_native_tool_calls: bool = True, supports_reasoning: bool = False, supports_streaming: bool = True, supports_image_input: bool = False, supports_audio_input: bool = False, supports_temperature: bool = True, default_output_mode: Literal['native', 'tools', 'prompted'] = 'native', native_structured_kind: Literal['openai_response_format', 'databricks_constrained_decoding', 'anthropic_tool_use', 'google_response_schema', 'bedrock_tool_use', 'openai_compatible_tools', 'unknown'] = 'unknown', schema_transformer_name: str | None = None, reasoning_effort_param: str | None = None, thinking_tags: tuple[str, str] | None = None, reasoning_display_default: Literal['summarized', 'omitted'] | None = None, reasoning_request_style: Literal['adaptive_effort', 'thinking_budget', 'reasoning_effort'] | None = None, preferred_transport: Literal['native', 'pydantic-ai'] | None = None, strict_mode_default: bool = True, supports_system_role: bool = True, drop_unsupported_params: bool = True, unsupported_request_params: frozenset[str] = frozenset(), max_tools: int | None = None, max_schema_keys: int | None = None, max_context_tokens: int | None = None, max_output_tokens: int | None = None)
Describes capabilities and configuration for a specific model.
This dataclass captures provider-specific capabilities and quirks, allowing the LLM layer to make informed decisions about: - Which structured output mode to use - How to transform schemas for compatibility - What parameters are supported
class-attribute
instance-attribute
¶
supports_schema_guided_output: bool = False
class-attribute
instance-attribute
¶
supports_json_only_output: bool = True
class-attribute
instance-attribute
¶
supports_native_tool_calls: bool = True
class-attribute
instance-attribute
¶
default_output_mode: Literal['native', 'tools', 'prompted'] = 'native'
class-attribute
instance-attribute
¶
native_structured_kind: Literal['openai_response_format', 'databricks_constrained_decoding', 'anthropic_tool_use', 'google_response_schema', 'bedrock_tool_use', 'openai_compatible_tools', 'unknown'] = 'unknown'
class-attribute
instance-attribute
¶
schema_transformer_name: str | None = None
class-attribute
instance-attribute
¶
reasoning_effort_param: str | None = None
class-attribute
instance-attribute
¶
reasoning_display_default: Literal['summarized', 'omitted'] | None = None
class-attribute
instance-attribute
¶
reasoning_request_style: Literal['adaptive_effort', 'thinking_budget', 'reasoning_effort'] | None = None
class-attribute
instance-attribute
¶
preferred_transport: Literal['native', 'pydantic-ai'] | None = None
class-attribute
instance-attribute
¶
unsupported_request_params: frozenset[str] = frozenset()
dataclass
¶
LLMEvent(event_type: str, timestamp: datetime, provider: str, model: str, trace_id: str | None = None, extra: dict[str, Any] | None = None)
Event emitted during LLM operations for observability.
classmethod
¶
create(event_type: str, provider: str, model: str, *, trace_id: str | None = None, **extra: Any) -> LLMEvent
Create an event with current timestamp.
TelemetryHooks()
Pluggable telemetry for LLM operations.
Integrates with PenguiFlow's metrics system and supports external observability platforms (MLflow, Prometheus, OpenTelemetry).
Usage
hooks = get_telemetry_hooks() hooks.register(my_callback)
In your code¶
hooks.emit(LLMEvent.create("request_start", "openai", "gpt-4o"))
register(callback: TelemetryCallback) -> None
Register a telemetry callback.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
TelemetryCallback
|
Function to call with each event. |
required |
unregister(callback: TelemetryCallback) -> bool
Unregister a telemetry callback.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
TelemetryCallback
|
The callback to remove. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if callback was found and removed. |
TimingContext(provider: str, model: str, trace_id: str | None = None)
Context manager for timing LLM operations.
dataclass
¶
RetryConfig(max_retries: int = 3, retry_on_validation: bool = True, retry_on_parse: bool = True, retry_on_provider_errors: bool = True, initial_backoff_s: float = 1.0, max_backoff_s: float = 30.0, backoff_multiplier: float = 2.0)
Configuration for retry behavior.
dataclass
¶
RetryState(attempt: int = 0, total_cost: float = 0.0, errors: list[Exception] = list())
ModelRetry(message: str, validation_errors: list[dict[str, Any]] | None = None)
ValidationRetry(errors: list[dict[str, Any]], raw_content: str)
dataclass
¶
LLMCallSpan(provider: str, model: str, stream: bool = False, response_format_kind: str | None = None, attempts: int | None = None, content_chars: int | None = None, input_tokens: int | None = None, output_tokens: int | None = None, cost_usd: float | None = None, latency_ms: float | None = None, error_type: str | None = None, error_message: str | None = None)
Mutable record describing one LLM completion call.
Request metadata is set when the span opens; outcome fields are filled in
before it closes. error_* fields are populated when the call raises.
outcome_attributes() -> dict[str, Any]
Outcome fields that have a value, as a flat attribute dict.
Bases: Protocol
Receives one span per LLM completion call.
span(call: LLMCallSpan) -> AbstractContextManager[Any]
Open a span for call. Outcome fields are set before exit.
LoggingLLMTraceSink(*, level: int = logging.INFO, logger_name: str = 'penguiflow.llm.trace')
Zero-dependency sink emitting one structured log line per LLM call.
MlflowLLMTraceSink(*, span_name: str = 'llm.complete')
Sink emitting MLflow Tracing spans (span_type="LLM").
Requires mlflow>=2.14 (the tracing fluent API). mlflow is imported
lazily on first use; if it is missing or too old the sink degrades to a
no-op with a single warning — it never fails the LLM call. Spans nest
under any active MLflow trace (e.g. an autologged agent run) or start a
new trace when none is active.
dataclass
¶
LLMError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = False, raw: Any = None)
Bases: Exception
Base class for all LLM errors.
All LLM errors carry: - message: A clean, user-readable error message - provider: The provider that raised the error (optional) - status_code: HTTP status code if applicable (optional) - retryable: Whether this error should be retried - raw: The original exception or response for debugging
dataclass
¶
LLMAuthError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = False, raw: Any = None)
dataclass
¶
LLMCancelledError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None)
dataclass
¶
LLMContextLengthError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = False, raw: Any = None, max_tokens: int | None = None, current_tokens: int | None = None)
dataclass
¶
LLMInvalidRequestError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = False, raw: Any = None)
dataclass
¶
LLMParseError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None, raw_content: str = '')
dataclass
¶
LLMRateLimitError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None, retry_after: float | None = None)
dataclass
¶
LLMServerError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None)
dataclass
¶
LLMTimeoutError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None)
dataclass
¶
LLMValidationError(message: str, provider: str | None = None, status_code: int | None = None, retryable: bool = True, raw: Any = None, validation_errors: list[dict[str, Any]] = list())
async
¶
generate_structured(model: str, messages: list[LLMMessage], response_model: type[T], *, api_key: str | None = None, fallback: ModelFallbackConfig | None = None, cooldown_store: CooldownStore | None = None, **kwargs: Any) -> T
Convenience function for one-shot structured generation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier. |
required |
messages
|
list[LLMMessage]
|
Conversation messages. |
required |
response_model
|
type[T]
|
Pydantic model for structured output. |
required |
api_key
|
str | None
|
API key. |
None
|
fallback
|
ModelFallbackConfig | None
|
Optional model fallback chain applied on 429s. |
None
|
cooldown_store
|
CooldownStore | None
|
Optional shared cooldown store used with fallback. |
None
|
**kwargs
|
Any
|
Additional arguments for LLMClient.generate(). |
{}
|
Returns:
| Type | Description |
|---|---|
T
|
Parsed Pydantic model instance. |
extract_single_tool_call(message: LLMMessage, expected_name: str | None = None) -> ToolCallPart
Extract a single tool call from a message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
LLMMessage
|
The message to extract from. |
required |
expected_name
|
str | None
|
If provided, verify the tool call has this name. |
None
|
Returns:
| Type | Description |
|---|---|
ToolCallPart
|
The tool call part. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no tool call found or multiple tool calls present. |
strip_markdown_fences(text: str) -> str
Strip markdown code fences from text if present.
create_provider(model: str, *, api_key: str | None = None, base_url: str | None = None, **kwargs: Any) -> Provider
Create a provider instance based on model string.
Model string formats: - "openai/gpt-4o" or "gpt-4o" -> OpenAI - "nim/qwen/qwen3.5-397b-a17b", "nvidia/qwen/qwen3.5-397b-a17b", or "nvidia/nemotron-3-nano-30b-a3b" -> NIM - "anthropic/claude-3-5-sonnet" or "claude-" -> Anthropic - "google/gemini-2.0-flash" or "gemini-" -> Google - "bedrock/anthropic.claude-3-5-sonnet" or "anthropic." -> Bedrock - "databricks/databricks-dbrx-instruct" or "databricks-" -> Databricks - "openrouter/anthropic/claude-3-5-sonnet" -> OpenRouter
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier with optional provider prefix. |
required |
api_key
|
str | None
|
API key (uses environment variable if not provided). |
None
|
base_url
|
str | None
|
Base URL override. |
None
|
**kwargs
|
Any
|
Provider-specific configuration. |
{}
|
Returns:
| Type | Description |
|---|---|
Provider
|
Configured provider instance. |
choose_output_mode(profile: ModelProfile, schema: dict[str, Any]) -> tuple[OutputMode, SchemaPlan]
Choose the best output mode for a schema and provider.
Follows the mode preference ladder: 1. Profile's default mode (if compatible) 2. NATIVE (if supported and compatible) 3. TOOLS (if supported and compatible) 4. PROMPTED (always available)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
profile
|
ModelProfile
|
The model profile. |
required |
schema
|
dict[str, Any]
|
The JSON schema for structured output. |
required |
Returns:
| Type | Description |
|---|---|
tuple[OutputMode, SchemaPlan]
|
Tuple of (chosen mode, schema plan). |
plan_schema(profile: ModelProfile, schema: dict[str, Any], *, mode: OutputMode | None = None) -> SchemaPlan
Plan schema transformation for a specific provider and mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
profile
|
ModelProfile
|
The model profile with capabilities. |
required |
schema
|
dict[str, Any]
|
The JSON schema to plan for. |
required |
mode
|
OutputMode | None
|
Optional specific mode to plan for. |
None
|
Returns:
| Type | Description |
|---|---|
SchemaPlan
|
A SchemaPlan with transformation results and compatibility info. |
create_native_adapter(model: str | Mapping[str, Any], *, temperature: float | None = None, json_schema_mode: bool = True, max_retries: int = 3, timeout_s: float = 360.0, streaming_enabled: bool = True, use_native_reasoning: bool = True, reasoning_effort: str | None = None, reasoning_display: ReasoningDisplay | None = None, fallback: Any | None = None, cooldown_store: Any | None = None, trace_sink: LLMTraceSink | None = None, transport: str | None = None, multimodal_inline_data_limit_bytes: int = INLINE_MULTIMODAL_DATA_LIMIT_BYTES, **kwargs: Any) -> Any
Factory function to create a NativeLLMAdapter (or fallback client).
Accepts the same configuration style as the existing _LiteLLMJSONClient for easy migration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str | Mapping[str, Any]
|
Model identifier string or config dict. |
required |
temperature
|
float | None
|
Default temperature. |
None
|
json_schema_mode
|
bool
|
Enable JSON schema mode. |
True
|
max_retries
|
int
|
Maximum retry attempts. |
3
|
timeout_s
|
float
|
Request timeout. |
360.0
|
streaming_enabled
|
bool
|
Enable streaming. |
True
|
use_native_reasoning
|
bool
|
Enable native reasoning for supported models. |
True
|
reasoning_effort
|
str | None
|
Reasoning effort level (e.g., "low", "medium", "high"). |
None
|
reasoning_display
|
ReasoningDisplay | None
|
Reasoning display mode ("summarized" or "omitted"). |
None
|
fallback
|
Any | None
|
Optional |
None
|
cooldown_store
|
Any | None
|
Optional shared |
None
|
trace_sink
|
LLMTraceSink | None
|
Optional |
None
|
transport
|
str | None
|
|
None
|
multimodal_inline_data_limit_bytes
|
int
|
Maximum inline byte size for each image/audio content part before provider serialization. Defaults to 32 KiB and is forwarded to every fallback-chain adapter. |
INLINE_MULTIMODAL_DATA_LIMIT_BYTES
|
**kwargs
|
Any
|
Additional provider configuration. |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
A |
Any
|
Both satisfy the |
parse_model_string(model: str) -> ParsedModel
Parse a model string into provider and model components.
Supported formats: - "gpt-4o" -> (openai, gpt-4o) - "openai/gpt-4o" -> (openai, gpt-4o) - "nim/qwen/qwen3.5-397b-a17b" -> (nim, qwen/qwen3.5-397b-a17b) - "nvidia/qwen/qwen3.5-397b-a17b" -> (nim, qwen/qwen3.5-397b-a17b) - "claude-3-5-sonnet" -> (anthropic, claude-3-5-sonnet) - "anthropic/claude-3-5-sonnet" -> (anthropic, claude-3-5-sonnet) - "gemini-2.0-flash" -> (google, gemini-2.0-flash) - "google/gemini-2.0-flash" -> (google, gemini-2.0-flash) - "anthropic.claude-3-5-sonnet-v2" -> (bedrock, anthropic.claude-3-5-sonnet-v2) - "bedrock/anthropic.claude-3-5-sonnet" -> (bedrock, anthropic.claude-3-5-sonnet) - "databricks-dbrx-instruct" -> (databricks, databricks-dbrx-instruct) - "databricks/databricks-dbrx-instruct" -> (databricks, databricks-dbrx-instruct) - "openrouter/anthropic/claude-3-5-sonnet" -> (openrouter, anthropic/claude-3-5-sonnet, anthropic)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
ParsedModel
|
ParsedModel with provider and model_id. |
build_model_string(provider: ProviderType, model_id: str, *, sub_provider: str | None = None) -> str
Build a full model string from components.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
provider
|
ProviderType
|
Provider type. |
required |
model_id
|
str
|
Model identifier. |
required |
sub_provider
|
str | None
|
Sub-provider for OpenRouter. |
None
|
Returns:
| Type | Description |
|---|---|
str
|
Full model string. |
normalize_model_id(model: str) -> str
Normalize a model ID by removing provider prefix.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Model ID without provider prefix. |
get_provider_for_model(model: str) -> ProviderType
Get the provider type for a model string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
ProviderType
|
Provider type string. |
estimate_context_window(model: str) -> int
Estimate the context window size for a model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
int
|
Estimated context window in tokens. |
is_reasoning_model(model: str) -> bool
Check if a model supports native reasoning/thinking.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the model supports native reasoning. |
is_vision_model(model: str) -> bool
Check if a model supports vision/image input.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier string. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the model supports vision. |
get_profile(model: str) -> ModelProfile
Get profile for a model, with fallback to defaults.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
The model identifier (e.g., "gpt-4o", "claude-3-5-sonnet"). |
required |
Returns:
| Type | Description |
|---|---|
ModelProfile
|
The model profile, or a default profile if not found. |
register_profile(model: str, profile: ModelProfile) -> None
Register a custom profile for a model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
The model identifier. |
required |
profile
|
ModelProfile
|
The model profile. |
required |
calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float
Calculate cost for a completion in USD.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier. |
required |
input_tokens
|
int
|
Number of input tokens. |
required |
output_tokens
|
int
|
Number of output tokens. |
required |
Returns:
| Type | Description |
|---|---|
float
|
Total cost in USD. |
calculate_cost_from_usage(model: str, usage: Usage) -> Cost
get_pricing(model: str) -> tuple[float, float]
Get pricing for a model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier. |
required |
Returns:
| Type | Description |
|---|---|
tuple[float, float]
|
Tuple of (input_price_per_1k, output_price_per_1k) in USD. |
register_pricing(model: str, input_price: float, output_price: float) -> None
Register custom pricing for a model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str
|
Model identifier. |
required |
input_price
|
float
|
Price per 1K input tokens in USD. |
required |
output_price
|
float
|
Price per 1K output tokens in USD. |
required |
create_mlflow_callback() -> TelemetryCallback
Create a callback that logs to MLflow.
Returns:
| Type | Description |
|---|---|
TelemetryCallback
|
Callback function for MLflow integration. |
Example
from penguiflow.llm.telemetry import get_telemetry_hooks, create_mlflow_callback hooks = get_telemetry_hooks() hooks.register(create_mlflow_callback())
create_prometheus_callback() -> TelemetryCallback
Create a callback that updates Prometheus metrics.
Returns:
| Type | Description |
|---|---|
TelemetryCallback
|
Callback function for Prometheus integration. |
Example
from penguiflow.llm.telemetry import get_telemetry_hooks, create_prometheus_callback hooks = get_telemetry_hooks() hooks.register(create_prometheus_callback())
get_telemetry_hooks() -> TelemetryHooks
Get the global telemetry hooks instance.
set_telemetry_hooks(hooks: TelemetryHooks) -> None
Replace the global telemetry hooks instance (for testing).
async
¶
call_with_retry(provider: Provider, base_messages: list[LLMMessage], response_model: type[T], output_strategy: Any, *, config: RetryConfig | None = None, on_retry: Callable[[int, Exception], None] | None = None, timeout_s: float | None = None, cancel: CancelToken | None = None, stream: bool = False, on_stream_event: StreamCallback | None = None, pricing_fn: Callable[[str, int, int], float] | None = None, build_request: Callable[[list[LLMMessage]], LLMRequest] | None = None, profile: Any = None, plan: Any = None, temperature: float | None = None) -> tuple[T, float]
Execute LLM call with automatic retry and cost accounting.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
provider
|
Provider
|
The LLM provider to use. |
required |
base_messages
|
list[LLMMessage]
|
Initial conversation messages. |
required |
response_model
|
type[T]
|
Pydantic model for structured output. |
required |
output_strategy
|
Any
|
Strategy for building requests and parsing responses. |
required |
config
|
RetryConfig | None
|
Retry configuration. |
None
|
on_retry
|
Callable[[int, Exception], None] | None
|
Callback called on each retry (attempt_num, error). |
None
|
timeout_s
|
float | None
|
Request timeout in seconds. |
None
|
cancel
|
CancelToken | None
|
Cancellation token. |
None
|
stream
|
bool
|
Whether to enable streaming. |
False
|
on_stream_event
|
StreamCallback | None
|
Streaming callback. |
None
|
pricing_fn
|
Callable[[str, int, int], float] | None
|
Function to calculate cost (model, input_tokens, output_tokens). |
None
|
build_request
|
Callable[[list[LLMMessage]], LLMRequest] | None
|
Custom request builder function. |
None
|
profile
|
Any
|
Model profile for building requests. |
None
|
plan
|
Any
|
Schema plan for building requests. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[T, float]
|
Tuple of (parsed_response, total_cost). |
Raises:
| Type | Description |
|---|---|
LLMError
|
If all retries fail. |
ValidationError
|
If final retry fails validation. |
resolve_trace_sink_from_env() -> LLMTraceSink | None
Build a trace sink from PENGUIFLOW_LLM_TRACING (transparent opt-in).
Supported values: mlflow (MLflow Tracing spans) and log /
logging (structured log lines). Unset or falsy values disable
tracing. Unknown values are ignored with a warning so a typo can never
break a production run.
is_context_length_error(error: Exception | str) -> bool
Check if an error is related to context length.
map_status_to_error(status_code: int, message: str, provider: str | None = None, raw: Any = None) -> LLMError
Map an HTTP status code to the appropriate LLMError subclass.