ai_workflows.llm_utilities module

Utilities for interacting with LLMs in AI workflows.

class ai_workflows.llm_utilities.JSONSchemaCache

Bases: object

Cache for JSON schemas.

static get_json_schema(json_description: str) → str

Retrieve cached schema from JSON description.

Parameters:

json_description (str) – Description of the JSON format.

Returns:

JSON schema or empty string if not found.

Return type:

str

static put_json_schema(json_description: str, json_schema: str)

Cache a JSON schema.

Parameters:

• json_description (str) – Description of the JSON format.

• json_schema (str) – JSON schema to cache.

class ai_workflows.llm_utilities.LLMInterface(openai_api_key: str | None = None, openai_model: str | None = None, temperature: float = 0.0, total_response_timeout_seconds: int = 600, number_of_retries: int = 2, seconds_between_retries: int = 5, azure_api_key: str | None = None, azure_api_engine: str | None = None, azure_api_base: str | None = None, azure_api_version: str | None = None, langsmith_api_key: str | None = None, langsmith_project: str = 'ai_workflows', langsmith_endpoint: str = 'https://api.smith.langchain.com', json_retries: int = 2, anthropic_api_key: str | None = None, anthropic_model: str | None = None, bedrock_model: str | None = None, bedrock_region: str = 'us-east-1', bedrock_aws_profile: str | None = None, max_tokens: int = -1, reasoning_effort: str | None = None, system_prompt: str = '', maintain_history: bool = False, starting_chat_history: list[tuple] | None = None)

Bases: object

Utility class for interacting with LLMs in AI workflows.

__init__(openai_api_key: str | None = None, openai_model: str | None = None, temperature: float = 0.0, total_response_timeout_seconds: int = 600, number_of_retries: int = 2, seconds_between_retries: int = 5, azure_api_key: str | None = None, azure_api_engine: str | None = None, azure_api_base: str | None = None, azure_api_version: str | None = None, langsmith_api_key: str | None = None, langsmith_project: str = 'ai_workflows', langsmith_endpoint: str = 'https://api.smith.langchain.com', json_retries: int = 2, anthropic_api_key: str | None = None, anthropic_model: str | None = None, bedrock_model: str | None = None, bedrock_region: str = 'us-east-1', bedrock_aws_profile: str | None = None, max_tokens: int = -1, reasoning_effort: str | None = None, system_prompt: str = '', maintain_history: bool = False, starting_chat_history: list[tuple] | None = None)

Initialize the LLM interface for LLM interactions.

This function sets up the interface for interacting with various LLMs, including OpenAI and Azure, and configures the necessary parameters for API access and response handling.

Parameters:

• openai_api_key (str) – OpenAI API key for accessing the LLM. Default is None.

• openai_model (str) – OpenAI model name. Default is None.

• temperature (float) – Temperature setting for the LLM. Default is 0.0.

• total_response_timeout_seconds (int) – Timeout for LLM responses in seconds. Default is 600.

• number_of_retries (int) – Number of retries for LLM calls. Default is 2.

• seconds_between_retries (int) – Seconds between retries for LLM calls. Default is 5.

• azure_api_key (str) – API key for Azure LLM. Default is None.

• azure_api_engine (str) – Azure API engine name (deployment name; assumed to be the same as the OpenAI model name). Default is None.

• azure_api_base (str) – Azure API base URL. Default is None.

• azure_api_version (str) – Azure API version. Default is None.

• langsmith_api_key (str) – API key for LangSmith. Default is None.

• langsmith_project (str) – LangSmith project name. Default is ‘ai_workflows’.

• langsmith_endpoint (str) – LangSmith endpoint URL. Default is ‘https://api.smith.langchain.com’.

• json_retries (int) – Number of automatic retries for invalid JSON responses. Default is 2.

• anthropic_api_key (str) – API key for Anthropic. Default is None.

• anthropic_model (str) – Anthropic model name. Default is None.

• bedrock_model (str) – AWS Bedrock model name. Default is None.

• bedrock_region (str) – AWS Bedrock region. Default is “us-east-1”.

• bedrock_aws_profile (str) – AWS profile for Bedrock access. Default is None.

• max_tokens (int) – Maximum tokens for LLM responses. Default is -1, which sets to 50000 for reasoning models (GPT-5, o1, o3, o4-mini) or 16000 for other models. Specify a value to override.

• reasoning_effort (str) – Reasoning effort level for reasoning models (‘low’, ‘medium’, ‘high’). Default is None, which sets to ‘low’ for reasoning models. Ignored for non-reasoning models.

• system_prompt (str) – System prompt to add to all LLM calls. Default is “”.

• maintain_history (bool) – Whether to maintain a history of LLM interactions (and include the history in each call to the LLM). Default is False.

• starting_chat_history (list[tuple]) – Starting chat history to use for the conversation chain (or None for none). Should be tuples, each with a human and an AI message.

async a_count_tokens(text: str) → int

Count the number of tokens in a text string (async version).

Parameters:

text (str) – Text to count tokens in.

Returns:

Number of tokens in the text.

Return type:

int

async a_enforce_max_tokens(text: str, max_tokens: int) → str

Truncate a string as necessary to fit within a maximum number of tokens (async version).

Parameters:

• text (str) – Text to potentially truncate.

• max_tokens (int) – Maximum number of tokens to allow.

Returns:

Original or truncated string.

Return type:

str

async a_generate_json_schema(json_output_spec: str) → str

Generate a JSON schema, adequate for JSON validation, based on a human-language JSON output specification (async version).

Parameters:

json_output_spec (str) – Human-language JSON output specification.

Returns:

JSON schema suitable for JSON validation purposes.

Return type:

str

async a_get_json_response(prompt: str | list, json_validation_schema: str = '', bypass_history_and_system_prompt=False, *, config=None) → tuple[dict | None, str, str]

Call out to LLM for structured JSON response (async version).

This function sends a prompt to the LLM and returns the response in JSON format.

Parameters:

• prompt (str | list) – Prompt to send to the LLM.

• json_validation_schema (str) – JSON schema for validating the JSON response (optional). Default is “”, which means no validation.

• bypass_history_and_system_prompt (bool) – Whether to bypass the history and system prompt. Default is False.

Returns:

Tuple with parsed JSON response, raw LLM response, and error message (if any).

Return type:

tuple(dict | None, str, str)

async a_get_llm_response(prompt: str | list, bypass_history_and_system_prompt=False, json_mode: bool = False, *, config=None) → str

Call out to LLM for a response to a prompt (async version).

Parameters:

• prompt (str | list) – Prompt to send to the LLM (simple string or list of user and assistant messages).

• bypass_history_and_system_prompt (bool) – Whether to bypass the history and system prompt. Default is False.

• json_mode (bool) – Whether to use model’s “JSON mode” (if available). Default is False.

Returns:

Content of the LLM response.

Return type:

str

static ai_message(ai_message: str) → dict

Generate an AI message.

This function takes an AI message and returns it in the message format expected by the LLM.

Parameters:

ai_message (str) – AI message to format.

Returns:

Formatted AI message.

Return type:

dict

count_tokens(text: str) → int

Count the number of tokens in a text string (synchronous version).

Parameters:

text (str) – Text to count tokens in.

Returns:

Number of tokens in the text.

Return type:

int

enforce_max_tokens(text: str, max_tokens: int) → str

Truncate a string as necessary to fit within a maximum number of tokens (synchronous version).

Parameters:

• text (str) – Text to potentially truncate.

• max_tokens (int) – Maximum number of tokens to allow.

Returns:

Original or truncated string.

Return type:

str

static extract_json(text: str, *, config=None) → list[dict]

Extract JSON content from a string, handling various formats.

Parameters:

text (str) – Text containing potential JSON content.

Returns:

A list of extracted JSON objects, or [] if none could be parsed or found.

Return type:

list[dict]

generate_json_schema(json_output_spec: str) → str

Generate a JSON schema, adequate for JSON validation, based on a human-language JSON output specification (synchronous version).

Parameters:

json_output_spec (str) – Human-language JSON output specification.

Returns:

JSON schema suitable for JSON validation purposes.

Return type:

str

static get_image_bytes(image: Image, output_format: str = 'PNG', max_bytes: int | None = None, current_dpi: int | None = None) → bytes

Convert a PIL Image to bytes in the specified format.

This function takes a PIL Image object and converts it to a byte array in the specified format.

Parameters:

• image (Image.Image) – PIL Image to convert.

• output_format (str) – Output format for the image (default is ‘PNG’).

• max_bytes (int | None) – Maximum size in bytes for the image. If the image is larger than this, it will be resized to fit within the limit. Default is None, which means no limit. (If you specify a limit, you must also specify the current DPI of the image.)

• current_dpi (int | None) – Current DPI of the image. Defaults to None but must be set if max_bytes is specified.

Returns:

Bytes representing the image in the specified format.

Return type:

bytes

get_json_response(prompt: str | list, json_validation_schema: str = '', json_validation_desc: str = '', bypass_history_and_system_prompt=False, *, config=None) → tuple[dict | None, str, str]

Call out to LLM for structured JSON response (synchronous version).

This function sends a prompt to the LLM and returns the response in JSON format.

Parameters:

• prompt (str | list) – Prompt to send to the LLM.

• json_validation_schema (str) – JSON schema for validating the JSON response (optional). Default is “”.

• json_validation_desc – Description of the JSON schema for validating response (optional). Default is “”. If supplied, will be converted into a JSON schema, cached in-memory, and used for validation.

• bypass_history_and_system_prompt (bool) – Whether to bypass the history and system prompt. Default is False.

Returns:

Tuple with parsed JSON response, raw LLM response, and error message (if any).

Return type:

tuple(dict | None, str, str)

get_llm_response(prompt: str | list, bypass_history_and_system_prompt=False, json_mode: bool = False, *, config=None) → str

Call out to LLM for a response to a prompt (synchronous version).

Parameters:

• prompt (str | list) – Prompt to send to the LLM (simple string or list of user and assistant messages).

• bypass_history_and_system_prompt (bool) – Whether to bypass the history and system prompt. Default is False.

• json_mode (bool) – Whether to use model’s “JSON mode” (if available). Default is False.

Returns:

Content of the LLM response.

Return type:

str

reset_history()

Reset the conversation history.

This function clears the conversation history maintained by the LLM interface.

system_message(system_message: str) → dict

Generate a system message.

This function takes a system message and returns it in the message format expected by the LLM.

Parameters:

system_message (str) – System message to format.

Returns:

Formatted system message.

Return type:

dict

static user_message(user_message: str) → dict

Generate a user message.

This function takes a user message and returns it in the message format expected by the LLM.

Parameters:

user_message (str) – User message to format.

Returns:

Formatted user message.

Return type:

dict

user_message_with_image(user_message: str, image: Image, max_bytes: int | None = None, current_dpi: int | None = None) → dict

Generate a user message with an embedded image.

This function takes a user message and an image and returns a combined message with the image embedded.

Parameters:

• user_message (str) – User message to include with the image.

• image (Image.Image) – Image to embed in the message.

• max_bytes (int | None) – Maximum size in bytes for the image. If the image is larger than this, it will be resized to fit within the limit. Default is None, which means no limit. (If you specify a limit, you must also specify the current DPI of the image.)

• current_dpi (int | None) – Current DPI of the image. Defaults to None but must be set if max_bytes is specified.

Returns:

Combined message with the image embedded.

Return type:

dict