欢迎光临
我们一直在努力

Nanbeige4.1-3B实操手册:webui.py源码关键修改点——支持历史会话持久化

Nanbeige4.1-3B实操手册:webui.py源码关键修改点——支持历史会话持久化

1. 引言:为什么需要历史会话持久化?

想象一下这个场景:你正在和Nanbeige4.1-3B模型进行一场深入的对话,讨论一个复杂的技术问题。聊了十几轮,模型给出了很多有价值的见解,你正准备把这些内容整理成文档。突然,浏览器崩溃了,或者你需要重启WebUI服务。当你重新打开页面时,发现刚才所有的对话记录都消失了——那种感觉,是不是特别让人抓狂?

这就是我们今天要解决的问题。

Nanbeige4.1-3B自带的WebUI界面功能很强大,但它有一个明显的短板:不支持历史会话的持久化保存。每次刷新页面或重启服务,所有的对话记录都会丢失。对于需要长期跟踪对话、积累知识库、或者进行多轮调试的用户来说,这无疑是一个巨大的痛点。

好消息是,这个问题完全可以通过修改webui.py源码来解决。在本文中,我将带你一步步分析源码,找到关键修改点,实现历史会话的自动保存和加载功能。无论你是Python新手还是有经验的开发者,都能跟着这个教程,让你的Nanbeige4.1-3B WebUI变得更加强大和实用。

2. 理解WebUI的会话管理机制

在开始修改代码之前,我们先要搞清楚Nanbeige4.1-3B WebUI是如何管理会话的。这就像修车一样,你得先知道车的结构,才能知道在哪里动手。

2.1 当前会话管理的局限性

打开webui.py文件,你会发现对话历史通常是以内存变量的形式存储的。比如,代码中可能会有类似这样的结构:

# 当前WebUI中常见的会话存储方式(简化示例)
conversation_history = [] # 这是一个全局变量或函数内的局部变量

def chat_function(user_input):
# 将用户输入添加到历史
conversation_history.append({"role": "user", "content": user_input})

# 调用模型生成回复
response = generate_response(conversation_history)

# 将模型回复添加到历史
conversation_history.append({"role": "assistant", "content": response})

return response

这种设计有几个明显的问题:

  • 数据易失性:变量存储在内存中,程序重启就没了
  • 缺乏隔离性:所有用户的对话都混在一起(如果是多用户场景)
  • 无法追溯:没有时间戳,不知道对话是什么时候发生的
  • 容量有限:长时间对话可能导致内存占用过高
  • 2.2 我们需要什么样的持久化方案?

    在动手修改之前,先明确我们的目标。一个好的历史会话持久化方案应该具备:

    • 自动保存:每次对话后自动保存,无需手动操作
    • 按会话隔离:每个对话会话独立存储,互不干扰
    • 支持恢复:重启后能加载之前的对话继续
    • 简单易用:对最终用户透明,不需要额外配置
    • 性能友好:不能明显拖慢WebUI的响应速度

    基于这些要求,我选择了JSON文件存储作为解决方案。原因很简单:JSON格式易读易写,Python原生支持,而且足够满足我们的需求。

    3. 源码分析:找到关键修改点

    现在让我们深入webui.py的源码,找到需要修改的关键位置。我会用实际的代码片段来说明,你可以对照自己的文件进行查找。

    3.1 定位会话存储变量

    首先,我们需要找到存储对话历史的数据结构。在webui.py中搜索以下关键词:

    # 可能的关键词
    history = []
    messages = []
    conversation = []
    chat_history = []

    通常,你会在chat函数或类似的对话处理函数中找到类似这样的代码:

    def predict(input_text, history):
    # history参数就是当前的对话历史
    # 它通常是一个列表,每个元素是一个元组 (用户输入, 模型回复)

    # 处理用户输入
    # 调用模型生成回复
    # 更新history

    return response, updated_history

    在Nanbeige4.1-3B的WebUI中,这个history变量就是我们要持久化的核心数据。

    3.2 理解Gradio的ChatInterface结构

    Nanbeige4.1-3B的WebUI基于Gradio构建,特别是使用了gr.ChatInterface。这是Gradio提供的一个高级组件,专门用于构建聊天界面。

    关键代码结构通常如下:

    import gradio as gr

    def chat_fn(message, history):
    """处理聊天消息的核心函数"""
    # history的格式:[(用户消息1, 助手回复1), (用户消息2, 助手回复2), …]

    # 构建模型需要的消息格式
    messages = []
    for human, assistant in history:
    messages.append({"role": "user", "content": human})
    messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": message})

    # 调用模型生成回复
    response = call_model(messages)

    # 返回回复和更新后的历史
    return response

    # 创建聊天界面
    demo = gr.ChatInterface(
    fn=chat_fn,
    title="Nanbeige4.1-3B Chat",
    description="与Nanbeige4.1-3B模型对话"
    )

    这里的关键是chat_fn函数的history参数,它包含了所有的对话历史。我们的任务就是把这个history保存到文件中,并在需要时加载回来。

    4. 实战修改:添加会话持久化功能

    理解了基本原理后,现在开始动手修改代码。我会提供完整的代码片段,你可以直接复制使用。

    4.1 第一步:添加必要的导入和配置

    在webui.py文件的开头部分,添加以下导入语句和配置:

    import json
    import os
    from datetime import datetime
    from pathlib import Path

    # 会话存储配置
    SESSION_DIR = Path("chat_sessions") # 会话存储目录
    SESSION_DIR.mkdir(exist_ok=True) # 确保目录存在

    def get_session_filename(session_id=None):
    """生成会话文件名"""
    if session_id is None:
    # 如果没有提供session_id,使用时间戳生成
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    session_id = f"session_{timestamp}"

    return SESSION_DIR / f"{session_id}.json"

    def save_session(history, session_id=None):
    """保存会话历史到文件"""
    try:
    filename = get_session_filename(session_id)

    # 转换历史格式为可序列化的结构
    session_data = {
    "session_id": session_id or filename.stem,
    "save_time": datetime.now().isoformat(),
    "history": history,
    "message_count": len(history)
    }

    with open(filename, 'w', encoding='utf-8') as f:
    json.dump(session_data, f, ensure_ascii=False, indent=2)

    print(f"会话已保存到: {filename}")
    return filename.stem # 返回session_id
    except Exception as e:
    print(f"保存会话失败: {e}")
    return None

    def load_session(session_id):
    """从文件加载会话历史"""
    try:
    filename = get_session_filename(session_id)

    if not filename.exists():
    print(f"会话文件不存在: {filename}")
    return []

    with open(filename, 'r', encoding='utf-8') as f:
    session_data = json.load(f)

    print(f"已加载会话: {session_id} (包含{session_data['message_count']}条消息)")
    return session_data["history"]
    except Exception as e:
    print(f"加载会话失败: {e}")
    return []

    def list_sessions():
    """列出所有可用的会话"""
    sessions = []
    for file in SESSION_DIR.glob("*.json"):
    try:
    with open(file, 'r', encoding='utf-8') as f:
    data = json.load(f)
    sessions.append({
    "id": file.stem,
    "time": data.get("save_time", "未知时间"),
    "message_count": data.get("message_count", 0)
    })
    except:
    continue

    # 按时间倒序排列
    sessions.sort(key=lambda x: x["time"], reverse=True)
    return sessions

    这段代码创建了一个完整的会话管理模块,提供了保存、加载、列出会话的功能。

    4.2 第二步:修改聊天函数支持自动保存

    接下来,我们需要修改核心的聊天函数,让它支持自动保存。找到webui.py中的chat_fn或类似函数,进行如下修改:

    # 添加一个全局变量或状态来跟踪当前会话ID
    current_session_id = None

    def chat_fn(message, history):
    """处理聊天消息的核心函数(已添加自动保存)"""
    global current_session_id

    # 如果是新会话且没有session_id,创建一个
    if not history and current_session_id is None:
    current_session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
    print(f"开始新会话: {current_session_id}")

    # 原有的聊天逻辑保持不变
    # 构建消息格式
    messages = []
    for human, assistant in history:
    messages.append({"role": "user", "content": human})
    messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": message})

    # 调用模型生成回复(这里用你的实际模型调用代码)
    # 注意:以下是一个示例,你需要替换为实际的模型调用代码
    response = "这是模型的回复" # 替换为实际的模型调用

    # 更新历史
    updated_history = history + [(message, response)]

    # 自动保存会话(每5轮对话保存一次,避免频繁IO)
    if len(updated_history) % 5 == 0 or len(updated_history) <= 1:
    save_session(updated_history, current_session_id)

    return response

    这里的关键点是:

  • 添加了current_session_id来跟踪当前会话
  • 在新会话开始时自动生成session_id
  • 每5轮对话自动保存一次(可调整这个频率)
  • 保持原有的聊天逻辑不变
  • 4.3 第三步:添加会话管理界面组件

    为了让用户能够方便地管理会话,我们需要在WebUI中添加一些控制组件。在创建Gradio界面的部分,添加以下代码:

    import gradio as gr

    # 原有的聊天界面创建代码
    # demo = gr.ChatInterface(…)

    # 我们将在原有基础上扩展功能

    def create_enhanced_interface():
    """创建增强版的聊天界面"""

    # 状态变量
    current_session = gr.State(value=None) # 存储当前session_id
    chat_history = gr.State(value=[]) # 存储聊天历史

    with gr.Blocks(title="Nanbeige4.1-3B Chat (增强版)", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🚀 Nanbeige4.1-3B 智能对话")
    gr.Markdown("支持历史会话持久化保存和加载")

    # 会话管理面板
    with gr.Row():
    with gr.Column(scale=1):
    gr.Markdown("### 会话管理")

    # 新会话按钮
    new_btn = gr.Button("🆕 开始新会话", variant="primary")

    # 保存会话按钮
    save_btn = gr.Button("💾 保存当前会话")

    # 会话列表
    session_dropdown = gr.Dropdown(
    label="加载历史会话",
    choices=[],
    interactive=True
    )

    # 加载按钮
    load_btn = gr.Button("📂 加载选中会话")

    # 刷新列表按钮
    refresh_btn = gr.Button("🔄 刷新会话列表")

    with gr.Column(scale=3):
    # 聊天界面
    chatbot = gr.Chatbot(label="对话历史", height=500)
    msg = gr.Textbox(label="输入消息", placeholder="在这里输入您的问题…")
    clear_btn = gr.Button("清空输入")

    # 原有的聊天函数(稍作修改)
    def enhanced_chat_fn(message, history, session_state):
    """增强版的聊天函数"""
    # 如果历史为空且没有session_id,创建新会话
    if not history and session_state is None:
    session_state = datetime.now().strftime("%Y%m%d_%H%M%S")
    print(f"开始新会话: {session_state}")

    # 构建消息(使用你的实际模型调用代码)
    # 这里简化处理,实际使用时替换为模型调用
    response = f"收到消息: {message}\\n(这是模拟回复,请替换为实际模型调用)"

    # 更新历史
    updated_history = history + [(message, response)]

    # 更新session_state
    return response, updated_history, session_state

    # 连接聊天功能
    msg.submit(
    enhanced_chat_fn,
    [msg, chatbot, current_session],
    [msg, chatbot, current_session]
    ).then(
    lambda: "", # 清空输入框
    None,
    [msg]
    )

    # 清空按钮功能
    clear_btn.click(lambda: "", None, [msg])

    # 新会话功能
    def start_new_session():
    new_id = datetime.now().strftime("%Y%m%d_%H%M%S")
    return new_id, [] # 返回新的session_id和空历史

    new_btn.click(
    start_new_session,
    None,
    [current_session, chatbot]
    )

    # 保存会话功能
    def save_current_session(history, session_id):
    if session_id and history:
    save_session(history, session_id)
    return gr.Info(f"会话已保存: {session_id}")
    return gr.Warning("没有可保存的会话")

    save_btn.click(
    save_current_session,
    [chatbot, current_session],
    None
    )

    # 刷新会话列表
    def refresh_session_list():
    sessions = list_sessions()
    choices = [f"{s['id']} ({s['time'][:10]}, {s['message_count']}条消息)"
    for s in sessions]
    return gr.Dropdown(choices=choices, value=choices[0] if choices else None)

    refresh_btn.click(
    refresh_session_list,
    None,
    [session_dropdown]
    )

    # 初始加载时刷新列表
    demo.load(refresh_session_list, None, [session_dropdown])

    # 加载会话功能
    def load_selected_session(selection):
    if not selection:
    return None, [], gr.Warning("请先选择一个会话")

    # 从选择项中提取session_id
    session_id = selection.split(" ")[0]
    history = load_session(session_id)

    if history:
    return session_id, history, gr.Info(f"已加载会话: {session_id}")
    else:
    return None, [], gr.Warning("加载会话失败")

    load_btn.click(
    load_selected_session,
    [session_dropdown],
    [current_session, chatbot, session_dropdown]
    )

    return demo

    # 使用增强版界面
    demo = create_enhanced_interface()

    这段代码创建了一个功能完整的会话管理界面,包括:

    • 开始新会话
    • 保存当前会话
    • 查看历史会话列表
    • 加载历史会话
    • 自动保存功能

    4.4 第四步:集成实际的模型调用

    上面的代码中,我用了简化的响应来演示。现在,我们需要将实际的Nanbeige4.1-3B模型调用集成进去。找到你原来的模型调用代码,将其整合到enhanced_chat_fn函数中:

    def enhanced_chat_fn(message, history, session_state):
    """增强版的聊天函数(集成实际模型调用)"""
    # 如果历史为空且没有session_id,创建新会话
    if not history and session_state is None:
    session_state = datetime.now().strftime("%Y%m%d_%H%M%S")
    print(f"开始新会话: {session_state}")

    # 构建模型需要的消息格式
    messages = []
    for human, assistant in history:
    messages.append({"role": "user", "content": human})
    messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": message})

    try:
    # 调用Nanbeige4.1-3B模型生成回复
    # 这是你原有的模型调用代码,确保它正确工作
    input_ids = tokenizer.apply_chat_template(
    messages,
    return_tensors="pt"
    ).to(model.device)

    outputs = model.generate(
    input_ids,
    max_new_tokens=512,
    temperature=0.6,
    top_p=0.95,
    do_sample=True
    )

    response = tokenizer.decode(
    outputs[0][len(input_ids[0]):],
    skip_special_tokens=True
    )

    except Exception as e:
    response = f"模型调用出错: {str(e)}"
    print(f"模型调用错误: {e}")

    # 更新历史
    updated_history = history + [(message, response)]

    # 自动保存(每3轮对话保存一次)
    if len(updated_history) % 3 == 0:
    save_session(updated_history, session_state)

    return response, updated_history, session_state

    注意:你需要确保tokenizer和model在全局作用域中可用,或者通过其他方式传递进来。

    5. 完整代码示例与部署

    为了让你更清楚地看到完整的修改效果,我提供一个简化版的完整webui.py示例:

    #!/usr/bin/env python3
    """
    Nanbeige4.1-3B WebUI with Session Persistence
    支持历史会话持久化的增强版Web界面
    """

    import json
    import os
    import gradio as gr
    import torch
    from datetime import datetime
    from pathlib import Path
    from transformers import AutoModelForCausalLM, AutoTokenizer

    # ==================== 配置部分 ====================
    MODEL_PATH = "/root/ai-models/nanbeige/Nanbeige4___1-3B"
    SESSION_DIR = Path("chat_sessions")
    SESSION_DIR.mkdir(exist_ok=True)

    # ==================== 会话管理函数 ====================
    def get_session_filename(session_id=None):
    """生成会话文件名"""
    if session_id is None:
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    session_id = f"session_{timestamp}"
    return SESSION_DIR / f"{session_id}.json"

    def save_session(history, session_id=None):
    """保存会话历史到文件"""
    try:
    if not session_id:
    session_id = datetime.now().strftime("%Y%m%d_%H%M%S")

    filename = get_session_filename(session_id)
    session_data = {
    "session_id": session_id,
    "save_time": datetime.now().isoformat(),
    "history": history,
    "message_count": len(history)
    }

    with open(filename, 'w', encoding='utf-8') as f:
    json.dump(session_data, f, ensure_ascii=False, indent=2)

    print(f"✅ 会话已保存: {filename}")
    return session_id
    except Exception as e:
    print(f"❌ 保存失败: {e}")
    return None

    def load_session(session_id):
    """从文件加载会话历史"""
    try:
    filename = get_session_filename(session_id)
    if not filename.exists():
    return []

    with open(filename, 'r', encoding='utf-8') as f:
    data = json.load(f)

    print(f"✅ 已加载会话: {session_id}")
    return data.get("history", [])
    except Exception as e:
    print(f"❌ 加载失败: {e}")
    return []

    def list_sessions():
    """列出所有会话"""
    sessions = []
    for file in SESSION_DIR.glob("*.json"):
    try:
    with open(file, 'r', encoding='utf-8') as f:
    data = json.load(f)
    sessions.append({
    "id": data.get("session_id", file.stem),
    "time": data.get("save_time", "未知"),
    "count": data.get("message_count", 0)
    })
    except:
    continue

    sessions.sort(key=lambda x: x["time"], reverse=True)
    return sessions

    # ==================== 模型加载 ====================
    print("正在加载模型和分词器…")
    tokenizer = AutoTokenizer.from_pretrained(
    MODEL_PATH,
    trust_remote_code=True
    )

    model = AutoModelForCausalLM.from_pretrained(
    MODEL_PATH,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True
    )
    print("✅ 模型加载完成")

    # ==================== 聊天函数 ====================
    def chat_with_model(message, history, session_id):
    """与模型聊天并自动保存历史"""
    # 构建消息
    messages = []
    for human, assistant in history:
    messages.append({"role": "user", "content": human})
    messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": message})

    # 生成回复
    try:
    input_ids = tokenizer.apply_chat_template(
    messages,
    return_tensors="pt"
    ).to(model.device)

    outputs = model.generate(
    input_ids,
    max_new_tokens=512,
    temperature=0.6,
    top_p=0.95,
    do_sample=True
    )

    response = tokenizer.decode(
    outputs[0][len(input_ids[0]):],
    skip_special_tokens=True
    )
    except Exception as e:
    response = f"抱歉,生成回复时出错: {str(e)}"

    # 更新历史
    updated_history = history + [(message, response)]

    # 自动保存(每2轮保存一次)
    if session_id and len(updated_history) % 2 == 0:
    save_session(updated_history, session_id)

    return response, updated_history, session_id

    # ==================== 创建Web界面 ====================
    def create_web_interface():
    """创建Web界面"""

    with gr.Blocks(title="Nanbeige4.1-3B Chat Pro", theme=gr.themes.Soft()) as demo:
    # 状态变量
    current_session = gr.State(value=None)

    gr.Markdown("""
    # 🤖 Nanbeige4.1-3B 智能对话系统
    **支持历史会话持久化 | 自动保存 | 会话管理**
    """)

    with gr.Row():
    # 左侧控制面板
    with gr.Column(scale=1):
    gr.Markdown("### 会话控制")

    # 会话管理按钮
    new_btn = gr.Button("🆕 新会话", size="sm")
    save_btn = gr.Button("💾 保存会话", size="sm")
    load_btn = gr.Button("📂 加载会话", size="sm")
    refresh_btn = gr.Button("🔄 刷新列表", size="sm")

    # 会话列表
    session_list = gr.Dropdown(
    label="历史会话",
    choices=[],
    interactive=True
    )

    # 会话信息显示
    session_info = gr.Markdown("当前会话: 无")

    # 右侧聊天区域
    with gr.Column(scale=3):
    chatbot = gr.Chatbot(height=450, label="对话记录")
    msg = gr.Textbox(
    label="输入消息",
    placeholder="输入您的问题,按Enter发送…",
    lines=3
    )
    clear_btn = gr.Button("清空对话", variant="secondary")

    # 事件处理函数
    def on_message_submit(message, history, session_id):
    """处理消息提交"""
    if not message.strip():
    return "", history, session_id, gr.Warning("消息不能为空")

    # 如果是第一条消息且没有会话ID,创建新会话
    if not history and not session_id:
    session_id = datetime.now().strftime("%Y%m%d_%H%M%S")

    # 调用模型
    response, new_history, session_id = chat_with_model(
    message, history, session_id
    )

    # 更新会话信息
    info_text = f"当前会话: {session_id or '新会话'} | 消息数: {len(new_history)}"

    return "", new_history, session_id, info_text

    def on_new_session():
    """开始新会话"""
    new_id = datetime.now().strftime("%Y%m%d_%H%M%S")
    return new_id, [], f"当前会话: {new_id} | 消息数: 0"

    def on_save_session(history, session_id):
    """保存当前会话"""
    if not session_id:
    return gr.Warning("请先开始一个会话")
    if not history:
    return gr.Warning("没有对话内容可保存")

    save_session(history, session_id)
    return gr.Info(f"会话已保存: {session_id}")

    def on_refresh_list():
    """刷新会话列表"""
    sessions = list_sessions()
    if not sessions:
    return gr.Dropdown(choices=["暂无历史会话"], value=None), "当前会话: 无"

    choices = [f"{s['id']} | {s['time'][:10]} | {s['count']}条消息"
    for s in sessions[:10]] # 只显示最近10个

    return gr.Dropdown(choices=choices, value=choices[0]), "请选择要加载的会话"

    def on_load_session(selection):
    """加载选中的会话"""
    if not selection or "暂无" in selection:
    return None, [], "当前会话: 无"

    # 提取session_id
    session_id = selection.split(" | ")[0]
    history = load_session(session_id)

    if history:
    info = f"当前会话: {session_id} | 消息数: {len(history)}"
    return session_id, history, info
    else:
    return None, [], "加载失败,会话可能已损坏"

    def on_clear_chat():
    """清空聊天记录"""
    return [], "当前会话: 已清空"

    # 绑定事件
    msg.submit(
    on_message_submit,
    [msg, chatbot, current_session],
    [msg, chatbot, current_session, session_info]
    )

    new_btn.click(
    on_new_session,
    None,
    [current_session, chatbot, session_info]
    )

    save_btn.click(
    on_save_session,
    [chatbot, current_session],
    None
    )

    refresh_btn.click(
    on_refresh_list,
    None,
    [session_list, session_info]
    )

    load_btn.click(
    on_load_session,
    [session_list],
    [current_session, chatbot, session_info]
    )

    clear_btn.click(
    on_clear_chat,
    None,
    [chatbot, session_info]
    )

    # 页面加载时刷新会话列表
    demo.load(on_refresh_list, None, [session_list, session_info])

    return demo

    # ==================== 启动应用 ====================
    if __name__ == "__main__":
    print("启动Nanbeige4.1-3B WebUI…")
    print(f"会话存储目录: {SESSION_DIR.absolute()}")

    demo = create_web_interface()
    demo.launch(
    server_name="0.0.0.0",
    server_port=7860,
    share=False
    )

    这个完整示例包含了所有必要的功能,你可以直接使用或根据自己的需求进行调整。

    6. 部署与使用指南

    6.1 部署步骤

  • 备份原文件:

    cd /root/nanbeige-webui
    cp webui.py webui.py.backup

  • 应用修改:

    • 将上面的完整代码保存为新的webui.py
    • 或者按照前面的步骤逐步修改你现有的文件
  • 创建会话存储目录:

    mkdir -p /root/nanbeige-webui/chat_sessions

  • 重启WebUI服务:

    # 如果使用Supervisor
    supervisorctl restart nanbeige-webui

    # 或者直接运行
    cd /root/nanbeige-webui
    python webui.py

  • 6.2 使用说明

    启动WebUI后,你会看到增强版的界面:

  • 开始新会话:点击"新会话"按钮,系统会自动生成一个会话ID
  • 正常对话:在输入框中输入消息,按Enter发送
  • 自动保存:系统会每2轮对话自动保存一次(可修改频率)
  • 手动保存:随时点击"保存会话"按钮手动保存
  • 查看历史:点击"刷新列表"查看所有历史会话
  • 加载会话:从下拉列表选择历史会话,点击"加载会话"
  • 6.3 会话文件管理

    所有会话都保存在chat_sessions目录中,每个会话是一个JSON文件:

    {
    "session_id": "session_20250225_143022",
    "save_time": "2025-02-25T14:30:22.123456",
    "message_count": 15,
    "history": [
    ["你好", "你好!我是Nanbeige4.1-3B,很高兴为您服务。"],
    ["你能做什么", "我可以回答各种问题、协助编程、进行对话等…"],

    ]
    }

    你可以:

    • 直接查看JSON文件了解对话内容
    • 备份整个目录以防数据丢失
    • 删除不需要的会话文件释放空间

    7. 高级功能与优化建议

    7.1 数据库存储方案

    如果你需要更强大的会话管理功能,可以考虑使用数据库。这里提供一个SQLite的示例:

    import sqlite3
    from datetime import datetime

    class SessionDB:
    def __init__(self, db_path="sessions.db"):
    self.conn = sqlite3.connect(db_path)
    self.create_table()

    def create_table(self):
    """创建会话表"""
    self.conn.execute("""
    CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    created_at TIMESTAMP,
    last_updated TIMESTAMP,
    title TEXT,
    message_count INTEGER,
    history TEXT
    )
    """)

    def save_session(self, session_id, history, title=None):
    """保存会话到数据库"""
    history_json = json.dumps(history, ensure_ascii=False)
    now = datetime.now().isoformat()

    # 检查是否已存在
    cursor = self.conn.execute(
    "SELECT id FROM sessions WHERE id = ?",
    (session_id,)
    )

    if cursor.fetchone():
    # 更新现有会话
    self.conn.execute("""
    UPDATE sessions
    SET last_updated = ?, history = ?, message_count = ?
    WHERE id = ?
    """, (now, history_json, len(history), session_id))
    else:
    # 插入新会话
    self.conn.execute("""
    INSERT INTO sessions
    (id, created_at, last_updated, title, message_count, history)
    VALUES (?, ?, ?, ?, ?, ?)
    """, (session_id, now, now, title, len(history), history_json))

    self.conn.commit()
    return True

    def load_session(self, session_id):
    """从数据库加载会话"""
    cursor = self.conn.execute(
    "SELECT history FROM sessions WHERE id = ?",
    (session_id,)
    )
    row = cursor.fetchone()

    if row:
    return json.loads(row[0])
    return []

    def list_sessions(self, limit=50):
    """列出所有会话"""
    cursor = self.conn.execute("""
    SELECT id, created_at, last_updated, title, message_count
    FROM sessions
    ORDER BY last_updated DESC
    LIMIT ?
    """, (limit,))

    return [
    {
    "id": row[0],
    "created": row[1],
    "updated": row[2],
    "title": row[3] or "未命名会话",
    "count": row[4]
    }
    for row in cursor.fetchall()
    ]

    def close(self):
    """关闭数据库连接"""
    self.conn.close()

    # 使用示例
    db = SessionDB()
    db.save_session("test_session", [["你好", "你好"]], "测试会话")
    sessions = db.list_sessions()
    db.close()

    7.2 会话标题自动生成

    为了让会话更容易识别,可以添加自动生成标题的功能:

    def generate_session_title(history):
    """根据对话历史自动生成会话标题"""
    if not history:
    return "新会话"

    # 取前3轮对话作为标题生成依据
    sample_text = ""
    for i, (user_msg, _) in enumerate(history[:3]):
    sample_text += user_msg + " "

    # 简单规则:取第一句话的前20个字符
    if history and history[0]:
    first_message = history[0][0]
    if len(first_message) > 20:
    return first_message[:20] + "…"
    return first_message

    return "未命名会话"

    # 在保存会话时使用
    title = generate_session_title(history)
    save_session_with_title(history, session_id, title)

    7.3 性能优化建议

  • 延迟保存:不要每次对话都立即保存,可以设置一个时间间隔或对话轮数阈值
  • 增量保存:只保存新增的对话内容,而不是整个历史
  • 压缩存储:对历史记录进行压缩,减少磁盘占用
  • 内存缓存:在内存中缓存最近使用的会话,减少文件读取
  • import gzip
    import pickle

    def save_session_compressed(history, session_id):
    """压缩保存会话"""
    filename = get_session_filename(session_id)
    compressed_data = gzip.compress(pickle.dumps(history))

    with open(filename, 'wb') as f:
    f.write(compressed_data)

    print(f"会话已压缩保存: {filename} (原始大小: {len(str(history))}B, 压缩后: {len(compressed_data)}B)")

    7.4 安全考虑

  • 输入验证:对用户输入进行适当的清理和验证
  • 文件权限:确保会话文件有适当的权限设置
  • 大小限制:限制单个会话的大小,防止恶意攻击
  • 备份机制:定期备份重要的会话数据
  • import re

    def sanitize_input(text):
    """清理用户输入,防止路径遍历等攻击"""
    # 移除可能的路径遍历字符
    text = re.sub(r'\\.\\./', '', text)
    text = re.sub(r'[<>:"|?*]', '', text)
    # 限制长度
    if len(text) > 10000:
    text = text[:10000] + "…[已截断]"
    return text

    8. 总结

    通过本文的详细讲解,你已经掌握了如何为Nanbeige4.1-3B的WebUI添加历史会话持久化功能。让我们回顾一下关键要点:

    8.1 核心修改总结

  • 找到了会话存储的关键位置:在chat_fn函数中的history参数
  • 实现了JSON文件存储:简单易用,无需额外依赖
  • 添加了会话管理界面:让用户可以方便地保存、加载、管理会话
  • 保持了原有功能:所有修改都是增量式的,不影响原有的聊天功能
  • 8.2 实际效果

    修改后的WebUI具有以下优势:

    • 数据不丢失:重启服务或刷新页面后,对话历史依然存在
    • 会话可管理:可以创建多个独立的会话,互不干扰
    • 操作简便:自动保存+手动保存双保险,用户无需关心底层实现
    • 易于扩展:基于文件的存储方案可以轻松迁移到数据库

    8.3 进一步优化方向

    如果你对这个功能有更高的要求,可以考虑:

  • 添加搜索功能:让用户能够搜索历史会话中的内容
  • 支持导出导入:将会话导出为Markdown、PDF等格式
  • 添加标签分类:为会话添加标签,方便分类管理
  • 实现云端同步:将会话同步到云端,多设备访问
  • 8.4 最后的建议

    在实际部署时,建议你先在测试环境验证所有功能,确保稳定后再应用到生产环境。同时,记得定期备份chat_sessions目录,防止数据丢失。

    历史会话持久化虽然是一个看似简单的功能,但它极大地提升了用户体验。现在,你可以放心地与Nanbeige4.1-3B进行长时间的深度对话,所有的思考过程、灵感火花都会被完整地保存下来,成为你宝贵的知识资产。

    希望这个教程对你有所帮助。如果你在实施过程中遇到任何问题,或者有更好的改进建议,欢迎在实践中不断优化这个方案。技术总是在不断迭代中进步的,最重要的是开始行动并持续改进。


    获取更多AI镜像

    想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。

    赞(0)
    未经允许不得转载:171主机测评 » Nanbeige4.1-3B实操手册:webui.py源码关键修改点——支持历史会话持久化
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址