Skip to main content
Aerospike 是一款高性能的分布式 NoSQL 数据库，专为大规模实时应用而设计。它提供亚毫秒级延迟、强一致性以及跨集群的自动数据分布，非常适合需要快速、可靠状态持久化的 AI 工作负载。
Aerospike 采用无共享架构，支持线性水平扩展，具备混合内存架构（DRAM 索引配合 SSD/NVMe 数据存储）、可按命名空间调节的强一致性或宽松一致性、内置 TTL 以自动过期记录、主动-主动跨数据中心复制，以及支持键值、文档、图和向量数据的多种模型。 Aerospike LangGraph 集成提供了一个检查点保存器（checkpointer）用于持久化图执行状态，以及一个存储（store）用于长期代理数据，例如用户画像、提取的实体和缓存的工具输出。

​
入门指南

该实现位于 aerospike-community/aerospike-langgraph 仓库中。它包含基于 Docker 的 Aerospike 环境配置、AerospikeSaver 和 AerospikeStore 的参考实现，以及用于测试检查点恢复、TTL 行为和命名空间范围存储访问的基础测试。 要试用，请启动 Aerospike 容器，在 LangGraph 应用中配置 saver 和 store，然后运行包含的示例以观察执行恢复和代理状态持久化。

​
安装

安装以下两个包： 
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

​
配置

Store 和 checkpointer 均使用相同的 Aerospike 连接设置： 
export AEROSPIKE_HOST=127.0.0.1
export AEROSPIKE_PORT=3000

​
LangGraph 检查点保存器

Aerospike checkpointer 用于持久化 LangGraph 执行状态，并支持从任意检查点恢复。 
import aerospike
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 store 用于长期代理数据，例如用户画像、提取的实体和缓存的工具输出。 
import 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",
)

# 写入数据
store.put(
    namespace=("users", "profiles"),
    key="user_123",
    value={"name": "Alice", "age": 30},
)

# 读取数据
item = store.get(
    namespace=("users", "profiles"),
    key="user_123",
)
print(item.value)

# 批量操作
results = store.batch([
    PutOp(namespace=("documents",), key="doc1", value={"status": "draft"}),
    GetOp(namespace=("documents",), key="doc1"),
])

# 在命名空间中搜索
search_results = store.search(
    namespace_prefix=("documents",),
    filter={"status": {"$eq": "draft"}},
    limit=10,
)

# 删除数据
store.delete(namespace=("users", "profiles"), key="user_123")

​
资源

在 GitHub 上编辑此页面 或 提交问题。
通过 MCP 将这些文档连接到 Claude、VSCode 等工具，以获取实时答案。