Flask中为每个用户会话创建独立Chatbot实例的实现方案

兄弟，我看你遇到的问题是多实例运行时所有会话共用同一个Bot，而用g对象又会每次请求新建实例——这其实是没搞清楚Flask里g和session的区别，还有全局变量的坑。我给你拆解下问题，再给你一套可行的方案：

问题根源分析 #

1. 全局变量bot2 = Bot2()：这个变量是所有请求共享的，不管哪个用户的会话都会用同一个Bot实例，所以多会话时必然串了。
2. before_request里每次新建g.bot2：g是请求级的变量，每个HTTP请求都会重新创建g对象，所以每次用户发消息都会生成新Bot，自然没法保持会话状态。
3. 每次请求重置session['uid']：你现在每次请求都生成新的uuid，等于每个请求都是新会话，根本没法关联同一个用户的多次请求。

解决方案思路 #

核心是：用会话ID（session['uid']）唯一标识每个用户会话，在服务器端存储会话ID和Bot实例的映射，同一个会话的所有请求复用同一个Bot实例。

具体实现步骤 #

1. 正确初始化会话ID：只有当会话中没有uid时才生成，保证同一个用户的会话ID不变。
2. 服务器端存储Bot实例：用一个全局字典绑定会话ID和Bot实例，同时加线程锁保证多线程环境下的安全。
3. 请求时绑定当前会话的Bot到g：在before_request中根据会话ID获取或创建Bot实例，绑定到g对象，方便后续代码调用。
4. 替换所有全局Bot的引用：把代码里所有用bot2的地方换成g.bot2。

修改后的完整代码 #

from flask import Flask, render_template, request, session, g
import uuid
import threading
import time

app = Flask(__name__) 
items = ["CommonName","Title", "Department","Address","City","PhoneNum"] 
app.config.from_object(__name__)

# 全局存储：会话ID -> Bot实例
bot_instances = {}
# 线程锁，保证多线程下操作字典的安全
bot_lock = threading.Lock()
# 记录每个会话的最后活跃时间，用于清理过期实例
last_active = {}
# 30分钟未活跃则清理实例
INACTIVE_TIMEOUT = 1800

def cleanup_inactive_bots():
    """定期清理长时间未活跃的Bot实例，避免内存泄漏"""
    while True:
        time.sleep(3600)  # 每小时执行一次清理
        current_time = time.time()
        with bot_lock:
            # 找出超过超时时间的会话ID
            expired_uids = [uid for uid, timestamp in last_active.items() if current_time - timestamp > INACTIVE_TIMEOUT]
            for uid in expired_uids:
                del bot_instances[uid]
                del last_active[uid]
            print(f"清理了{len(expired_uids)}个过期Bot实例")

# 启动后台清理线程（守护线程，随主进程退出）
cleanup_thread = threading.Thread(target=cleanup_inactive_bots, daemon=True)
cleanup_thread.start()

@app.before_request
def before_request():
    # 初始化会话ID：只有会话中没有uid时才生成
    if 'uid' not in session:
        session['uid'] = str(uuid.uuid4())
    uid = session['uid']
    
    with bot_lock:
        # 为当前会话创建或获取Bot实例
        if uid not in bot_instances:
            bot_instances[uid] = Bot2()
        # 将Bot实例绑定到g对象，方便当前请求的其他函数使用
        g.bot2 = bot_instances[uid]
        # 更新最后活跃时间
        last_active[uid] = time.time()
    print(f"当前会话ID: {uid}")

@app.route("/", methods=['GET'])
def home():
    return render_template("index.html")

@app.route("/tables")
def show_tables():
    # 使用当前会话的Bot实例
    data = g.bot2.df
    if data.size == 0:
        return render_template('sad.html')
    return render_template('view.html', tables=[data.to_html(classes='df')], titles=items)

@app.route("/get")
def get_bot_response():
    userText = request.args.get('msg')
    # 使用当前会话的Bot实例处理消息
    g.bot2.input(str(userText))
    print(f"Bot回复内容: {g.bot2.message}")
    
    if g.bot2.export:
        return f"{g.bot2.message}<br/>Would you like to narrow your results?"
    else:
        return str(g.bot2.message)

if __name__ == "__main__":
    app.secret_key = 'super secret key' 
    app.run(threaded=True)

关键细节说明 #

• 线程锁bot_lock：因为你用了threaded=True启动Flask，多线程环境下对全局字典的读写可能会有并发问题，加锁能保证操作的原子性。
• 过期实例清理：如果用户关闭浏览器或者会话过期，对应的Bot实例会一直占内存，后台线程定期清理能避免内存泄漏。
• g对象的作用：g是请求级的容器，把Bot实例绑定到g后，同一个请求里的所有函数都能直接用g.bot2，不用每次从字典里取，代码更简洁。

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