基于LlamaIndex与FastAPI的RAG聊天机器人多用户会话隔离问题求助

问题背景 #

当前基于LlamaIndex与FastAPI搭建的RAG聊天机器人，因全局共享单个chat_engine实例，导致多用户同时访问时会话历史串扰——比如用户A告知机器人自己名字后，用户B询问名字会得到属于A的回复。

解决方案 #

需要在FastAPI层面做会话隔离，结合LlamaIndex的聊天引擎特性实现用户会话独立，以下是三种可行方案：

方案1：为每个用户创建独立ChatEngine实例 #

核心思路：不再全局初始化chat_engine，而是为每个用户（通过user_id标识）维护独立的聊天引擎实例，会话历史自然隔离。

1. 修改请求模型，让前端传递用户ID：

class Question(BaseModel):
    user_id: str
    question: str

2. 新增用户聊天引擎的存储与获取逻辑：

# 内存存储用户专属chat_engine，生产环境建议替换为Redis等分布式存储
user_chat_engines = {}

def get_user_chat_engine(user_id: str):
    if user_id not in user_chat_engines:
        # 为每个用户创建独立的聊天引擎
        chat_engine = res_index.as_chat_engine(
            chat_mode="context",
            system_prompt="你的系统提示词内容"
        )
        user_chat_engines[user_id] = chat_engine
    return user_chat_engines[user_id]

3. 更新接口逻辑，绑定用户ID到对应聊天引擎：

@app.post('/ask')
def ask(question : Question):
    chat_engine = get_user_chat_engine(question.user_id)
    response = chat_engine.chat(question.question)
    return {"response": response.response}

方案2：手动传递对话历史（无状态实现） #

核心思路：放弃后端维护会话历史，由前端每次请求携带完整对话记录，后端调用时传入历史上下文，实现无状态服务。

1. 修改请求模型，支持接收对话历史：

from llama_index.core.chat_engine.types import ChatMessage

class ChatHistoryItem(BaseModel):
    role: str  # 取值为"user"或"assistant"
    content: str

class Question(BaseModel):
    question: str
    chat_history: list[ChatHistoryItem]

2. 更新接口逻辑，手动注入对话历史：

@app.post('/ask')
def ask(question : Question):
    # 将前端传入的历史转换为LlamaIndex兼容的ChatMessage格式
    history = [ChatMessage(role=item.role, content=item.content) for item in question.chat_history]
    # 调用聊天引擎时传入历史上下文
    response = chat_engine.chat(question.question, chat_history=history)
    # 返回更新后的对话历史给前端，方便下次请求携带
    updated_history = history + [
        ChatMessage(role="user", content=question.question),
        response
    ]
    return {
        "response": response.response,
        "updated_history": [{"role": msg.role, "content": msg.content} for msg in updated_history]
    }

方案3：基于Cookie会话ID自动隔离用户 #

核心思路：通过FastAPI的Cookie机制自动生成会话ID，无需前端手动传用户ID，后端用会话ID绑定独立聊天引擎。

1. 更新接口逻辑，处理会话ID：

from fastapi import Cookie, Response
import os

# 复用方案1的user_chat_engines与get_user_chat_engine函数

@app.post('/ask')
def ask(question : Question, session_id: str = Cookie(None), response: Response = None):
    # 若没有会话ID，生成新ID并写入Cookie
    if not session_id:
        session_id = os.urandom(16).hex()
        response.set_cookie(key="session_id", value=session_id)
    # 获取当前会话对应的聊天引擎
    chat_engine = get_user_chat_engine(session_id)
    response_data = chat_engine.chat(question.question)
    return {"response": response_data.response}

方案对比 #

• 方案1：开发简单，用户体验无感知，但内存随用户量增长，生产环境需搭配分布式存储。
• 方案2：无状态服务，适合大规模部署，无需后端存储会话数据，但需要前端配合维护对话历史。
• 方案3：用户无需主动提供标识，依赖Cookie自动识别，但需处理跨域Cookie、会话过期等问题。

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