基于Qdrant的LangGraph语义智能体记忆与跨线程检索机制
利用 Qdrant 为 LangGraph 智能体提供长期记忆功能：嵌入用户信息，跨线程 ID 进行检索，并通过负载过滤器实现多租户隔离。
来源: **Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.**
在构建任何东西之前，先明确为什么智能体需要语义记忆，以及它与情景记忆有何不同，这是很有帮助的。
主要有三个原因：
- 跨会话持续性。标准的 LangGraph 检查点工具会将状态与某个
thread_id相关联。当会话结束或开始新对话时，上下文会重置为零。 - 亚线性检索能力。半年的原始对话记录无法全部放入上下文窗口中，这会导致延迟急剧上升，准确性因“中间丢失”效应而下降，同时还会超出令牌限制。
- 意图匹配优于精确关键词匹配。当用户重新表述某个概念时，基于关键词的查询会失效，而向量相似度则能弥补这种概念上的差异。
Qdrant 可以通过 Docker Desktop 在本地部署，也可以作为托管服务在 Qdrant Cloud 上运行，或是部署在 Kubernetes 环境中用于生产环境。下文将介绍本地部署方式。
环境配置
首先创建一个 Python 虚拟环境。依赖项会放在 requirements.txt 文件中：
fastapi>=0.110.0
uvicorn>=0.28.0
pydantic>=2.6.0
python-dotenv>=1.0.1
langchain-core>=0.1.30
langchain-openai>=0.1.0
langgraph>=0.0.30
qdrant-client>=1.10.0
在 Ubuntu 或 Debian 上创建环境的标准方法是使用 venv 模块：
#Install venv module (Ubuntu/Debian)
$sudo apt update && sudo apt install python3-venv
#Create a virtual environment
$python3 -m venv myenv
#Activate the environment
$source myenv/bin/activate
#Install saved dependencies
$pip install -r requirements.txt
#Save dependencies
#pip freeze > requirements.txt
#Deactivate when finished:
$deactivate
#Delete the environment:
$rm -rf myenv
使用 Astral 的快速包安装工具及项目管理工具 uv 完成相同任务的效果如下：
#Install uv
$curl -LsSf [https://astral.sh/uv/install.sh](https://astral.sh/uv/install.sh) | sh
#Create a virtual environment
$uv venv myenv
#Activate the environment
$source myenv/bin/activate
#Install saved dependencies:
$uv pip install -r requirements.txt
#Save dependencies:
$uv pip freeze > requirements.txt
#Sync dependencies from lockfile
$uv sync
#Add a new package and auto-update file
$uv add <package_name>
使用 Docker Compose 运行 Qdrant
为了进行本地测试，docker-compose.yaml 文件用于启动数据库：
version: '3.8'
services:
qdrant:
image: qdrant/qdrant:latest #image name download from docker.io
container_name: sbi_semantic_memory #container name
ports:
- "6333:6333" # REST HTTP API & Web Dashboard
- "6334:6334" # High-speed gRPC API
volumes:
- ./qdrant_storage:/qdrant/storage #Binds local directory to container for data persistence across restarts
networks:
- agent-network #Attaches container to isolated network for communication
healthcheck: #Executes internal HTTP ping against health check endpoint
test: ["CMD", "curl", "-f", "http://localhost:6333/healthz"]
interval: 10s #Run health check every 10 seconds
timeout: 5s #Fail if check takes longer than 5 seconds
retries: 5 #Startup grace period before recording failures
networks: #Private bridged network for secure container-to-container communication
agent-network:
driver: bridge #Standard single-host bridge driver
Docker Compose允许在单个文件中描述多容器应用程序。无需输入包含大量参数的冗长docker run命令，容器、网络、环境变量及卷等所有配置都可在docker-compose.yml中声明。此处即使用了官方的Qdrant镜像：
image: qdrant/qdrant:latest
将其拿起来检查容器：
$docker compose up -d
$ docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
8922a12f1029 qdrant/qdrant:latest "./entrypoint.sh" 45 hours ago Up 45 hours (unhealthy) 0.0.0.0:6333-6334->6333-6334/tcp, [::]:6333-6334->6333-6334/tcp sbi_semantic_memory
$ curl http://localhost:6333/healthz
healthz check passed
#View recent logs
$docker logs sbi_semantic_memory
#Tail live logs in real time
$docker logs -f sbi_semantic_memory
#View the last 50 lines of logs
$docker logs --tail 50 sbi_semantic_memory
#Inspect health check failure reasons
$docker inspect --format='{{json .State.Health}}' sbi_semantic_memory
#Check container resource consumption (CPU/RAM):
$docker stats sbi_semantic_memory
#Inspect container runtime details and exit codes:
$docker inspect sbi_semantic_memory
#Execute an interactive shell inside the container
$docker exec -it sbi_semantic_memory sh
#Restart the container:
$docker restart sbi_semantic_memory
#Stop, remove, and recreate the container:
$docker compose down && docker compose up -d
#Force-kill a stuck container:
$docker kill sbi_semantic_memory
向量存储
在容器运行且健康检查通过后，下一个文件是vector_store.py。该文件中的Python类会建立与本地或远程Qdrant实例的客户端连接，并确保存在名为agent_memories的集合，且该集合具有正确的向量几何结构。向量的维度为1536，这是OpenAI的text-embedding-3-small和text-embedding-ada-002模型所使用的标准维度，而距离则是通过余弦相似度来计算的。
import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.http import models
# Load environment variables
load_dotenv()
class SemanticMemory:
def __init__(self):
host = os.getenv("QDRANT_HOST", "localhost")
port = int(os.getenv("QDRANT_PORT", 6333))
self.client = QdrantClient(host=host, port=port)
self.collection_name = "agent_memories"
self._ensure_collection()
def _ensure_collection(self):
collections = self.client.get_collections().collections
exists = any(c.name == self.collection_name for c in collections)
if not exists:
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=models.VectorParams(
size=1536,
distance=models.Distance.COSINE
),
)
memory_vault = SemanticMemory()
余弦相似度是通过计算两个向量之间夹角的余弦值来得出的。由于它会除以向量的长度，因此长度差异会被忽略，只有方向才起作用。这衡量的是概念上的匹配程度，而非单词数量或句子长度。该图展示了不同相似度指标的对比以及它们各自最适用的情况：
.env 文件用于存储 OpenAI 密钥以及 Qdrant 的连接设置：
# OpenAI API Key
OPENAI_API_KEY=
# Qdrant Database Configuration
QDRANT_HOST=localhost
QDRANT_PORT=6333
QDRANT_COLLECTION_NAME=agent_memories
# FastAPI Server Setup
HOST=0.0.0.0
PORT=8000
图表
graph.py 定义了三个节点并构建了工作流程。
recall_node 负责检索记忆并将其注入系统。它使用 text-embedding-3-small 将用户的最新查询转换为1536维向量，从 agent_memories 集合中查找最相关的三条事实，然后将包含这些事实的系统指令插入到 state["messages"] 的索引0位置。
agent_node 负责推理并给出答案。它会带着更新后的历史记录和注入的记忆异步调用 gpt-4o，随后将模型的回复添加到 state["messages"] 中。
memorize_node功能会过滤并保存数据。它会遍历state[“messages”]，仅找出最新的原始用户语句，跳过系统指令和AI回复。随后，它会为该输入创建一个向量，并使用唯一的UUID、文本、thread_id以及元数据将一个新的PointStruct插入到Qdrant中。
用于测试的API
main.py提供了一个FastAPI接口，可供测试使用：
from typing import Dict, Any, List
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
# Ensure environment variables are loaded at application start
load_dotenv(override=True)
from graph import app_graph
app = FastAPI(title="Qdrant Semantic Memory Service")
class ChatRequest(BaseModel):
message: str
thread_id: str
metadata: Dict[str, Any] = {}
class ChatResponse(BaseModel):
thread_id: str
messages: List[Dict[str, Any]]
@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(req: ChatRequest):
try:
initial_state = {
"messages": [{"role": "user", "content": req.message}],
"thread_id": req.thread_id,
"metadata": req.metadata
}
final_state = await app_graph.ainvoke(initial_state)
return ChatResponse(
thread_id=req.thread_id,
messages=final_state["messages"]
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
使用 uvicorn main:app --reload 启动服务器，然后在端口 8000 的 /docs 地址打开交互式文档。首先是线程 001 中发送的陈述事实的请求：
{
"message": "My favorite color is Obsidian Blue.",
"thread_id": "thread-001",
"metadata": {"source": "mobile_app"}
}
回复内容：
{
"thread_id": "thread-001",
"messages": [
{
"role": "system",
"content": "You have access to the following long-term memory facts about the user:\n- My favorite color is Obsidian Blue.\n- My favorite color is Obsidian Blue.\n- Obsidian Blue is a deep, rich shade of blue that often resembles the color of the volcanic glass obsidian. It's a striking and elegant color choice! If you have any questions or need assistance related to colors or anything else, feel free to ask!\n\nUse the facts above to directly answer the user's question."
},
{
"role": "user",
"content": "My favorite color is Obsidian Blue."
},
{
"role": "assistant",
"content": "That's a great choice! Obsidian Blue is a deep, rich shade of blue that resembles the color of volcanic glass. It's both striking and elegant. If you have any questions or need assistance related to colors or anything else, feel free to ask!"
}
]
}
然后在另一个帖子 thread-002 中有一个问题：
{
"message": "What is my favorite color?",
"thread_id": "thread-002",
"metadata": {"source": "web_dashboard" }
}
答案与之前帖子中提到的颜色一致：
{
"thread_id": "thread-002",
"messages": [
{
"role": "system",
"content": "You have access to the following long-term memory facts about the user:\n- What is my favorite color?\n- My favorite color is Obsidian Blue.\n- My favorite color is Obsidian Blue.\n\nUse the facts above to directly answer the user's question."
},
{
"role": "user",
"content": "What is my favorite color?"
},
{
"role": "assistant",
"content": "Your favorite color is Obsidian Blue."
}
]
}
为何召回会涉及线程
所有嵌入数据都存储在同一个集合 agent_memories 中。每条记录在其载荷中都会存储一个 thread_id，但在 recall_node 中进行搜索时并不会依据该值进行过滤：
# 1. Generate query vector from user's message
query_vector = embeddings.embed_query(user_query)
# 2. Search Qdrant WITHOUT a payload filter
relevant_docs = memory_vault.client.query_points(
collection_name="agent_memories",
query=query_vector,
limit=2 # Grabs top 2 matches globally across all threads
).points
由于 query_points 只接收集合名称和查询向量，Qdrant 会将该查询与数据库中的每一个点进行比对，无论这些点是由哪个线程创建的。thread-002 生成的查询会被转换为一个 1,536 维的向量，Qdrant 会计算该向量与所有存储的向量（包括 thread-001 生成的向量）之间的余弦角度。由于“我喜欢的颜色是什么？”与“我喜欢的颜色是黑曜石蓝”之间的相似度超过了匹配阈值，因此 Qdrant 会返回对应的点，而 recall_node 会在 GPT-4o 运行之前将相关文本插入到新的提示词中。
若要限制智能体仅使用当前线程，可在 thread_id 有效载荷字段上添加 Filter：
from qdrant_client.http import models
# Returns matches ONLY if thread_id matches current thread
relevant_docs = memory_vault.client.query_points(
collection_name="agent_memories",
query=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="thread_id",
match=models.MatchValue(value=state["thread_id"])
)
]
),
limit=2
).points
其核心优势
首要优势在于跨会话的长期记忆保存。存储在 PostgreSQL 或 SQLite 检查点中的短期情景记忆会在 thread_id 结束或变更时消失，而 Qdrant 能在独立的线程之间全局保留这些记忆。
其次，搜索能够遵循用户意图。嵌入模型将文本映射到 1,536 维空间，通过余弦距离根据语义而非精确字符串来查找记忆。
第三，检索范围可由用户选择。不使用负载过滤器可实现全局跨线程检索；若需要在特定情况下进行严格隔离，则可在 user_id 或 tenant_id 等键上添加 models.Filter。
多租户过滤功能
针对多租户场景，经过更新的 graph.py 会在每次查询时应用负载过滤器：
import os
import uuid
from typing import TypedDict, List, Dict, Any
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from qdrant_client.http import models
from langgraph.graph import StateGraph, START, END
from app.memory.vector_store import memory_vault
load_dotenv()
class AgentState(TypedDict):
messages: List[Dict[str, Any]]
thread_id: str
tenant_id: str # Enforces multi-tenancy boundaries
user_id: str # Identifies the end-user
metadata: Dict[str, Any]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
async def recall_node(state: AgentState) -> dict:
"""Retrieves semantic memories STRICTLY bounded by tenant_id and user_id."""
user_query = state["messages"][-1]["content"]
query_vector = embeddings.embed_query(user_query)
# Multi-Tenant Payload Filter Enforcement
tenant_filter = models.Filter(
must=[
models.FieldCondition(
key="tenant_id",
match=models.MatchValue(value=state["tenant_id"])
),
models.FieldCondition(
key="user_id",
match=models.MatchValue(value=state["user_id"])
)
]
)
# Scoped Query Execution
relevant_docs = memory_vault.client.query_points(
collection_name="agent_memories",
query=query_vector,
query_filter=tenant_filter, # Prevents cross-tenant data leaks
limit=3
).points
if relevant_docs:
memory_context = "\n".join([f"- {hit.payload['text']}" for hit in relevant_docs])
system_instruction = {
"role": "system",
"content": (
"You have access to the following long-term memory facts about the user:\n"
f"{memory_context}\n\n"
"Use the facts above to directly answer the user's question."
)
}
state["messages"].insert(0, system_instruction)
return {"messages": state["messages"]}
async def agent_node(state: AgentState) -> dict:
"""LLM reasoning node."""
response = await llm.ainvoke(state["messages"])
state["messages"].append({"role": "assistant", "content": response.content})
return {"messages": state["messages"]}
async def memorize_node(state: AgentState) -> dict:
"""Embeds user facts along with mandatory multi-tenant metadata payloads."""
user_message = next(
(m["content"] for m in reversed(state["messages"]) if m.get("role") == "user"),
None
)
if user_message:
vector = embeddings.embed_query(user_message)
memory_vault.client.upsert(
collection_name="agent_memories",
points=[
models.PointStruct(
id=str(uuid.uuid4()),
vector=vector,
payload={
"text": user_message,
"tenant_id": state["tenant_id"], # Tenant payload tag
"user_id": state["user_id"], # User payload tag
"thread_id": state["thread_id"],
"metadata": state.get("metadata", {})
}
)
]
)
return state
# Graph Workflow Wiring
workflow = StateGraph(AgentState)
workflow.add_node("recall", recall_node)
workflow.add_node("agent", agent_node)
workflow.add_node("memorize", memorize_node)
workflow.add_edge(START, "recall")
workflow.add_edge("recall", "agent")
workflow.add_edge("agent", "memorize")
workflow.add_edge("memorize", END)
app_graph = workflow.compile()
其优势十分实用。用户或组织只能获取自己的记忆点，因此一个租户的上下文绝不会泄露给其他租户。单个集合即可满足所有人的需求，从而避免了需要数千个独立集合所带来的内存和CPU消耗问题。而且一个运行的Qdrant实例就能服务数千个租户，无需为每个客户端都配备专用节点。
相关阅读
- 使用Ollama和LangChain在本地LLM上运行的QuickBooks财务智能体 — 基于Qwen 3.5 9B模型和RTX 4070的ReAct智能体可在无需使用云服务的情况下读取实时QuickBooks数据。文中包含代码、运行结果以及已知限制。