haive.core.common.mixins.engine_mixin

Engine management mixin for tracking and accessing Haive engines.

This module provides a sophisticated mixin for managing different types of engines within the Haive framework. It enables tracking engines by name and type, collecting usage statistics, and provides rich visualization capabilities.

Usage:

from pydantic import BaseModel from haive.core.common.mixins import EngineStateMixin from haive.core.engine.llm import LLMEngine

class MyState(EngineStateMixin, BaseModel):

# Other fields pass

# Create state and add engines state = MyState()

# Add an LLM engine llm = LLMEngine(name=”my_llm”, model=”gpt-4”, provider=”openai”) state.add_engine(llm)

# Later, retrieve the engine retrieved_llm = state.get_engine(“my_llm”)

# Get all LLM engines all_llms = state.get_llms()

# Display engine information state.display_engines()

Classes

EngineStateMixin

Mixin providing comprehensive engine management capabilities with validation.

Module Contents

class haive.core.common.mixins.engine_mixin.EngineStateMixin(/, **data)

Bases: pydantic.BaseModel

Mixin providing comprehensive engine management capabilities with validation.

This mixin allows StateSchema to manage engines by name and type, providing rich access patterns, performance tracking, and debugging capabilities.

The mixin maintains multiple indexes for efficient access: - By name: Fast lookup of specific engines - By type: Group engines by their type (LLM, Retriever, etc.)

It also tracks metadata about engines including access patterns and performance metrics, and provides rich visualization for debugging.

Parameters:

data (Any)

engines

Dictionary of engines indexed by name.

engines_by_type

Dictionary of engines organized by type.

engine_metadata

Dictionary containing additional metadata for each engine.

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.

add_engine(engine, name=None)

Add an engine to the state with automatic organization by type.

This method adds an engine to both the main engines dictionary and the type-based index. It also initializes metadata tracking for the engine.

Parameters:

• engine (haive.core.engine.base.Engine) – The engine to add.

• name (str | None) – Optional name override (defaults to engine.name).

Return type:

None

change_provider(name, provider)

Change the provider of an LLM engine.

This is a convenience method specifically for changing the provider of an engine (e.g., switching from ‘openai’ to ‘anthropic’).

Parameters:

• name (str) – Engine name.

• provider (str) – New provider name.

Raises:

ValueError – If the engine is not found.

Return type:

None

debug_engine_access(limit=10)

Show debug information about engine access patterns.

This method displays access statistics and a recent access log to help with debugging engine usage patterns and performance.

Parameters:

limit (int) – Maximum number of access log entries to show.

Return type:

None

display_engine_details(name)

Display detailed information about a specific engine.

This method creates a rich visual representation of all the details for a specific engine, including its configuration, metadata, and performance metrics.

Parameters:

name (str) – Engine name.

Return type:

None

display_engines(show_metadata=False, show_performance=False)

Display all engines in a rich tree view organized by type.

This method creates a rich visual representation of engines organized by type, with optional metadata and performance information.

Parameters:

• show_metadata (bool) – Whether to show additional metadata.

• show_performance (bool) – Whether to show performance metrics.

Return type:

None

get_agents()

Get all agent engines.

Returns:

List of agent engines.

Return type:

list[haive.core.engine.base.Engine]

get_all_engines()

Get all engines as a dictionary.

Returns:

Dictionary of all engines by name.

Return type:

dict[str, haive.core.engine.base.Engine]

get_embeddings()

Get all embeddings engines.

Returns:

List of embeddings engines.

Return type:

list[haive.core.engine.base.Engine]

get_engine(name)

Get an engine by name with access logging.

This method retrieves an engine by name and updates access metrics including access count, timestamp, and performance data.

Parameters:

name (str) – Engine name.

Returns:

Engine if found, None otherwise.

Return type:

haive.core.engine.base.Engine | None

get_engine_routes(name)

Get tool routes from an engine.

This method tries to extract routing information from an engine by checking various attribute names where routes might be stored.

Parameters:

name (str) – Engine name.

Returns:

Dictionary of routes if available, empty dict otherwise.

Return type:

dict[str, str]

get_engine_summary()

Get a summary of all engines.

This method generates a summary of engine statistics including counts by type, access patterns, and performance information.

Returns:

Dictionary with engine statistics.

Return type:

dict[str, Any]

get_engine_tools(name)

Get tools from an engine (for agents/LLMs with tools).

This method tries to extract tools from an engine by checking various attribute names where tools might be stored.

Parameters:

name (str) – Engine name.

Returns:

List of tools if available, empty list otherwise.

Return type:

list[Any]

get_engines_by_type(engine_type)

Get all engines of a specific type.

Parameters:

engine_type (haive.core.engine.base.types.EngineType) – The engine type to filter by.

Returns:

List of engines of the specified type.

Return type:

list[haive.core.engine.base.Engine]

get_llms()

Get all LLM/AugLLM engines.

Returns:

List of LLM engines.

Return type:

list[haive.core.engine.base.Engine]

get_retrievers()

Get all retriever engines.

Returns:

List of retriever engines.

Return type:

list[haive.core.engine.base.Engine]

get_tools()

Get all tool engines.

Returns:

List of tool engines.

Return type:

list[haive.core.engine.base.Engine]

get_vector_stores()

Get all vector store engines.

Returns:

List of vector store engines.

Return type:

list[haive.core.engine.base.Engine]

remove_engine(name)

Remove an engine from the state.

This method removes an engine from both the main engines dictionary and the type-based index, as well as removing its metadata.

Parameters:

name (str) – Engine name to remove.

Returns:

True if removed, False if not found.

Return type:

bool

update_engine(name, **kwargs)

Update engine attributes dynamically.

This method allows updating specific attributes of an engine by providing them as keyword arguments.

Parameters:

• name (str) – Engine name.

• **kwargs – Attributes to update.

Raises:

ValueError – If the engine is not found.

Return type:

None

validate_and_organize_engines()

Ensure engines are properly organized by type and validated.

This validator rebuilds the engines_by_type index to ensure consistency and initializes metadata for any engines that don’t have it yet.

Returns:

Self with validated engine organization.

Return type:

Self

model_config

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