Skip to content

如何将语义搜索添加到您的 LangGraph 部署中

本指南解释了如何将语义搜索添加到您的 LangGraph 部署的跨线程 存储 中，以便您的代理可以通过语义相似性搜索记忆和其他文档。

准备工作

  • 一个 LangGraph 部署（参见如何部署）
  • 嵌入提供商的 API 密钥（本例中为 OpenAI）
  • langchain >= 0.3.8（如果你使用下面的字符串格式指定）

步骤

  1. 更新您的 langgraph.json 配置文件，以包含存储配置：
{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

此配置：

  • 使用 OpenAI 的 text-embeddings-3-small 模型生成嵌入
  • 将嵌入维度设置为 1536（与模型输出匹配）

  • 索引存储数据中的所有字段（["$"] 表示索引所有内容，或指定特定字段如 ["text", "metadata.title"]）

  • 要使用上述字符串嵌入格式，请确保您的依赖项包含 langchain >= 0.3.8：

# 在 pyproject.toml 中
[project]
dependencies = [
    "langchain>=0.3.8"
]

或者，如果使用 requirements.txt：

langchain>=0.3.8

使用方法

配置完成后，您可以在LangGraph节点中使用语义搜索。存储需要一个命名空间元组来组织记忆：

def search_memory(state: State, *, store: BaseStore):
    # 使用语义相似性搜索存储
    # 命名空间元组有助于组织不同类型的记忆
    # 例如，("user_facts", "preferences") 或 ("conversation", "summaries")
    results = store.search(
        namespace=("memory", "facts"),  # 按类型组织记忆
        query="your search query",
        limit=3  # 返回的结果数量
    )
    return results

在上述代码中，namespace参数用于指定记忆的类型，query参数用于输入查询内容，limit参数用于限制返回的结果数量。

自定义嵌入

如果你希望使用自定义嵌入，可以传递一个指向自定义嵌入函数的路径：

{
    ...
    "store": {
        "index": {
            "embed": "path/to/embedding_function.py:embed",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

部署时会在指定路径中查找该函数。该函数必须是异步的，并接受一个字符串列表：

# path/to/embedding_function.py
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def aembed_texts(texts: list[str]) -> list[list[float]]:
    """自定义嵌入函数，必须满足以下条件：
    1. 是异步的
    2. 接受一个字符串列表
    3. 返回一个浮点数组列表（嵌入）
    """
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [e.embedding for e in response.data]

通过API查询

您也可以使用LangGraph SDK查询存储。由于SDK使用异步操作：

from langgraph_sdk import get_client

async def search_store():
    client = get_client()
    results = await client.store.search_items(
        ("memory", "facts"),
        query="your search query",
        limit=3  # 返回的结果数量
    )
    return results

# 在异步上下文中使用
results = await search_store()

Comments