[docs]@beta(message="This API is in beta and can change in future.")classAmazonQ(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]=Noneconversation_id:Optional[str]=Nonechat_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_nameself.credentials=credentialsself.client=clientorself.validate_environment()self.application_id=application_idself.parent_message_id=parent_message_idself.conversation_id=conversation_idself.chat_mode=chat_modedefinvoke(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,}ifself.conversation_id:request.update({'conversationId':self.conversation_id,'parentMessageId':self.parent_message_id,})# Call Amazon Qresponse=self.client.chat_sync(**request)# Extract the response textif'systemMessage'inresponse:returnAIMessage(content=response["systemMessage"],response_metadata=response)else:raiseValueError("Unexpected response format from Amazon Q")exceptExceptionase:if"Prompt Length"instr(e):logger.info(f"Prompt Length: {len(input)}")logger.info(f"""Prompt:{input}""")raiseValueError(f"Error raised by Amazon Q service: {e}")defvalidate_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 providedifself.credentialsisNone:raiseValueError("Either the credentials or the client needs to be provided.")"""Validate that AWS credentials to and python package exists in environment."""try:importboto3try:ifself.region_nameisnotNone:client=boto3.client('qbusiness',self.region_name,**self.credentials)else:# use default regionclient=boto3.client('qbusiness',**self.credentials)exceptExceptionase:raiseValueError("Could not load credentials to authenticate with AWS client. ""Please check that credentials in the specified ""profile name are valid.")fromeexceptImportError:raiseImportError("Could not import boto3 python package. ""Please install it with `pip install boto3`.")returnclientdefconvert_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 stringiftype(input)isstr:returninputreturninput.to_string()