Utils Module

The utils module provides utility functions and helpers used throughout Pantheon.

Submodules

LLM Utilities

async pantheon.utils.llm.acompletion_openai(messages: list[dict], model: str, tools: list[dict] | None = None, response_format: Any | None = None, process_chunk: Callable | None = None, retry_times: int = 3, base_url: str | None = None, model_params: dict | None = None)

async pantheon.utils.llm.acompletion_responses(messages: list[dict], model: str, tools: list[dict] | None = None, response_format: Any | None = None, process_chunk: Callable | None = None, base_url: str | None = None, api_key: str | None = None, model_params: dict | None = None, num_retries: int = 3) → dict

Call OpenAI Responses API with streaming.

Used for models that require the Responses API (e.g. codex-mini-latest). Returns a normalised message dict compatible with extract_message_from_response.

pantheon.utils.llm.stream_chunk_builder(chunks: list[dict]) → Any

Assemble streaming chunks into a complete response object.

Aggregates content deltas, tool_call deltas, and usage from collected chunks into a SimpleNamespace that mimics the shape of a chat completion response.

Replaces the stream_chunk_builder from external dependencies.

async pantheon.utils.llm.acompletion(messages: list[dict], model: str, tools: list[dict] | None = None, response_format: Any | None = None, process_chunk: Callable | None = None, base_url: str | None = None, api_key: str | None = None, model_params: dict | None = None, num_retries: int = 3)

Call LLM via provider adapters.

Two modes of operation:

1. PROVIDER MODE: - The requested provider has explicit credentials configured - Calls use the provider’s native or OpenAI-compatible adapter as usual

2. OPENAI-COMPATIBLE FALLBACK MODE: - LLM_API_BASE + LLM_API_KEY are configured - OpenAI models use that fallback endpoint instead of the official base URL

pantheon.utils.llm.remove_parsed(messages: list[dict]) → list[dict]

pantheon.utils.llm.remove_reasoning_content(messages: list[dict]) → list[dict]

Remove reasoning fields from messages (prevent context pollution).

Removes both reasoning_content (unified field) and thinking_blocks (Claude-specific).

pantheon.utils.llm.convert_tool_message(messages: list[dict]) → list[dict]

pantheon.utils.llm.remove_raw_content(messages: list[dict]) → list[dict]

pantheon.utils.llm.filter_base64_in_tool_result(result: dict) → dict

Filter base64 data from a dict.

pantheon.utils.llm.filter_tool_messages(messages: list[dict]) → list[dict]

Filter tool-returned messages

pantheon.utils.llm.remove_unjsonifiable_raw_content(messages: list[dict]) → list[dict]

pantheon.utils.llm.remove_extra_fields(messages: list[dict]) → list[dict]

pantheon.utils.llm.remove_ui_fields(messages: list[dict]) → list[dict]

Remove UI-only fields that should not be sent to LLM.

These fields are added for frontend display/processing and would waste tokens if sent to the LLM. They can also confuse the LLM model.

Fields removed: - Attachment metadata (detected_attachments - only kept for UI/frontend) - Timing information (timestamp, start_timestamp, end_timestamp, generation_duration, execution_duration) - Internal IDs (id, message_id, chunk_index, transfer)

pantheon.utils.llm.remove_metadata(messages: list[dict]) → list[dict]

Strip messages down to only standard OpenAI fields before sending to LLM.

Strict providers like Groq reject ANY unknown field (chat_id, _metadata, _llm_content, _user_metadata, detected_attachments, etc.) and also reject null values for optional fields like tool_calls.

pantheon.utils.llm.process_messages_for_model(messages: list[dict], model: str) → list[dict]

Process messages for model consumption.

Processing steps (order matters): 1. remove_parsed - Remove parsed fields 2. remove_raw_content - Remove raw_content (structured tool outputs) 3. filter_base64_in_tool_messages - Filter base64 in tool messages 4. remove_extra_fields - Remove agent_name, tool_name 5. remove_ui_fields - Remove UI-only fields

pantheon.utils.llm.process_messages_for_store(messages: list[dict]) → list[dict]

Process messages before storing in memory.

Ensures all messages have a unique ID for later reference (e.g., revert operations).

pantheon.utils.llm.process_messages_for_hook_func(messages: list[dict]) → list[dict]

async pantheon.utils.llm.openai_embedding(texts: list[str], model: str = 'text-embedding-3-large') → list[list[float]]

Get embeddings using the effective OpenAI-routed configuration.

Parameters:

• texts – List of texts to embed

• model – Embedding model to use

Returns:

List of embedding vectors

pantheon.utils.llm.remove_hidden_fields(content: dict) → dict

Remove hidden fields from dict content.

If content is not a dict, return as-is.

pantheon.utils.llm.process_tool_result(result: Any, max_length: int | None = None, tool_name: str | None = None) → Any

Process tool result with optional truncation.

Parameters:

• result – Raw tool result

• max_length – Global max length for truncation (fallback)

• tool_name – Tool name for per-tool threshold lookup

Returns:

Processed result

class pantheon.utils.llm.TimingTracker

Track execution time for different phases.

Provides: - Manual start/end tracking - Context manager for automatic timing - Aggregate timing report

Examples

>>> tracker = TimingTracker()
>>> tracker.start("phase1")
>>> time.sleep(0.1)
>>> duration = tracker.end("phase1")
>>> print(tracker.get_all())
{'phase1': 0.10...}

>>> async with tracker.measure("phase2"):
...     await asyncio.sleep(0.1)
>>> tracker.get_all()
{'phase1': 0.10..., 'phase2': 0.10...}

__init__()

Initialize timing tracker.

start(phase: str) → None

Mark the start of a phase.

Parameters:

phase – Phase name

Raises:

ValueError – If phase already started

end(phase: str) → float

End a phase and record duration.

Parameters:

phase – Phase name

Returns:

Duration in seconds

Raises:

ValueError – If phase not started

get_all() → dict[str, float]

Get all recorded timings.

Returns:

Dictionary of phase -> duration

measure(phase: str)

Measure timing for a phase using context manager.

Parameters:

phase – Phase name

Examples

>>> async with tracker.measure("api_call"):
...     await some_async_function()

pantheon.utils.llm.calculate_total_cost_from_messages(messages: list[dict]) → float

Calculate total cost from message list

Uniformly processes all message types, calculating current_cost only once per message.

Note: Should pass in the full message list (for_llm=False), including compressed messages. The system naturally calculates only once because: - Compressed assistant messages have their own current_cost - Compression messages only record the cost of compression itself - No duplicate calculations

Parameters:

messages – Message list (recommended to use memory.get_messages(for_llm=False))

Returns:

Total cost (rounded to 4 decimals)

pantheon.utils.llm.collect_message_stats_lightweight(message: dict, messages: list[dict], model: str) → None

Lightweight statistics collection - read usage from _debug fields

Only collects essential fields: - total_tokens: actual size of current context - current_cost: current message cost - max_tokens: model’s maximum context

Data source priority: 1. Read from _debug_usage/_debug_cost (populated by call_llm_provider) 2. Fallback: manually calculate new messages

pantheon.utils.llm.count_tokens_in_messages(messages: list[dict], model: str, tools: list[dict] | None = None, assistant_message: dict | None = None) → dict

Count tokens with per-role breakdown and context usage metrics.

Separates system prompt (first system message) and tools definition from other roles.

pantheon.utils.llm.format_token_visualization(token_info: dict, bar_width: int = 50) → tuple[str, str, str]

Format token distribution bar with per-role breakdown and warning if usage >= 90%.

Vision Utilities

Vision utilities for multimodal agent input.

This module provides: 1. VisionInput: Pydantic model for vision input 2. ImageStore: Disk-based image storage with deduplication 3. Utilities for converting between paths, base64, and OpenAI format

class pantheon.utils.vision.VisionInput(*, images: list[str], prompt: str)

Vision input model containing images and prompt.

images: list[str]

prompt: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pantheon.utils.vision.vision_input(prompt: str, image_paths: list[str] | list[Path] | str | Path, from_path: bool = False) → VisionInput

Create a VisionInput from prompt and image paths/URLs.

Parameters:

• prompt – Text prompt

• image_paths – Image paths or URLs (can be str, Path, or list of either)

• from_path – If True, use file:// paths (will be expanded to Base64 before LLM call)

Returns:

VisionInput instance

Note

When from_path=True, paths are stored as file:// URIs. The actual Base64 conversion happens in expand_image_references_for_llm() just before the LLM API call, ensuring efficient Memory storage.

pantheon.utils.vision.vision_to_openai(vision: VisionInput) → list[dict]

Convert VisionInput to OpenAI message format.

Parameters:

vision – VisionInput instance

Returns:

List of message dicts in OpenAI format

pantheon.utils.vision.get_image_base64(file_path: str, max_size: int = 1568) → str

Read a local image file and return its base64 data URI.

Automatically resizes large images to reduce transmission cost. All images are processed through PIL for consistent output.

Parameters:

• file_path – Path to image file (with or without file:// prefix)

• max_size – Maximum dimension (width or height). Default: 1568px

Returns:

image/…;base64,…)

Return type:

Data URI string (data

pantheon.utils.vision.path_to_image_url(file_path: str, max_size: int = 1568) → str

Read a local image file and return its base64 data URI.

Automatically resizes large images to reduce transmission cost. All images are processed through PIL for consistent output.

Parameters:

• file_path – Path to image file (with or without file:// prefix)

• max_size – Maximum dimension (width or height). Default: 1568px

Returns:

image/…;base64,…)

Return type:

Data URI string (data

pantheon.utils.vision.path_to_vision(prompt: str, image_paths: list[str] | str | Path | list[Path]) → VisionInput

Create VisionInput from local image paths.

Parameters:

• prompt – Text prompt

• image_paths – One or more local image paths

Returns:

// paths (expanded to Base64 before LLM call)

Return type:

VisionInput with file

pantheon.utils.vision.parse_image_mentions(message: str, workspace: Path | str | None = None) → list[dict]

Parse @image:path tokens from message and build OpenAI multimodal format.

Parameters:

• message – User input string (may contain @image: tokens)

• workspace – Workspace directory for resolving relative paths. If None, uses workspace from settings

Returns:

List of message dicts in OpenAI format

Example

>>> parse_image_mentions("@image:photo.png describe this")
[{"role": "user", "content": [
    {"type": "text", "text": "describe this"},
    {"type": "image_url", "image_url": {"url": "file:///abs/path/photo.png"}}
]}]

class pantheon.utils.vision.ImageStore(storage_root: str | Path | None = None)

Manages storage of images for chat sessions.

Storage location: <pantheon_dir>/images/<chat_id>/<md5_hash>.<ext>

Handles: 1. Saving base64 images to disk (deduplicated by hash) 2. Validating and resolving local file paths 3. Processing message dicts to convert images to file:// references

__init__(storage_root: str | Path | None = None)

save_base64_image(chat_id: str, base64_data: str) → str

Save a base64 string image to disk.

Parameters:

• chat_id – The ID of the chat

• base64_data – Full data URI (data:image/png;base64,…)

Returns:

Absolute local file path to the saved image

normalize_local_path(path_str: str) → str

Normalize and verify a local file path.

Parameters:

path_str – Raw path (e.g. “file:///tmp/a.png”, “/tmp/a.png”)

Returns:

Absolute file path

Raises:

FileNotFoundError – If file does not exist

process_message_images(message: dict, chat_id: str) → None

Process a single message dict in-place.

• Extracts content from message

• Skips if content is not a list (plain text)

• For each image_url item: - Base64 → save to disk → replace with file:// path - Local path → verify → standardize to file:// path - HTTP URL → pass through

pantheon.utils.vision.get_image_store() → ImageStore

Get or create global ImageStore instance.

pantheon.utils.vision.expand_image_references_for_llm(messages: list[dict]) → list[dict]

Expand file:// image references to Base64 data URIs for LLM consumption.

Called just before sending messages to the LLM API.

Parameters:

messages – List of message dicts (will be deep copied)

Returns:

// paths converted to base64 data URIs

Return type:

New list with file

Logging Utilities

pantheon.utils.log.temporary_log_level(level: str)

Context manager to temporarily set log level for loguru logger

Usage:

with temporary_log_level(“WARNING”):

agent.run() # Only WARNING and ERROR will be logged

pantheon.utils.log.set_level(level: str)

Set the logging level for the console handler.

This safely replaces only the console handler, preserving other handlers (like file handlers).

pantheon.utils.log.disable_all()

Completely disable all logging. Cannot be re-enabled.

pantheon.utils.log.setup_file_logging(log_dir: Path | None = None, level: str = 'INFO', session_name: str = 'repl') → Path

Setup file logging to save logs to a file.

This is useful in REPL mode where console logging is suppressed, but you still want to capture logs for debugging.

The file log level defaults to INFO, which captures INFO, WARNING, and ERROR logs while avoiding verbose DEBUG output. This provides a good balance between having useful diagnostic information and avoiding excessive log size.

Parameters:

• log_dir – Directory for log files. Defaults to settings.logs_dir (.pantheon/logs)

• level – Log level for file handler (default: INFO - captures most useful logs)

• session_name – Prefix for log file name (default: “repl”)

Returns:

Path to the created log file

pantheon.utils.log.suppress_aiohttp_warnings(loop, context) → None

Custom asyncio exception handler to suppress aiohttp cleanup warnings.

aiohttp prints warnings via asyncio’s exception handler during GC. Use with: loop.set_exception_handler(suppress_aiohttp_warnings)

Miscellaneous Utilities

pantheon.utils.misc.generate_service_id(id_hash: str) → str

Generate a service ID from id_hash using SHA256.

This function matches the logic in NATSRemoteWorker (pantheon/remote/backend/nats.py) to ensure consistent service_id generation across different components.

Parameters:

id_hash – The hash string to generate service_id from

Returns:

A 64-character hexadecimal string (SHA256 hash)

Example

>>> service_id = generate_service_id("my-stable-id-123")
>>> len(service_id)
64
>>> all(c in '0123456789abcdef' for c in service_id)
True

async pantheon.utils.misc.call_endpoint_method(endpoint_service: Any, endpoint_method_name: str, **kwargs) → Any

Call a method on endpoint service, supporting both Endpoint instances and remote services.

This function handles both: - Direct Endpoint instances: calls method directly - Remote services: uses invoke() for RPC

async pantheon.utils.misc.run_func(func: Callable, *args, **kwargs)

pantheon.utils.misc.desc_to_openai_dict(desc: Description, skip_params: List[str] = [], relaxed_schema: bool = False) → dict

pantheon.utils.misc.print_agent_message(agent_name: str, message: dict, console: Console | None = None, print_tool_call: bool = True, print_assistant_message: bool = True, print_tool_response: bool = True, print_markdown: bool = True, max_tool_call_message_length: int | None = 1000)

pantheon.utils.misc.find_free_port(start_port: int = 3100, host: str = '127.0.0.1', max_attempts: int = 100) → int

Find a free port starting from start_port.

Tries to bind to ports sequentially until an available one is found.

Parameters:

• start_port – Port number to start searching from.

• host – Host address to bind to.

• max_attempts – Maximum number of ports to try.

Returns:

First available port number.

Raises:

RuntimeError – If no free port is found within the range.

pantheon.utils.misc.unwrap_single_layer(data: Any) → Any

Unwrap one layer of JSON strings in the data structure.

This handles cases where JSON data is nested as strings. Only unwraps one layer to avoid excessive processing.

Parameters:

data – Data structure to unwrap

Returns:

Unwrapped data (one layer only)

Examples

>>> unwrap_single_layer({"result": "[1, 2, 3]"})
{"result": [1, 2, 3]}

>>> unwrap_single_layer({"data": '{"key": "value"}'})
{"data": {"key": "value"}}

Overview

The utils module contains helper functions for:

• LLM interaction and message processing

• Vision/image handling

• Logging configuration

• General utility functions

LLM Utilities

The llm module provides functions for interacting with language models.

Common Functions

from pantheon.utils.llm import (
    acompletion_openai,
    acompletion,  # adapter-based completion
    process_messages_for_model,
    remove_hidden_fields
)

# Process messages for specific model
messages = process_messages_for_model(
    messages=[
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ],
    model="gpt-4"
)

# Make completion request
response = await acompletion_openai(
    model="gpt-4",
    messages=messages,
    temperature=0.7
)

Message Processing

# Remove hidden fields from messages
clean_messages = remove_hidden_fields(messages)

# Process for specific hooks
hook_messages = process_messages_for_hook_func(
    messages,
    hook_function
)

Vision Utilities

The vision module handles image inputs for agents.

VisionInput Class

from pantheon.utils.vision import VisionInput, vision_to_openai

# Create vision input
vision_input = VisionInput(
    text="What's in this image?",
    images=["path/to/image.jpg", "path/to/image2.png"]
)

# Convert to OpenAI format
openai_messages = vision_to_openai(vision_input)

Image Handling

# Agent with vision support
agent = Agent(
    name="vision_agent",
    instructions="Analyze images and answer questions"
)

# Use with vision input
result = await agent.run(
    VisionInput(
        text="Describe these images",
        images=["image1.jpg", "image2.jpg"]
    )
)

Logging Configuration

The log module provides consistent logging across Pantheon.

Logger Setup

from pantheon.utils.log import logger

# Use the pre-configured logger
logger.info("Starting agent")
logger.debug("Debug information")
logger.error("Error occurred", exc_info=True)

Custom Logging

import logging
from pantheon.utils.log import setup_logging

# Custom logging configuration
setup_logging(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    log_file='pantheon.log'
)

Miscellaneous Utilities

Helper Functions

The misc module contains various helper functions:

from pantheon.utils.misc import (
    desc_to_openai_dict,
    run_func
)

# Convert function description to OpenAI format
func_desc = parse_func(my_function)
openai_dict = desc_to_openai_dict(func_desc)

# Run function with timeout
result = await run_func(
    async_function,
    timeout=30,
    *args,
    **kwargs
)

Utility Examples

Rate Limiting

from pantheon.utils.misc import RateLimiter

# Create rate limiter
limiter = RateLimiter(
    max_calls=100,
    period=60  # 100 calls per minute
)

# Use with agent
@limiter.limit
async def make_api_call():
    return await agent.run("Task")

Retry Logic

from pantheon.utils.misc import retry_with_backoff

@retry_with_backoff(
    max_retries=3,
    initial_delay=1.0,
    backoff_factor=2.0
)
async def unreliable_operation():
    # Operation that might fail
    pass

Token Counting

from pantheon.utils.llm import count_tokens

# Count tokens in messages
messages = [
    {"role": "user", "content": "Hello, how are you?"},
    {"role": "assistant", "content": "I'm doing well!"}
]

token_count = count_tokens(messages, model="gpt-4")
print(f"Total tokens: {token_count}")

Best Practices

1. Use Provided Utilities: Leverage existing utilities instead of reimplementing

2. Consistent Logging: Use the configured logger for all logging

3. Error Handling: Utility functions include proper error handling

4. Type Safety: Utilities use type hints for better IDE support

5. Performance: Utilities are optimized for common use cases

Integration with Agents

Utilities in Agent Creation

from pantheon.agent import Agent
from pantheon.utils.log import logger
from pantheon.utils.vision import VisionInput

class CustomAgent(Agent):
    async def run(self, input_data):
        # Use logger
        logger.info(f"Processing input: {type(input_data)}")

        # Handle vision input
        if isinstance(input_data, VisionInput):
            logger.debug("Processing vision input")
            messages = vision_to_openai(input_data)
        else:
            messages = [{"role": "user", "content": str(input_data)}]

        # Process with utilities
        return await super().run(messages)

Environment Variables

Utilities respect environment variables:

# Logging level
export PANTHEON_LOG_LEVEL=DEBUG

# API timeouts
export PANTHEON_API_TIMEOUT=30

# Token limits
export PANTHEON_MAX_TOKENS=4096

Future Enhancements

Planned utility additions:

• Caching utilities

• Serialization helpers

• Validation functions

• Performance profiling

• Metrics collection