实战-AI智能伴侣-会话管理
实战-AI智能伴侣-会话管理
把会话存进文件实现"关掉网页还能接着聊":新建/保存/加载/删除会话,以及 os、json、datetime 的配合使用
要解决的问题
前面几节的聊天记录只存在 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.json为什么要保存 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 osimport 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 已废弃) |
with st.sidebar: 用的是 Python 的上下文管理器语法。它和 with open(...) 是同一个语法机制:进入代码块时切到侧边栏,退出时切回主区域。同理 with col1: 表示”接下来的组件放进第一列”。
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 stimport osfrom openai import OpenAIfrom datetime import datetimeimport 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("删除会话失败!")
# 大标题 / Logost.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()、____()、____()、____()、____()(保存 / 列出 / 加载 / 删除) - 每个会话文件里保存四样东西:
messages、、、____ - 保存会话:
json.____(session_data, f, ensure_ascii=False, indent=2) - 存盘前先判断并创建目录:
if not os.path.____("sessions"): os.____("sessions") - 列出历史会话:
os.____("sessions"),再用filename.____(".json")筛出会话文件 - 让当前会话的按钮高亮:
type="primary" if session == st.session_state.____ else "secondary" - 循环里给每个按钮加
key=f"load_{session}",是为了避免 ____ 报错 - 数据变了要立刻刷新页面:
st.____() - 删除的正好是当前会话时,还要把
st.session_state.messages清空并把current_session换成 ____
填空答案(做完再点开)
- save_session、load_sessions、load_session、delete_session 2.
nick_name、nature、current_session3. dump 4. exists / mkdir 5. listdir / endswith 6. current_session 7. DuplicateWidgetID(key 重复) 8. rerun 9. 新的会话标识
二、裸写题
-
2-1 保存会话 写
save_session():把 messages、昵称、性格、会话标识打包成字典,存进sessions/<会话标识>.json(目录不存在先创建,中文不转义、有缩进)。提示(先自己想,实在想不出再点开)一级 · 思路:拼字典 → 确保目录存在 → 用 json 写文件 二级 · 方法:
os.path.exists/os.mkdir/json.dump/with open(...)三级 · 骨架:文件名用f"sessions/{st.session_state.current_session}.json"参考答案(做完再点开)import osimport jsonimport streamlit as stfrom datetime import datetimedef 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) -
2-2 加载历史会话 写
load_sessions()(返回所有会话标识列表,按倒序)和load_session(name)(把读到的数据写回st.session_state),并用 try/except 包住、失败时提示。提示一级 · 思路:列目录得到”有哪些”,读单个文件得到”内容是什么” 二级 · 方法:
os.listdir/json.load/session_list.sort(reverse=True)三级 · 骨架:st.session_state.messages = session_data["____"]参考答案(做完再点开)import osimport jsonimport streamlit as stdef 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_listdef 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_nameexcept Exception:st.error("加载会话失败!") -
2-3 删除会话(含边界情况) 写
delete_session(name):删掉对应文件;如果删的正好是当前会话,还要清空消息列表并生成新的会话标识。提示一级 · 思路:先删文件,再判断”删的是不是当前会话” 二级 · 方法:
os.remove/ 条件判断 /generate_session_name()三级 · 骨架:if session_name == st.session_state.____:参考答案(做完再点开)import osimport streamlit as stfrom datetime import datetimedef 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("删除会话失败!")
评论区
如果你喜欢,那么欢迎来到我的世界!
了解更多













