haive.core.engine.aug_llm.utils¶

Utility functions for working with AugLLMConfig instances.

This module provides helper functions for composing, managing, and utilizing augmented LLM runnables within the Haive framework. The utilities here enable flexible configuration management, runtime composition, and chain construction for augmented language models.

Key functionalities include: - Creating runnables from configuration objects - Managing collections of runnables - Chaining runnables together - Merging configuration dictionaries with special handling for LLM settings

Functions¶

`chain_runnables`(runnables[, override_config])	Chain multiple runnables together with optional configuration overrides.
`compose_runnable`(aug_llm_config[, runnable_config])	Compose a runnable LLM chain from an AugLLMConfig.
`compose_runnables_from_dict`(runnables[, runnable_config])	Compose and return a dictionary of runnables from a dictionary of configs.
`create_runnables_dict`(runnables)	Create a dictionary mapping names to runnable configs from various input formats.
`merge_configs`(base_config, override_config)	Merge two runnable configs, with override taking precedence.

Module Contents¶

haive.core.engine.aug_llm.utils.chain_runnables(runnables, override_config=None)[source]¶

Chain multiple runnables together with optional configuration overrides.

Creates a composite runnable that passes the output of each runnable to the input of the next one in sequence. This is useful for creating processing pipelines that transform data through multiple stages.

Parameters:

runnables (List[Runnable]) – List of runnables to chain together in sequence. The output of each runnable is passed as input to the next one.
override_config (Optional[Dict[str, Any]]) – Optional configuration overrides to apply to the final chained runnable.

Returns:

A composite runnable representing the chain of all input runnables.

Return type:

Runnable

Raises:

ValueError – If the input list is empty.

Examples

>>> from langchain_core.runnables import Runnable
>>> # Define some runnables
>>> prompt_formatter = RunnablePromptFormatter()
>>> llm = RunnableLLM()
>>> output_parser = RunnableOutputParser()
>>>
>>> # Chain them together
>>> pipeline = chain_runnables([prompt_formatter, llm, output_parser])
>>>
>>> # Use the chained runnable
>>> result = pipeline.invoke({"question": "What is the capital of France?"})

haive.core.engine.aug_llm.utils.compose_runnable(aug_llm_config, runnable_config=None)[source]¶

Compose a runnable LLM chain from an AugLLMConfig.

Creates a ready-to-use runnable object from the provided configuration, applying any runtime configuration overrides as needed.

Parameters:

aug_llm_config (AugLLMConfig) – Configuration object for the LLM chain, containing all the necessary settings and components.
runnable_config (Optional[RunnableConfig]) – Optional runtime configuration overrides to apply when creating the runnable.

Returns:

A fully composed, executable LLM chain.

Return type:

Runnable

Raises:

Exception – Any exceptions from the underlying creation process are logged and re-raised.

Examples

>>> from haive.core.engine.aug_llm.config import AugLLMConfig
>>> config = AugLLMConfig(name="my_llm", model="gpt-4")
>>> runnable = compose_runnable(config)
>>> result = runnable.invoke("Hello, world!")

haive.core.engine.aug_llm.utils.compose_runnables_from_dict(runnables, runnable_config=None)[source]¶

Compose and return a dictionary of runnables from a dictionary of configs.

Takes a dictionary mapping names to configurations and returns a dictionary mapping the same names to fully composed, executable runnables. This is useful for bulk-creating multiple runnables with consistent configuration.

Parameters:

runnables (Dict[str, Union[AugLLMConfig, Any]]) – Dictionary mapping names to configuration objects. Objects must have a create_runnable method.
runnable_config (Optional[RunnableConfig]) – Optional runtime configuration overrides to apply to all runnables.

Returns:

Dictionary mapping names to composed runnables. Entries that fail to compose are omitted from the result.

Return type:

Dict[str, Runnable]

Examples

>>> from haive.core.engine.aug_llm.config import AugLLMConfig
>>> configs = {
...     "qa": AugLLMConfig(name="qa_llm", model="gpt-4"),
...     "summarizer": AugLLMConfig(name="summary_llm", model="claude-3")
... }
>>> runnables = compose_runnables_from_dict(configs)
>>> # Use the composed runnables
>>> qa_response = runnables["qa"].invoke("What is the capital of France?")
>>> summary = runnables["summarizer"].invoke("Summarize this long text...")

haive.core.engine.aug_llm.utils.create_runnables_dict(runnables)[source]¶

Create a dictionary mapping names to runnable configs from various input formats.

Normalizes different input formats (list, dictionary, or single config) into a consistent dictionary format mapping names to configurations. This is useful for managing collections of runnables in a standardized way.

Parameters:: runnables (Union[List[AugLLMConfig], Dict[str, AugLLMConfig], AugLLMConfig]) – LLM configs in various formats: - List of AugLLMConfig objects (uses their names as keys) - Dictionary mapping names to AugLLMConfig objects - Single AugLLMConfig object (uses its name as key)
Returns:: Dictionary mapping names to AugLLMConfig objects. Returns an empty dictionary if the input cannot be processed.
Return type:: Dict[str, AugLLMConfig]

Examples

>>> from haive.core.engine.aug_llm.config import AugLLMConfig
>>> config1 = AugLLMConfig(name="llm1", model="gpt-4")
>>> config2 = AugLLMConfig(name="llm2", model="claude-3")
>>>
>>> # From list
>>> configs_dict = create_runnables_dict([config1, config2])
>>> list(configs_dict.keys())
['llm1', 'llm2']
>>>
>>> # From dictionary
>>> configs_dict = create_runnables_dict({"custom_name": config1})
>>> list(configs_dict.keys())
['custom_name']
>>>
>>> # From single config
>>> configs_dict = create_runnables_dict(config1)
>>> list(configs_dict.keys())
['llm1']

haive.core.engine.aug_llm.utils.merge_configs(base_config, override_config)[source]¶

Merge two runnable configs, with override taking precedence.

Performs a deep merge of two configuration dictionaries, with special handling for nested structures like engine_configs. The override values take precedence over base values when keys conflict, except for nested dictionaries which are merged recursively.

Parameters:

base_config (Optional[RunnableConfig]) – Base configuration dictionary to start with.
override_config (Optional[RunnableConfig]) – Configuration dictionary whose values will override corresponding values in the base config.

Returns:

A new configuration dictionary containing the merged values.

Return type:

RunnableConfig

Examples

>>> base = {
...     "configurable": {
...         "temperature": 0.7,
...         "engine_configs": {
...             "gpt-4": {"max_tokens": 1000}
...         }
...     }
... }
>>> override = {
...     "configurable": {
...         "temperature": 0.5,
...         "engine_configs": {
...             "gpt-4": {"temperature": 0.9}
...         }
...     }
... }
>>> merged = merge_configs(base, override)
>>> merged["configurable"]["temperature"]
0.5
>>> merged["configurable"]["engine_configs"]["gpt-4"]
{'max_tokens': 1000, 'temperature': 0.9}