一个基于Streamlit​框架开发的本地化智能助手系统

一、程序概述与核心定位

一个基于Streamlit​框架开发的本地化智能助手系统，集成了身份认证、多格式文档处理、上下文感知对话、会话管理等功能，支持用户通过上传 PDF/TXT/DOCX 文档构建私有知识库，并基于大语言模型（LLM）实现文档问答与通用对话。其核心价值在于：将通用 LLM 能力与本地私有文档结合，同时通过轻量级身份验证保障访问安全，适用于实验室、小团队的内部知识管理与智能问答场景。

程序采用模块化设计，分为配置初始化、身份验证、客户端配置、文档处理、侧边栏管理、聊天界面六大功能模块，通过 Streamlit 的 session_state实现会话级数据持久化，整体架构轻量但完整，可直接部署于本地服务器或单机环境。

二、功能模块详解

（一）配置与初始化模块

1. 页面基础配置

st.set_page_config(
page_title="labAI 智能助手",
layout="wide",
initial_sidebar_state="expanded"
)

• 作用：设置 Streamlit 应用的全局属性，layout="wide"启用宽屏布局，适配多列组件；initial_sidebar_state="expanded"默认展开侧边栏，便于用户操作文档和查看统计。

• 技术细节：Streamlit 的 set_page_config必须在其他 Streamlit 命令前调用，此处作为程序入口的第一行配置，确保页面样式统一。

（二）身份验证模块

1. 用户凭证加载（load_users函数）

def load_users():
users = {}
try:
if os.path.exists("auth_users.txt"):
with open("auth_users.txt", "r") as f:
for line in f:
if ":" in line:
user, pwd = line.strip().split(":", 1)
users[user] = pwd
except Exception as e:
st.error(f"凭证加载失败: {str(e)}")
return users

• 功能：从本地文件 auth_users.txt加载用户凭证，文件格式为 用户名:密码（每行一条）。

• 设计逻辑：采用本地文件存储凭证（非数据库），简化部署复杂度，适合小团队场景；通过 try-except捕获文件不存在或格式错误异常，避免程序崩溃。

• 局限性：明文存储密码，未做哈希加密，生产环境需优化（如添加 hashlib加密）。

2. 登录界面（login_form函数）

def login_form():
st.title("lab 身份验证")
with st.form("auth"):
user = st.text_input("工号/学号")
password = st.text_input("密码", type="password")
if st.form_submit_button("登录"):
users = load_users()
if users.get(user) == password:
st.session_state.logged_in = True
st.session_state.user = user
st.rerun()
else:
st.error("验证失败，请检查凭证")
st.stop()

• 交互流程：通过 Streamlit 表单（st.form）封装输入框和提交按钮，提交后校验凭证：若匹配成功，设置 st.session_state.logged_in = True并记录用户名，通过 st.rerun()刷新页面进入主界面；失败则提示错误。

• 关键机制：st.stop()确保未登录用户无法访问后续功能，强制停留在登录页。

3. 登录状态校验

if not st.session_state.get("logged_in"):
login_form()

• 作用：程序启动时的第一道关卡，通过 Streamlit 会话状态（session_state）判断用户是否已登录。session_state是 Streamlit 用于跨页面刷新存储数据的核心机制，此处实现“会话级身份验证”。

（三）客户端配置模块

client = openai.OpenAI(
base_url="",
api_key="",
http_client=httpx.Client()
)

• 功能：初始化 OpenAI 兼容的 API 客户端，用于调用大语言模型。

• 参数说明：

  • base_url：LLM 服务地址（留空需用户手动填写，可对接本地部署的 QwQ-32B 或其他兼容 OpenAI API 的模型）；

  • api_key：API 密钥（实际使用时需填入真实密钥）；

  • http_client=httpx.Client()：自定义 HTTP 客户端，支持超时、代理等高级配置。

• 扩展性：通过修改 base_url，可无缝切换至不同 LLM 服务（如 OpenAI、本地 Ollama 服务等）。

（四）文档处理模块（核心功能之一）

1. 通用文本提取函数（extract_text）

def extract_text(file):
try:
# 临时文件创建
suffix = { … }.get(file.type, "")
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
tmp.write(file.getvalue())
tmp_path = tmp.name

# 按类型解析
if file.type == "application/pdf":
reader = PdfReader(tmp_path)
text = "\\n".join([p.extract_text() for p in reader.pages])
elif file.type == "text/plain":
with open(tmp_path, "r", encoding="utf-8") as f:
text = f.read()
elif file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
doc = Document(tmp_path)
text = "\\n".join([p.text for p in doc.paragraphs])
else:
return None

os.unlink(tmp_path) # 清理临时文件
return text[:15000] # 限制长度
except Exception as e:
st.error(f"文档解析失败: {str(e)}")
return None

• 功能：支持 PDF、TXT、DOCX 三种格式的文档文本提取，返回纯文本内容。

• 实现逻辑：

• 临时文件处理：由于 PyPDF2、python-docx 等库需文件路径而非字节流，通过 tempfile.NamedTemporaryFile创建临时文件，写入上传文件的二进制数据，避免磁盘残留。

• 格式适配：

  • PDF：使用 PyPDF2.PdfReader逐页提取文本，合并为字符串；

  • TXT：直接读取 UTF-8 编码文本；

  • DOCX：通过 python-docx.Document遍历段落（paragraphs）提取文本。

• 安全限制：返回文本截断至前15000字符，防止过长上下文导致 LLM 调用超时或超 token 限制。

• 异常处理：捕获所有解析异常，通过 st.error提示用户，避免程序崩溃。

（五）侧边栏管理模块

侧边栏是用户与程序交互的核心入口，集成文档管理、会话统计、系统控制三大功能：

def sidebar_manager():
with st.sidebar:
# 用户信息
st.write(f"用户：`{st.session_state.user}`")

# 文档管理（上传、展示、删除）
st.header("文档管理")
uploaded_files = st.file_uploader(…) # 多文件上传组件

# 处理新上传文件（提取文本并存入 session_state）
if uploaded_files:
for f in uploaded_files:
if f.name not in [d["name"] for d in st.session_state.get("docs", [])]:
content = extract_text(f)
if content:
st.session_state.docs.append({…}) # 存储文档元数据

# 显示文档列表（含删除按钮）
if st.session_state.get("docs"):
for doc in st.session_state.docs:
cols = st.columns([6, 2])
cols[0].caption(f"▪ {doc['name']} ({doc['size'] // 1024}KB)")
if cols[1].button("×", key=f"del_{doc['name']}"):
st.session_state.docs = [d for d in st.session_state.docs if d["name"] != doc["name"]]
st.rerun()

# 对话统计
st.header("对话统计")
msg_counts = {
"user": len([m for m in st.session_state.messages if m["role"] == "user"]),
"assistant": len([m for m in st.session_state.messages if m["role"] == "assistant"])
}
st.metric("用户消息", msg_counts["user"])
st.metric("助手回复", msg_counts["assistant"])

# 管理功能（清空对话、退出登录）
st.header("⚙管理")
if st.button("清空对话"):
st.session_state.messages = [st.session_state.messages[0]] if st.session_state.messages else []
st.rerun()
if st.button("退出登录"):
st.session_state.clear()
st.rerun()

• 文档管理：

  • 去重机制：通过检查文件名是否已存在于 st.session_state.docs，避免重复上传同一文档；

  • 动态更新：删除文档时通过 st.rerun()实时刷新界面，确保状态同步。

• 会话统计：基于 st.session_state.messages统计用户与助手的消息数量，直观展示对话活跃度。

• 会话控制：清空对话保留系统提示（第一条消息），退出登录清除所有会话状态，返回登录页。

（六）聊天界面模块（核心功能之二）

1. 上下文提示构建（build_prompt函数）

def build_prompt(user_input):
context = []
if st.session_state.get("docs"):
context.append("## 参考文档")
for i, doc in enumerate(st.session_state.docs, 1):
context.append(f"【文档{i}】{doc['name']}\\n内容摘要：{doc['content'][:2000]}…")
context.append(f"## 用户问题\\n{user_input}")
return "\\n\\n".join(context)

• 作用：将用户问题与上传文档内容拼接为 LLM 可理解的上下文提示，实现“基于文档的问答”。

• 设计逻辑：

  • 若存在上传文档，优先插入文档摘要（截断至前 2000 字符，避免超 token 限制）；

  • 若无文档，仅保留用户问题，由 LLM 使用通用知识库回答。

• 优势：通过结构化提示（Markdown 标题分隔），引导 LLM 区分“参考文档”与“用户问题”，提升回答准确性。

2. 系统提示设置（可配置）

with st.expander("⚙ 系统设置", expanded=False):
default_prompt = """你是一个学术助手，根据以下规则响应：
1. 当存在参考文档时，严格基于文档内容回答
2. 没有文档时使用通用知识库
3. 回答语言与提问语言一致"""
sys_prompt = st.text_area("系统指令", value=…, height=150)
# 快捷模式按钮（恢复默认、文档模式、通用模式）
if cols[0].button("恢复默认"): …
if cols[1].button("文档模式"): …
if cols[2].button("通用模式"): …
if st.button("保存设置"):
st.session_state.messages[0]["content"] = sys_prompt # 更新系统提示

• 功能：允许用户自定义 LLM 的系统指令，控制回答风格。系统提示存储在 st.session_state.messages[0]（会话消息的第一条），确保每次调用 LLM 时生效。

3. 消息显示与用户输入

# 历史消息展示
for msg in st.session_state.messages[1:]:
role = msg["role"]
time = datetime.fromisoformat(msg["timestamp"]).strftime("%m/%d %H:%M")
with st.chat_message(role):
st.markdown(f"`[{time}]`\\n{msg['content']}")
if role == "user" and "docs_used" in msg:
st.caption(f"关联文档：{', '.join(msg['docs_used']) or '无'}")

# 用户输入处理
if prompt := st.chat_input("输入问题…"):
full_prompt = build_prompt(prompt)
user_msg = {
"role": "user",
"content": prompt,
"full_prompt": full_prompt, # 存储带上下文的完整提示
"timestamp": datetime.now().isoformat(),
"docs_used": [d["name"] for d in st.session_state.get("docs", [])]
}
st.session_state.messages.append(user_msg)

• 消息结构：每条消息包含 role（角色）、content（内容）、timestamp（时间戳）、full_prompt（完整上下文，仅用户消息）、docs_used（关联的文档名）。

• 用户体验：显示消息发送时间、关联文档，增强对话可追溯性。

4. LLM 调用与流式输出

response = client.chat.completions.create(
model="QwQ-32B", # 模型名称（可替换）
messages=[{"role": m["role"], "content": m["full_prompt"] if m["role"] == "user" else m["content"]}
for m in [st.session_state.messages[0]] + st.session_state.messages[1:]],
stream=True, # 流式输出
timeout=45 # 超时时间（秒）
)

# 流式渲染
response_text = ""
container = st.empty()
for chunk in response:
if chunk.choices[0].delta.content:
response_text += chunk.choices[0].delta.content
container.markdown(response_text + "▌") # 实时显示光标
container.markdown(response_text) # 最终输出

• 流式调用：通过 stream=True启用流式响应，逐块接收 LLM 输出并实时渲染，避免用户等待过久；

• 消息拼接：将系统提示（messages[0]）与历史对话拼接为完整上下文，确保 LLM 理解对话历史；

• 超时控制：设置 45 秒超时，防止因网络或模型响应慢导致的页面卡死。

三、数据结构设计

（一）会话状态（st.session_state）核心结构

键名

类型

描述

（二）临时数据结构

• 用户凭证字典：users = {"user1": "pwd1", "user2": "pwd2"}，由 load_users从文件加载；

• 文档解析临时路径：tmp_path，通过 tempfile.NamedTemporaryFile生成，使用后自动删除。

四、核心算法与逻辑

（一）文档去重算法

if f.name not in [d["name"] for d in st.session_state.get("docs", [])]:
# 处理新文档

• 逻辑：通过列表推导式遍历已上传文档的名称列表，判断新上传文件的文件名是否存在，实现简单高效的去重。时间复杂度 O(n)，适合小规模文档场景。

（二）上下文窗口管理

（三）流式输出渲染算法

response_text = ""
container = st.empty() # 创建空容器
for chunk in response:
if chunk.choices[0].delta.content:
response_text += chunk.choices[0].delta.content
container.markdown(response_text + "▌") # 追加光标
container.markdown(response_text) # 移除光标，显示最终结果

• 原理：利用 Streamlit 的 st.empty()创建可更新的容器，每接收到一个 chunk 就更新容器内容，模拟打字机效果，提升用户体验。

五、程序执行流程

六、技术特点与优化建议

（一）技术特点

（二）优化建议

七、总结

一个功能完整、架构清晰的本地化智能助手系统，核心价值在于将私有文档​与大语言模型​有机结合，同时通过轻量级身份验证保障安全。其设计兼顾了易用性与扩展性，既适合实验室内部知识管理，也可作为二次开发的基础框架。通过优化安全性和性能，可进一步应用于更复杂的场景（如企业知识库、教育辅助系统等）。

源代码

import streamlit as st
import openai
import httpx
import json
import os
import tempfile
from datetime import datetime
from PyPDF2 import PdfReader
from docx import Document

# ————————–
# 配置与初始化
# ————————–
st.set_page_config(
page_title="labAI 智能助手",
layout="wide",
initial_sidebar_state="expanded"
)

# ————————–
# 身份验证模块
# ————————–
def load_users():
"""加载用户凭证"""
users = {}
try:
if os.path.exists("auth_users.txt"):
with open("auth_users.txt", "r") as f:
for line in f:
if ":" in line:
user, pwd = line.strip().split(":", 1)
users[user] = pwd
except Exception as e:
st.error(f"凭证加载失败: {str(e)}")
return users

def login_form():
"""显示登录界面"""
st.title("lab 身份验证")
with st.form("auth"):
user = st.text_input("工号/学号")
password = st.text_input("密码", type="password")
if st.form_submit_button("登录"):
users = load_users()
if users.get(user) == password:
st.session_state.logged_in = True
st.session_state.user = user
st.rerun()
else:
st.error("验证失败，请检查凭证")
st.stop()

# ————————–
# 登录检查
# ————————–
if not st.session_state.get("logged_in"):
login_form()

# ————————–
# 客户端配置
# ————————–
client = openai.OpenAI(
base_url="",
api_key="",
http_client=httpx.Client()
)

# ————————–
# 文档处理模块
# ————————–
def extract_text(file):
"""通用文本提取函数"""
try:
# 创建临时文件
suffix = {
"application/pdf": ".pdf",
"text/plain": ".txt",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx"
}.get(file.type, "")

with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
tmp.write(file.getvalue())
tmp_path = tmp.name

# 按类型解析
if file.type == "application/pdf":
reader = PdfReader(tmp_path)
text = "\\n".join([p.extract_text() for p in reader.pages])
elif file.type == "text/plain":
with open(tmp_path, "r", encoding="utf-8") as f:
text = f.read()
elif file.type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
doc = Document(tmp_path)
text = "\\n".join([p.text for p in doc.paragraphs])
else:
return None

os.unlink(tmp_path) # 清理临时文件
return text[:15000] # 限制长度

except Exception as e:
st.error(f"文档解析失败: {str(e)}")
return None

# ————————–
# 侧边栏组件
# ————————–
def sidebar_manager():
"""侧边栏管理功能"""
with st.sidebar:
st.write(f"用户：`{st.session_state.user}`")

# 文档管理
st.header("文档管理")
uploaded_files = st.file_uploader(
"上传文档 (PDF/TXT/DOCX)",
type=["pdf", "txt", "docx"],
accept_multiple_files=True
)

# 处理新上传文件
if uploaded_files:
for f in uploaded_files:
if f.name not in [d["name"] for d in st.session_state.get("docs", [])]:
content = extract_text(f)
if content:
if "docs" not in st.session_state:
st.session_state.docs = []
st.session_state.docs.append({
"name": f.name,
"content": content,
"size": f.size,
"time": datetime.now().strftime("%m/%d %H:%M")
})

# 显示文档列表
if st.session_state.get("docs"):
st.subheader("已加载文档")
for doc in st.session_state.docs:
cols = st.columns([6, 2])
cols[0].caption(f"▪ {doc['name']} ({doc['size'] // 1024}KB)")
if cols[1].button("×", key=f"del_{doc['name']}"):
st.session_state.docs = [d for d in st.session_state.docs if d["name"] != doc["name"]]
st.rerun()

# 统计信息
st.header(" 对话统计")
msg_counts = {
"user": len([m for m in st.session_state.messages if m["role"] == "user"]),
"assistant": len([m for m in st.session_state.messages if m["role"] == "assistant"])
}
st.metric("用户消息", msg_counts["user"])
st.metric("助手回复", msg_counts["assistant"])

# 管理功能
st.header("⚙管理")
if st.button("清空对话"):
st.session_state.messages = [st.session_state.messages[0]] if st.session_state.messages else []
st.rerun()

if st.button(" 退出登录"):
st.session_state.clear()
st.rerun()

# ————————–
# 聊天界面
# ————————–
def build_prompt(user_input):
"""构建带上下文的提示"""
context = []
if st.session_state.get("docs"):
context.append("## 参考文档")
for i, doc in enumerate(st.session_state.docs, 1):
context.append(f"【文档{i}】{doc['name']}\\n内容摘要：{doc['content'][:2000]}…")
context.append(f"## 用户问题\\n{user_input}")
return "\\n\\n".join(context)

def chat_interface():
"""主聊天界面"""
st.title(" labAI 智能助手")

# 系统提示设置
with st.expander("⚙ 系统设置", expanded=False):
default_prompt = """你是一个学术助手，根据以下规则响应：
1. 当存在参考文档时，严格基于文档内容回答
2. 没有文档时使用通用知识库
3. 回答语言与提问语言一致"""

sys_prompt = st.text_area(
"系统指令",
value=st.session_state.messages[0]["content"] if st.session_state.messages else default_prompt,
height=150
)

cols = st.columns(3)
if cols[0].button("恢复默认"):
sys_prompt = default_prompt
if cols[1].button("文档模式"):
sys_prompt = "严格基于提供的文档内容回答问题，超出文档范围时明确说明"
if cols[2].button("通用模式"):
sys_prompt = "使用通用知识库回答问题，保持简洁专业"

if st.button("保存设置"):
if not st.session_state.messages:
st.session_state.messages = [{"role": "system", "content": sys_prompt}]
else:
st.session_state.messages[0]["content"] = sys_prompt
st.rerun()

# 消息显示
for msg in st.session_state.messages[1:]:
role = msg["role"]
time = datetime.fromisoformat(msg["timestamp"]).strftime("%m/%d %H:%M")
with st.chat_message(role):
st.markdown(f"`[{time}]`\\n{msg['content']}")
if role == "user" and "docs_used" in msg:
st.caption(f"关联文档：{', '.join(msg['docs_used']) or '无'}")

# 用户输入
if prompt := st.chat_input("输入问题…"):
full_prompt = build_prompt(prompt)

user_msg = {
"role": "user",
"content": prompt,
"full_prompt": full_prompt,
"timestamp": datetime.now().isoformat(),
"docs_used": [d["name"] for d in st.session_state.get("docs", [])]
}
st.session_state.messages.append(user_msg)

# 显示消息
with st.chat_message("user"):
st.markdown(prompt)
if user_msg["docs_used"]:
st.caption(f" 已关联 {len(user_msg['docs_used'])} 个文档")

# 生成回答
with st.chat_message("assistant"):
try:
response = client.chat.completions.create(
model="QwQ-32B",
messages=[
{"role": m["role"], "content": m["full_prompt"] if m["role"] == "user" else m["content"]}
for m in [st.session_state.messages[0]] + st.session_state.messages[1:]
],
stream=True,
timeout=45
)

response_text = ""
container = st.empty()
for chunk in response:
if chunk.choices[0].delta.content:
response_text += chunk.choices[0].delta.content
container.markdown(response_text + "▌")

container.markdown(response_text)
st.session_state.messages.append({
"role": "assistant",
"content": response_text,
"timestamp": datetime.now().isoformat()
})

except Exception as e:
st.error(f"生成失败: {str(e)}")

# ————————–
# 主程序
# ————————–
if __name__ == "__main__":
# 初始化会话状态
if "messages" not in st.session_state:
st.session_state.messages = []

# 界面布局
sidebar_manager()
chat_interface()