What is LangGraph?

LangGraph: A Comprehensive Guide for Beginners

LangGraph is a library for building stateful, multi-actor applications with LLMs. It extends the LangChain Expression Language with the ability to coordinate multiple chains (or actors) across multiple steps of computation in a cyclic manner. It is inspired by Pregel and Apache Beam. The current interface exposed is one inspired by NetworkX.

The main use is for adding cycles to your LLM application. Crucially, LangGraph is NOT optimized for acyclic, or Directed Acyclic Graph (DAG), workflows. If you want to build a DAG, you should just use LangChain Expression Language.

Cycles are important for agent-like behaviours, where you call an LLM in a loop, asking it what action to take next.

Some key features of LangGraph include:

• Graph-based data structures: Define complex data structures using graph theory concepts.
• Functional programming: Focus on immutability, recursion, and higher-order functions for concise code.
• Type-safe: Ensure type correctness through a statically typed language.
• Dynamic typing: Allow for flexible typing at runtime (optional).
• Integration with popular libraries: Seamlessly integrate with popular libraries for data science, machine learning, and more.

Installation

pip install langgraph

Quick start

• One of the central concepts of LangGraph is state. Each graph execution creates a state that is passed between nodes in the graph as they execute, and each node updates this internal state with its return value after it executes. The way that the graph updates its internal state is defined by either the type of graph chosen or a custom function.
• State in LangGraph can be pretty general, but to keep things simpler to start, we’ll show off an example where the graph’s state is limited to a list of chat messages using the built-in MessageGraph class. This is convenient when using LangGraph with LangChain chat models because we can return chat model output directly.

A Simple Example

Now, let’s go over a more general example with a cycle. We will recreate the AgentExecutor class from LangChain. The agent itself will use chat models and function calling. This agent will represent all its state as a list of messages.

We will need to install some LangChain packages, as well as Tavily to use as an example tool.

• First, install the LangChain OpenAI integration package:

pip install -U langchain langchain_openai tavily-python

• We also need to export some additional environment variables for OpenAI and Tavily API access.

export OPENAI_API_KEY=sk-...
export TAVILY_API_KEY=tvly-...

Step 1: Set up the tools

• As above, we will first define the tools we want to use. For this simple example, we will use a built-in search tool via Tavily. However, it is really easy to create your own tools — see documentation here on how to do that.

from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolExecutor

tools = [TavilySearchResults(max_results=1)]
tool_executor = ToolExecutor(tools)

Step 2: Set up the model

• Now we need to load the chat model we want to use. This time, we’ll use the older function calling interface. This walkthrough will use OpenAI, but we can choose any model that supports OpenAI function calling.

from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0, streaming=True)

• We should make sure the model knows that it has these tools available to call. We can do this by converting the LangChain tools into the format for OpenAI tool calling using the bind_tools() method.

model = model.bind_tools(tools)

Step 3: Define the Graph State

• The state we will track will just be a list of messages. We want each node to just add messages to that list. Therefore, we will use a TypedDict with one key (messages) and annotate it so that the messages attribute is always added to the second parameter (operator.add). (Note: the state can be any type, including pydantic BaseModel's).

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    # The `add_messages` function within the annotation defines
    # *how* updates should be merged into the state.
    messages: Annotated[list, add_messages]

Step 4: Define Nodes

We now need to define a few different nodes in our graph. In langgraph, a node can be either a function or a runnable. There are two main nodes we need for this:

1. The agent: responsible for deciding what (if any) actions to take.
2. A function to invoke tools: if the agent decides to take an action, this node will then execute that action.

We will also need to define some edges. Some of these edges may be conditional. The reason they are conditional is that based on the output of a node, one of several paths may be taken. The path that is taken is not known until that node is run (the LLM decides).

1. Conditional Edge: after the agent is called, we should either:

• a. If the agent said to take an action, then the function to invoke tools should be called
• b. If the agent said that it was finished, then it should finish

2. Normal Edge: after the tools are invoked, it should always go back to the agent to decide what to do next

Let’s define the nodes, as well as a function to decide how what conditional edge to take.

from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage

# Define the function that determines whether to continue or not
def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    # If there is no function call, then we finish
    if "function_call" not in last_message.additional_kwargs:
        return "end"
    # Otherwise if there is, we continue
    else:
        return "continue"

# Define the function that calls the model
def call_model(state):
    messages = state['messages']
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}

# Define the function to execute tools
def call_tool(state):
    messages = state['messages']
    last_message = messages[-1]
    action = ToolInvocation(
        tool=last_message.additional_kwargs["function_call"]["name"],
        tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
    )
    response = tool_executor.invoke(action)
    function_message = FunctionMessage(content=str(response), name=action.tool)
    
    return {"messages": [function_message]}

Step 5: Define the graph

• We can now put it all together and define the graph!

from langgraph.graph import StateGraph, END

# Define a new graph
workflow = StateGraph(AgentState)

# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # Otherwise we finish.
        "end": END
    }
)

workflow.add_edge('action', 'agent')

Step 6: Compile and Run the Graph

• Finally, we compile our graph and run it with some input.

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable

from langchain_core.messages import HumanMessage

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
app.invoke(inputs)

Conclusion

LangGraph is a versatile tool for building complex, stateful applications with LLMs. By understanding its core concepts and working through simple examples, beginners can start to leverage its power for their projects. Remember to pay attention to state management, conditional edges, and ensuring there are no dead-end nodes in your graph. Happy coding!