向量数据库

使用

使用 InMemoryStore 快速创建一个向量数据库

from langgraph.store.memory import InMemoryStore
from langchain_huggingface import HuggingFaceEmbeddings

store = InMemoryStore(
    index={
        "embed": HuggingFaceEmbeddings(model_name="thenlper/gte-large-zh"),
        # 使用向量数据库的维度
        "dims": 1024,
    }
)

store 可以跨多个会话使用

只需要在编译图的时候，传递 store 对象

graph = builder.compile(checkpointer=MemorySaver(), store=store)

在节点中，通过 store 获取

from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore

def node(state: State, config: RunnableConfig, *, store: BaseStore):
    print(store)

当添加数据时，需要定义两个属性：

对象的命名空间，一个元组（类似于目录）
对象键（类似于文件民）

user_id = "123"
namespace = ("memories", user_id)
store.put(namespace, "1", { "text": "我的名字叫kingmusi" })
store.put(namespace, "2", { "text": "我喜欢打篮球、音乐、跳舞" })

在上面例子中，使用 ("memories", <user_id>) 作为命名空间，使用自增id（1）作为每个新记忆的键

如果put 的命名空间和键完全一样，则后添加的会覆盖前面的

通过 search，实现自然语言搜索存储的向量

store.search(namespace, query="我的名字是什么", limit=1)

高级用法

1、定向嵌入范围

把嵌入数据和存储数据分开，以提高召回率或省略部分嵌入数据

store = InMemoryStore(
    index={
        "embed": HuggingFaceEmbeddings(model_name="thenlper/gte-large-zh"),
        # 使用向量数据库的维度
        "dims": 1024,
        # 定义要进行搜索的字段
        "fields": ["text"]
    }
)

namespace = ("memories", "user_123")
store.put(namespace, "1", { "text": "我的名字叫kingmusi", "no_indexed": "我爱吃三文鱼" })
store.put(namespace, "2", { "text": "我喜欢打篮球、音乐、跳舞", "no_indexed": "我不爱吃胡萝卜" })
store.put(namespace, "3", { "text": "我爱吃拉面", "no_indexed": "我是个程序员" })

res = store.search(namespace, query="我的食物爱好是什么", limit=1)
print(res)
# [Item(namespace=['memories', 'user_123'], key='3', value={'text': '我爱吃拉面', 'no_indexed': '我是个程序员'}, created_at='2025-05-04T02:17:13.173811+00:00', updated_at='2025-05-04T02:17:13.173811+00:00', score=0.5673106250040881)]

2、存储时指定嵌入字段

store = InMemoryStore(
    index={
        "embed": HuggingFaceEmbeddings(model_name="thenlper/gte-large-zh"),
        # 使用向量数据库的维度
        "dims": 1024,
        # 定义要进行搜索的字段
        "fields": ["text"]
    }
)

namespace = ("memories", "user_123")
store.put(namespace, "1", { "text": "我的名字叫kingmusi", "no_indexed": "我爱吃三文鱼" })
store.put(namespace, "2", { "text": "我喜欢打篮球、音乐、跳舞", "no_indexed": "我不爱吃胡萝卜" })

# 指定此次存储的嵌入字段
store.put(namespace, "3", { "text": "我爱吃拉面", "no_indexed": "我是个程序员" }, index=["no_indexed"])

res = store.search(namespace, query="我的食物爱好是什么", limit=1)
print(res)
# [Item(namespace=['memories', 'user_123'], key='3', value={'text': '我爱吃拉面', 'no_indexed': '我是个程序员'}, created_at='2025-05-04T02:17:13.173811+00:00', updated_at='2025-05-04T02:17:13.173811+00:00', score=0.5673106250040881)]

3、存储时禁止嵌入数据

store = InMemoryStore(
    index={
        "embed": HuggingFaceEmbeddings(model_name="thenlper/gte-large-zh"),
        # 使用向量数据库的维度
        "dims": 1024,
        # 定义要进行搜索的字段
        "fields": ["text"]
    }
)

namespace = ("memories", "user_123")
store.put(namespace, "1", { "text": "我的名字叫kingmusi", "no_indexed": "我爱吃三文鱼" })
store.put(namespace, "2", { "text": "我喜欢打篮球、音乐、跳舞", "no_indexed": "我不爱吃胡萝卜" })

# 指定不进行嵌入
store.put(namespace, "3", { "text": "我爱吃拉面", "no_indexed": "我是个程序员" }, index=False)

res = store.search(namespace, query="我的食物爱好是什么", limit=1)
print(res)
# [Item(namespace=['memories', 'user_123'], key='2', value={'text': '我喜欢打篮球、音乐、跳舞', 'no_indexed': '我不 爱吃胡萝卜'}, created_at='2025-05-04T02:21:59.696452+00:00', updated_at='2025-05-04T02:21:59.696452+00:00', score=0.4840481757123507)]