### Configure InMemoryStore for Semantic Search

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Initialize an InMemoryStore with an embedding model and specify the dimensions and fields to be embedded for semantic search capabilities.

```python
from langchain.embeddings import init_embeddings

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),  # Embedding provider
        "dims": 1536,                              # Embedding dimensions
        "fields": ["food_preference", "$"]              # Fields to embed
    }
)
```

--------------------------------

### Store Memories with Custom Embedding Indexing

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Demonstrates how to store memories, either by specifying particular fields for embedding or by explicitly disabling embedding for certain data, controlling what is searchable.

```python
# Store with specific fields to embed
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {
        "food_preference": "I love Italian cuisine",
        "context": "Discussing dinner plans"
    },
    index=["food_preference"]  # Only embed "food_preferences" field
)

# Store without embedding (still retrievable, but not searchable)
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {"system_info": "Last updated: 2024-01-01"},
    index=False
)
```

--------------------------------

### Enable Semantic Search in Memory Store

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Initialize an InMemoryStore with embeddings to enable semantic search. Use `store.put` to add items and `store.search` to retrieve semantically similar items.

```python
from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 1536,
    }
)

store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})

items = store.search(
    ("user_123", "memories"), query="I'm hungry", limit=1
)
```

--------------------------------

### Initialize In-Memory Vector Store with OpenAI Embeddings (Python)

Source: https://docs.langchain.com/oss/python/langgraph/agentic-rag

This snippet sets up an in-memory vector store for semantic search, indexing pre-split documents using OpenAI embeddings. The resulting retriever can be used to fetch relevant documents based on a query.

```python
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
```

--------------------------------

### Search and Retrieve User Memories for Model Call (Python)

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Illustrates how to use the Runtime object to search for user memories based on the latest message and prepare them for use in a model call.

```python
from dataclasses import dataclass
from langgraph.runtime import Runtime

@dataclass
class Context:
    user_id: str

async def call_model(state: MessagesState, runtime: Runtime[Context]):
    # Get the user id from the runtime context
    user_id = runtime.context.user_id

    # Namespace the memory
    namespace = (user_id, "memories")

    # Search based on the most recent message
    memories = await runtime.store.asearch(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    info = "\n".join([d.value["memory"] for d in memories])

    # ... Use memories in the model call
```

--------------------------------

### Invoke Graph with Valid Input

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Invokes the graph with a valid input dictionary, where the key 'a' maps to a string value, demonstrating successful execution.

```python
graph.invoke({"a": "hello"})
```

--------------------------------

### Perform Semantic Search on Memories

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Execute a semantic search query against the configured store to find relevant memories based on natural language meaning, returning a specified number of top matches.

```python
# Find memories about food preferences
# (This can be done after putting memories into the store)
memories = store.search(
    namespace_for_memory,
    query="What does the user like to eat?",
    limit=3  # Return top 3 matches
)
```

--------------------------------

### Define a Namespace for LangGraph Memories

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Defines a tuple-based namespace for organizing memories, allowing for flexible categorization (e.g., by user ID and memory type) within the LangGraph store.

```python
user_id = "1"
namespace_for_memory = (user_id, "memories")
```

--------------------------------

### Access Memory Store in Node Functions via Runtime (Python)

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Explains how to access the `InMemoryStore` within a node function using the `Runtime` object, allowing for searching and storing memories. It also shows how to pass context during graph invocation.

```python
from dataclasses import dataclass
from langgraph.runtime import Runtime
from langgraph.graph import StateGraph, MessagesState, START
import uuid

@dataclass
class Context:
    user_id: str

async def call_model(state: MessagesState, runtime: Runtime[Context]):  # [!code highlight]
    user_id = runtime.context.user_id  # [!code highlight]
    namespace = (user_id, "memories")

    # Search for relevant memories
    memories = await runtime.store.asearch(  # [!code highlight]
        namespace, query=state["messages"][-1].content, limit=3
    )
    info = "\n".join([d.value["data"] for d in memories])

    # ... Use memories in model call

    # Store a new memory
    await runtime.store.aput(  # [!code highlight]
        namespace, str(uuid.uuid4()), {"data": "User prefers dark mode"}
    )

builder = StateGraph(MessagesState, context_schema=Context)  # [!code highlight]
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(store=store)

# Pass context at invocation time
graph.invoke(
    {"messages": [{"role": "user", "content": "hi"}]},
    {"configurable": {"thread_id": "1"}},
    context=Context(user_id="1"),  # [!code highlight]
)
```

--------------------------------

### Add Long-Term Memory with InMemoryStore (Python)

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Illustrates how to initialize an `InMemoryStore` and compile a `StateGraph` with it to enable long-term memory capabilities.

```python
from langgraph.store.memory import InMemoryStore  # [!code highlight]
from langgraph.graph import StateGraph

store = InMemoryStore()  # [!code highlight]

builder = StateGraph(...)
graph = builder.compile(store=store)  # [!code highlight]
```

--------------------------------

### Create a Cyclic Pregel Graph with Conditional Termination in Python

Source: https://docs.langchain.com/oss/python/langgraph/pregel

This snippet illustrates how to introduce a cycle where a node writes back to its input channel, with execution continuing until a `None` value is written, demonstrating iterative processing.

```python
from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry

example_node = (
    NodeBuilder().subscribe_only("value")
    .do(lambda x: x + x if len(x) < 10 else None)
    .write_to(ChannelWriteEntry("value", skip_none=True))
)

app = Pregel(
    nodes={"example_node": example_node},
    channels={
```

--------------------------------

### Update User Memory with LangGraph Runtime (Python)

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Demonstrates how to access the Runtime object within a node to retrieve the user_id from context and asynchronously store new memories.

```python
from langgraph.runtime import Runtime
from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

async def update_memory(state: MessagesState, runtime: Runtime[Context]):

    # Get the user id from the runtime context
    user_id = runtime.context.user_id

    # Namespace the memory
    namespace = (user_id, "memories")

    # ... Analyze conversation and create a new memory

    # Create a new memory ID
    memory_id = str(uuid.uuid4())

    # We create a new memory
    await runtime.store.aput(namespace, memory_id, {"memory": memory})
```

--------------------------------

### Paginate LangGraph Store Search Results

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Illustrates how to paginate through a large namespace using `store.search` with `limit` and `offset` parameters, allowing retrieval of items in batches.

```python
page_size = 50
offset = 0
while True:
    page = store.search(("alice", "memories"), limit=page_size, offset=offset)
    if not page:
        break
    for item in page:
        pass
    offset += page_size
```

--------------------------------

### Define a Simple Looping Graph with State Aggregation

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

This snippet defines a `StateGraph` with two nodes (`a`, `b`) and a conditional edge that creates a loop, terminating when the `aggregate` list in the state reaches a length of 7.

```python
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# Define edges
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

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

--------------------------------

### Define Conditional Entry Point with Output Mapping (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Map the output of the routing function to specific starting nodes when defining a conditional entry point, allowing for dynamic graph initiation.

```python
graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})
```

--------------------------------

### Search LangGraph Store with Item Limit

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Demonstrates using `store.search` with a `limit` argument to retrieve a specified maximum number of items from a given namespace prefix, useful for controlling result set size.

```python
# Return up to 100 items stored under ("alice", "memories").
items = store.search(("alice", "memories"), limit=100)
```

--------------------------------

### Initialize Anthropic Chat Model (init_chat_model)

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Initializes an Anthropic chat model using the `init_chat_model` utility function. Requires `ANTHROPIC_API_KEY` to be set in the environment.

```python
import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")
```

--------------------------------

### Retrieve Memories from LangGraph Store using Search

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Retrieves memories from a specified namespace using `store.search`, returning a list of `Item` objects. The example shows accessing the last item and converting it to a dictionary.

```python
memories = store.search(namespace_for_memory)
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}
```

--------------------------------

### Initialize HuggingFace Chat Model

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Initialize a HuggingFace chat model using either `init_chat_model` or by directly instantiating `ChatHuggingFace` with a `HuggingFaceEndpoint`. Set the `HUGGINGFACEHUB_API_TOKEN` environment variable.

```python
import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)
```

```python
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)
```

--------------------------------

### Invoke Graph with Invalid Input for Pydantic Validation

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Attempts to invoke the graph with an invalid input type (integer instead of string) to demonstrate Pydantic's automatic validation and error handling.

```python
try:
    graph.invoke({"a": 123})  # Should be a string
except Exception as e:
    print("An exception was raised because `a` is an integer rather than a string.")
    print(e)
```

--------------------------------

### Define a LangGraph with Runtime Configuration

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

This snippet demonstrates how to define a `ContextSchema` for runtime values, integrate it into a node's function signature, and access these values within the node logic.

```python
from langgraph.graph import END, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

# 1. Specify config schema
class ContextSchema(TypedDict):
    my_runtime_value: str

# 2. Define a graph that accesses the config in a node
class State(TypedDict):
    my_state_value: str

def node(state: State, runtime: Runtime[ContextSchema]):  # [!code highlight]
    if runtime.context["my_runtime_value"] == "a":  # [!code highlight]
        return {"my_state_value": 1}
    elif runtime.context["my_runtime_value"] == "b":  # [!code highlight]
        return {"my_state_value": 2}
    else:
        raise ValueError("Unknown values.")

builder = StateGraph(State, context_schema=ContextSchema)  # [!code highlight]
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)

graph = builder.compile()
```

--------------------------------

### Synchronous LangGraph with OracleDB Memory

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

This snippet shows how to integrate a BaseStore for memory management into a synchronous LangGraph StateGraph. It demonstrates storing and retrieving user-specific memories and streaming responses.

```python
              state: MessagesState,
              config: RunnableConfig,
              *,
              store: BaseStore,  # [!code highlight]
          ):
              user_id = config["configurable"]["user_id"]
              namespace = ("memories", user_id)
              memories = store.search(namespace, query=str(state["messages"][-1].content))  # [!code highlight]
              info = "\n".join([d.value["data"] for d in memories])
              system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

              # Store new memories if the user asks the model to remember
              last_message = state["messages"][-1]
              if "remember" in last_message.content.lower():
                  memory = "User name is Bob"
                  store.put(namespace, str(uuid.uuid4()), {"data": memory})  # [!code highlight]

              response = model.invoke(
                  [{"role": "system", "content": system_msg}] + state["messages"]
              )
              return {"messages": response}

          builder = StateGraph(MessagesState)
          builder.add_node(call_model)
          builder.add_edge(START, "call_model")

          graph = builder.compile(
              checkpointer=checkpointer,
              store=store,  # [!code highlight]
          )

          config = {
              "configurable": {
                  "thread_id": "1",  # [!code highlight]
                  "user_id": "1",  # [!code highlight]
              }
          }
          for chunk in graph.stream(
              {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
              config,  # [!code highlight]
              stream_mode="values",
          ):
              chunk["messages"][-1].pretty_print()

          config = {
              "configurable": {
                  "thread_id": "2",  # [!code highlight]
                  "user_id": "1",
              }
          }

          for chunk in graph.stream(
              {"messages": [{"role": "user", "content": "what is my name?"}]},
              config,  # [!code highlight]
              stream_mode="values",
          ):
              chunk["messages"][-1].pretty_print()
```

--------------------------------

### Multi-Turn Conversation with State Accumulation

Source: https://docs.langchain.com/oss/python/langgraph/use-subgraphs

Illustrates how an agent maintains state across multiple invocations, allowing it to remember past conversations in a multi-turn interaction.

```python
config = {"configurable": {"thread_id": "1"}}

# First call
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about apples"}]},
    config=config,
)
# Subagent message count: 4

# Second call - subagent REMEMBERS apples conversation
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Now tell me about bananas"}]},
    config=config,
)
# Subagent message count: 8 (accumulated!)
```

--------------------------------

### Add Conditional Edges with Output Mapping (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Provide a dictionary to `add_conditional_edges` to map the routing function's output to specific next node names, allowing for explicit control over the next step.

```python
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
```

--------------------------------

### Configure LangGraph Store Indexing for Semantic Search (JSON)

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Example langgraph.json configuration for enabling semantic search by specifying embedding model, dimensions, and fields for indexing.

```json
{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}
```

--------------------------------

### Stream Graph Updates with New Thread and Existing User Context (Python)

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Shows how to invoke a graph on a new thread while maintaining the same user_id in the context to access existing memories.

```python
# Invoke the graph on a new thread
config = {"configurable": {"thread_id": "2"}}

# Let's say hi again
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]},
    config,
    stream_mode="updates",
    context=Context(user_id="1"),
):
    print(update)
```

--------------------------------

### Initialize Anthropic Chat Model (ChatAnthropic Class)

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Initializes an Anthropic chat model directly using the `ChatAnthropic` class. Requires `ANTHROPIC_API_KEY` to be set in the environment.

```python
import os
from langchain_anthropic import ChatAnthropic

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = ChatAnthropic(model="claude-sonnet-4-6")
```

--------------------------------

### Defining State with `operator.add` Reducer (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

This `TypedDict` uses `Annotated` with `operator.add` for the `bar` key, causing list updates to be appended instead of overwritten. The `foo` key retains default overwrite behavior.

```python
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]
```

--------------------------------

### Consume Multiple Projections Concurrently with astream_events (Python)

Source: https://docs.langchain.com/oss/python/langgraph/event-streaming

For asynchronous code, use "astream_events" and "asyncio.gather" to concurrently consume multiple projections like messages and subgraphs from the same event stream.

```python
import asyncio

stream = await graph.astream_events(input, version="v3")

async def consume_messages():
    async for message in stream.messages:
        print(f"[llm] node={message.node}")

async def consume_subgraphs():
    async for subgraph in stream.subgraphs:
        print(f"[subgraph] path={subgraph.path}")

await asyncio.gather(consume_messages(), consume_subgraphs())
```

--------------------------------

### Pass Private State Between Nodes (Python)

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Illustrates how to pass private data between specific nodes in a sequential graph, ensuring it's not part of the overall public state accessible to all nodes.

```python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# The overall state of the graph (this is the public state shared across nodes)
class OverallState(TypedDict):
    a: str

# Output from node_1 contains private data that is not part of the overall state
class Node1Output(TypedDict):
    private_data: str

# The private data is only shared between node_1 and node_2
def node_1(state: OverallState) -> Node1Output:
    output = {"private_data": "set by node_1"}
    print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Node 2 input only requests the private data available after node_1
class Node2Input(TypedDict):
    private_data: str

def node_2(state: Node2Input) -> OverallState:
    output = {"a": "set by node_2"}
    print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Node 3 only has access to the overall state (no access to private data from node_1)
def node_3(state: OverallState) -> OverallState:
    output = {"a": "set by node_3"}
    print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {output}")
    return output

# Connect nodes in a sequence
# node_2 accepts private data from node_1, whereas
# node_3 does not see the private data.
builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()

# Invoke the graph with the initial state
response = graph.invoke(
    {
        "a": "set at start",
    }
)

print()
print(f"Output of graph invocation: {response}")
```

--------------------------------

### Define Input and Output Schemas for `StateGraph` in Python

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Demonstrates how to define distinct `TypedDict` schemas for graph input, output, and an overall internal state, allowing for structured data handling.

```python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# Define the schema for the input
class InputState(TypedDict):
    question: str

# Define the schema for the output
class OutputState(TypedDict):
    answer: str

# Define the overall schema, combining both input and output
class OverallState(InputState, OutputState):
    pass
```

--------------------------------

### Set Graph Durability Mode to Sync (Python)

Source: https://docs.langchain.com/oss/python/langgraph/persistence

Demonstrates how to specify the "sync" durability mode when executing a LangGraph stream, ensuring checkpoints are written synchronously.

```python
graph.stream(
    {"input": "test"},
    durability="sync"
)
```

--------------------------------

### Install Anthropic Integration

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Installs the necessary LangChain package for Anthropic chat model integration.

```shell
pip install -U "langchain[anthropic]"
```

--------------------------------

### Download SQLite Sample Database (Python)

Source: https://docs.langchain.com/oss/python/langgraph/sql-agent

Downloads the `Chinook.db` SQLite sample database from a public GCS bucket if the file does not already exist in the local directory.

```python
import pathlib
import requests

url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
local_path = pathlib.Path("Chinook.db")

if local_path.exists():
    print(f"{local_path} already exists, skipping download.")
else:
    response = requests.get(url, timeout=60)
    if response.status_code == 200:
        local_path.write_bytes(response.content)
        print(f"File downloaded and saved as {local_path}")
    else:
        print(f"Failed to download the file. Status code: {response.status_code}")
```

--------------------------------

### Use Topic Channel for Accumulation in Pregel with Python

Source: https://docs.langchain.com/oss/python/langgraph/pregel

This snippet demonstrates using a `Topic` channel with `accumulate=True` to collect multiple values written to the same channel by different nodes, showcasing parallel writes and aggregation.

```python
from langgraph.channels import EphemeralValue, Topic
from langgraph.pregel import Pregel, NodeBuilder

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b", "c")
)

node2 = (
    NodeBuilder().subscribe_to("b")
    .do(lambda x: x["b"] + x["b"])
    .write_to("c")
)

app = Pregel(
    nodes={"node1": node1, "node2": node2},
    channels={
        "a": EphemeralValue(str),
        "b": EphemeralValue(str),
        "c": Topic(str, accumulate=True)
    },
    input_channels=["a"],
    output_channels=["c"]
)

app.invoke({"a": "foo"})
```

--------------------------------

### Extend MessagesState with Summary Key (LangGraph)

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Shows how to add a `summary` field to LangGraph's `MessagesState` to store conversation summaries, allowing the graph state to track a running summary.

```python
from langgraph.graph import MessagesState
class State(MessagesState):
    summary: str
```

--------------------------------

### Initialize HuggingFace Chat Model

Source: https://docs.langchain.com/oss/python/langgraph/sql-agent

Demonstrates two methods for initializing a HuggingFace chat model: using the `init_chat_model` utility or by directly instantiating `ChatHuggingFace` with `HuggingFaceEndpoint`.

```python
import os
from langchain.chat_models import init_chat_model

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

model = init_chat_model(
    "microsoft/Phi-3-mini-4k-instruct",
    model_provider="huggingface",
    temperature=0.7,
    max_tokens=1024,
)
```

```python
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint

os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."

llm = HuggingFaceEndpoint(
    repo_id="microsoft/Phi-3-mini-4k-instruct",
    temperature=0.7,
    max_length=1024,
)
model = ChatHuggingFace(llm=llm)
```

--------------------------------

### Extended Example: Streaming from Subgraphs with v2

Source: https://docs.langchain.com/oss/python/langgraph/streaming

Demonstrates building a parent graph that invokes a subgraph and streams all updates, including those from the subgraph, using `subgraphs=True` and `version='v2'`. It shows how to interpret the `ns` field for identifying the source of updates.

```python
from langgraph.graph import START, StateGraph
from typing import TypedDict

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            print(f"Subgraph {chunk['ns']}: {chunk['data']}")
        else:
            print(f"Root: {chunk['data']}")

```

--------------------------------

### Install LangChain HuggingFace Integration

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Install the necessary LangChain HuggingFace integration package to use HuggingFace models.

```shell
pip install -U "langchain[huggingface]"
```

--------------------------------

### Surface Unexpected Errors in LangGraph

Source: https://docs.langchain.com/oss/python/langgraph/thinking-in-langgraph

Demonstrates how to explicitly re-raise unexpected exceptions within a node, allowing them to bubble up for debugging rather than being silently caught and potentially masked.

```python
def send_reply(state: EmailAgentState):
        try:
            email_service.send(state["draft_response"])
        except Exception:
            raise  # Surface unexpected errors
```

--------------------------------

### Define Graph with RemainingSteps for Custom Termination

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Illustrates how to define a LangGraph state with `RemainingSteps` and use it in a conditional edge to control graph termination based on the number of steps remaining, rather than relying solely on `GraphRecursionError`.

```python
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed.is_last_step import RemainingSteps

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    remaining_steps: RemainingSteps

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

# Define edges
def route(state: State) -> Literal["b", END]:
    if state["remaining_steps"] <= 2:
        return END
    else:
        return "b"

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

# Test it out
result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
print(result)
```

--------------------------------

### Output of Single Pregel Node Invocation

Source: https://docs.langchain.com/oss/python/langgraph/pregel

Expected output when invoking the single-node Pregel application with 'foo' as input, showing the doubled value in channel 'b'.

```con
{'b': 'foofoo'}
```

--------------------------------

### Configure InMemorySaver for Parent Graph with Subgraphs (Python)

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Shows how to set up a `StateGraph` with a subgraph and compile the parent graph with an `InMemorySaver` checkpointer. The checkpointer is automatically propagated to the subgraph.

```python
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict

class State(TypedDict):
    foo: str

# Subgraph

def subgraph_node_1(state: State):
    return {"foo": state["foo"] + "bar"}

subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()  # [!code highlight]

# Parent graph

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # [!code highlight]
builder.add_edge(START, "node_1")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)  # [!code highlight]
```

--------------------------------

### Full Example: Subgraph with Different State Schemas (Python)

Source: https://docs.langchain.com/oss/python/langgraph/use-subgraphs

Demonstrates a complete setup for integrating a subgraph with a distinct state schema into a parent graph, including state definition, node functions, and graph compilation for both.

```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START

# Define subgraph
class SubgraphState(TypedDict):
    # note that none of these keys are shared with the parent graph state
    bar: str
    baz: str

def subgraph_node_1(state: SubgraphState):
    return {"baz": "baz"}

def subgraph_node_2(state: SubgraphState):
    return {"bar": state["bar"] + state["baz"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

def node_2(state: ParentState):
    # Transform the state to the subgraph state
    response = subgraph.invoke({"bar": state["foo"]})
    # Transform response back to the parent state
    return {"foo": response["bar"]}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

stream = graph.stream_events({"foo": "foo"}, version="v3")
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])
```

--------------------------------

### Connect and Query SQLite Database (Python)

Source: https://docs.langchain.com/oss/python/langgraph/sql-agent

Connects to the `Chinook.db` using Python's `sqlite3` module, lists available tables, and fetches the first 5 rows from the 'Artist' table.

```python
import sqlite3

con = sqlite3.connect("Chinook.db")
cursor = con.cursor()

cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]

print("Dialect: sqlite")
print(f"Available tables: {tables}")

cursor.execute("SELECT * FROM Artist LIMIT 5;")
print(f"Sample output: {cursor.fetchall()}")
con.close()
```

--------------------------------

### Fault Tolerance Flowchart with Mermaid

Source: https://docs.langchain.com/oss/python/langgraph/fault-tolerance

Illustrates the fault tolerance mechanisms in LangGraph, showing the flow from node execution through retries and error handling.

```mermaid
%%{init:{'theme':'base','themeVariables':{'lineColor':'#40668D','primaryColor':'#E5F4FF','primaryTextColor':'#030710','primaryBorderColor':'#006DDD'}}}%%
flowchart LR
    start([Attempt starts]) --> exec[Run node]
    exec -->|"success"| done([Continue graph])
    exec -->|"any exception<br/>including NodeTimeoutError"| retry{retry_policy<br/>matches?}
    retry -->|"yes, attempts left"| exec
    retry -->|"exhausted or absent"| handler{error_handler?}
    handler -->|"yes"| run_handler["Invoke handler<br/>with NodeError"]
    run_handler --> route([Update state +<br/>Command goto])
    handler -->|"no"| bubble([Exception<br/>bubbles up])

    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef decision fill:#FDF3FF,stroke:#7E65AE,stroke-width:2px,color:#504B5F
    classDef alert fill:#F8E8E6,stroke:#B27D75,stroke-width:2px,color:#634643
    classDef output fill:#EBD0F0,stroke:#885270,stroke-width:2px,color:#441E33

    class exec,run_handler process
    class retry,handler decision
    class bubble alert
    class done,route,start output
```

--------------------------------

### Invoke Parallel Graph and Observe State Accumulation in Python

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

This snippet invokes the previously defined graph with an initial empty aggregate list, demonstrating how the `operator.add` reducer accumulates values from parallel nodes.

```python
graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})
```

--------------------------------

### Split Documents into Chunks for Indexing

Source: https://docs.langchain.com/oss/python/langgraph/agentic-rag

Initializes a `RecursiveCharacterTextSplitter` with a tiktoken encoder to break down the fetched documents into smaller, overlapping chunks suitable for vector store indexing.

```python
from langchain_text_splitters import RecursiveCharacterTextSplitter

docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)
```

--------------------------------

### Initialize Anthropic Chat Model for Tool-Calling

Source: https://docs.langchain.com/oss/python/langgraph/sql-agent

Configure and instantiate an Anthropic chat model using either `init_chat_model` or the `ChatAnthropic` class, setting the API key from environment variables.

```python
import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = init_chat_model("claude-sonnet-4-6")
```

```python
import os
from langchain_anthropic import ChatAnthropic

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

model = ChatAnthropic(model="claude-sonnet-4-6")
```

--------------------------------

### LangGraph State Update Formats with add_messages (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Illustrates the two supported dictionary formats for sending graph inputs or state updates to the `messages` channel when using the `add_messages` reducer function, allowing for direct `HumanMessage` objects or dictionary representations.

```python
# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}
```

--------------------------------

### Initialize LangGraph with Oracle Memory Store

Source: https://docs.langchain.com/oss/python/langgraph/add-memory

Initializes `OracleStore` and `OracleSaver` for a LangGraph agent, configuring the store with embeddings for semantic search capabilities in an Oracle AI Database.

```python
import uuid

from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import OracleSaver
from langgraph_oracledb.store.oracle import OracleStore

model = init_chat_model(model="claude-haiku-4-5-20251001")
embeddings = init_embeddings("openai:text-embedding-3-small")

DB_URI = "user/password@localhost:1521/FREEPDB1"

with (
    OracleStore.from_conn_string(
        DB_URI,
        index={"embed": embeddings, "dims": 1536},
    ) as store,
    OracleSaver.from_conn_string(DB_URI) as checkpointer,
):
    store.setup()
    checkpointer.setup()

    def call_model(
```

--------------------------------

### Subclass MessagesState to Add Custom Fields (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Demonstrates how to extend the prebuilt `MessagesState` by subclassing it and adding additional fields, such as a list of documents, to the graph's state.

```python
from langgraph.graph import MessagesState

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

--------------------------------

### Invoke Agents with Namespace Isolation

Source: https://docs.langchain.com/oss/python/langgraph/use-subgraphs

Demonstrates invoking multiple sub-agents (fruit and veggie experts) that maintain independent state across calls due to proper namespace isolation.

```python
config = {"configurable": {"thread_id": "1"}}

# First call - LLM calls both fruit and veggie experts
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about cherries and broccoli"}]},
    config=config,
)
# Fruit subagent message count: 4
# Veggie subagent message count: 4

# Second call - both agents accumulate independently
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Now tell me about oranges and carrots"}]},
    config=config,
)
# Fruit subagent message count: 8 (remembers cherries!)
# Veggie subagent message count: 8 (remembers broccoli!)
```

--------------------------------

### Restricting Streamed Output with output_keys

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Shows how to restrict the streamed values to a specific set of channels by using the output_keys parameter with stream_mode="values", effectively hiding private channels from the stream output.

```python
for chunk in graph.stream(
    {"user_input": "My"},
    stream_mode="values",
    output_keys=["graph_output"]  # [!code highlight]
):
    print(chunk)
# {'graph_output': 'My name is Lance'}
```

--------------------------------

### Setup imports and OpenAI client

Source: https://docs.langchain.com/oss/python/langgraph/streaming

Required imports for async streaming, type hints, LangGraph, and OpenAI client initialization with model name configuration.

```python
import operator
import json

from typing import TypedDict
from typing_extensions import Annotated
from langgraph.graph import StateGraph, START

from openai import AsyncOpenAI

openai_client = AsyncOpenAI()
model_name = "gpt-5.4-mini"
```

--------------------------------

### Route Streaming Tokens to Graph Nodes

Source: https://docs.langchain.com/oss/python/langgraph/frontend/graph-execution

Use `useMessages` with a `SubgraphDiscoverySnapshot` to extract and display messages scoped to a specific graph node. This allows for granular display of streaming content per node.

```tsx
import { AIMessage } from "langchain";
import { useMessages, type AnyStream, type SubgraphDiscoverySnapshot } from "@langchain/react";

function NodeCard({
  node,
  stream,
}: {
  node: SubgraphDiscoverySnapshot;
  stream: AnyStream;
}) {
  const messages = useMessages(stream, node);
  const lastAIMessage = messages.find(AIMessage.isInstance);
  const streamingContent = lastAIMessage?.text ?? "";

  return <NodeCardBody node={node} content={streamingContent} />;
}
```

--------------------------------

### Configure LLM and System Message at Runtime

Source: https://docs.langchain.com/oss/python/langgraph/use-graph-api

Demonstrates how to extend `ContextSchema` to allow dynamic configuration of both the LLM provider and a system message during graph invocation.

```python
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage
from langgraph.graph import END, MessagesState, StateGraph, START
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

@dataclass
class ContextSchema:
    model_provider: str = "anthropic"
    system_message: str | None = None

MODELS = {
    "anthropic": init_chat_model("claude-haiku-4-5-20251001"),
    "openai": init_chat_model("gpt-5.4-mini"),
}

def call_model(state: MessagesState, runtime: Runtime[ContextSchema]):
    model = MODELS[runtime.context.model_provider]
    messages = state["messages"]
    if (system_message := runtime.context.system_message):
        messages = [SystemMessage(system_message)] + messages
    response = model.invoke(messages)
    return {"messages": [response]}

builder = StateGraph(MessagesState, context_schema=ContextSchema)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

# Usage
input_message = {"role": "user", "content": "hi"}
response = graph.invoke({"messages": [input_message]}, context={"model_provider": "openai", "system_message": "Respond in Italian."})
for message in response["messages"]:
    message.pretty_print()
```

--------------------------------

### Route with Send Objects from Conditional Edges (Python)

Source: https://docs.langchain.com/oss/python/langgraph/graph-api

Return `Send` objects from a routing function in `add_conditional_edges` to dynamically route to nodes and pass specific state to each, useful for map-reduce patterns.

```python
from langgraph.types import Send

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

graph.add_conditional_edges("node_a", continue_to_jokes)
```

--------------------------------

### Define State TypedDict with message accumulation

Source: https://docs.langchain.com/oss/python/langgraph/streaming

State schema using TypedDict with Annotated list that accumulates messages using operator.add for proper state merging in graph execution.

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