当前位置：首页 > news >正文

Nanbeige 4.1-3B保姆级教学：为像素终端添加离线模式与本地缓存机制

news 2026/3/27 1:54:43

Nanbeige 4.1-3B保姆级教学：为像素终端添加离线模式与本地缓存机制

1. 项目背景与目标

Nanbeige 4.1-3B像素冒险聊天终端是一款充满复古游戏风格的AI对话前端，但当前版本存在一个明显短板：完全依赖网络连接。本教程将手把手教你实现两大核心功能升级：

离线模式：在网络不稳定或完全断网时仍能使用基础对话功能
本地缓存：自动保存对话历史，避免意外刷新导致聊天记录丢失

完成改造后，你的像素终端将获得以下增强能力：

断网时可继续查看历史对话
网络恢复后自动同步未发送的消息
浏览器关闭后仍保留完整对话记录
显著降低API调用次数

2. 环境准备与改造规划

2.1 所需工具清单

已部署的Nanbeige 4.1-3B像素终端（基于Streamlit）
Python 3.8+
新增依赖库：
```
pip install localstorage cachetools
```

2.2 架构改造示意图

原架构： [前端] → [网络] → [Nanbeige API] 新架构： [前端] → [本地缓存] → [离线队列] ↗ [网络恢复时] → [批量同步]

3. 分步实现离线功能

3.1 检测网络状态

在main.py中添加网络检测模块：

import requests def check_online(): try: requests.get('https://api.nanbeige.cn/health', timeout=3) return True except: return False

3.2 创建离线消息队列

在Streamlit session state中初始化队列：

if 'offline_queue' not in st.session_state: st.session_state.offline_queue = [] if 'is_online' not in st.session_state: st.session_state.is_online = check_online()

4. 实现本地缓存系统

4.1 使用localstorage保存对话

创建cache_manager.py：

from localstorage import LocalStorage class ChatCache: def __init__(self): self.storage = LocalStorage() self.CACHE_KEY = "nanbeige_chat_history" def save(self, messages): self.storage.setItem(self.CACHE_KEY, json.dumps(messages)) def load(self): data = self.storage.getItem(self.CACHE_KEY) return json.loads(data) if data else []

4.2 集成到主对话流程

修改消息处理逻辑：

cache = ChatCache() def handle_message(user_input): if st.session_state.is_online: # 正常API调用 response = api_call(user_input) st.session_state.messages.append(response) cache.save(st.session_state.messages) else: # 离线模式处理 offline_msg = {"user": user_input, "status": "pending"} st.session_state.offline_queue.append(offline_msg) st.warning("当前处于离线模式，消息将在网络恢复后发送")

5. 网络恢复自动同步

添加定时检查任务：

from cachetools import TTLCache sync_cache = TTLCache(maxsize=100, ttl=300) def sync_offline_messages(): if st.session_state.offline_queue and st.session_state.is_online: for msg in list(st.session_state.offline_queue): try: response = api_call(msg["user"]) st.session_state.messages.append(response) st.session_state.offline_queue.remove(msg) except: break cache.save(st.session_state.messages)

在Streamlit中每30秒检查一次：

import time if 'last_sync' not in st.session_state: st.session_state.last_sync = 0 current_time = time.time() if current_time - st.session_state.last_sync > 30: sync_offline_messages() st.session_state.last_sync = current_time

6. 界面适配改造

6.1 添加网络状态指示器

在侧边栏添加状态显示：

online_status = "🟢 在线" if st.session_state.is_online else "🔴 离线" st.sidebar.markdown(f"**网络状态:** {online_status}")

6.2 离线消息提醒

修改消息显示逻辑：

for msg in st.session_state.messages: if msg.get("status") == "pending": with st.chat_message("user", avatar="⏳"): st.write(f"[待发送] {msg['content']}") else: # 正常显示消息

7. 完整代码整合

创建offline_manager.py整合所有功能：

import json import streamlit as st from localstorage import LocalStorage from cachetools import TTLCache import requests class OfflineManager: def __init__(self): self.storage = LocalStorage() self.sync_cache = TTLCache(maxsize=100, ttl=300) def check_online(self): try: requests.get('https://api.nanbeige.cn/health', timeout=3) return True except: return False def save_chat(self, messages): self.storage.setItem("nanbeige_chat", json.dumps(messages)) def load_chat(self): data = self.storage.getItem("nanbeige_chat") return json.loads(data) if data else [] def sync_messages(self, offline_queue): success_count = 0 for msg in list(offline_queue): try: response = api_call(msg["content"]) offline_queue.remove(msg) success_count += 1 except: continue return success_count