calute.core.utils

Utility functions for the Calute framework.

This module provides helper functions for debugging, data merging, and function introspection. It includes utilities for converting Python functions to JSON schema format, merging response chunks, and debug printing with timestamps.

class calute.core.utils.CaluteBase

Bases: BaseModel

Base Pydantic model for Calute configuration classes.

This class provides a strict configuration base that enforces data validation and prevents accidental attribute errors. All Calute configuration models should inherit from this class.

model_config

Pydantic configuration that forbids extra attributes, validates default values, and uses enum values directly.

Example

>>> class MyConfig(CaluteBase):
...     name: str
...     count: int = 0
>>> config = MyConfig(name="test")
>>> config.name
'test'

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'use_enum_values': True, 'validate_default': True}

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

calute.core.utils.T

Generic type variable for type-safe operations.

alias of TypeVar(‘T’)

calute.core.utils.debug_print(debug: bool, *args: str) → None

Print debug messages with timestamp if debug mode is enabled.

Parameters

• debug – Whether debug mode is enabled.

• *args – Variable number of string arguments to print.

Returns

None

Example

>>> debug_print(True, "Processing", "function", "call")
[2025-01-27 10:30:45] Processing function call

calute.core.utils.estimate_messages_tokens(messages: list[dict[str, Any]]) → int

Estimate the total number of tokens in a list of messages.

Parameters

messages – List of message dictionaries with ‘role’ and ‘content’ keys.

Returns

Estimated total number of tokens.

Example

>>> msgs = [{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there!"}]
>>> estimate_messages_tokens(msgs)
5

calute.core.utils.estimate_tokens(text: str, chars_per_token: float = 4.0) → int

Estimate the number of tokens in a text string.

Uses a simple character-based heuristic. For more accurate counting, consider using tiktoken or a provider-specific tokenizer.

Parameters

• text – The text to estimate tokens for.

• chars_per_token – Average characters per token (default: 4.0 for English).

Returns

Estimated number of tokens.

Example

>>> estimate_tokens("Hello, world!")
4

calute.core.utils.function_to_json(func: callable) → dict

Convert a Python function into a JSON-serializable dictionary.

Extracts the function’s signature, including its name, description (from docstring), and parameters with their types and descriptions. This is used to generate function schemas for LLM tool calling.

If the function has a __calute_schema__ attribute, it will be used directly instead of extracting from the signature. This is useful for dynamically created functions like MCP tool wrappers.

Parameters

func – The function to be converted.

Returns

A dictionary representing the function’s signature in JSON format, compatible with OpenAI function calling schema.

Raises

ValueError – If the function signature cannot be extracted.

Example

>>> def greet(name: str, age: int = 0) -> str:
...     '''Greet a person.
...     name: Person's name
...     age: Person's age
...     '''
...     return f"Hello {name}"
>>> schema = function_to_json(greet)
>>> print(schema["function"]["name"])
greet

calute.core.utils.get_callable_public_name(func: callable) → str

Return the public tool name exposed to the model/runtime.

Calute tools may carry a __calute_schema__ dict whose name key differs from the Python callable’s __name__. This allows compatibility aliases and dotted tool names without changing the underlying implementation name.

Parameters

func – The callable whose public tool name should be resolved.

Returns

The public name string. If the callable carries a __calute_schema__ dictionary with a non-empty name entry, that value is returned. Otherwise falls back to func.__name__ or str(func).

Example

>>> def my_tool():
...     pass
>>> get_callable_public_name(my_tool)
'my_tool'

calute.core.utils.merge_chunk(final_response: dict, delta: dict) → None

Merge a streaming response chunk into the final response.

Handles special processing for tool_calls and removes role field.

Parameters

• final_response – The accumulated response dictionary.

• delta – The new chunk to merge.

Returns

None (modifies final_response in place)

Note

This function is specifically designed for merging streaming API response chunks that may contain tool calls.

Example

>>> response = {"content": "Hello", "tool_calls": [{"name": "func"}]}
>>> delta = {"content": " World"}
>>> merge_chunk(response, delta)
>>> print(response["content"])
Hello World

calute.core.utils.merge_fields(target: dict, source: dict) → None

Recursively merge fields from source dictionary into target dictionary.

For string values, concatenates them. For dict values, merges recursively.

Parameters

• target – The target dictionary to merge into.

• source – The source dictionary to merge from.

Returns

None (modifies target in place)

Example

>>> target = {"text": "Hello", "nested": {"key": "value"}}
>>> source = {"text": " World", "nested": {"key": "2"}}
>>> merge_fields(target, source)
>>> print(target)
{"text": "Hello World", "nested": {"key": "value2"}}

calute.core.utils.run_sync(coro: Coroutine[Any, Any, T]) → T

Run an async coroutine synchronously, handling nested event loops.

This function safely executes a coroutine from synchronous code, handling the case where an event loop is already running (e.g., in Jupyter notebooks or async frameworks).

Parameters

coro – The coroutine to execute.

Returns

The result of the coroutine.

Example

>>> async def fetch_data():
...     return "data"
>>> result = run_sync(fetch_data())
>>> print(result)
data