LangGraph

LangGraph is a framework that allows you to define execution flows. It is designed for building agentic systems.

from time import sleep
from typing import TypedDict, Literal

import langgraph
from langgraph.types import Command
from langgraph.graph import END, START, StateGraph

State

The state is the object that moves through the entire graph. It provides the input information to the nodes and passes the information from the current node to the next one.

The schema of the State must be provided when defining of the langgraph.graph.StateGraph. Supported ways to define a schema include: typing.TypedDict, dataclasses.dataclass, or pydantic.BaseModel.

---

Consider a case in which the state is defined as a dataclass. The following cell creates a graph that processes a State dataclass instance and increments the inp attribute.

from dataclasses import dataclass

@dataclass
class State:
    inp: int

def my_node(state: State):
    print(state)
    state.inp += 1
    return state

builder = StateGraph(State)
builder.add_node("my", my_node)

builder.add_edge(START, "my")
builder.add_edge("my", END)

graph = builder.compile()

The following cell invokes the graph for State(inp=4).

graph.invoke(State(inp=4))

State(inp=4)

{'inp': 5}

Parallel execution

If a node has multiple outgoing edges, the flows defined by those edges will execute in parallel.

---

The following cell defines and displays this type of graph. The nodes of the graph display information about the start and end of exection, and block the flow for a while.

class State(TypedDict):
    pass

builder = StateGraph(State)

def node_a(state: State):
    print("executing a")
    sleep(10)
    print("finishing a")

def node_b(state: State):
    print("executing b")
    sleep(10)
    print("execution b")

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")
graph = builder.compile()
graph

The project’s invocation is represented in the following cell:

graph.invoke(State())

executing a
executing b
finishing a
execution b

It follows from the result that flows was executed in parallel.

Merging output

If two graph flows join at the same node and both return a value, there will be a conflict. You must specify a strategy for merging the outputs using a reducer function.

THe reducer function can be defined as metatdata of the typing.Annotated for the specific attribute. This function will be applied if the outputs from those nodes need to be processed in some way. The most common way is to use the operator.add function, which is equivalent to simply apply the + operator.

---

The following cell shows a regular graph with two flows join at the END node, and both return some information.

class State(TypedDict):
    out: str

def node_a(state: State) -> State:
    return State(out="A output")

def node_b(state: State) -> State:
    return State(out="B output")

builder = StateGraph(State)

builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph

Executing such a graph results in the error displayed in the following cell.

try:
    graph.invoke(State(out="start"))
except Exception as e:
    print(type(e).__name__, ":", e)

InvalidUpdateError : At key 'out': Can receive only one value per step. Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/INVALID_CONCURRENT_GRAPH_UPDATE

The following cell shows the definition of the state that applies operator.add when there are two outputs in a node.

import operator
from typing import Annotated

class State(TypedDict):
    out: Annotated[str, operator.add]

Here is an example of invoking the same graph with an updated state schema.

def node_a(state: State) -> State:
    return State(out="A output")

def node_b(state: State) -> State:
    return State(out="B output")

builder = StateGraph(State)

builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph.invoke(State(out="value"))

{'out': 'valueA outputB output'}

As a result, the outputs of the different nodes were simply concatenated.

Conditional edges

Conditional edges allow you to define a graph that will go one way or another depending on the conditions.

You can define the conditional node using:

• Adding the node that returns langgraph.types.Command with goto specifying the name of the next node.

• Adding the edge with add_conditional_edge method of the graph builder.

---

The following cell defines the conditional node that directs the execution to either the END or "a" node, depending on the corresponding value of the state["goto"].

class State(TypedDict):
    goto: Literal["a", "__end__"]

def node_a(state: State) -> State:
    print("a node is invoked")
    return state

def conditional_node(state: State) -> Command[Literal["a", "__end__"]]:
    return Command(update=state, goto=state["goto"])

graph = (
    StateGraph(State)
    .add_node("conditional", conditional_node)
    .add_node("a", node_a)

    .add_edge(START, "conditional")
    .add_edge("a", END)
    .compile()
)
graph

The following cell invokes the graph with to be executed with the "a" node.

_ = graph.invoke(State(goto="a"))

a node is invoked

And alternatively routing directly to the __end__.

_ = graph.invoke(State(goto="__end__"))

Alternative defition

The add_conditional_edges method of the builder, takes the method that must return the names of the possible subsequent nodes.

---

The following code illustrates how to create a graph with conditional edge using the add_conditional_edges method.

class State(TypedDict):
    goto: Literal["a", "__end__"]

def node_a(state: State) -> State:
    print("a node is invoked")
    return state

def conditional_edge(state: State) -> Literal["a", "__end__"]:
    return state["goto"]

(
    StateGraph(State)
    .add_node("a", node_a)

    .add_conditional_edges(START, conditional_edge)
    .add_edge("a", END)
    .compile()
)