ChainBuilder package

Submodules

ChainBuilder.ChainBuilder module

academic_metrics.ChainBuilder.ChainBuilder.FirstCallRequired

TypeVar representing a dictionary that is required on first call but optional after.

Type Structure:

TypeVar bound to Dict[str, Any]

Usage:

Used to enforce that a dictionary parameter must be provided on first call but can be optional on subsequent calls.

alias of TypeVar(‘FirsCallRequired’, bound=Dict[str, Any])

academic_metrics.ChainBuilder.ChainBuilder.ParserUnion

Union type representing valid parser types.

Type Structure:

Union[PydanticOutputParser, JsonOutputParser, StrOutputParser]

alias of PydanticOutputParser | JsonOutputParser | StrOutputParser

class academic_metrics.ChainBuilder.ChainBuilder.ParserType

TypeVar representing the main parser type.

Type Structure:

TypeVar bound to Union[PydanticOutputParser, JsonOutputParser, StrOutputParser]

Usage:

Used to enforce type consistency for the primary parser in chain operations.

alias of TypeVar(‘ParserType’, bound=PydanticOutputParser | JsonOutputParser | StrOutputParser)

class academic_metrics.ChainBuilder.ChainBuilder.FallbackParserType

TypeVar representing the fallback parser type.

Type Structure:

TypeVar bound to Union[PydanticOutputParser, JsonOutputParser, StrOutputParser]

Usage:

Used to enforce type consistency for the fallback parser in chain operations.

alias of TypeVar(‘FallbackParserType’, bound=PydanticOutputParser | JsonOutputParser | StrOutputParser)

class academic_metrics.ChainBuilder.ChainBuilder.ChainBuilder(*, chat_prompt, llm, parser=None, fallback_parser=None, logger=None, log_to_console=True)

Bases: object

A builder class for constructing and configuring LangChain chains with logging and parsing capabilities.

This class provides functionality to construct LangChain chains with customizable prompts, language models, and output parsers. It includes support for logging, fallback chains, and various parser types.

log_file_path

Path to the log file.

Type:

Path

logger

Logger instance for the chain builder.

Type:

Logger

chat_prompt

Template for chat prompts.

Type:

ChatPromptTemplate

parser

Primary output parser.

Type:

ParserType | None

fallback_parser

Fallback output parser.

Type:

ParserType | None

llm

Language model instance.

Type:

Union[ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI]

chain

Primary chain instance.

Type:

Runnable

fallback_chain

Fallback chain instance.

Type:

Runnable

get_chain() → :class:`Runnable`

Returns the primary chain instance.

get_fallback_chain() → :class:`Runnable`

Returns the fallback chain instance.

__str__() → str

Returns a string representation of the ChainBuilder.

__repr__() → str

Returns a string representation of the ChainBuilder.

_run_pydantic_parser_logging() → None

Logs Pydantic parser configuration.

_build_chain() → None

Constructs the primary chain.

_build_fallback_chain() → None

Constructs the fallback chain.

__init__(*, chat_prompt, llm, parser=None, fallback_parser=None, logger=None, log_to_console=True)

Initialize a ChainBuilder instance.

Parameters:

• chat_prompt (ChatPromptTemplate) – Template for structuring chat interactions.

• llm (ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI) – Language model to use.

• parser (ParserType | None) – Primary output parser. Defaults to None.

• fallback_parser (FallbackParserType | None) – Fallback output parser. Defaults to None.

• logger (Logger | None) – Custom logger instance. Defaults to None.

• log_to_console (bool | None) – Whether to log to console. Defaults to LOG_TO_CONSOLE.

Returns:

None

__str__()

Returns a string representation of the ChainBuilder object.

The string includes the class names of the chat_prompt, llm, and parser attributes.

Returns:

A string representation of the ChainBuilder object.

Return type:

str

__repr__()

Returns a string representation of the object for debugging purposes.

This method calls the __str__() method to provide a human-readable representation of the object.

Returns:

A string representation of the object.

Return type:

str

_run_pydantic_parser_logging()

Logs the required fields and their default values (if any) for a Pydantic model if the parser is an instance of PydanticOutputParser. Additionally, logs the JSON schema of the Pydantic model.

This method performs the following steps: 1. Checks if the parser is an instance of PydanticOutputParser. 2. Retrieves the Pydantic model from the parser. 3. Logs each field’s name and whether it is required or optional. 4. Logs the default value of each field if it is not None. 5. Logs the JSON schema of the Pydantic model.

Return type:

None

Returns:

None

_build_chain()

Builds and returns a chain of Runnable objects.

The chain is constructed by combining a RunnablePassthrough object with the chat_prompt and llm attributes. If a parser is provided, it is added to the chain and pydantic parser logging is executed.

Returns:

The constructed chain of Runnable objects.

Return type:

Runnable

_build_fallback_chain()

Builds the fallback chain for the Runnable.

This method constructs a fallback chain by combining a RunnablePassthrough instance with the chat prompt and language model (llm). If a fallback parser is provided, it adds the parser to the chain and logs the parser usage.

Returns:

The constructed fallback chain.

Return type:

Runnable

get_chain()

Retrieves the current chain.

Returns:

The current chain instance.

Return type:

Runnable

get_fallback_chain()

Retrieve the fallback chain.

Returns:

The fallback chain instance.

Return type:

Runnable

class academic_metrics.ChainBuilder.ChainBuilder.ChainWrapper(*, chain, fallback_chain, parser=None, fallback_parser=None, preprocessor=None, postprocessor=None, logger=None)

Bases: object

A wrapper class for managing and executing chains with optional fallback chains.

This class is designed to handle the execution of primary chains and, if necessary, fallback chains in case of errors or unexpected outputs. It supports preprocessing and postprocessing of input and output data, as well as logging for debugging and monitoring purposes.

chain

The primary chain to be executed.

Type:

Runnable

fallback_chain

The fallback chain to be executed if the primary chain fails.

Type:

Runnable

parser

The parser to be used for processing the output of the primary chain. Defaults to None.

Type:

ParserType | None

fallback_parser

The parser to be used for processing the output of the fallback chain. Defaults to None.

Type:

FallbackParserType | None

preprocessor

A function to preprocess input data before passing it to the chain. Defaults to None.

Type:

Callable | None

postprocessor

A function to postprocess the output data from the chain. Defaults to None.

Type:

Callable | None

logger

A logger instance for logging information and debugging. Defaults to None.

Type:

Logger | None

__init__(chain, fallback_chain, parser=None, fallback_parser=None,

preprocessor=None, postprocessor=None, logger=None):

Initializes the ChainWrapper instance with the provided parameters.

__str__() → str

Returns a string representation of the ChainWrapper object, including the chain, parser type, and fallback parser type.

__repr__() → str

Returns a string representation of the ChainWrapper object for debugging purposes.

run_chain(input_data=None, is_last_chain=False) → Any

Executes the primary chain with the provided input data. If an error occurs, attempts to execute the fallback chain. Supports preprocessing and postprocessing.

Parameters:

• input_data (dict | None) – The input data for the chain Type: Dict[str, Any]

• is_last_chain (bool) – Whether this is the last chain in sequence Defaults to False

get_parser_type() → str | None

Returns the type of the parser as a string if the parser exists, otherwise None.

get_fallback_parser_type() → str | None

Returns the type of the fallback parser as a string if it exists, otherwise None.

__init__(*, chain, fallback_chain, parser=None, fallback_parser=None, preprocessor=None, postprocessor=None, logger=None)

Initialize the ChainBuilder.

Parameters:

• chain (Runnable) – The primary chain to be executed.

• fallback_chain (Runnable) – The fallback chain to be executed if the primary chain fails.

• parser (ParserType | None) – The parser to be used for the primary chain.

• fallback_parser (FallbackParserType | None) – The parser to be used for the fallback chain.

• preprocessor (Callable | None) – A function to preprocess input data before passing it to the chain. Defaults to None.

• postprocessor (Callable | None) – A function to postprocess the output data from the chain. Defaults to None.

• logger (Logger | None) – A logger instance for logging information. Defaults to None.

__str__()

Returns a string representation of the ChainWrapper object.

The string includes the chain, the type of the parser, and the type of the fallback parser.

Returns:

A string representation of the ChainWrapper object.

Return type:

str

__repr__()

Returns a string representation of the object for debugging purposes.

This method calls the __str__() method to provide a user-friendly string representation of the object. It is intended to be used for debugging and development, rather than for end-user display.

Returns:

A string representation of the object.

Return type:

str

run_chain(*, input_data=None, is_last_chain=False)

Executes the primary chain with the provided input data.

If an error occurs in the primary chain, and a fallback chain is provided, it attempts to execute the fallback chain.

Parameters:

• input_data (dict | None) – The input data required to run the chain. Type: Dict[str, Any] Defaults to None.

• is_last_chain (bool | None) – Indicates if this is the last chain in the sequence. Defaults to False.

Returns:

The output from the chain execution, potentially processed by a postprocessor.

Return type:

Any

Raises:

• ValueError – If no input data is provided.

• json.JSONDecodeError – If a JSON decode error occurs in both the main and fallback chains.

• ValidationError – If a validation error occurs in both the main and fallback chains.

• TypeError – If a type error occurs in both the main and fallback chains.

Notes

• If the output is a Pydantic model and this is not the last chain, the model is converted to a dictionary.

• If a postprocessor is defined, it processes the output before returning.

get_parser_type()

Returns the type of the parser as a string if the parser exists, otherwise returns None.

Returns:

The type of the parser as a string, or None if the parser

does not exist.

Return type:

str | None

get_fallback_parser_type()

Returns the type name of the fallback parser if it exists, otherwise returns None.

Returns:

The type name of the fallback parser as a string if it exists,

otherwise None.

Return type:

str | None

class academic_metrics.ChainBuilder.ChainBuilder.ChainComposer(logger=None)

Bases: object

Manages a sequence of chain operations.

ChainComposer is a class that manages a sequence of chain operations, allowing for the addition of chains and running them in sequence with provided data.

__init__(logger

Logger | None): Initializes the ChainComposer instance with an optional logger. Type: Logger | None Defaults to None.

__str__() → str

Returns a string representation of the ChainComposer instance.

__repr__() → str

Returns a string representation of the ChainComposer instance.

add_chain(chain_wrapper, output_passthrough_key_name)

Adds a chain to the chain sequence.

Parameters:

• chain_wrapper (ChainWrapper) – The chain wrapper to add

• output_passthrough_key_name (str | None) – Optional output key name Defaults to None.

run(data_dict, data_dict_update_function)

Runs the chain sequence with the provided data dictionary and optional update function.

Parameters:

• data_dict (dict) – The input data dictionary Type: Dict[str, Any]

• data_dict_update_function (callable | None) – Optional update function Type: Callable[[Dict[str, Any]], None] Defaults to None.

Returns:

The updated data dictionary

Type: Dict[str, Any]

Return type:

dict

__init__(logger=None)

Initializes the ChainBuilder instance.

Parameters:

logger (Logger | None) – A logger instance to be used for logging. Type: logging.Logger | None If not provided, a default logger will be created. Defaults to None.

logger

The logger instance used for logging. Type: logging.Logger

Type:

Logger

chain_sequence

A list to store the chain sequence. Type: List[Tuple[ChainWrapper, str | None]]

Type:

list

__str__()

Returns a string representation of the ChainComposer object.

The string representation includes the index and the wrapper of each element in the chain_sequence.

Returns:

A string representation of the ChainComposer object.

Return type:

str

__repr__()

Returns a string representation of the object for debugging purposes.

This method calls the __str__() method to provide a human-readable representation of the object.

Returns:

A string representation of the object.

Return type:

str

add_chain(*, chain_wrapper, output_passthrough_key_name=None)

Adds a chain to the chain sequence.

Parameters:

• chain_wrapper (ChainWrapper) – The chain wrapper to be added.

• output_passthrough_key_name (str | None) – The key name for output passthrough. Type: str | None Defaults to None.

Return type:

None

Returns:

None

run(*, data_dict, data_dict_update_function=None)

Executes a sequence of chains, updating the provided data dictionary with the results of each chain.

Parameters:

• data_dict (dict) – The initial data dictionary containing input variables for the chains. Type: Dict[str, Any]

• data_dict_update_function (callable | None) – An optional function to update the data dictionary after each chain execution. Type: Callable[[Dict[str, Any]], None] Defaults to None.

Returns:

The updated data dictionary after all chains have been executed.

Type: Dict[str, Any]

Return type:

dict

Raises:

UserWarning – If the provided data dictionary is empty.

Notes

• The method iterates over a sequence of chains (self.chain_sequence), executing each chain and updating the data_dict with the output.

• If an output_name is provided for a chain, the output is stored in data_dict under that name; otherwise, it is stored under the key _last_output.

• If data_dict_update_function is provided, it is called with the updated data_dict after each chain execution.

class academic_metrics.ChainBuilder.ChainBuilder.ChainManager(llm_model, api_key, llm_temperature=0.7, preprocessor=None, postprocessor=None, log_to_console=False, verbose=False, llm_kwargs=None, words_to_ban=None)

Bases: object

Class responsible for managing and orchestrating a sequence of chain layers,

ChainManager is a class responsible for managing and orchestrating a sequence of chain layers, each of which can process input data and produce output data. It supports various types of language models (LLMs) and parsers, and allows for pre-processing and post-processing of data.

api_key

The API key for accessing the LLM service.

Type:

str

llm_model

The name of the LLM model to use.

Type:

str

llm_model_type

The type of the LLM model (e.g., “openai”, “anthropic”, “google”). Type: Literal[“openai”, “anthropic”, “google”]

Type:

str

llm_temperature

The temperature setting for the LLM model.

Type:

float

llm_kwargs

Additional keyword arguments for the LLM model. Type: Dict[str, Any]

Type:

dict

llm

The initialized LLM instance.

Type:

Union[ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI]

chain_composer

The composer for managing the sequence of chain layers.

Type:

ChainComposer

chain_variables

A dictionary of variables used in the chain layers. Type: Dict[str, Any]

Type:

dict

chain_variables_update_overwrite_warning_counter

Counter for tracking variable overwrite warnings.

Type:

int

preprocessor

Optional preprocessor function.

Type:

Callable, optional

postprocessor

Optional postprocessor function.

Type:

Callable, optional

logger

Logger for logging information and debugging.

Type:

Logger

__str__()

Returns a string representation of the ChainManager instance.

__repr__()

Returns a string representation of the ChainManager instance.

_get_llm_model_type(llm_model

str) -> str: Determines the type of the LLM model based on its name.

_initialize_llm(api_key

str, llm_model_type: str, llm_model: str, llm_temperature: float, llm_kwargs: Dict[str, Any]) -> Union[ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI]: Initializes the LLM instance based on the model type.

_create_openai_llm(api_key

str, llm_model: str, llm_temperature: float, llm_kwargs: Dict[str, Any]) -> ChatOpenAI: Creates an OpenAI LLM instance.

_create_anthropic_llm(api_key

str, llm_model: str, llm_temperature: float, llm_kwargs: Dict[str, Any]) -> ChatAnthropic: Creates an Anthropic LLM instance.

_create_google_llm(api_key

str, llm_model: str, llm_temperature: float, llm_kwargs: Dict[str, Any]) -> ChatGoogleGenerativeAI: Creates a Google Generative AI LLM instance.

_initialize_parser(parser_type

str, pydantic_output_model: Optional[BaseModel] = None) -> ParserUnion: Initializes a parser based on the specified type.

_create_pydantic_parser(pydantic_output_model

pydantic.BaseModel) -> PydanticOutputParser: Creates a Pydantic parser.

_create_json_parser(pydantic_output_model

Optional[BaseModel]) -> JsonOutputParser:

) -> JsonOutputParser

Creates a JSON parser.

_create_str_parser() → :class:`~langchain_core.output_parsers.StrOutputParser`

Creates a string parser.

_run_chain_validation_checks(output_passthrough_key_name

str | None, ignore_output_passthrough_key_name_error: bool, parser_type: Literal[“pydantic”, “json”, “str”] | None, pydantic_output_model: BaseModel | None, fallback_parser_type: Literal[“pydantic”, “json”, “str”] | None, fallback_pydantic_output_model: BaseModel | None) -> None: Runs validation checks for the chain configuration.

_format_chain_sequence(chain_sequence

List[Tuple[ChainWrapper, Optional[str]]]) -> None: Formats and prints the chain sequence.

_run_validation_checks(prompt_variables_dict

Union[FirstCallRequired, None]) -> None: Runs validation checks for the prompt variables.

add_chain_layer(system_prompt

str, human_prompt: str, output_passthrough_key_name: str | None = None, ignore_output_passthrough_key_name_error: bool = False, preprocessor: Callable[[Dict[str, Any]], Dict[str, Any]] | None = None, postprocessor: Callable[[Any], Any] | None = None, parser_type: Literal[“pydantic”, “json”, “str”] | None = None, fallback_parser_type: Literal[“pydantic”, “json”, “str”] | None = None, pydantic_output_model: BaseModel | None = None, fallback_pydantic_output_model: BaseModel | None = None) -> None: Adds a chain layer to the chain composer.

_format_overwrite_warning(overwrites

Dict[str, Dict[str, Any]]) -> None: Formats a warning message for variable overwrites.

_check_first_time_overwrites(prompt_variables_dict

Dict[str, Any]) -> None: Checks for first-time overwrites of global variables and issues warnings.

_update_chain_variables(prompt_variables_dict

Dict[str, Any]) -> None: Updates global variables with new values, warning on first-time overwrites.

get_chain_sequence() → List[Tuple[ChainWrapper, str | None]]

Returns the current chain sequence.

print_chain_sequence() → None

Prints the current chain sequence.

get_chain_variables() → Dict[str, Any]

Returns the current chain variables.

print_chain_variables() → None

Prints the current chain variables.

run(prompt_variables_dict

Union[FirstCallRequired, None] = None) -> None: Runs the chain composer with the provided prompt variables.

__init__(llm_model, api_key, llm_temperature=0.7, preprocessor=None, postprocessor=None, log_to_console=False, verbose=False, llm_kwargs=None, words_to_ban=None)

Initializes the ChainBuilder class.

Parameters:

• llm_model (str) – The name of the language model to be used.

• api_key (str) – The API key for accessing the language model.

• llm_temperature (float, optional) – The temperature setting for the language model. Defaults to 0.7.

• preprocessor (Callable | None) – A function to preprocess input data. Defaults to None.

• postprocessor (Callable | None) – A function to postprocess output data. Defaults to None.

• log_to_console (bool | None) – Flag to enable logging to console. Defaults to False.

• llm_kwargs (dict | None) – Additional keyword arguments for the language model. Type: Dict[str, Any]

• words_to_ban (list) – A list of words to ban from the language model’s output. Type: List[str] Defaults to None.

__str__()

Returns a string representation of the ChainManager object.

The string includes the values of the llm, chain_composer, and chain_variables attributes.

Returns:

A string representation of the ChainManager object.

Return type:

str

__repr__()

Returns a string representation of the object for debugging purposes.

This method calls the __str__() method to provide a user-friendly string representation of the object.

Returns:

A string representation of the object.

Return type:

str

_validate_words_to_ban(words_to_ban)

Validate the words to ban list.

Parameters:

words_to_ban (list) – The list of words to ban. Type: List[str]

Return type:

None

_get_logit_bias_dict(llm_model)

Get the logit bias dictionary for the words to ban.

Return type:

Dict[int, int]

_get_llm_model_type(*, llm_model)

Determine the type of LLM (Large Language Model) based on the provided model name.

Parameters:

llm_model (str) – The name of the LLM model.

Returns:

The type of the LLM model. Possible values are “openai”, “anthropic”, and “google”.

Return type:

str

Raises:

ValueError – If the provided LLM model name does not match any of the supported types.

_initialize_llm(*, api_key=None, llm_model_type=None, llm_model=None, llm_temperature=None, llm_kwargs=None, logit_bias_dict=None)

Initializes a language model based on the specified type.

Parameters:

• api_key (str | None) – The API key for accessing the language model service. Defaults to None.

• llm_model_type (str | None) – The type of the language model (e.g., “openai”, “anthropic”, “google”). Defaults to None.

• llm_model (str | None) – The specific model to use within the chosen type. Defaults to None.

• llm_temperature (float | None) – The temperature setting for the language model, affecting randomness. Defaults to None.

• llm_kwargs (dict | None) – Additional keyword arguments specific to the language model. Type: Dict[str, Any] Defaults to None.

• logit_bias_dict (dict | None) – A dictionary mapping token IDs to logit bias values. Type: Dict[int, int] Defaults to None.

Returns:

An instance of the initialized language model.

Return type:

Union[ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI]

Raises:

ValueError – If the specified llm_model_type is not supported.

_recreate_llm()

Recreates the LLM with the current parameters.

Return type:

None

_create_openai_llm(api_key, llm_model, llm_temperature, llm_kwargs, logit_bias_dict=None)

Creates an instance of the ChatOpenAI language model.

Parameters:

• api_key (str) – The API key for authenticating with the OpenAI service.

• llm_model (str) – The identifier of the language model to use.

• llm_temperature (float) – The temperature setting for the language model, which controls the randomness of the output.

• llm_kwargs (dict) – Additional keyword arguments to pass to the ChatOpenAI constructor. Type: Dict[str, Any]

• logit_bias_dict (dict | None) – A dictionary mapping token IDs to logit bias values. Type: Dict[int, int] | None Defaults to None.

Returns:

An instance of the ChatOpenAI language model configured with

the specified parameters.

Return type:

ChatOpenAI

_create_anthropic_llm(api_key, llm_model, llm_temperature, **llm_kwargs)

Creates an instance of the ChatAnthropic language model.

Parameters:

• api_key (str) – The API key for authenticating with the Anthropic service.

• llm_model (str) – The identifier of the language model to use.

• llm_temperature (float) – The temperature setting for the language model, controlling the randomness of the output.

• llm_kwargs (dict) – Additional keyword arguments to pass to the ChatAnthropic constructor. Type: Dict[str, Any]

Returns:

An instance of the ChatAnthropic language model.

Return type:

ChatAnthropic

_create_google_llm(api_key, llm_model, llm_temperature, **llm_kwargs)

Creates an instance of ChatGoogleGenerativeAI with the specified parameters.

Parameters:

• api_key (str) – The API key for authenticating with the Google LLM service.

• llm_model (str) – The model identifier for the Google LLM.

• llm_temperature (float) – The temperature setting for the LLM, which controls the randomness of the output.

• llm_kwargs (dict) – Additional keyword arguments to pass to the ChatGoogleGenerativeAI constructor. Type: Dict[str, Any]

Returns:

An instance of the ChatGoogleGenerativeAI class configured

with the provided parameters.

Return type:

ChatGoogleGenerativeAI

_initialize_parser(parser_type, pydantic_output_model=None)

Initializes and returns a parser based on the specified parser type.

Parameters:

• parser_type (str) – The type of parser to initialize. Must be one of “pydantic”, “json”, or “str”. Type: Literal[“pydantic”, “json”, “str”]

• pydantic_output_model (BaseModel | None) – The Pydantic model to use for the parser. Required if parser_type is “pydantic”. Defaults to None.

Returns:

An instance of the specified parser type.

Return type:

ParserUnion

Raises:

ValueError – If an invalid parser_type is provided.

_create_pydantic_parser(*, pydantic_output_model)

Creates a Pydantic output parser.

Parameters:

pydantic_output_model (BaseModel | None) – The Pydantic model to be used for parsing output.

Returns:

An instance of PydanticOutputParser initialized with

the provided Pydantic model.

Return type:

PydanticOutputParser

Raises:

ValueError – If pydantic_output_model is not provided.

Notes

• pydantic_output_model must be provided for parser_type ‘pydantic’.

_create_json_parser(*, pydantic_output_model)

Creates a JSON parser for the chain layer output.

Parameters:

pydantic_output_model (BaseModel | None) – An optional Pydantic model to enforce typing on the JSON output. Type: pydantic.BaseModel | None Defaults to None.

Returns:

An instance of JsonOutputParser. If pydantic_output_model is provided, the parser will enforce the model’s schema on the output.

Return type:

JsonOutputParser

Raises:

UserWarning – If pydantic_output_model is not provided, a warning is issued recommending its use for proper typing of the output.

_create_str_parser()

Creates an instance of StrOutputParser.

Returns:

An instance of the StrOutputParser class.

Return type:

StrOutputParser

_run_chain_validation_checks(*, output_passthrough_key_name, ignore_output_passthrough_key_name_error, parser_type, pydantic_output_model, fallback_parser_type, fallback_pydantic_output_model)

Validates chain configuration parameters before execution.

Performs validation checks on chain configuration parameters to ensure proper setup and compatibility between different components.

Parameters:

• output_passthrough_key_name (str | None) – Optional key name for passing chain output to next layer. Defaults to None.

• ignore_output_passthrough_key_name_error (bool) – Whether to ignore missing output key name. Defaults to False.

• parser_type (str | None) – Type of parser to use. Type: Literal[“pydantic”, “json”, “str”] | None Defaults to None.

• pydantic_output_model (BaseModel | None) – Pydantic model for output validation. Defaults to None.

• fallback_parser_type (str | None) – Type of fallback parser. Type: Literal[“pydantic”, “json”, “str”] | None Defaults to None.

• fallback_pydantic_output_model (BaseModel | None) – Pydantic model for fallback parser. Defaults to None.

Raises:

ValueError – If validation fails for: - Missing output key name when required - Invalid parser type combinations - Missing required models - Duplicate parser types - Same models used for main and fallback

Warning

UserWarning: For non-critical issues like:

Return type:

None - Missing output key name when ignored - Missing recommended Pydantic models - Unused provided models

_format_chain_sequence(chain_sequence)

Formats and prints the details of each chain in the given chain sequence.

Parameters:

chain_sequence (list) – A list of tuples where each tuple contains a ChainWrapper object and an optional output name. Type: List[Tuple[ChainWrapper, str | None]]

Return type:

None

Returns:

None

_run_validation_checks(*, prompt_variables_dict)

Validates the input parameters for the chain execution.

Parameters:

prompt_variables_dict (dict | None) – A dictionary containing the variables to be passed to the chain layers. Type: Union[FirstCallRequired, None] - On the first call to run(), this parameter must be provided. - On subsequent calls, it can be omitted if there are no new variables to pass.

Raises:

• ValueError – If prompt_variables_dict is None on the first call to run().

• TypeError – If prompt_variables_dict is not a dictionary when provided.

Return type:

None

Notes

• The prompt_variables_dict should contain keys that match the variable names used in the chain layers.

• The output_passthrough_key_name parameter in the add_chain_layer method is used to identify the output of the chain layer and assign it to a variable.

• If output_passthrough_key_name is not specified, the output of the chain layer will not be assigned to a variable and will not be available to the next chain layer.

• The ignore_output_passthrough_key_name_error parameter can be set to True if the output of the chain layer is not needed for the next chain layer, such as when running a chain layer solely for its side effects or if it is the last chain layer in a multi-layer chain.

• Ensure that the placeholder variable names in your prompt strings match the keys in prompt_variables_dict passed into the ChainManager.run() method.

add_chain_layer(*, system_prompt, human_prompt, output_passthrough_key_name=None, ignore_output_passthrough_key_name_error=False, preprocessor=None, postprocessor=None, parser_type=None, fallback_parser_type=None, pydantic_output_model=None, fallback_pydantic_output_model=None)

Adds a chain layer to the chain composer.

This method configures and adds a new chain layer to the chain composer, allowing for the processing of input data through specified prompts and parsers.

Parameters:

• system_prompt (str) – The system prompt template for the chain layer.

• human_prompt (str) – The human prompt template for the chain layer.

• output_passthrough_key_name (str | None) – Key name for passing chain output to the next layer. Defaults to None.

• ignore_output_passthrough_key_name_error (bool) – Flag to ignore missing output key name errors. Defaults to False.

• preprocessor (python:typing.Callable | None) – Function to preprocess input data. Defaults to None.

• postprocessor (python:typing.Callable | None) – Function to postprocess output data. Defaults to None.

• parser_type (str | None) – Type of parser to use. Type: Literal[“pydantic”, “json”, “str”] | None Defaults to None.

• fallback_parser_type (str | None) – Type of fallback parser. Type: Literal[“pydantic”, “json”, “str”] | None Defaults to None.

• pydantic_output_model (BaseModel | None) – Pydantic model for output validation. Defaults to None

• fallback_pydantic_output_model (BaseModel | None) – Pydantic model for fallback parser. Defaults to None.

Return type:

None

Returns:

None

_format_overwrite_warning(overwrites)

Formats a warning message for overwritten values.

Parameters:

overwrites (dict) – A dictionary where the key is the name of the overwritten item, and the value is another dictionary with ‘old’ and ‘new’ keys representing the old and new values respectively. Type: Dict[str, Dict[str, Any]]

Returns:

A formatted string that lists each overwritten item with its old and

new values.

Return type:

str

_check_first_time_overwrites(prompt_variables_dict)

Checks and warns if any global chain variables are being overwritten for the first time.

This method compares the keys in the provided prompt_variables_dict with the existing chain_variables. If any keys match, it indicates that an overwrite is occurring. A warning is issued the first time this happens, detailing the old and new values of the overwritten variables. Subsequent overwrites will not trigger warnings.

Parameters:

prompt_variables_dict (dict) – A dictionary containing the new values for the chain variables that may overwrite existing ones. Type: Dict[str, Any]

Return type:

None

Returns:

None

_update_chain_variables(prompt_variables_dict)

Update global variables with new values, warning on first-time overwrites.

Parameters:

prompt_variables_dict (dict) – A dictionary containing the new values for the global variables. Type: Dict[str, Any]

Return type:

None

Returns:

None

get_chain_sequence()

Retrieves the chain sequence from the chain composer.

Returns:

A list of tuples where each tuple contains a ChainWrapper object and

the output key name if output_passthrough_key_name was provided to add_chain_layer. Type: List[Tuple[academic_metrics.ChainBuilder.ChainBuilder.ChainWrapper, str | None]]

Return type:

list

print_chain_sequence()

Prints the chain sequence by formatting it.

This method retrieves the chain sequence from the chain composer and formats it using the _format_chain_sequence method.

Return type:

None

Returns:

None

get_chain_variables()

Retrieve the chain variables.

Returns:

A dictionary containing the chain variables.

Type: Dict[str, Any]

Return type:

dict

print_chain_variables()

Prints the chain variables in a formatted manner.

This method prints the chain variables stored in the chain_variables attribute of the class. The output is formatted with a header and footer consisting of dashes, and each key-value pair is printed on a new line.

Return type:

None

Returns:

None

set_words_to_ban(words_to_ban)

Updates the list of words to ban and recreates the OpenAI LLM with new logit bias.

Parameters:

words_to_ban (list) – List of words to ban from LLM output Type: List[str]

Return type:

None

Returns:

None

run(*, prompt_variables_dict=None)

Executes the chain builder process.

This method performs validation checks, updates chain variables if provided, and runs the chain composer with the current chain variables.

Parameters:

prompt_variables_dict (dict | None) – A dictionary containing prompt variables. If provided, it will be used to update the chain variables. Type: Union[FirstCallRequired, None] Defaults to None.

Returns:

The result of running the chain composer.

Return type:

str

Module contents