LangGraph Python Integration

Use langgraph-oracledb to add Oracle-backed checkpoints, durable stores, and vector-searchable memory to LangGraph Python applications.

Contents

Install the Package

Install langgraph-oracledb with LangGraph and any model or embedding packages that your application uses.

pip install -qU langgraph langgraph-oracledb

For vector search with an OpenAI embedding model, install the OpenAI integration package.

pip install -qU langchain-openai

For development from a local clone, install the package in editable mode.

pip install -e /path/to/langgraph-oracledb

Connect to Oracle AI Database

The package uses the python-oracledb driver. Use the connection string format user/password@dsn with the from_conn_string() helpers.

conn_string = "myuser/mypassword@localhost:1521/FREEPDB1"

You can also create an oracledb.Connection or oracledb.ConnectionPool directly and pass it to OracleSaver or OracleStore.

import oracledb

pool = oracledb.create_pool(
    user="myuser",
    password="mypassword",
    dsn="localhost:1521/FREEPDB1",
    min=1,
    max=5,
)

Use Oracle AI Database with AI Vector Search enabled when you configure vector indexes.

Use Oracle Checkpointing

OracleSaver implements LangGraph checkpoint persistence. Pass it to compile(checkpointer=...) so LangGraph writes graph state to Oracle AI Database.

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph_oracledb.checkpoint.oracle import OracleSaver


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


def reply(state: State) -> dict[str, list[AIMessage]]:
    return {"messages": [AIMessage(content="Stored this step in Oracle.")]}


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

conn_string = "myuser/mypassword@localhost:1521/FREEPDB1"

with OracleSaver.from_conn_string(conn_string) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "demo-thread"}}
    graph.invoke(
        {"messages": [HumanMessage(content="Start the thread.")]},
        config,
    )

Call setup() before first use. The method is idempotent, so you can call it during application startup.

Resume a Graph Thread

LangGraph identifies checkpoint state with the thread_id in the runnable configuration. Invoke the graph again with the same thread_id to resume the thread from Oracle AI Database.

config = {"configurable": {"thread_id": "demo-thread"}}

graph.invoke(
    {"messages": [HumanMessage(content="Continue the same thread.")]},
    config,
)

Use a different thread_id when you want an independent checkpoint history.

Use the Oracle Store

OracleStore implements the LangGraph BaseStore interface. Use it for durable JSON values that are keyed by namespace and key.

from langgraph_oracledb.store.oracle import OracleStore

conn_string = "myuser/mypassword@localhost:1521/FREEPDB1"

with OracleStore.from_conn_string(conn_string) as store:
    store.setup()

    namespace = ("users", "123")
    store.put(namespace, "preferences", {"theme": "dark", "locale": "en-US"})

    item = store.get(namespace, "preferences")
    print(item.value)

    results = store.search(("users",), filter={"theme": "dark"}, limit=10)

The store supports put, get, delete, search, list_namespaces, and batch operations. The asynchronous store provides matching aput, aget, adelete, asearch, alist_namespaces, and abatch methods.

Use Vector Search in the Store

Enable semantic search by passing an index configuration when you create the store. The embed value must implement the LangChain Embeddings interface.

from langchain_openai import OpenAIEmbeddings
from langgraph_oracledb.store.oracle import OracleStore

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

index_config = {
    "dims": 1536,
    "embed": embeddings,
    "fields": ["text"],
    "index_type": {
        "type": "hnsw",
        "neighbors": 16,
        "efconstruction": 200,
        "distance_metric": "COSINE",
    },
}

with OracleStore.from_conn_string(conn_string, index=index_config) as store:
    store.setup()

    namespace = ("research", "papers")
    store.put(
        namespace,
        "paper-1",
        {
            "text": "The paper studies long-term memory for LLM agents.",
            "title": "Example Paper",
            "year": 2026,
        },
    )

    hits = store.search(namespace, query="agent memory", limit=3)
    for hit in hits:
        print(hit.value["title"])

Use fields to control which JSON fields are embedded. Fields that are not embedded remain available as stored metadata.

Note: Vector search requires Oracle AI Database with AI Vector Search enabled.

Use Typed Index Configurations

The package exports OracleIndexConfig, HNSWConfig, and IVFConfig typed dictionaries. Use them in place of plain dictionaries so IDE autocomplete, type checkers, and refactoring tools can validate index parameters at the call site.

from langchain_openai import OpenAIEmbeddings
from langgraph_oracledb.store.oracle import (
    HNSWConfig,
    IVFConfig,
    OracleIndexConfig,
    OracleStore,
)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# HNSW: fast queries, higher memory, best for read-heavy workloads.
hnsw_index: OracleIndexConfig = {
    "dims": 1536,
    "embed": embeddings,
    "fields": ["text"],
    "index_type": HNSWConfig(
        type="hnsw",
        neighbors=16,
        efconstruction=200,
        distance_metric="COSINE",
    ),
}

# IVF: smaller memory footprint, configurable partitioning, useful for very large stores.
ivf_index: OracleIndexConfig = {
    "dims": 1536,
    "embed": embeddings,
    "fields": ["text"],
    "index_type": IVFConfig(
        type="ivf",
        neighbor_partitions=128,
        samples_per_partition=1024,
        distance_metric="COSINE",
    ),
}

with OracleStore.from_conn_string(conn_string, index=hnsw_index) as store:
    store.setup()

The plain-dictionary form shown earlier remains supported. The typed form is preferred for new code because misspelled keys, missing required fields, and incompatible parameter combinations surface at type-check time rather than at setup() time.

Use Asynchronous Persistence

Use AsyncOracleSaver and AsyncOracleStore in asynchronous applications.

import asyncio

from langgraph_oracledb.checkpoint.oracle import AsyncOracleSaver
from langgraph_oracledb.store.oracle import AsyncOracleStore


async def main() -> None:
    conn_string = "myuser/mypassword@localhost:1521/FREEPDB1"

    async with AsyncOracleSaver.from_conn_string(conn_string) as checkpointer:
        await checkpointer.setup()

    async with AsyncOracleStore.from_conn_string(conn_string) as store:
        await store.setup()
        await store.aput(("demo",), "item-1", {"text": "hello"})
        item = await store.aget(("demo",), "item-1")
        print(item.value)


asyncio.run(main())

Pass AsyncOracleSaver to graphs that you invoke with ainvoke(). Pass AsyncOracleStore to agents or tools that use asynchronous store methods.

Use Connection Pools

For production applications, create a pool or use pool_config with the connection-string helpers.

from langgraph_oracledb.checkpoint.oracle import OracleSaver
from langgraph_oracledb.store.oracle import OracleStore

pool_config = {"min_size": 2, "max_size": 10}

with OracleSaver.from_conn_string(
    conn_string,
    pool_config=pool_config,
) as checkpointer:
    checkpointer.setup()

with OracleStore.from_conn_string(
    conn_string,
    pool_config=pool_config,
) as store:
    store.setup()

You can also create one oracledb.ConnectionPool and pass it to both classes.

checkpointer = OracleSaver(pool)
store = OracleStore(pool, index=index_config)

checkpointer.setup()
store.setup()

Sharing one pool keeps checkpoint state and store data in the same database access path.

Use Stores with LangGraph Agents

Pass the store to a LangGraph or LangChain agent so tools can read and write long-term memory.

import uuid
from typing_extensions import Annotated

from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedStore
from langgraph.store.base import BaseStore

MEMORY_NAMESPACE = ("research", "notes")


@tool
def save_note(
    text: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Save a note for later semantic search."""
    store.put(MEMORY_NAMESPACE, uuid.uuid4().hex, {"text": text})
    return "Saved note."


@tool
def search_notes(
    query: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Search prior notes."""
    hits = store.search(MEMORY_NAMESPACE, query=query, limit=5)
    return "\n".join(hit.value["text"] for hit in hits) or "No notes found."


agent = create_agent(
    model,
    tools=[save_note, search_notes],
    checkpointer=checkpointer,
    store=store,
)

The checkpointer stores short-term state for one thread_id. The store keeps long-term memory that tools can search across threads.

Coordinate Multiple Agents with a Supervisor

For workloads where one agent is not enough — for example, a planner-assistant that needs both a domain analyst and a policy specialist — wrap several agents inside a supervisor with the langgraph-supervisor package. The supervisor decides which specialist to delegate to, collects their summaries, and writes the final answer. The checkpointer stores the supervisor’s per-thread state; the store keeps memory that the specialists can read across threads.

Install the additional package.

pip install -qU langgraph-supervisor

Compile two specialists, share one AsyncOracleSaver checkpointer and one AsyncOracleStore, then wrap everything in a supervisor.

import asyncio
from typing_extensions import Annotated

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import InjectedStore
from langgraph.store.base import BaseStore
from langgraph_oracledb.checkpoint.oracle import AsyncOracleSaver
from langgraph_oracledb.store.oracle import AsyncOracleStore
from langgraph_supervisor import create_supervisor


@tool
async def search_demand_reports(
    query: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Search vector knowledge for relevant demand reports."""
    hits = await store.asearch(("knowledge", "demand"), query=query, limit=5)
    return "\n".join(hit.value["text"] for hit in hits) or "No matches."


@tool
async def get_user_memory(
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """Read the planner's saved preferences."""
    item = await store.aget(("users", user_id, "memories"), "pref")
    return item.value["note"] if item else "No saved preferences."


async def main() -> None:
    conn_string = "myuser/mypassword@localhost:1521/FREEPDB1"
    model = ChatOpenAI(model="gpt-4o-mini")

    async with (
        AsyncOracleSaver.from_conn_string(conn_string) as checkpointer,
        AsyncOracleStore.from_conn_string(conn_string, index=index_config) as store,
    ):
        await checkpointer.setup()
        await store.setup()

        demand_analyst = create_agent(
            model,
            tools=[search_demand_reports],
            name="demand_analyst",
            prompt="You are the demand analyst. Use vector search to ground claims.",
        )
        policy_agent = create_agent(
            model,
            tools=[get_user_memory],
            name="policy_agent",
            prompt="You are the policy agent. Always consult the user's saved memory.",
        )

        supervisor = create_supervisor(
            agents=[demand_analyst, policy_agent],
            model=model,
            prompt=(
                "You are the supervisor. Delegate questions about demand to "
                "demand_analyst and questions about policy or preferences to "
                "policy_agent. Synthesise their findings into one answer."
            ),
        ).compile(checkpointer=checkpointer, store=store)

        result = await supervisor.ainvoke(
            {"messages": [("user", "Should we stock more cleats for Q3?")]},
            config={"configurable": {"thread_id": "planner-2026Q3"}},
        )
        print(result["messages"][-1].content)


asyncio.run(main())

The supervisor and both specialists share one Oracle AI Database. Vector retrieval, planner preferences, and per-thread agent state all participate in the same connection pool and the same transaction surface. For a full reference implementation with a chat UI, see the Supply-Chain Demand Planning Agent workshop in the Oracle AI Developer Hub.

Manage Tables and Migrations

The setup() methods create and upgrade the required Oracle tables.

Persistence object Tables
OracleSaver and AsyncOracleSaver checkpoints, checkpoint_blobs, checkpoint_writes, checkpoint_migrations
OracleStore and AsyncOracleStore store_<suffix>, store_migrations_<suffix>, store_configs
Store with vector search store_vectors_<suffix>, vector_migrations_<suffix>, and an Oracle vector index

When you do not set table_suffix, the store generates one. Non-vector stores use novec. Vector stores derive the suffix from the embedding and index configuration.

Set table_suffix only when you need predictable table names, such as in tests or isolated deployments.

Troubleshoot Common Issues

Review these common issues before debugging application code.

Symptom Cause Resolution
setup() fails during vector index creation Oracle AI Vector Search is not available or enabled. Use Oracle AI Database with AI Vector Search enabled, or create the store without an index configuration.
A graph does not resume prior messages The invocation uses a new or missing thread_id. Pass the same {"configurable": {"thread_id": "..."}} value on each invocation.
Semantic store search returns no matches The store was created without index, or the written values were not indexed. Create the store with an index configuration and write values with index enabled.
A vector store rejects a reused suffix The stored configuration has different dimensions, distance metric, or index parameters. Use a different table_suffix or keep the same vector configuration.
Async methods fail in a synchronous context The async store or saver requires a running event loop. Use OracleSaver or OracleStore in synchronous code, or call async methods inside an event loop.