基于Python-Flask的LLM聊天机器人多用户会话上下文隔离方案问询

核心思路 #

LLM本身无状态，要实现用户独立会话，需解决三个核心问题：

• 为每个用户分配唯一标识（登录用户用用户ID，匿名用户用会话UUID）
• 服务器端持久化存储每个用户的聊天历史
• 每次请求时，根据用户标识取出对应历史传入Llama Index

步骤1：修复Flask Session问题（改用服务器端存储） #

Flask默认会话存在客户端Cookie，易丢失且无法跨设备共享，必须改用服务器端会话存储，推荐用AWS Elasticache Redis或DynamoDB：

依赖安装 #

pip install flask flask-session redis llama-index openai

Flask会话配置（Redis后端示例） #

from flask import Flask, session, request, jsonify
from flask_session import Session
import redis
from uuid import uuid4

app = Flask(__name__)

# 生产环境建议用AWS Secrets Manager或环境变量存储敏感信息
app.config['SECRET_KEY'] = 'your-secure-secret-key'
app.config['SESSION_TYPE'] = 'redis'
# 替换为AWS Elasticache Redis的端点、端口和密码
app.config['SESSION_REDIS'] = redis.Redis(
    host='your-redis-endpoint',
    port=6379,
    password='your-redis-password'
)
app.config['SESSION_PERMANENT'] = True  # 持久化会话，支持跨设备
app.config['SESSION_USE_SIGNER'] = True  # 防止会话ID被篡改

Session(app)

步骤2：实现用户聊天历史的存储与加载 #

Llama Index的ChatMemoryBuffer支持序列化/反序列化，可直接对接Redis存储：

from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI
from llama_index.core.chat_engine import SimpleChatEngine
import json

# 初始化OpenAI LLM
llm = OpenAI(model="gpt-3.5-turbo")

# 从Redis加载用户聊天历史
def load_chat_history(user_id):
    redis_client = app.session_interface.redis
    history_data = redis_client.get(f"chat_history:{user_id}")
    if history_data:
        return ChatMemoryBuffer.from_dict(json.loads(history_data))
    # 无历史则创建新会话
    return ChatMemoryBuffer.from_defaults(token_limit=1500)

# 将更新后的聊天历史保存到Redis
def save_chat_history(user_id, memory):
    redis_client = app.session_interface.redis
    redis_client.set(f"chat_history:{user_id}", json.dumps(memory.to_dict()))

步骤3：核心聊天接口实现 #

@app.route('/chat', methods=['POST'])
def chat():
    data = request.get_json()
    user_message = data.get('message')
    if not user_message:
        return jsonify({'error': '缺少消息内容'}), 400

    # 获取或生成用户唯一标识
    if 'user_id' in session:
        user_id = session['user_id']
    else:
        user_id = str(uuid4())
        session['user_id'] = user_id

    # 加载用户专属聊天历史
    memory = load_chat_history(user_id)
    # 创建聊天引擎并生成回复
    chat_engine = SimpleChatEngine.from_defaults(llm=llm, memory=memory)
    response = chat_engine.chat(user_message)
    # 保存更新后的历史
    save_chat_history(user_id, memory)

    return jsonify({'response': str(response)})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

关键注意事项 #

1. 跨设备会话共享：匿名用户仅能通过同一会话ID（Cookie中的session标识）共享历史，若要实现跨设备，必须添加用户登录系统，用用户ID作为唯一标识。
2. AWS部署适配：
   • Redis使用AWS Elasticache时，需确保Flask应用所在VPC与Elasticache集群网络连通
   • 若选择DynamoDB，可替换Flask-Session后端为dynamodb，或自行实现存储逻辑
3. 生产环境优化：
   • 敏感配置（SECRET_KEY、Redis密码）避免硬编码，用环境变量或AWS Secrets Manager管理
   • 限制聊天历史的token数量，避免LLM调用超时或费用过高
4. Flask Session排错：
   • 检查是否设置了有效SECRET_KEY（长度不少于16位）
   • 确认服务器端存储（Redis/DynamoDB）连接正常
   • 确保SESSION_PERMANENT设为True，会话不会自动过期

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