Use the Graph API

from typing_extensions import TypedDict
from langchain.messages import AnyMessage

class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("Hello!")
    return {"messages": messages + [new_message], "extra_field": 10}

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

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int  # Default: overwrite

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

from langgraph.types import Overwrite

def replace_messages(state: State):
    return {"messages": Overwrite(["replacement message"])}

class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class OverallState(InputState, OutputState):
    pass

graph = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)

class Node1Output(TypedDict):
    private_data: str

def node_1(state: OverallState) -> Node1Output:
    return {"private_data": "set by node_1"}

class Node2Input(TypedDict):
    private_data: str

def node_2(state: Node2Input) -> OverallState:
    return {"a": "set by node_2"}

from pydantic import BaseModel

class OverallState(BaseModel):
    a: str

from langgraph.runtime import Runtime

class ContextSchema(TypedDict):
    llm_provider: str = "openai"

def node(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    ...

graph = StateGraph(State, context_schema=ContextSchema)
graph.invoke(inputs, context={"llm_provider": "anthropic"})

from langgraph.types import RetryPolicy

builder.add_node("node_name", node_function, retry_policy=RetryPolicy(max_attempts=3))

builder.add_node("model", call_model, timeout=1.0)

from langgraph.errors import NodeError
from langgraph.types import Command

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(update={"status": f"compensated: {error.error}"}, goto="finalize")

def my_node(state: State, runtime: Runtime) -> State:
    thread_id = runtime.execution_info.thread_id
    attempt = runtime.execution_info.node_attempt
    ...

from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache

builder.add_node("expensive", expensive_node, cache_policy=CachePolicy(ttl=120))
graph = builder.compile(cache=InMemoryCache())

builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_node(step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")

builder.add_edge(START, "call_llm_1")
builder.add_edge(START, "call_llm_2")
builder.add_edge(START, "call_llm_3")
builder.add_edge("call_llm_1", "aggregator")
builder.add_edge("call_llm_2", "aggregator")
builder.add_edge("call_llm_3", "aggregator")

from langgraph.types import Send
from typing import Annotated
import operator

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

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

from langgraph.types import Command
from typing import Literal

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(update={"foo": "bar"}, goto="my_other_node")

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",
        graph=Command.PARENT
    )

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))