Flask集成LangChain多会话聊天Bot的会话管理问题排查

核心问题分析 #

你当前的代码每次请求都会重新实例化RetrievalQA和ConversationBufferMemory，导致会话记录无法留存；同时RetrievalQA对象不可序列化，无法直接存入Flask Session。下面提供两种可行的解决思路：

方案一：服务器端全局维护会话映射（单进程场景） #

将会话实例存储在服务器端的全局字典中，通过唯一会话ID关联用户请求，避免重复创建对象并留存记忆。同时将重复初始化的资源（如Chroma、LLM、Retriever）移到路由外，提升性能。

完整代码示例

from flask import Flask, request, jsonify, session
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.settings import Settings
import threading
import uuid

app = Flask(__name__)
app.secret_key = "your-custom-secret-key"  # 必须设置才能启用Session

# 全局初始化复用资源，避免每次请求重建
embeddings = OpenAIEmbeddings()
CHROMA_SETTINGS = Settings(
    chroma_db_impl='duckdb+parquet',
    persist_directory='db',
    anonymized_telemetry=False
)
db = Chroma(persist_directory='db', embedding_function=embeddings, client_settings=CHROMA_SETTINGS)
retriever = db.as_retriever(search_kwargs={"k": target_source_chunks})
llm = ChatOpenAI()

# 会话模板
template = """
Use the following context (delimited by <ctx></ctx>) and the chat history (delimited by <hs></hs>) to answer the question:
------
<ctx>
{context}
</ctx>
------
<hs>
{history}
</hs>
------
{question}
Answer:
"""
prompt = PromptTemplate(
    input_variables=["history", "context", "question"],
    template=template,
)

# 服务器端会话存储：key为会话ID，value为RetrievalQA实例
session_store = {}
store_lock = threading.Lock()  # 多线程环境下加锁保证线程安全

@app.route('/answer', methods=['POST'])
def answer():
    # 生成/获取当前用户的唯一会话ID
    if "session_id" not in session:
        session["session_id"] = str(uuid.uuid4())
    session_id = session["session_id"]
    query = request.form["query"]

    with store_lock:
        # 检查会话是否已存在，不存在则创建QA实例
        if session_id not in session_store:
            qa = RetrievalQA.from_chain_type(
                llm=llm,
                chain_type='stuff',
                retriever=retriever,
                verbose=True,
                chain_type_kwargs={
                    "verbose": True,
                    "prompt": prompt,
                    "memory": ConversationBufferMemory(
                        memory_key="history",
                        input_key="question"
                    ),
                }
            )
            session_store[session_id] = qa
        else:
            qa = session_store[session_id]

    # 执行查询，自动更新会话记忆
    result = qa.run({"query": query})
    return jsonify({'answer': result})

注意事项

• 该方案仅适用于单进程部署（如Flask开发服务器），多进程部署时全局字典无法跨进程共享，需改用Redis等分布式存储替代全局字典。

方案二：序列化会话记忆到Flask Session（多进程兼容） #

将会话记忆序列化后存入Flask Session，每次请求时重建ConversationBufferMemory并加载历史记录，无需在服务器端维护全局实例，天然支持多进程部署。

完整代码示例

from flask import Flask, request, jsonify, session
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.settings import Settings

app = Flask(__name__)
app.secret_key = "your-custom-secret-key"

# 全局初始化复用资源
embeddings = OpenAIEmbeddings()
CHROMA_SETTINGS = Settings(
    chroma_db_impl='duckdb+parquet',
    persist_directory='db',
    anonymized_telemetry=False
)
db = Chroma(persist_directory='db', embedding_function=embeddings, client_settings=CHROMA_SETTINGS)
retriever = db.as_retriever(search_kwargs={"k": target_source_chunks})
llm = ChatOpenAI()

# 会话模板
template = """
Use the following context (delimited by <ctx></ctx>) and the chat history (delimited by <hs></hs>) to answer the question:
------
<ctx>
{context}
</ctx>
------
<hs>
{history}
</hs>
------
{question}
Answer:
"""
prompt = PromptTemplate(
    input_variables=["history", "context", "question"],
    template=template,
)

@app.route('/answer', methods=['POST'])
def answer():
    query = request.form["query"]
    # 从Session加载历史对话，默认空列表
    history_messages = session.get("chat_history", [])

    # 重建会话记忆并加载历史
    memory = ConversationBufferMemory(
        memory_key="history",
        input_key="question"
    )
    for msg in history_messages:
        if msg["type"] == "human":
            memory.chat_memory.add_user_message(msg["content"])
        elif msg["type"] == "ai":
            memory.chat_memory.add_ai_message(msg["content"])

    # 创建QA实例（复用全局资源，仅记忆为会话专属）
    qa = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type='stuff',
        retriever=retriever,
        verbose=True,
        chain_type_kwargs={
            "verbose": True,
            "prompt": prompt,
            "memory": memory,
        }
    )

    # 执行查询并更新会话记忆
    result = qa.run({"query": query})
    updated_messages = [{"type": msg.type, "content": msg.content} for msg in memory.chat_memory.messages]
    session["chat_history"] = updated_messages

    return jsonify({'answer': result})

注意事项

• Flask默认将Session存储在客户端Cookie中，若会话历史过长会导致Cookie体积超限，建议使用Flask-Session扩展将Session存储到服务器端（如Redis、数据库）。

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