LangGraph

Hello and welcome back to part 3 of this tutorial series. In this part, we’ll be getting started with LangGraph. Instead of having a lot of explanation before we start, we’ll see how stuff works as we go along. So without further ado, let’s just jump right in.

Let’s start by actually installing LangGraph, as it doesn’t get installed by default with LangChain. To install LangGraph, you can use the following command in your terminal:

pipenv install langgraph==0.0.30 langchainhub==0.1.15

Once you’ve installed LangGraph, let’s start by creating a new file called simple_langgraph.py:

📂 FINX_LANGGRAPH
    📂 images
    📂 tools
    📄 .env
    📄 langchain_basics.py
    📄 Pipfile
    📄 Pipfile.lock
    📄 setup_environment.py
    📄 simple_langgraph.py    ✨New file

Over the next three parts, we’ll be looking at different ways in which you can use LangGraph to chain LLMs and tools together. In this first part we’ll be looking at a simple classic LLM –> goes to a tool executor –> and then back to LLM type setup.

Open up simple_langgraph.py and let’s start by importing the necessary modules:

import operator
from typing import Annotated, TypedDict, Union

from colorama import Fore, Style
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_core.agents import AgentAction, AgentActionMessageLog, AgentFinish
from langchain_core.messages import BaseMessage
from langchain_core.runnables.base import Runnable
from langchain_openai.chat_models import ChatOpenAI
from langgraph.graph import END, StateGraph
from langgraph.prebuilt.tool_executor import ToolExecutor

from setup_environment import set_environment_variables
from tools import generate_image, get_weather

That is a lot of stuff! Don’t worry, most of it is actually not as complex as it seems. Usually, I’ll go over all the imports before we get started, but as there are quite a few to go through, I’ll cover each import when we get to the part where it’s used instead. For now, just have them copied.

Next, we’ll set the environment variables and define a couple of constants:

set_environment_variables("LangGraph Basics")

LLM = ChatOpenAI(model="gpt-3.5-turbo-0125", streaming=True)
TOOLS = [get_weather, generate_image]
PROMPT = hub.pull("hwchase17/openai-functions-agent")

We reused our set_environment_variables function from the previous part to set the environment variables and set the name for the LangSmith traces to LangGraph Basics. We then define our LLM just like we did in part 1, also setting the streaming parameter to True. We then define a list of tools which is literally just a list containing the two tools that we wrote.

The LangChain Hub

For the prompt template, we pull it from the LangChain Hub this time, mostly because I want to show you that it exists! The LangChain Hub is kind of like a mini-GitHub for storing LangChain ChatPromptTemplates just like the simple ones we wrote in part 1. You can push new commits to your templates and pull them like we just did here, kind of like GitHub.

You can go to https://smith.langchain.com/ and scroll down to find the Hub button:

Click it to visually browse the prompts available on the hub:

You can use this as a convenient place to store your prompts. You can also set them to private if you don’t want to share them with the world and you can even fork other public prompts that you like to your own repositories. It’s a handy tool for development. For production or highly sensitive company data, you might want to store your prompts in a more secure location.

If we look up the prompt we just pulled, we can see that it is a fairly simple prompt:

It has an extremely basic system message of "You are a helpful assistant" and we can see that it has placeholders for chat_history, human input and an agent_scratchpad. The chat_history and input are kind of self-explanatory in that they hold the chat history so far and the human input, but what about this agent_scratchpad?

The agent_scratchpad is kind of like a place where the agent can take notes while going through its reasoning process of what action should be taken next and what functions should be called. Think of it as a notepad where the LLM can jot down its thoughts. Think of it kind of like the following:

user:
    "Can you recommend me a zombie game from the year 2022?"

    > Entering new AgentExecutor chain...
    Thought: Oh, I love zombie games! There are so many great ones out there. Let me think about the best zombie game from 2022.
    Action: use_search_engine
    Action Input: "best zombie game 2022"

    Observation:[{list of search result objects for query "best zombie game 2022"}]
    There are three great zombie games from 2022 that I found: Zombie Cure Lab, Zombie Survivors, and SurrounDead. Let me think about which one to recommend.
    Action: use_search_engine
    Action Input: "Zombie Cure Lab"

    Observation:[{list of search result objects for query "Zombie Cure Lab"}]
    Zombie Cure Lab is a game where you manage a lab and try to cure the zombie virus. (Bunch more info here yadayada...) I recommend Zombie Cure Lab as the best zombie game from 2022.

    Final Answer: The best zombie game from 2022 is Zombie Cure Lab.

This is just a conceptual example here to describe the idea, but the agent takes reasoning steps and makes observations along the way, first deciding to call a search engine tool to better answer the user question, then deciding to call the search engine tool to get more information on one of the games in particular, and then finally deciding that it has enough information to answer the user question.

So the agent_scratchpad is used to store these intermediate observations on what action to take next, but also to decide when the agent is done, so that it doesn’t just keep looping indefinitely. We’ll get back to how we can see when the agent is done in a moment.

The State Object

Ok, we have an LLM, some tools, and a prompt template. The next thing we need is a state object to keep track of the state for each step along our graph. So a LangGraph is kind of like a state machine, and it is going to take this state object and pass it along each node of the graph. Let’s look at a simplified example:

# Simplified example
StateObject():
    user_input = "please do a for me"
    chat_history = [list of previous chat messages for context...]
    am_i_done = False
    steps_taken = []

So say we have this state object above. We have received the user input question, and whatever chat history has come before if we have decided to implement memory. We have a flag am_i_done which is obviously set to False at the start, and we have a list of steps_taken which is empty at the start. Now we hand this state object to node A in our graph ->

# Simplified example Node A
StateObject():
    user_input = "please do a for me"
    chat_history = [list of previous chat messages for context...]
    am_i_done = False
    steps_taken = ["action_a was taken"]

It does some action we will just call action_a, which has taken it a step closer to answering the user question but it is not quite done yet so the am_i_done flag is still set to false. Now node A passes this state object to node B in our graph ->

# Simplified example Node B
StateObject():
    user_input = "please do a for me"
    chat_history = [list of previous chat messages for context...]
    am_i_done = True
    steps_taken = ["action_a was taken", "action_b was taken"]

This node does some action_b stuff and now has the final answer it needs to give to the user. It sets the am_i_done flag to True because it is done. We can use this am_i_done flag to test if the graph is completed yet (e.g. the user question or request has been fully answered).

So as the graph traverses over the nodes we define, each node will receive the state object, update it where needed, and then pass it along to the next node, or perhaps back to the previous one, or sideways to node D if a certain condition is met. So let’s define the real state object that we will be using:

class AgentState(TypedDict):
    input: str
    chat_history: list[BaseMessage]
    agent_outcome: Union[AgentAction, AgentFinish, None]
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

We use a TypedDict to define a specific dictionary structure, defining the keys that this dictionary will have and the types of values that will be stored for each of those keys. The first entry is simply the user input, which is a str string value.

The second entry is the chat history, which is a list of BaseMessage objects. A BaseMessage object is just any one of the lines of this object below where you have a message and the originator of the message like “system”, “human”, or “ai”:

# Example BaseMessages
("system", "You are a helpful AI bot. Your name is {name}."),
("human", "Hello, how are you doing?"),
("ai", "I'm doing well, thanks!"),
("human", "{user_input}"),

The third item in the state object will be agent_outcome. The agent here will do its thing and then either return an AgentAction object or an AgentFinish object to us.

• AgentAction: An AgentAction object simply contains the name of the tool the agent wants to call and the input arguments for that tool call, maybe like get_weather and {"location": "New York"}.
• AgentFinish: An AgentFinish object simply means that the agent considers its task finished and holds the final return_values inside.

Using this agent_outcome object we can see what the next step is or if it is done.

The fourth and last entry in the AgentState object is a bit easier to read from the inside. We have a list of tuples where each tuple contains an AgentAction object and a str string. The AgentAction here is the same object that we described in the step above, containing a tool to be called and its input arguments. The difference here is that the step is already taken and the string which is the second item in the tuple is the tool output after it was called. So something like this:

## Fictional example object
[
    (
        AgentAction(tool="get_weather", input={"location": "New York"}),
        "{API response JSON object...}",
    ),
    (
        AgentAction(tool="generate_image", input={"image_description": "cat"}),
        "Path/to/image.png",
    ),
]

The Annotated type hint is used to add metadata to the type hint. In this case, we are using the operator.add function to tell the type checker that this list will be added to, so we are describing the AgentState object’s intermediate_steps list as a list that will be added to, like the example above.

The Agent

Now that we have our state object defined, we will define our agent that will have access to both the generate_image and get_weather tools:

runnable_agent: Runnable = create_openai_functions_agent(LLM, TOOLS, PROMPT)

We use the create_openai_functions_agent function we imported from LangChain to create an agent that has access to the LLM, the tools, and the prompt we defined so far. LangChain will make this into an OpenAI compatible agent by combining them for us into a Runnable type object. We have seen this Runnable object before in part 1 in the form of our chains. All Runnable type objects have the invoke, stream, and batch methods just like the chains we used in part 1.

Before we move on with the nodes and graph let’s test the agent we have so far. We’ll manually create a quick input here (as we haven’t built our graph yet) and then call invoke on the agent:

inputs = {
    "input": "give me the weather for New York please.",
    "chat_history": [],
    "intermediate_steps": [],
}

agent_outcome = runnable_agent.invoke(inputs)
print(agent_outcome)

Now go ahead and run this to test the agent so far and you should see something like this:

API Keys loaded and tracing set with project name:  LangGraph Basics
tool='get_weather' tool_input={'location': 'New York'}