LangGraph Multi-Modal Agent with Function Calling

%pip install -U langchain-community langgraph langchain-chroma langchain_aws pandas  ipywidgets pillow

import getpass
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()
os.environ["LANGCHAIN_PROJECT"] = "lc-multimodal-agent"

from langchain_aws import ChatBedrock
import boto3

<h1>---- ⚠️ Update region for your AWS setup ⚠️ ----</h1>
bedrock_client = boto3.client("bedrock-runtime", region_name="us-east-1")

import pandas as pd
from collections import Counter
from langchain_core.tools import tool


def read_travel_data(file_path: str = "synthetic_travel_data.csv") -> pd.DataFrame:
    """Read travel data from CSV file"""
    try:
        df = pd.read_csv(file_path)
        return df
    except FileNotFoundError:
        return pd.DataFrame(
            columns=[
                "Name",
                "Current_Location",
                "Age",
                "Past_Travel_Destinations",
                "Number_of_Trips",
                "Flight_Number",
                "Departure_City",
                "Arrival_City",
                "Flight_Date",
            ]
        )


@tool
def compare_and_recommend_destination(name: str) -> str:
    """This tool is used to check which destinations user has already traveled. 
    Use name of the user to fetch the information about that user.
    If user has already been to a city then do not recommend that city. 

    Args:
        name (str): Name of the user.
    Returns:
        str: Destination to be recommended.

    """

    df = read_travel_data()

    if name not in df["Name"].values:
        return "User not found in the travel database."

    user_data = df[df["Name"] == name].iloc[0]
    current_location = user_data["Current_Location"]
    age = user_data["Age"]
    past_destinations = user_data["Past_Travel_Destinations"].split(", ")

    # Get all past destinations of users with similar age (±5 years) and same current location
    similar_users = df[
        (df["Current_Location"] == current_location)
        & (df["Age"].between(age - 5, age + 5))
    ]
    all_destinations = [
        dest
        for user_dests in similar_users["Past_Travel_Destinations"].str.split(", ")
        for dest in user_dests
    ]

    # Count occurrences of each destination
    destination_counts = Counter(all_destinations)

    # Remove user's current location and past destinations from recommendations
    for dest in [current_location] + past_destinations:
        if dest in destination_counts:
            del destination_counts[dest]

    if not destination_counts:
        return f"No new recommendations found for users in {current_location} with similar age."

    # Get the most common destination
    recommended_destination = destination_counts.most_common(1)[0][0]

    return f"Based on your current location ({current_location}), age ({age}), and past travel data, we recommend visiting {recommended_destination}."

tools = [compare_and_recommend_destination]

llm_with_tools = llm.bind_tools(tools)

from langchain_aws import ChatBedrockConverse

llm = ChatBedrockConverse(
    model="anthropic.claude-3-sonnet-20240229-v1:0",
    # model="anthropic.claude-3-5-sonnet-20240620-v1:0",
    temperature=0,
    max_tokens=None,
    client=bedrock_client,
    # other params...
)

import operator
from typing import Annotated, Sequence, TypedDict

from langchain_core.messages import BaseMessage


class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

from langchain_core.messages import ToolMessage

from langgraph.prebuilt import ToolInvocation

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, MessagesState, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)


def chatbot(state: State):
    """Always use tools to fulfill user requests.
    1. If you do not have enough inputs to execute a tool then you can ask for more information.
    2. If user has uploaded image and 'image_processing_node' has returned city and activity then use that information to call 'chatbot'
    """
    # Filter out messages with image type
    # text_messages = [msg for msg in state["messages"] if msg['content'][0].get("type") != "image"]
    text_messages = [
        msg for msg in state["messages"] 
        if not (isinstance(msg.content, list) and msg.content[0].get("type") == "image_url")
    ]

    # Invoke LLM with only text messages
    return {"messages": [llm_with_tools.invoke(text_messages)]}


graph_builder.add_node("chatbot", chatbot)

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(tools)
graph_builder.add_node("tools", tool_node)

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
    {"tools": "tools", "__end__": "__end__"},
)

def process_image_input(state):
    """
    Process image input. This tool will return activity shown in this image. 
    """
    last_message = state["messages"][-1].content[0]
    input_image = last_message['image_url']['url']
    print(input_image)
    message = HumanMessage(
        content=[
            {"type": "text", "text": "Which activity is shown in this image?"},
            {
                "type": "image_url",
                # "image_url": {"url": {input_image}},
                "image_url": {"url": f"data:image/jpeg;base64,{input_image}"},
            },
        ],
    )
    response = llm.invoke([message])
    print(response.content)
    output = {
        "messages": [
            HumanMessage(
                content=f"Image information: {response.content}", name="image_description"
            )
        ],
    }
    return output

graph_builder.add_node("image_processing_node", process_image_input)

def is_image_node(state):
    messages = state["messages"]
    last_message = messages[-1]
    if hasattr(last_message, "content") and isinstance(last_message.content, list):
        for item in last_message.content:
            if isinstance(item, dict) and item.get("type") == "image_url":
                return True
    return False

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

<h1>Any time a tool is called, we return to the chatbot to decide the next step</h1>
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("image_processing_node", "chatbot")
graph_builder.add_conditional_edges(START, is_image_node, {True: "image_processing_node", False:"chatbot"})

graph = graph_builder.compile(checkpointer=memory)

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

from IPython.display import display, clear_output
from PIL import Image
import io
import ipywidgets as widgets
import base64

import ipywidgets as widgets
from IPython.display import display, clear_output
from PIL import Image
import io
import base64


class ImageUploader:
    def __init__(self):
        self.img_data = None
        self.img_str = None
        self.img_format = None

        self.uploader = widgets.FileUpload(
            accept="image/*",  # Accept all image types
            multiple=False,  # Allow only single file upload
        )
        self.uploader.observe(self.process_image, names="value")

    def process_image(self, change):
        clear_output(wait=True)  # Clear previous output
        if change["new"]:  # Check if there's a new file uploaded
            # Get the image data
            self.img_data = change["new"][0][
                "content"
            ]  # Access the first (and only) file

            # Open the image using PIL
            img = Image.open(io.BytesIO(self.img_data))

            # Display the image
            display(img)

            # Print image information
            print(f"Image size: {img.size}")
            print(f"Image format: {img.format}")
            print(f"Image mode: {img.mode}")

            # Store the image format
            self.img_format = img.format

            # Convert image to base64
            buffered = io.BytesIO()
            img.save(buffered, format=self.img_format)
            self.img_str = base64.b64encode(buffered.getvalue()).decode("utf-8")

            print("\nBase64 representation of the image:")
            print(
                self.img_str[:100] + "..." + self.img_str[-100:]
            )  # Print first and last 100 characters
            print(f"\nTotal length of base64 string: {len(self.img_str)} characters")

    def display_uploader(self):
        display(self.uploader)
        print("Upload an image to see its information and base64 representation.")

    def get_base64_string(self):
        return self.img_str

    def get_image_format(self):
        return self.img_format

<h1>Create an instance of ImageUploader</h1>
uploader = ImageUploader()

<h1>Display the upload widget</h1>
uploader.display_uploader()

from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "20"}}
message = HumanMessage(
    content=[
        {
            "type": "image_url",
            "image_url": {"url": uploader.img_str},
        },
    ],
)
print(graph.invoke({"messages": message}, config)["messages"][-1].content)

print(
    graph.invoke(
        {
            "messages": [
                (
                    "user",
                    "Suggest the travel location based on uploaded image",
                )
            ]
        },
        config,
    )["messages"][-1].content
)