Framework Utilities#

Supporting systems for advanced usage and development tooling.

Core Components#

models.get_model([provider, model_config, ...])

Create a configured LLM model instance for structured generation with PydanticAI.

models.get_chat_completion(message[, ...])

Execute direct chat completion requests across multiple AI providers.

logger.get_logger([source, component_name, ...])

Get a colored logger for any component with support for both structured and explicit APIs.

logger.ComponentLogger(base_logger, ...[, color])

Rich-formatted logger for framework and application components with color coding and message hierarchy.

streaming.get_streamer(source, component[, ...])

Get a stream writer for consistent streaming events.

streaming.StreamWriter(source, component[, ...])

Stream writer that provides consistent streaming events with automatic step counting.

Model Factory#

Multi-provider LLM model management for structured generation and direct completions.

framework.models.get_model(provider=None, model_config=None, model_id=None, api_key=None, base_url=None, timeout=None, max_tokens=100000)[source]#

Create a configured LLM model instance for structured generation with PydanticAI.

This factory function creates and configures LLM model instances optimized for structured generation workflows using PydanticAI agents. It handles provider-specific initialization, credential validation, HTTP client configuration, and proxy setup automatically based on environment variables and configuration files.

The function supports flexible configuration through multiple approaches: - Direct parameter specification for programmatic use - Model configuration dictionaries from YAML files - Automatic credential loading from configuration system - Environment-based HTTP proxy detection and configuration

Provider-specific behavior: - Anthropic: Requires API key and model ID, supports HTTP proxy - Google: Requires API key and model ID, supports HTTP proxy - OpenAI: Requires API key and model ID, supports HTTP proxy and custom base URLs - Ollama: Requires model ID and base URL, no API key needed, no proxy support - CBORG: Requires API key, model ID, and base URL, supports HTTP proxy

Parameters:

  • provider (str, optional) – AI provider name (‘anthropic’, ‘google’, ‘openai’, ‘ollama’, ‘cborg’)

  • model_config (dict, optional) – Configuration dictionary with provider, model_id, and other settings

  • model_id (str, optional) – Specific model identifier recognized by the provider

  • api_key (str, optional) – API authentication key, auto-loaded from config if not provided

  • base_url (str, optional) – Custom API endpoint URL, required for Ollama and CBORG

  • timeout (float, optional) – Request timeout in seconds, defaults to provider configuration

  • max_tokens (int) – Maximum tokens for generation, defaults to 100000

Raises:

  • ValueError – If required provider, model_id, api_key, or base_url are missing

  • ValueError – If provider is not supported

Returns:

Configured model instance ready for PydanticAI agent integration

Return type:

Union[OpenAIModel, AnthropicModel, GeminiModel]

Note

HTTP proxy configuration is automatically detected from the HTTP_PROXY environment variable for supported providers. Timeout and connection pooling are managed through shared HTTP clients when proxies are enabled.

Warning

API keys and base URLs are validated before model creation. Ensure proper configuration is available through the config system or direct parameter specification.

Examples

Basic model creation with direct parameters:

>>> from framework.models import get_model
>>> model = get_model(
...     provider="anthropic",
...     model_id="claude-3-sonnet-20240229",
...     api_key="your-api-key"
... )
>>> # Use with PydanticAI Agent
>>> agent = Agent(model=model, output_type=YourModel)

Using configuration dictionary from YAML:

>>> model_config = {
...     "provider": "cborg",
...     "model_id": "anthropic/claude-sonnet",
...     "max_tokens": 4096,
...     "timeout": 30.0
... }
>>> model = get_model(model_config=model_config)

Ollama local model setup:

>>> model = get_model(
...     provider="ollama",
...     model_id="llama3.1:8b",
...     base_url="http://localhost:11434"
... )

See also

get_chat_completion() : Direct chat completion without structured output configs.config.get_provider_config() : Provider configuration loading pydantic_ai.Agent : PydanticAI agent that uses these models Convention over Configuration: Configuration-Driven Registry Patterns : Complete model setup guide

framework.models.get_chat_completion(message, max_tokens=1024, model_config=None, provider=None, model_id=None, budget_tokens=None, enable_thinking=False, output_model=None, base_url=None)[source]#

Execute direct chat completion requests across multiple AI providers.

This function provides immediate access to LLM model inference with support for advanced features including extended thinking, structured outputs, and automatic TypedDict conversion. It handles provider-specific API differences, credential management, and HTTP proxy configuration transparently.

The function supports multiple interaction patterns: - Simple text-to-text completion for basic use cases - Structured output generation with Pydantic models or TypedDict - Extended thinking workflows for complex reasoning tasks - Enterprise proxy and timeout configuration

Provider-specific features: - Anthropic: Extended thinking with budget_tokens, content block responses - Google: Thinking configuration for enhanced reasoning - OpenAI: Structured outputs with beta chat completions API - Ollama: Local model inference with JSON schema validation - CBORG: OpenAI-compatible API with custom endpoints (LBNL-provided service)

Parameters:

  • message (str) – Input prompt or message for the LLM model

  • max_tokens (int) – Maximum tokens to generate in the response

  • model_config (dict, optional) – Configuration dictionary with provider and model settings

  • provider (str, optional) – AI provider name (‘anthropic’, ‘google’, ‘openai’, ‘ollama’, ‘cborg’)

  • model_id (str, optional) – Specific model identifier recognized by the provider

  • budget_tokens (int, optional) – Thinking budget for Anthropic/Google extended reasoning

  • enable_thinking (bool) – Enable extended thinking capabilities where supported

  • output_model (Type[BaseModel], optional) – Pydantic model or TypedDict for structured output validation

  • base_url (str, optional) – Custom API endpoint, required for Ollama and CBORG providers

Raises:

  • ValueError – If required provider, model_id, api_key, or base_url are missing

  • ValueError – If budget_tokens >= max_tokens or other invalid parameter combinations

  • pydantic.ValidationError – If output_model validation fails for structured outputs

  • anthropic.APIError – For Anthropic API-specific errors

  • openai.APIError – For OpenAI API-specific errors

  • ollama.ResponseError – For Ollama API-specific errors

Returns:

Model response in format determined by provider and output_model settings

Return type:

Union[str, BaseModel, list]

Note

Extended thinking is currently supported by Anthropic (with budget_tokens) and Google (with thinking_config). Other providers will log warnings if thinking parameters are provided.

Warning

When using structured outputs, ensure your prompt guides the model toward generating the expected structure. Not all models handle schema constraints equally well.

Examples

Simple text completion:

>>> from framework.models import get_chat_completion
>>> response = get_chat_completion(
...     message="Explain quantum computing in simple terms",
...     provider="anthropic",
...     model_id="claude-3-sonnet-20240229",
...     max_tokens=500
... )
>>> print(response)

Extended thinking with Anthropic:

>>> response = get_chat_completion(
...     message="Solve this complex reasoning problem...",
...     provider="anthropic",
...     model_id="claude-3-sonnet-20240229",
...     enable_thinking=True,
...     budget_tokens=1000,
...     max_tokens=2000
... )
>>> # Response includes thinking process and final answer

Structured output with Pydantic model:

>>> from pydantic import BaseModel
>>> class AnalysisResult(BaseModel):
...     summary: str
...     confidence: float
...     recommendations: list[str]
>>>
>>> result = get_chat_completion(
...     message="Analyze this data and provide structured results",
...     provider="openai",
...     model_id="gpt-4",
...     output_model=AnalysisResult
... )
>>> print(f"Confidence: {result.confidence}")

Using configuration dictionary:

>>> config = {
...     "provider": "ollama",
...     "model_id": "llama3.1:8b",
...     "max_tokens": 1000
... }
>>> response = get_chat_completion(
...     message="Hello, how are you?",
...     model_config=config,
...     base_url="http://localhost:11434"
... )

See also

get_model() : Create model instances for PydanticAI agents configs.config.get_provider_config() : Provider configuration loading pydantic.BaseModel : Base class for structured output models Convention over Configuration: Configuration-Driven Registry Patterns : Complete model configuration and usage guide

Developer Tools#

Component logging and streaming utilities for framework development.

Logging System#

configs.logger.get_logger(source=None, component_name=None, level=logging.INFO, *, name=None, color=None)[source]#

Get a colored logger for any component with support for both structured and explicit APIs.

Structured API (recommended):

source: Source type - either ‘framework’ or an application name (e.g., ‘als_expert’) component_name: Name of the component (e.g., ‘data_analysis’, ‘router’, ‘pv_finder’) level: Logging level

Explicit API (for custom loggers):

name: Direct logger name (keyword-only) color: Direct color specification (keyword-only) level: Logging level

Returns:

ComponentLogger instance

Return type:

ComponentLogger

Examples

# Structured API (recommended for framework/application components) logger = get_logger(“framework”, “orchestrator”) logger = get_logger(“als_expert”, “pv_finder”)

# Explicit API (for custom loggers or tests) logger = get_logger(name=”test_graph_execution”, color=”white”) logger = get_logger(name=”custom_component”, color=”blue”, level=logging.DEBUG)

class configs.logger.ComponentLogger(base_logger, component_name, color='white')[source]#

Bases: object

Rich-formatted logger for framework and application components with color coding and message hierarchy.

Message Types: - key_info: Important operational information - info: Normal operational messages - debug: Detailed tracing information - warning: Warning messages - error: Error messages - success: Success messages - timing: Timing information - approval: Approval messages - resume: Resume messages

Initialize component logger.

Parameters:

  • base_logger (Logger) – Underlying Python logger

  • component_name (str) – Name of the component (e.g., ‘data_analysis’, ‘router’, ‘mongo’)

  • color (str) – Rich color name for this component

__init__(base_logger, component_name, color='white')[source]#

Initialize component logger.

Parameters:

  • base_logger (Logger) – Underlying Python logger

  • component_name (str) – Name of the component (e.g., ‘data_analysis’, ‘router’, ‘mongo’)

  • color (str) – Rich color name for this component

key_info(message)[source]#

Important operational information.

info(message)[source]#

Normal operational information.

debug(message)[source]#

Detailed tracing information.

warning(message)[source]#

Warning messages.

error(message, exc_info=False)[source]#

Error messages.

success(message)[source]#

Success messages.

timing(message)[source]#

Timing messages.

approval(message)[source]#

Approval messages.

resume(message)[source]#

Resume messages.

critical(message, *args, **kwargs)[source]#
exception(message, *args, **kwargs)[source]#
log(level, message, *args, **kwargs)[source]#
property level: int#
property name: str#
setLevel(level)[source]#
isEnabledFor(level)[source]#
Return type:

bool

Streaming System#

configs.streaming.get_streamer(source, component, state=None)[source]#

Get a stream writer for consistent streaming events.

Parallels the get_logger() pattern for familiar usage.

Parameters:

  • source (str) – Source type (e.g., “framework”, “als_expert”)

  • component (str) – Component name (e.g., “orchestrator”, “python_executor”)

  • state (Any | None) – Optional AgentState for extracting execution context

Returns:

StreamWriter instance that handles event emission automatically

Return type:

StreamWriter

Example

streamer = get_streamer(“framework”, “orchestrator”, state) streamer.status(“Creating execution plan…”) streamer.success(“Plan created”)

class configs.streaming.StreamWriter(source, component, state=None)[source]#

Bases: object

Stream writer that provides consistent streaming events with automatic step counting.

Eliminates the need for manual if writer: checks and provides step context for task preparation phases.

Initialize stream writer with component context.

Parameters:

  • source (str) – Source type (e.g., “framework”, “als_expert”)

  • component (str) – Component name (e.g., “orchestrator”, “python_executor”)

  • state (Any | None) – Optional AgentState for extracting execution context

__init__(source, component, state=None)[source]#

Initialize stream writer with component context.

Parameters:

  • source (str) – Source type (e.g., “framework”, “als_expert”)

  • component (str) – Component name (e.g., “orchestrator”, “python_executor”)

  • state (Any | None) – Optional AgentState for extracting execution context

status(message)[source]#

Emit a status update event

error(message, error_data=None)[source]#

Emit an error event

warning(message)[source]#

Emit a warning event

See also

Orchestrator-First Architecture: Upfront Planning in Practice

Development utilities integration patterns and configuration conventions