要解决的问题
前面几节的聊天记录只存在 st.session_state 里——内存中存放的数据在计算机关机后就会消失,要永久保存数据,就需要将数据保存在文件中。
于是侧边栏要提供四个功能:
| 功能 | 说明 |
|---|---|
| 新建会话 | 保存当前会话 → 清空聊天 → 生成新的会话标识 |
| 保存会话 | 每轮对话结束后自动保存一份 |
| 加载历史会话 | 点历史列表里的某一项,把它的内容读回来 |
| 删除历史会话 | 删掉对应的文件 |
每个会话文件存什么
思考题:每一个历史会话文件中,需要保存哪些信息呢?
| 字段 | 含义 |
|---|---|
messages | 交互消息(完整的对话历史,会话记忆的基础) |
nick_name | 昵称 |
nature | 性格 |
current_session | 会话标识(名字,唯一) |
存成 JSON 文件的样子(sessions/2026-01-11_18-00-05.json):
{
"messages": [
{ "role": "user", "content": "你好" },
{ "role": "assistant", "content": "哈囉~今天過得怎麼樣呀?💕" }
],
"nick_name": "小美",
"nature": "温柔可爱的台湾腔姑娘",
"current_session": "2026-01-09_16-05-20"
}文件目录结构:
sessions
|- 2026-01-11_18-00-05.json
|- 2026-01-11_18-04-42.json
|- 2026-01-11_18-08-56.json
|- 2026-01-11_18-14-25.json
|- 2026-01-11_18-45-08.jsonTIP
为什么要保存 nick_name 和 nature?因为它们是生成 system 提示词的参数。只恢复聊天记录、不恢复伴侣设定的话,性格就变了——同一份对话换个性格继续聊,会很割裂。
五个核心函数
1. 生成会话标识
from datetime import datetime
def generate_session_name():
return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")2. 保存会话
import os
import json
def save_session():
if st.session_state.current_session:
# 构建新的会话对象
session_data = {
"nick_name": st.session_state.nick_name,
"nature": st.session_state.nature,
"current_session": st.session_state.current_session,
"messages": st.session_state.messages
}
# 如果 sessions 目录不存在, 则创建
if not os.path.exists("sessions"):
os.mkdir("sessions")
# 保存会话数据
with open(f"sessions/{st.session_state.current_session}.json", "w", encoding="utf-8") as f:
json.dump(session_data, f, ensure_ascii=False, indent=2)3. 加载会话列表
def load_sessions():
session_list = []
# 加载sessions目录下的文件
if os.path.exists("sessions"):
file_list = os.listdir("sessions")
for filename in file_list:
if filename.endswith(".json"):
session_list.append(filename[:-5]) # 去掉末尾的 .json
session_list.sort(reverse=True) # 排序, 降序排列(最新的在最上面)
return session_list4. 加载指定会话
def load_session(session_name):
try:
if os.path.exists(f"sessions/{session_name}.json"):
# 读取会话数据
with open(f"sessions/{session_name}.json", "r", encoding="utf-8") as f:
session_data = json.load(f)
st.session_state.messages = session_data["messages"]
st.session_state.nick_name = session_data["nick_name"]
st.session_state.nature = session_data["nature"]
st.session_state.current_session = session_name
except Exception:
st.error("加载会话失败!")5. 删除会话
def delete_session(session_name):
try:
if os.path.exists(f"sessions/{session_name}.json"):
os.remove(f"sessions/{session_name}.json") # 删除文件
# 如果删除的是当前会话, 则需要更新消息列表
if session_name == st.session_state.current_session:
st.session_state.messages = []
st.session_state.current_session = generate_session_name()
except Exception:
st.error("删除会话失败!")侧边栏界面
with st.sidebar: # with 是 streamlit 中的上下文管理器
st.subheader("AI控制面板")
# 新建会话
if st.button("新建会话", width="stretch", icon="✏️"):
# 1. 保存当前会话信息
save_session()
# 2. 创建新的会话
if st.session_state.messages: # 如果聊天信息非空, True; 否则, False
st.session_state.messages = []
st.session_state.current_session = generate_session_name()
save_session()
st.rerun() # 重新运行当前页面
# 会话历史
st.text("会话历史")
session_list = load_sessions()
for session in session_list:
col1, col2 = st.columns([4, 1])
with col1:
# 加载会话信息
# 三元运算符: 值1 if 条件 else 值2
if st.button(session, width="stretch", icon="📄", key=f"load_{session}",
type="primary" if session == st.session_state.current_session else "secondary"):
load_session(session)
st.rerun()
with col2:
# 删除会话信息
if st.button("", width="stretch", icon="❌️", key=f"delete_{session}"):
delete_session(session)
st.rerun()
st.divider()
# 伴侣信息
st.subheader("伴侣信息")
nick_name = st.text_input("昵称", placeholder="请输入昵称", value=st.session_state.nick_name)
if nick_name:
st.session_state.nick_name = nick_name
nature = st.text_area("性格", placeholder="请输入性格", value=st.session_state.nature)
if nature:
st.session_state.nature = nature关键点
| 代码 | 为什么必须这么写 |
|---|---|
key=f"load_{session}" | Streamlit 里每个按钮都需要唯一的 key。循环生成的按钮如果 key 相同会报 DuplicateWidgetID,把会话名拼进去即可保证唯一 |
type="primary" if ... else "secondary" | 三元运算符:当前会话的按钮高亮成主色,一眼能看出"现在在哪一个会话里" |
st.rerun() | 按钮点击后数据已改变,需要让脚本立刻重跑一遍,把新状态渲染出来 |
st.columns([4, 1]) | 把一行按 4:1 分成两列:左边放会话名按钮(宽),右边放删除按钮(窄) |
width="stretch" | 按钮宽度撑满所在列(旧写法 use_container_width=True 已废弃) |
NOTE
with st.sidebar: 用的是 Python 的上下文管理器语法。它和 with open(...) 是同一个语法机制:进入代码块时切到侧边栏,退出时切回主区域。同理 with col1: 表示"接下来的组件放进第一列"。
WARNING
save_session() 里判断目录要用 os.path.exists("sessions"),不能每次都 os.mkdir——目录已存在时 os.mkdir 会抛 FileExistsError。
数据流转全景
启动 ──> 初始化 session_state(messages / nick_name / nature / current_session)
│
├─ 侧边栏点「新建会话」 ──> save_session() 存盘 ──> 清空 ──> 新标识 ──> st.rerun()
├─ 侧边栏点某条历史 ──> load_session() 读盘 ──> 覆盖 session_state ──> st.rerun()
├─ 侧边栏点 ❌ ──> delete_session() 删文件 ──> st.rerun()
└─ 输入框发消息 ──> 调用大模型(流式) ──> append 进 messages ──> save_session()完整代码(最终版)
import streamlit as st
import os
from openai import OpenAI
from datetime import datetime
import json
# 设置页面的配置项
st.set_page_config(
page_title="AI智能伴侣",
page_icon="🤖",
layout="wide",
initial_sidebar_state="expanded",
menu_items={}
)
# 生成会话标识函数
def generate_session_name():
return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
# 保存会话信息函数
def save_session():
if st.session_state.current_session:
session_data = {
"nick_name": st.session_state.nick_name,
"nature": st.session_state.nature,
"current_session": st.session_state.current_session,
"messages": st.session_state.messages
}
if not os.path.exists("sessions"):
os.mkdir("sessions")
with open(f"sessions/{st.session_state.current_session}.json", "w", encoding="utf-8") as f:
json.dump(session_data, f, ensure_ascii=False, indent=2)
# 加载所有的会话列表信息
def load_sessions():
session_list = []
if os.path.exists("sessions"):
file_list = os.listdir("sessions")
for filename in file_list:
if filename.endswith(".json"):
session_list.append(filename[:-5])
session_list.sort(reverse=True)
return session_list
# 加载指定的会话信息
def load_session(session_name):
try:
if os.path.exists(f"sessions/{session_name}.json"):
with open(f"sessions/{session_name}.json", "r", encoding="utf-8") as f:
session_data = json.load(f)
st.session_state.messages = session_data["messages"]
st.session_state.nick_name = session_data["nick_name"]
st.session_state.nature = session_data["nature"]
st.session_state.current_session = session_name
except Exception:
st.error("加载会话失败!")
# 删除会话信息函数
def delete_session(session_name):
try:
if os.path.exists(f"sessions/{session_name}.json"):
os.remove(f"sessions/{session_name}.json")
if session_name == st.session_state.current_session:
st.session_state.messages = []
st.session_state.current_session = generate_session_name()
except Exception:
st.error("删除会话失败!")
# 大标题 / Logo
st.title("AI智能伴侣")
st.logo("resources/logo.png")
# 系统提示词
system_prompt = """
你叫 %s,现在是用户的真实伴侣,请完全代入伴侣角色。
规则:
1. 每次只回1条消息
2. 禁止任何场景或状态描述性文字
3. 匹配用户的语言
4. 回复简短,像微信聊天一样
5. 有需要的话可以用❤️🌸等emoji表情
6. 用符合伴侣性格的方式对话
7. 回复的内容, 要充分体现伴侣的性格特征
伴侣性格:
- %s
你必须严格遵守上述规则来回复用户。
"""
# 初始化会话状态
if "messages" not in st.session_state:
st.session_state.messages = []
if "nick_name" not in st.session_state:
st.session_state.nick_name = "小甜甜"
if "nature" not in st.session_state:
st.session_state.nature = "活泼开朗的东北姑娘"
if "current_session" not in st.session_state:
st.session_state.current_session = generate_session_name()
# 展示聊天信息
st.text(f"会话名称: {st.session_state.current_session}")
for message in st.session_state.messages:
st.chat_message(message["role"]).write(message["content"])
# 创建与AI大模型交互的客户端对象
client = OpenAI(api_key=os.environ.get('DEEPSEEK_API_KEY'), base_url="https://api.deepseek.com")
# 左侧的侧边栏
with st.sidebar:
st.subheader("AI控制面板")
# 新建会话
if st.button("新建会话", width="stretch", icon="✏️"):
save_session()
if st.session_state.messages:
st.session_state.messages = []
st.session_state.current_session = generate_session_name()
save_session()
st.rerun()
# 会话历史
st.text("会话历史")
session_list = load_sessions()
for session in session_list:
col1, col2 = st.columns([4, 1])
with col1:
if st.button(session, width="stretch", icon="📄", key=f"load_{session}",
type="primary" if session == st.session_state.current_session else "secondary"):
load_session(session)
st.rerun()
with col2:
if st.button("", width="stretch", icon="❌️", key=f"delete_{session}"):
delete_session(session)
st.rerun()
st.divider()
# 伴侣信息
st.subheader("伴侣信息")
nick_name = st.text_input("昵称", placeholder="请输入昵称", value=st.session_state.nick_name)
if nick_name:
st.session_state.nick_name = nick_name
nature = st.text_area("性格", placeholder="请输入性格", value=st.session_state.nature)
if nature:
st.session_state.nature = nature
# 消息输入框
prompt = st.chat_input("请输入您要问的问题")
if prompt:
st.chat_message("user").write(prompt)
st.session_state.messages.append({"role": "user", "content": prompt})
# 调用AI大模型(流式)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": system_prompt % (st.session_state.nick_name, st.session_state.nature)},
*st.session_state.messages
],
stream=True
)
response_message = st.empty()
full_response = ""
for chunk in response:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
response_message.chat_message("assistant").write(full_response)
st.session_state.messages.append({"role": "assistant", "content": full_response})
# 保存会话信息
save_session()本章总结
| 知识块 | 内容 |
|---|---|
| Streamlit 构建页面 | 页面配置、常用组件、侧边栏、列布局、session_state |
| 文件基本操作 | open / read / write / close、with 上下文管理器 |
| 大模型交互方案 | openai 库 + DeepSeek API、流式输出 |
| 大模型会话记忆方案 | messages 滚雪球 |
| 大模型部署方案 | 本地部署 / 官方开放 API / 云服务平台 |
| HTTP 协议 | 请求与响应的数据格式 |
| json 操作 | dump / load 序列化与反序列化 |
| os / datetime 模块 | 目录与文件判断、会话标识生成 |
| 伴侣性格定制 | system 提示词模板 + %s 占位符 |
相关
练习题
一、知识回顾(读完直接做下面的实践题)
- 五个核心函数:
generate_session_name()、save_session()、load_sessions()、load_session()、delete_session()(保存 / 列出 / 加载 / 删除) - 每个会话文件里保存四样东西:
messages、nick_name、nature、current_session - 保存会话:
json.dump(session_data, f, ensure_ascii=False, indent=2) - 存盘前先判断并创建目录:
if not os.path.exists("sessions"): os.mkdir("sessions") - 列出历史会话:
os.listdir("sessions"),再用filename.endswith(".json")筛出会话文件 - 让当前会话的按钮高亮:
type="primary" if session == st.session_state.current_session else "secondary" - 循环里给每个按钮加
key=f"load_{session}",是为了避免 DuplicateWidgetID(key 重复) 报错 - 数据变了要立刻刷新页面:
st.rerun() - 删除的正好是当前会话时,还要把
st.session_state.messages清空并把current_session换成新的会话标识
二、裸写题
- 提示(先自己想,实在想不出再点开)
一级 · 思路:拼字典 → 确保目录存在 → 用 json 写文件 二级 · 方法:
os.path.exists/os.mkdir/json.dump/with open(...);函数名save_session()三级 · 骨架:文件名用f"sessions/{st.session_state.current_session}.json"- 参考答案(做完再点开)
pythonimport os import json import streamlit as st from datetime import datetime def save_session(): if st.session_state.current_session: # 拼出要保存的四样东西 session_data = { "nick_name": st.session_state.nick_name, "nature": st.session_state.nature, "current_session": st.session_state.current_session, "messages": st.session_state.messages, } # 目录不存在就创建 if not os.path.exists("sessions"): os.mkdir("sessions") # 写盘:中文不转义、带缩进 with open(f"sessions/{st.session_state.current_session}.json", "w", encoding="utf-8") as f: json.dump(session_data, f, ensure_ascii=False, indent=2)- 提示(先自己想,实在想不出再点开)
一级 · 思路:列目录得到"有哪些",读单个文件得到"内容是什么" 二级 · 方法:函数名
load_sessions()/load_session(name);os.listdir/json.load/session_list.sort(reverse=True)三级 · 骨架:st.session_state.messages = session_data["____"]三级 · 骨架:st.session_state.messages = session_data["____"]- 参考答案(做完再点开)
pythonimport os import json import streamlit as st def load_sessions(): """返回所有会话标识,倒序(最新的在最前)""" session_list = [] if os.path.exists("sessions"): for filename in os.listdir("sessions"): if filename.endswith(".json"): session_list.append(filename[:-5]) session_list.sort(reverse=True) return session_list def load_session(session_name): try: if os.path.exists(f"sessions/{session_name}.json"): with open(f"sessions/{session_name}.json", "r", encoding="utf-8") as f: session_data = json.load(f) st.session_state.messages = session_data["messages"] st.session_state.nick_name = session_data["nick_name"] st.session_state.nature = session_data["nature"] st.session_state.current_session = session_name except Exception: st.error("加载会话失败!")- 提示(先自己想,实在想不出再点开)
一级 · 思路:先删文件,再判断"删的是不是当前会话" 二级 · 方法:函数名
delete_session(name);os.remove/ 条件判断 /generate_session_name()三级 · 骨架:if session_name == st.session_state.____:- 参考答案(做完再点开)
pythonimport os import streamlit as st from datetime import datetime def generate_session_name(): return datetime.now().strftime("%Y-%m-%d_%H-%M-%S") def delete_session(session_name): try: if os.path.exists(f"sessions/{session_name}.json"): os.remove(f"sessions/{session_name}.json") # 删的正好是当前会话:清空消息 + 换一个新标识 if session_name == st.session_state.current_session: st.session_state.messages = [] st.session_state.current_session = generate_session_name() except Exception: st.error("删除会话失败!")
