LangChain AWS

When using ValkeyStore, you can enable enhanced vector search by providing a ValkeyIndexConfig. This supports two algorithms:

• HNSW (Hierarchical Navigable Small World): Best for large datasets (>10K vectors). Offers fast approximate search with configurable speed/accuracy trade-offs.
• FLAT (Brute Force): Best for small datasets (<10K vectors). Provides exact search with perfect accuracy but slower performance on large datasets.

HNSW Tuning Parameters:

• hnsw_m: Connections per layer (4-64, default 16). Higher = better recall, more memory.
• hnsw_ef_construction: Construction search width (100-800, default 200). Higher = better index quality, slower construction.
• hnsw_ef_runtime: Query search width (10-500, default 10). Higher = better recall, slower queries.

from langgraph_checkpoint_aws.store.valkey import ValkeyIndexConfig

index_config = {
    "collection_name": "my_documents",
    "dims": 1536,
    "embed": embeddings,
    "fields": ["text", "content"],
    "timezone": "UTC",
    "index_type": "hnsw",  # "hnsw" or "flat"
    "hnsw_m": 16,
    "hnsw_ef_construction": 200,
    "hnsw_ef_runtime": 10,
}

You can enable semantic search by providing an index configuration in the DynamoDBStore constructor. This uses a DynamoDB vector index to perform approximate nearest neighbor search without a separate vector database.

Note: The vector index is eventually consistent; items written immediately before a search might not appear in results. If no index is provided, search() performs standard namespace listing and filtering.

from langgraph_checkpoint_aws import DynamoDBStore
from langchain_aws import BedrockEmbeddings

store = DynamoDBStore(
    table_name="agent-ltm",
    index={
        "embed": BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0"),
        "dims": 256,  # must match the embedding model's output
        "fields": ["text"],  # which value fields to embed
    },
)
store.setup()

# Embedding is computed automatically on put()
store.put(("user", "alice"), "mem1", {"text": "Alice prefers dark mode"})

# Semantic search returns results ranked by relevance
results = store.search(("user", "alice"), query="UI preferences")
for item in results:
    print(item.score, item.value)

All components in langgraph-checkpoint-aws support asynchronous operations. You can use AsyncBedrockSessionSaver, AsyncValkeySaver, and DynamoDBSaver (via ainvoke) for non-blocking workflows.

Common async patterns include:

• Creating Bedrock sessions using AsyncBedrockSessionSaver.
• Running LangGraph with DynamoDBSaver using ainvoke.
• Using AsyncValkeySaver within an async with context manager.
• Using AsyncValkeyStore for key-value operations with aput and aget.

from langgraph_checkpoint_aws.async_saver import AsyncBedrockSessionSaver
from langgraph_checkpoint_aws import AsyncValkeySaver, DynamoDBSaver, AsyncValkeyStore

# Async Bedrock usage
session_saver = AsyncBedrockSessionSaver(region_name="us-west-2")
session_id = (await session_saver.session_client.create_session()).session_id

# Async DynamoDB usage
checkpointer = DynamoDBSaver(table_name="my-checkpoints", region_name="us-west-2")
result = await graph.ainvoke(1, {"configurable": {"thread_id": "session-1"}})

# Async ValkeySaver usage
async with AsyncValkeySaver.from_conn_string("valkey://localhost:6379") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    result = await graph.ainvoke(1, {"configurable": {"thread_id": "session-1"}})

# Async ValkeyStore usage
async with AsyncValkeyStore.from_conn_string("valkey://localhost:6379") as store:
    namespace = ("example",)
    key = "key"
    data = {"message": "Sample message"}
    await store.setup()
    await store.aput(namespace, key, data)
    result = await store.aget(namespace, key)

For asynchronous workflows, use AsyncBedrockSessionSaver. This requires awaiting the session creation and using ainvoke on the compiled graph.

from langgraph.graph import StateGraph
from langgraph_checkpoint_aws.async_saver import AsyncBedrockSessionSaver

# Initialize the saver
session_saver = AsyncBedrockSessionSaver(
    region_name="us-west-2",  # Your AWS region
    credentials_profile_name="default",  # Optional: AWS credentials profile
)

# Create a session
session_create_response = await session_saver.session_client.create_session()
session_id = session_create_response.session_id

# Use with LangGraph
builder = StateGraph(int)
builder.add_node("add_one", lambda x: x + 1)
builder.set_entry_point("add_one")
builder.set_finish_point("add_one")

graph = builder.compile(checkpointer=session_saver)
config = {"configurable": {"thread_id": session_id}}
await graph.ainvoke(1, config)