>Aerospike 是一个高性能、分布式 NoSQL 数据库，专为大规模实时应用而设计。它提供亚毫秒级延迟、强一致性和跨集群自动数据分布，非常适合需要快速、可靠状态持久化的 AI 工作负载。
Aerospike features a shared-nothing architecture with linear horizontal scaling, a Hybrid Memory Architecture (DRAM indexes with SSD/NVMe data storage), tunable strong or relaxed consistency per namespace, built-in TTL for automatic record expiration, active-active cross-datacenter replication, and multi-model support for key-value, document, graph, and vector data.
Aerospike LangGraph 集成提供了检查点保存器，用于持久化图执行状态，以及存储用于长期代理数据（如用户配置文件、提取的实体和缓存的工具输出）。
开始使用
该实现可在 aerospike-community/aerospike-langgraph 仓库中找到。它包括基于 Docker 的 Aerospike 设置、 AerospikeSaver 和 AerospikeStore的参考实现，以及测试检查点恢复、TTL 行为和命名空间作用域存储访问的基本测试。
要试用，请启动 Aerospike 容器，在 LangGraph 应用中配置保存器和存储，并运行包含的示例以观察执行恢复和代理状态持久化。
安装
安装两个包：
pip install -U langgraph-store-aerospike langgraph-checkpoint-aerospike本地运行 Aerospike
使用 Aerospike Docker 镜像启动 Aerospike：
docker run -d --name aerospike \
-p 3000-3002:3000-3002 \
container.aerospike.com/aerospike/aerospike-server配置
存储和检查点保存器使用相同的 Aerospike 连接设置：
LangGraph 检查点保存器
Aerospike 检查点保存器持久化 LangGraph 执行状态并支持从任何检查点恢复。
from langgraph.checkpoint.aerospike import AerospikeSaver
client = aerospike.client(
{"hosts": [("127.0.0.1", 3000)]}
).connect()
saver = AerospikeSaver(
client=client,
namespace="langgraph",
)
compiled = graph.compile(checkpointer=saver)
compiled.invoke(
{"input": "hello"},
config={
"configurable": {
"thread_id": "demo-thread"
}
}
)LangGraph 存储
Aerospike 存储用于长期代理数据，如用户配置文件、提取的实体和缓存的工具输出。
from langgraph.store.aerospike import AerospikeStore
from langgraph.store.base import PutOp, GetOp, SearchOp
client = aerospike.client(
{"hosts": [("127.0.0.1", 3000)]}
).connect()
store = AerospikeStore(
client=client,
namespace="langgraph",
set="store",
)
# Write data
store.put(
namespace=("users", "profiles"),
key="user_123",
value={"name": "Alice", "age": 30},
)
# Read data
item = store.get(
namespace=("users", "profiles"),
key="user_123",
)
print(item.value)
# Batch operations
results = store.batch([
PutOp(namespace=("documents",), key="doc1", value={"status": "draft"}),
GetOp(namespace=("documents",), key="doc1"),
])
# Search within a namespace
search_results = store.search(
namespace_prefix=("documents",),
filter={"status": {"$eq": "draft"}},
limit=10,
)
# Delete data
store.delete(namespace=("users", "profiles"), key="user_123")