I am participating in a challenge that asks to building an agent system to answer questions based on a database. The competition provides a virtual environment where many packages are not the latest versions. Additionally, I need to deploy the solution on AWS and use services like ChatBedrock to interact with large language models (LLMs). This process requires logging into the cloud, and some packages assume the presence of environment variables for login information. For instance, the Embedchain package expects an AWS_REGION environment variable for authentication.

The aws_bedrock.py script includes a class AWSBedrockLlm, which is designed to interface with AWS services:

@register_deserializable
class AWSBedrockLlm(BaseLlm):
    def __init__(self, config: Optional[BaseLlmConfig] = None):
        super().__init__(config)

    def get_llm_model_answer(self, prompt) -> str:
        response = self._get_answer(prompt, self.config)
        return response

    def _get_answer(self, prompt: str, config: BaseLlmConfig) -> str:
        try:
            import boto3
        except ModuleNotFoundError:
            raise ModuleNotFoundError(
                "The required dependencies for AWSBedrock are not installed."
                'Please install with `pip install --upgrade "embedchain[aws-bedrock]"`'
            ) from None

        self.boto_client = boto3.client("bedrock-runtime", "us-west-2" or os.environ.get("AWS_REGION"))

        kwargs = {
            "model_id": config.model or "amazon.titan-text-express-v1",
            "client": self.boto_client,
            "model_kwargs": config.model_kwargs
            or {
                "temperature": config.temperature,
            },
        }

        if config.stream:
            from langchain.callbacks.streaming_stdout import \
                StreamingStdOutCallbackHandler

            callbacks = [StreamingStdOutCallbackHandler()]
            llm = Bedrock(**kwargs, streaming=config.stream, callbacks=callbacks)
        else:
            llm = Bedrock(**kwargs)

        return llm.invoke(prompt)

I encountered an issue logging into AWS using the existing code and wanted to experiment with the package. After researching, I discovered that I could override the method even after the object was instantiated.

First, I defined my new function in a way I could login in the AWS.

def new_get_llm_model_answer(self, prompt) -> str:
        '''
        Overide dfault get_llm_model_answer to login in chadbedrock using session
        '''
        session = boto3.Session()
        self.boto_client = session.client("bedrock-runtime")

        kwargs = {
            "model_id": self.config.model or "amazon.titan-text-express-v1",
            "client": self.boto_client,
            "model_kwargs": self.config.model_kwargs,
            "disable_streaming": True,
            "verbose": True
            or {
                "temperature": self.config.temperature,
            },
        }

        if self.config.stream:
            from langchain.callbacks.streaming_stdout import \
                StreamingStdOutCallbackHandler

            callbacks = [StreamingStdOutCallbackHandler()]
            llm = ChatBedrock(**kwargs, streaming=self.config.stream, callbacks=callbacks)
        else:
            llm = ChatBedrock(**kwargs,)

        response = llm.invoke(prompt).content
        return response

Next, I instantiated my Embedchain app and replaced one specific method:

app = App.from_config(config_path='src/rag/embedchain_youtube_config.yaml')
app.llm.get_answer_from_llm = new_get_llm_model_answer.__get__(app.llm, type(app.llm)) # <-- This is the code responsible to replace the old method.

The line new_get_llm_model_answer.__get__(app.llm, type(app.llm)) manually binds the new function new_get_llm_model_answer to the app.llm instance. The __get__ method takes two arguments:

• Instance (app.llm): The instance through which the attribute is accessed.
• Owner (type(app.llm)): The class of the instance.

By assigning this bound method back to app.llm.get_answer_from_llm, we can effectively override the existing method for this particular instance. It allows for dynamic modification of methods at runtime without altering the class definition.

To illustrate this concept abstractly imagine you have a class “Dog” that prints “WOOF” when you use the method “bark”. You can print a different type of bark without changing the class and other instances by using the aforementioned trick.

class Dog:
    def bark(self):
        print("WOOF")

# Create an instance of Dog
boby = Dog()
boby.bark()  # Outputs: WOOF

# Define a new bark function
def new_bark(self):
    print("WoOoOoF!!")

# Override the bark method for this instance
boby.bark = new_bark.__get__(boby, Dog)
boby.bark()  # Outputs: WoOoOoF!!

This trick to override a method allowed me to log into AWS and experiment with the Embedchain package. A very good addition to my collection of tricks.