prompttrail.core package

Submodules

prompttrail.core.cache module

class prompttrail.core.cache.CacheProvider

Bases: object

Abstract base class for cache providers.

Cache providers are responsible for storing and retrieving messages from a cache. Each provider must implement add() and search() methods.

abstract add(session: Session, message: Message) → None

Add a message to the cache.

Parameters:

  • session – Session associated with the message

  • message – Message to cache

Note

Cache should also consider Parameters when storing messages

abstract search(config: Config, session: Session) → Message | None

Search for a cached message.

Parameters:

  • config – Config associated with the message

  • session – Session to search for

Returns:

Cached message if found, None otherwise

class prompttrail.core.cache.LRUCacheProvider(n_items: int = 10000)

Bases: CacheProvider

LRU (Least Recently Used) cache implementation.

Caches messages in an LRU cache with a fixed maximum size.

__init__(n_items: int = 10000)

Initialize LRU cache.

Parameters:

n_items – Maximum number of items in cache (default: 10000)

add(session: Session, message: Message) → None

Add message to cache.

Parameters:

  • session – Session associated with the message

  • message – Message to cache

search(parameters: Config, session: Session) → Message | None

Search for cached message.

Parameters:

  • config – Config associated with the message

  • session – Session to search for

Returns:

Cached message if found, None otherwise

prompttrail.core.const module

exception prompttrail.core.const.BreakException

Bases: Exception

Exception raised when BreakTemplate is rendered.

exception prompttrail.core.const.ReachedEndTemplateException(farewell_message: str | None = None)

Bases: Exception

Exception raised when EndTemplate is rendered.

__init__(farewell_message: str | None = None) → None

prompttrail.core.errors module

exception prompttrail.core.errors.ConfigurationValidationError

Bases: Exception

Raised when a configuration validation fails.

exception prompttrail.core.errors.HookError

Bases: Exception

Raised when a hook execution fails.

exception prompttrail.core.errors.ParameterValidationError

Bases: Exception

Raised when a parameter validation fails.

exception prompttrail.core.errors.ProviderResponseError(message: str, response: Any)

Bases: Exception

Raised when a provider returns an error response.

__init__(message: str, response: Any) → None

Initialize ProviderResponseError.

Parameters:

  • message – The error message

  • response – The raw response from the provider

exception prompttrail.core.errors.RenderingError

Bases: Exception

Raised when template rendering fails.

exception prompttrail.core.errors.TemplateNotFoundError

Bases: Exception

Raised when a referenced template cannot be found.

prompttrail.core.mocks module

Mock providers for testing LLM interactions without calling actual APIs.

class prompttrail.core.mocks.EchoMockProvider(role: MessageRoleType)

Bases: FunctionalMockProvider

Mock provider that echoes back the last message with a specified role.

Parameters:

role (MessageRoleType) – Role to assign to the echo response messages

__init__(role: MessageRoleType)
class prompttrail.core.mocks.FunctionalMockProvider(func: Callable[[Session], Message])

Bases: MockProvider

Mock provider that generates responses using a custom function.

Parameters:

func (Callable[[Session], Message]) – Function that takes a session and returns a response message

__init__(func: Callable[[Session], Message])
call(session: Session) → Message

Return a mock response for the given session.

Parameters:

session (Session) – The conversation session including all messages

Returns:

The mocked response message

Return type:

Message

class prompttrail.core.mocks.MockProvider

Bases: ABC

Abstract base class for mock providers.

Mock providers simulate LLM responses without making actual API calls. Subclasses must implement the call() method to define response behavior.

abstract call(session: Session) → Message

Return a mock response for the given session.

Parameters:

session (Session) – The conversation session including all messages

Returns:

The mocked response message

Return type:

Message

class prompttrail.core.mocks.OneTurnConversationMockProvider(conversation_table: Dict[str, Message])

Bases: MockProvider

Mock provider that returns predefined responses based on the last message.

Parameters:

conversation_table (Dict[str, Message]) – Mapping of input messages to their predefined responses

__init__(conversation_table: Dict[str, Message])
call(session: Session) → Message

Return a mock response for the given session.

Parameters:

session (Session) – The conversation session including all messages

Returns:

The mocked response message

Return type:

Message

prompttrail.core.utils module

class prompttrail.core.utils.Debuggable

Bases: object

Base class adding logging capabilities to child classes.

__init__() → None

Initialize logging for the class.

critical(msg: str, *args, **kwargs)

Log a critical level message.

debug(msg: str, *args, **kwargs)

Log a debug level message.

disable_log()

Disable logging for this instance.

enable_log()

Enable logging for this instance.

error(msg: str, *args, **kwargs)

Log an error level message.

info(msg: str, *args, **kwargs)

Log an info level message.

log(level: int, msg: str, *args, **kwargs)

Log a message at specified level with class and function name context.

Parameters:

  • level – Logging level

  • msg – Message format string

  • args – Format string arguments

  • kwargs – Additional logging arguments

setup_logger_for_pydantic()

Setup logger after pydantic initialization. Should be called after pydantic initialization.

warning(msg: str, *args, **kwargs)

Log a warning level message.

prompttrail.core.utils.count_tokens(text: str, encoding_name: str) → int

Count the number of tokens in text using specified encoding.

Parameters:

  • text – Text to count tokens for

  • encoding_name – Name of the tiktoken encoding to use

Returns:

Number of tokens in text

prompttrail.core.utils.is_in_test_env() → bool

Check if the code is running in a test environment.

Returns:

True if running in a test environment, False otherwise.

Return type:

bool

Module contents

Core module for PromptTrail.

pydantic model prompttrail.core.Config

Bases: BaseModel

Unified configuration base class.

Centralizes configuration and parameter management with integrated validation.

Fields:

  • cache_provider (prompttrail.core.cache.CacheProvider | None)

  • max_tokens (int | None)

  • mock_provider (prompttrail.core.mocks.MockProvider | None)

  • model_name (str)

  • repetition_penalty (float | None)

  • temperature (float | None)

  • tools (List[Any] | None)

  • top_k (int | None)

  • top_p (float | None)

Validators:

  • validate_providers » all fields

field cache_provider: CacheProvider | None = None

Cache provider to cache the response from the model.

Validated by:

  • validate_providers

field max_tokens: int | None = 1024

Maximum number of tokens to generate.

Validated by:

  • validate_providers

field mock_provider: MockProvider | None = None

Mock provider to mock the response from the model.

Validated by:

  • validate_providers

field model_name: str [Required]

Name of the model to use.

Validated by:

  • validate_providers

field repetition_penalty: float | None = None

Repetition penalty parameter.

Validated by:

  • validate_providers

field temperature: float | None = 1.0

Temperature for sampling.

Validated by:

  • validate_providers

field tools: List[Tool] | None = None

Optional list of tools that can be used by the model.

Validated by:

  • validate_providers

field top_k: int | None = None

Top-k sampling parameter.

Validated by:

  • validate_providers

field top_p: float | None = None

Top-p (nucleus) sampling parameter.

Validated by:

  • validate_providers

validate_all(session: Session | None = None) → None

Integrated validation method

validator validate_providers  »  all fields

Validate that only one provider is used.

validate_session(session: Session, is_async: bool = False) → None

Perform basic session validation.

pydantic model prompttrail.core.Message

Bases: BaseModel

A message represents a single message from a user, model, or API etc…

Fields:

  • content (str)

  • metadata (prompttrail.core._core.Metadata)

  • role (Literal['system', 'user', 'assistant', 'tool_result', 'control'])

  • tool_use (Dict[str, Any] | None)

field content: str [Required]
field metadata: Metadata [Optional]
field role: Literal['system', 'user', 'assistant', 'tool_result', 'control'] [Required]
field tool_use: Dict[str, Any] | None = None
__init__(content: str, role: Literal['system', 'user', 'assistant', 'tool_result', 'control'], metadata: Dict[str, Any] | Metadata | None = None, tool_use: Dict[str, Any] | None = None)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

class prompttrail.core.Metadata(*args: Any, **kwargs: Any)

Bases: Dict[str, Any]

Base class providing type-safe metadata

__init__(*args: Any, **kwargs: Any) → None
copy() → a shallow copy of D
dict(*args: Any, **kwargs: Any) → Dict[str, Any]
model_copy(*args: Any, **kwargs: Any) → Metadata
pydantic model prompttrail.core.Model

Bases: BaseModel, ABC, Debuggable

Class defining the interface for interaction with LLM models.

Fields:

  • configuration (prompttrail.core._core.Config)

  • enable_logging (bool)

  • logger (logging.Logger | None)

field configuration: Config [Required]
field enable_logging: bool = True
field logger: Logger | None = None
after_send(session: Session | None, message: Message) → Message

Perform message post-processing.

format_tool(tool: Any) → Dict[str, Any]

Convert tool to model-specific format.

format_tool_result(result: Any) → Dict[str, Any]

Convert tool execution result to model-specific format.

is_cached() → bool

Return whether the model is using cache.

is_mocked() → bool

Return whether the model is mocked.

list_models() → List[str]

Return a list of available models.

model_post_init(*args, **kwargs)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

prepare(session: Session | None = None) → Session

Perform session preprocessing.

send(session: Session) → Message

Define standard procedure for sending messages to the model.

send_async(session: Session, yield_type: Literal['all', 'new'] = 'new') → Generator[Message, None, None]

Define standard procedure for sending messages asynchronously.

validate_session(session: Session, is_async: bool = False) → None

Perform session validation.

Parameters:

  • session – Session to validate

  • is_async – Whether validation is for asynchronous processing

Raises:

ParameterValidationError – Validation error

pydantic model prompttrail.core.Session

Bases: BaseModel

A session represents a conversation between a user and a model, or API etc…

Fields:

  • debug_mode (bool)

  • jump_to_id (str | None)

  • messages (List[prompttrail.core._core.Message])

  • metadata (prompttrail.core._core.Metadata)

  • runner (Any | None)

  • stack (List[Any])

field debug_mode: bool = False
field jump_to_id: str | None = None
field messages: List[Message] [Optional]
field metadata: Metadata [Optional]
field runner: Runner | None = None
field stack: List[Stack] [Optional]
__init__(messages: List[Message] = [], metadata: Dict[str, Any] | Metadata | None = None, runner: Any | None = None, debug_mode: bool = False, stack: List[Any] = [], jump_to_id: str | None = None) → None

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

append(message: Message) → None

Append a message to the session.

classmethod from_dict(data: dict) → Session

Create session from dictionary representation

get_current_template_id() → str | None

Get the ID of the current template.

get_jump() → str | None

Get the jump target template ID.

get_last() → Message

Get the last message in the session.

get_last_message() → Message

Alias for get_last().

head_stack() → Any

Get the top stack frame.

pop_stack() → Any

Pop a stack frame.

push_stack(stack: Any) → None

Push a stack frame.

set_jump(jump_to_id: str | None) → None

Set the jump target template ID.

to_dict() → dict

Convert session to dictionary representation