基于LangChain+OpenAI+ChromaDB构建记忆式RAG多会话系统求助

1. 构建带向量检索的记忆式RAG核心步骤 #

• 向量知识库搭建：将业务文档转换为向量存入ChromaDB，生成检索器（Retriever），用于查询时召回相关文档。
• 会话记忆管理：为每个用户的每个会话维护独立的聊天历史，确保上下文连贯性。
• 多模块融合：将用户当前查询、会话历史、检索到的知识库文档整合后输入大模型，生成基于上下文和知识库的精准回答。

2. 用Runnable Chain结合OpenAI与ChromaDB实现系统 #

核心实现与错误修正 #

步骤1：初始化向量数据库与检索器

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 加载持久化的ChromaDB
persist_directory = "./chroma_db"
embedding = OpenAIEmbeddings()
vectordb = Chroma(persist_directory=persist_directory, embedding_function=embedding)
retriever = vectordb.as_retriever(k=4)  # 召回4条相关文档

步骤2：设计包含记忆与检索内容的Prompt

Prompt需同时承载知识库上下文、会话历史和用户输入：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "结合以下知识库内容和会话历史回答用户问题：\n{context}"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

步骤3：构建核心Runnable Chain

先检索文档并格式化，再结合输入和历史传入Prompt，最后调用LLM：

from langchain_openai.chat_models import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 格式化检索到的文档为字符串
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 整合检索、输入、历史的Chain
retrieval_chain = (
    {"context": retriever | format_docs, "input": RunnablePassthrough(), "history": RunnablePassthrough()}
    | prompt
    | llm
)

步骤4：绑定多用户多会话记忆

通过RunnableWithMessageHistory为每个会话维护独立历史：

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import ConfigurableFieldSpec
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory

# 会话存储，用(user_id, conversation_id)作为唯一标识
store = {}

def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
    key = (user_id, conversation_id)
    if key not in store:
        store[key] = ChatMessageHistory()
    return store[key]

# 为Chain添加记忆支持
with_message_history = RunnableWithMessageHistory(
    runnable=retrieval_chain,
    get_session_history=get_session_history,
    input_messages_key="input",  # 与输入字段名对应
    history_messages_key="history",  # 与Prompt中的占位符对应
    history_factory_config=[
        ConfigurableFieldSpec(
            id="user_id",
            annotation=str,
            name="用户ID",
            description="用户唯一标识",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id",
            annotation=str,
            name="会话ID",
            description="会话唯一标识",
            default="",
            is_shared=True,
        ),
    ],
)

步骤5：测试调用

user_id = "user123"
conversation_id = "conv456"

# 首次提问
response = with_message_history.invoke(
    {"input": "法国的首都是什么？"},
    config={"configurable": {"user_id": user_id, "conversation_id": conversation_id}},
)
print(response.content)

# 跟进提问（测试记忆）
response2 = with_message_history.invoke(
    {"input": "它的人口大约是多少？"},
    config={"configurable": {"user_id": user_id, "conversation_id": conversation_id}},
)
print(response2.content)

关键错误修正说明 #

• 原代码直接拼接retriever|llm无效：Retriever输出的是文档对象，需先格式化为字符串才能传入Prompt。
• 统一字段名：确保输入字段、Prompt变量、记忆配置的key一致（如用"input"代替"query"），避免匹配错误。
• Chain结构逻辑：必须先完成文档检索与格式化，再将上下文、输入、历史一起传入Prompt，最后调用LLM。

内容的提问来源于stack exchange，提问作者streak