Types

from typing import Annotated
from langgraph.types import Send
from langgraph.graph import END, START
from langgraph.graph import StateGraph
import operator

class OverallState(TypedDict):
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]

def continue_to_jokes(state: OverallState):
    # Send multiple messages to invoke the node in parallel
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder = StateGraph(OverallState)
builder.add_node(
    "generate_joke",
    lambda state: {"jokes": [f"Joke about {state['subject']}"]}
)
builder.add_conditional_edges(START, continue_to_jokes)
builder.add_edge("generate_joke", END)
graph = builder.compile()

# Invoking with two subjects results in a generated joke for each
graph.invoke({"subjects": ["cats", "dogs"]})
# {'subjects': ['cats', 'dogs'], 'jokes': ['Joke about cats', 'Joke about dogs']}

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

class State(TypedDict):
    foo: str

def my_node(state: State):
    # Update state and navigate to a specific node
    return Command(
        update={"foo": "bar"},
        goto="next_node"
    )

def next_node(state: State):
    return {"foo": state["foo"] + "_processed"}

builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_node("next_node", next_node)
builder.add_edge(START, "my_node")
graph = builder.compile()

result = graph.invoke({"foo": "initial"})
print(result)  # {"foo": "bar_processed"}

import uuid
from typing import Optional
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command


class State(TypedDict):
    """The graph state."""
    foo: str
    human_value: Optional[str]
    """Human value will be updated using an interrupt."""


def node(state: State):
    answer = interrupt(
        # This value will be sent to the client
        # as part of the interrupt information.
        "what is your age?"
    )
    print(f"> Received an input from the interrupt: {answer}")
    return {"human_value": answer}


builder = StateGraph(State)
builder.add_node("node", node)
builder.add_edge(START, "node")

# A checkpointer must be enabled for interrupts to work!
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": uuid.uuid4(),
    }
}

for chunk in graph.stream({"foo": "abc"}, config):
    print(chunk)
# > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)}

command = Command(resume="some input from a human!!!")

for chunk in graph.stream(command, config):
    print(chunk)
# > Received an input from the interrupt: some input from a human!!!
# > {'node': {'human_value': 'some input from a human!!!'}}

from langgraph.types import RetryPolicy
from langgraph.graph import StateGraph

class State(TypedDict):
    value: int

def unreliable_node(state: State):
    # This might fail occasionally
    if random.random() < 0.5:
        raise ValueError("Random failure")
    return {"value": state["value"] + 1}

builder = StateGraph(State)
builder.add_node(
    "unreliable",
    unreliable_node,
    retry=RetryPolicy(
        initial_interval=1.0,
        max_attempts=5,
        backoff_factor=2.0,
        retry_on=ValueError
    )
)

from langgraph.types import CachePolicy
from langgraph.graph import StateGraph

def expensive_computation(state: State):
    # This is an expensive operation that we want to cache
    time.sleep(5)
    return {"result": state["input"] * 2}

builder = StateGraph(State)
builder.add_node(
    "expensive",
    expensive_computation,
    cache=CachePolicy(
        ttl=3600  # Cache for 1 hour
    )
)

from typing import Annotated
import operator
from langgraph.graph import StateGraph
from langgraph.types import Overwrite

class State(TypedDict):
    messages: Annotated[list, operator.add]

def node_a(state: State):
    # Normal update: uses the reducer (operator.add)
    return {"messages": ["a"]}

def node_b(state: State):
    # Overwrite: bypasses the reducer and replaces the entire value
    return {"messages": Overwrite(value=["b"])}

builder = StateGraph(State)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.set_entry_point("node_a")
builder.add_edge("node_a", "node_b")
graph = builder.compile()

# Without Overwrite in node_b, messages would be ["START", "a", "b"]
# With Overwrite, messages is just ["b"]
result = graph.invoke({"messages": ["START"]})
assert result == {"messages": ["b"]}