Source code for langchain_aws.runnables.q_business

import logging
from typing import Any, List, Optional, Union

from langchain_core._api.beta_decorator import beta
from langchain_core.messages.ai import AIMessage
from langchain_core.prompt_values import ChatPromptValue
from langchain_core.runnables import Runnable
from langchain_core.runnables.config import RunnableConfig
from pydantic import ConfigDict
from typing_extensions import Self

logger = logging.getLogger(__name__)



[docs]
@beta(message="This API is in beta and can change in future.")
class AmazonQ(Runnable[Union[str,ChatPromptValue, List[ChatPromptValue]], ChatPromptValue]):
    """Amazon Q Runnable wrapper.

    To authenticate, the AWS client uses the following methods to
    automatically load credentials:
    https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html

    Make sure the credentials / roles used have the required policies to
    access the Amazon Q service.
    """

    region_name: Optional[str] = None
    """AWS region name. If not provided, will be extracted from environment."""

    credentials: Optional[Any] = None
    """Amazon Q credentials used to instantiate the client if the client is not provided."""

    client: Optional[Any] = None
    """Amazon Q client."""

    application_id: str = None

    """Store the full response from Amazon Q."""

    parent_message_id: Optional[str] = None

    conversation_id: Optional[str] = None

    chat_mode: str = "RETRIEVAL_MODE"

    model_config = ConfigDict(
        extra="forbid",
    )

    def __init__(
        self,
        region_name: Optional[str] = None,
        credentials: Optional[Any] = None,
        client: Optional[Any] = None,
        application_id: str = None,
        parent_message_id: Optional[str] = None,
        conversation_id: Optional[str] = None,
        chat_mode: str = "RETRIEVAL_MODE",
    ):
        self.region_name = region_name
        self.credentials = credentials
        self.client = client or self.validate_environment()
        self.application_id = application_id
        self.parent_message_id = parent_message_id
        self.conversation_id = conversation_id
        self.chat_mode = chat_mode

    def invoke(
        self,
        input: Union[str,ChatPromptValue],
        config: Optional[RunnableConfig] = None,
        **kwargs: Any
    ) -> ChatPromptValue:
        """Call out to Amazon Q service.

        Args:
            input: The prompt to pass into the model.

        Returns:
            The string generated by the model.

        Example:
            .. code-block:: python

            model = AmazonQ(
                credentials=your_credentials,
                application_id=your_app_id
            )
            response = model.invoke("Tell me a joke")
        """
        try:
            # Prepare the request            
            request = {
                'applicationId': self.application_id,
                'userMessage': self.convert_langchain_messages_to_q_input(input), # Langchain's input comes in the form of an array of "messages". We must convert to a single string for Amazon Q's use
                'chatMode': self.chat_mode,
            }
            if self.conversation_id:
                request.update({
                    'conversationId': self.conversation_id,
                    'parentMessageId': self.parent_message_id,
                })

            # Call Amazon Q
            response = self.client.chat_sync(**request)

            # Extract the response text
            if 'systemMessage' in response:
                return AIMessage(content=response["systemMessage"], response_metadata=response)
            else:
                raise ValueError("Unexpected response format from Amazon Q")

        except Exception as e:
            if "Prompt Length" in str(e):
                logger.info(f"Prompt Length: {len(input)}")
                logger.info(f"""Prompt:
                {input}""")
            raise ValueError(f"Error raised by Amazon Q service: {e}")

    def validate_environment(self) -> Self:
        """Don't do anything if client provided externally"""
        #If the client is not provided, and the user_id is not provided in the class constructor, throw an error saying one or the other needs to be provided
        if self.credentials is None:
            raise ValueError(
                "Either the credentials or the client needs to be provided."
            )

        """Validate that AWS credentials to and python package exists in environment."""
        try:
            import boto3

            try:
                if self.region_name is not None:
                    client = boto3.client('qbusiness', self.region_name, **self.credentials)
                else:
                    # use default region
                    client = boto3.client('qbusiness', **self.credentials)

            except Exception as e:
                raise ValueError(
                    "Could not load credentials to authenticate with AWS client. "
                    "Please check that credentials in the specified "
                    "profile name are valid."
                ) from e

        except ImportError:
            raise ImportError(
                "Could not import boto3 python package. "
                "Please install it with `pip install boto3`."
            )
        return client
    def convert_langchain_messages_to_q_input(self, input: Union[str,ChatPromptValue,List[ChatPromptValue]]) -> str:
        #If it is just a string and not a ChatPromptTemplate collection just return string
        if type(input) is str:
            return input
        return input.to_string()