工业级大模型学习之路030：Streamlit 企业级智能体前端工作台

一、理论基础智能体前端技术选型与 Streamlit 架构深度解析1.1 智能体前端的核心需求与技术挑战与传统 Web 应用不同大模型智能体前端有其独特的核心需求流式输出支持需要实时展示大模型逐 token 生成的内容长会话管理需要持久化保存用户的对话历史和系统状态多模态展示需要支持 Markdown、代码高亮、图表、文件等多种内容格式实时状态反馈需要展示智能体的思考过程、工具调用、执行进度低延迟交互需要保证用户输入和模型响应的流畅性快速迭代能力需要能够快速调整界面和功能适应大模型的快速发展传统前端技术栈的痛点Vue/React 需要掌握 HTML、CSS、JavaScript、Node.js 等全套技术栈学习成本高需要单独开发后端 API 层前后端分离增加了系统复杂度流式输出需要自行实现 WebSocket 或 SSE开发难度大大模型相关的组件如聊天界面、代码高亮需要自行开发或引入第三方库1.2 主流智能体前端技术栈深度对比基于工业界 2026 年的最新实践我们对四种主流智能体前端技术栈进行了全面对比技术栈学习成本开发速度与 Python 后端集成流式输出支持多模态能力企业级适用性生产案例Streamlit⭐ 极低纯 Python⭐⭐⭐⭐⭐ 最快⭐⭐⭐⭐⭐ 原生集成⭐⭐⭐⭐⭐ 完美支持⭐⭐⭐⭐ 优秀⭐⭐⭐⭐ 适合内部工具、原型、MVP字节跳动、OpenAI、Anthropic 内部工具Chainlit⭐⭐ 低⭐⭐⭐⭐ 快⭐⭐⭐⭐ 专为 LangChain 设计⭐⭐⭐⭐⭐ 完美支持⭐⭐⭐⭐ 优秀⭐⭐⭐⭐ 适合 LangChain 生态LangChain 官方推荐Gradio⭐ 极低⭐⭐⭐⭐ 快⭐⭐⭐⭐ 原生集成⭐⭐⭐⭐ 良好⭐⭐⭐⭐ 优秀⭐⭐⭐ 适合 AI 演示、模型测试Hugging Face SpacesVue/React⭐⭐⭐⭐ 高⭐⭐ 慢⭐⭐ 需要 API 封装⭐⭐⭐ 需自行实现⭐⭐⭐⭐⭐ 最强⭐⭐⭐⭐⭐ 适合面向 C 端的产品ChatGPT、Claude工业级结论对于我们的课程和绝大多数企业内部智能体项目Streamlit 是无可争议的最佳选择零前端经验要求所有 Python 开发者都能在 1 小时内上手纯 Python 编写与我们的后端代码无缝集成不需要额外的 API 层原生支持流式输出完美匹配大模型的生成特性丰富的组件生态开箱即用的聊天、表单、图表、文件上传等功能部署简单一行命令即可启动服务活跃的社区和官方支持持续更新大模型相关的功能1.3.1 Streamlit 运行机制脚本重运行每当用户与界面交互如点击按钮、输入文本Streamlit 会从头到尾重新运行整个 Python 脚本状态隔离每个浏览器标签页对应一个独立的会话拥有自己的st.session_state增量更新Streamlit 会比较前后两次运行的输出差异只更新变化的部分保证界面流畅组件状态自动管理所有输入组件如st.text_input、st.slider会自动管理自己的状态不需要手动编写事件处理函数1.3.2 会话状态管理机制st.session_state是 Streamlit 应用状态管理的核心它是一个字典 - like 的对象用于在脚本重运行之间保存数据。核心特性会话隔离每个用户会话拥有独立的st.session_state互不干扰跨页面共享在多页面应用中st.session_state在所有页面之间共享自动序列化Streamlit 会自动序列化st.session_state中的数据支持大多数 Python 原生类型组件双向绑定通过key参数可以将组件与st.session_state中的变量进行双向绑定工业级最佳实践使用st.session_state.get(key, default_value)进行健壮的状态初始化为所有输入组件设置唯一的key参数不要在st.session_state中存储不可序列化的对象如数据库连接、文件句柄定期清理不需要的状态避免内存泄漏1.4 Streamlit 与 LangGraph 集成的核心优势LangGraph 负责多智能体的逻辑编排Streamlit 负责可视化和交互两者结合是目前最快实现可观测多智能体应用的技术栈状态共享Streamlit 的st.session_state可以直接存储 LangGraph 的状态和配置流式输出Streamlit 原生支持 LangGraph 的流式执行结果实时状态展示可以实时展示 LangGraph 的执行节点、工具调用、中间结果人类介入可以轻松实现 LangGraph 的人类介入功能在关键节点等待用户审核快速迭代可以快速调整界面和智能体逻辑不需要重新部署后端服务二、核心实战搭建企业级智能体工作台2.1 第一步环境搭建与项目结构设计2.1.1 设计思想模块化设计将前端代码按功能划分为多个页面便于维护和扩展前后端一体化直接复用后端的RAGService不需要额外的 API 层配置化所有可配置项都放在统一的地方便于调整工业级标准遵循 Python 编码规范添加类型注解和异常处理2.1.2 环境安装在项目虚拟环境中执行以下命令安装所有必需的依赖pip install streamlit1.38.0 streamlit-markdown1.0.3 streamlit-code-editor0.1.162.1.3 项目结构设计在项目根目录创建frontend文件夹采用 Streamlit 官方推荐的多页面应用结构langchain-2026/ ├── core/ # 原有后端核心代码完全复用 │ ├── rag_service.py │ ├── llm_factory.py │ ├── tools.py │ ├── react_agent.py │ ├── multi_agent.py │ └── checkpoint.py ├── config/ # 原有配置完全复用 │ └── settings.py ├── data/ # 原有数据目录完全复用 ├── frontend/ # 新增Streamlit前端 │ ├── pages/ # 多页面应用目录 │ │ ├── 1_chat.py # 智能聊天页面 │ │ ├── 2_report.py # 报告生成页面第9天集成 │ │ ├── 3_code.py # 代码生成页面第10天集成 │ │ └── 4_admin.py # 系统管理页面 │ └── app.py # 主应用入口 ├── main8.py # 原有FastAPI后端可选前端可直接调用后端服务 └── requirements.txt # 依赖清单2.2 第二步主应用入口与全局配置2.2.1 设计思想全局配置优先所有全局配置都放在主应用入口确保所有页面都能继承样式统一自定义全局 CSS 样式打造企业级界面效果服务单例化将RAGService初始化为全局单例避免重复初始化异常处理添加全局异常处理保证应用的稳定性2.2.2 核心代码实现创建frontend/app.pyimport streamlit as st import sys import os from typing import Optional # 将项目根目录添加到Python路径解决导入问题 # 这是Streamlit应用的标准做法确保能够正确导入后端模块 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 导入统一日志配置避免重复配置 from utils.logger import logger # # 全局页面配置必须放在最前面 # st.set_page_config( page_title企业级智能体平台, layoutwide, initial_sidebar_stateexpanded, menu_items{ Get Help: https://docs.streamlit.io, Report a bug: None, About: # 企业级智能体平台\n基于LangChain和LangGraph构建的工业级智能体平台\n版本v1.0.0 } ) # # 自定义全局CSS样式 # st.markdown( style /* 主容器样式 */ .main .block-container { padding-top: 2rem; padding-bottom: 2rem; max-width: 90%; } /* 消息气泡样式 */ .user-message { background-color: #e3f2fd; padding: 1rem 1.2rem; border-radius: 1rem 1rem 0 1rem; margin-bottom: 1rem; text-align: right; box-shadow: 0 1px 3px rgba(0,0,0,0.12); } .assistant-message { background-color: #f5f5f5; padding: 1rem 1.2rem; border-radius: 1rem 1rem 1rem 0; margin-bottom: 1rem; box-shadow: 0 1px 3px rgba(0,0,0,0.12); } /* 代码块样式优化 */ pre { background-color: #272822 !important; color: #f8f8f2 !important; padding: 1rem !important; border-radius: 0.5rem !important; overflow-x: auto !important; font-size: 0.9rem !important; } code { background-color: #f0f0f0 !important; padding: 0.1rem 0.3rem !important; border-radius: 0.2rem !important; font-size: 0.9rem !important; } /* 侧边栏样式 */ .css-1d391kg { background-color: #f8f9fa; border-right: 1px solid #e9ecef; } /* 按钮样式优化 */ .stButtonbutton { width: 100%; border-radius: 0.5rem; height: 2.5rem; font-weight: 500; transition: all 0.2s ease; } .stButtonbutton:hover { transform: translateY(-1px); box-shadow: 0 4px 6px rgba(0,0,0,0.1); } /* 标题样式 */ h1, h2, h3, h4 { color: #1976d2; font-weight: 600; } /* 进度条样式 */ .stProgress div div div { background-color: #1976d2; } /* 隐藏侧边栏默认的 app 标题 */ [data-testidstSidebarNav] div:first-child { display: none !important; } /* 自定义侧边栏标题 */ [data-testidstSidebarNav]::before { content: 企业级智能体平台; display: block; padding: 1rem; font-size: 1.2rem; font-weight: 600; color: #1976d2; border-bottom: 1px solid #e9ecef; margin-bottom: 0.5rem; } /style , unsafe_allow_htmlTrue) # 自定义浏览器标签页标题覆盖Streamlit默认标题 st.markdown( script document.title 企业级智能体平台; /script , unsafe_allow_htmlTrue) # # 全局服务初始化单例模式 # def init_services() - None: 初始化全局服务确保只初始化一次 if rag_service not in st.session_state: with st.spinner(正在初始化智能体服务...): try: from core.rag_service import RAGService st.session_state.rag_service RAGService() logger.info(RAG服务初始化成功) except Exception as e: logger.error(fRAG服务初始化失败: {str(e)}, exc_infoTrue) st.error(f智能体服务初始化失败: {str(e)}) st.stop() # 初始化其他全局状态 if current_user not in st.session_state: st.session_state.current_user admin if current_session_id not in st.session_state: st.session_state.current_session_id None if messages not in st.session_state: st.session_state.messages [] # 执行服务初始化 init_services() # # 应用主界面 # st.title(企业级智能体平台) st.divider() # 侧边栏导航 st.sidebar.title(导航菜单) page st.sidebar.radio( 选择功能, [智能聊天, 报告生成, 代码生成, 系统管理], index0, help选择要使用的智能体功能 ) # 系统状态显示 st.sidebar.divider() st.sidebar.subheader(系统状态) if rag_service in st.session_state: st.sidebar.success(智能体服务正常运行) else: st.sidebar.error(智能体服务异常) st.sidebar.info(f当前用户{st.session_state.current_user}) st.sidebar.info(f当前版本v1.0.0) # 根据选择加载对应页面 if page 智能聊天: st.switch_page(pages/1_智能聊天.py) elif page 报告生成: st.switch_page(pages/2_报告生成.py) elif page 代码生成: st.switch_page(pages/3_代码生成.py) elif page 系统管理: st.switch_page(pages/4_系统管理.py)2.2.3 代码解释与注意事项路径处理通过sys.path.append将项目根目录添加到 Python 路径确保能够正确导入后端模块单例初始化通过检查st.session_state确保RAGService只初始化一次避免重复加载模型和向量数据库全局样式通过自定义 CSS 统一界面风格打造企业级视觉效果异常处理添加服务初始化异常处理保证应用能够优雅地处理错误状态隔离每个用户会话拥有独立的状态互不干扰2.3 第三步核心聊天界面实现2.3.1 设计思想流式输出优先使用 Streamlit 原生的st.write_stream实现流畅的打字机效果消息历史管理使用st.session_state保存消息历史支持多轮对话多模式支持支持普通 RAG 和 ReAct Agent 两种模式满足不同场景需求用户体验优化添加加载状态、错误提示、消息气泡等用户体验元素2.3.2 核心代码实现创建frontend/pages/1_chat.pyimport streamlit as st import time from langchain_core.messages import HumanMessage, AIMessage from typing import Generator # 页面配置 st.title( 智能聊天) st.divider() # # 侧边栏配置 # with st.sidebar: st.subheader(聊天配置) # 智能体模式选择 agent_mode st.selectbox( 智能体模式, [普通RAG, ReAct Agent], index0, help普通RAG仅使用知识库回答问题ReAct Agent可以使用工具解决复杂问题 ) # 模型参数调整 temperature st.slider( 温度, min_value0.0, max_value1.0, value0.1, step0.1, help值越低回答越准确、越保守值越高回答越有创造性、越多样化 ) max_tokens st.number_input( 最大生成长度, min_value100, max_value4096, value2048, step100, help模型生成的最大token数量 ) st.divider() # 会话管理 st.subheader(会话管理) if st.button(➕ 新建会话, typeprimary, use_container_widthTrue): st.session_state.current_session_id fsession_{int(time.time())} st.session_state.messages [] st.rerun() # 历史会话列表后续课程将实现持久化 st.subheader(历史会话) st.info(历史会话持久化功能将在后续课程中实现) # # 聊天消息显示区域 # chat_container st.container(height600, borderTrue) with chat_container: # 显示历史消息 for message in st.session_state.messages: if isinstance(message, HumanMessage): with st.chat_message(user, avatar): st.markdown(message.content) elif isinstance(message, AIMessage): with st.chat_message(assistant, avatar): st.markdown(message.content) # # 聊天输入与处理 # if prompt : st.chat_input(输入你的问题我会尽力为你解答...): # 添加用户消息到历史 st.session_state.messages.append(HumanMessage(contentprompt)) # 显示用户消息 with chat_container: with st.chat_message(user, avatar): st.markdown(prompt) # 生成AI回答 with chat_container: with st.chat_message(assistant, avatar): message_placeholder st.empty() full_response try: # 调用后端服务获取流式输出 use_agent (agent_mode ReAct Agent) # 生成流式响应 stream: Generator[str, None, None] st.session_state.rag_service.stream_query( questionprompt, user_idst.session_state.current_user, use_agentuse_agent, temperaturetemperature, max_tokensmax_tokens ) # 使用st.write_stream展示流式输出Streamlit 1.30推荐方法 full_response st.write_stream(stream) except Exception as e: full_response f❌ 抱歉发生了一个错误{str(e)} message_placeholder.markdown(full_response) st.error(f生成回答失败: {str(e)}) # 添加AI消息到历史 st.session_state.messages.append(AIMessage(contentfull_response))2.3.3 代码解释与注意事项流式输出使用 Streamlit 1.30 新增的st.write_stream方法这是展示大模型流式输出的最佳实践比手动循环更新更流畅、性能更好消息类型使用 LangChain 的HumanMessage和AIMessage类型保存消息历史与后端保持一致参数传递将前端配置的模型参数温度、最大生成长度传递给后端服务实现灵活的参数调整异常处理添加全局异常处理保证应用在出错时能够给出友好的提示用户体验添加头像、消息气泡、加载状态等元素提升用户体验2.4 第四步其他页面基础框架实现2.4.1 报告生成页面import streamlit as st st.title( 智能报告生成) st.divider() # 侧边栏配置 with st.sidebar: st.subheader(报告配置) report_type st.selectbox( 报告类型, [双智能体报告, 三智能体评审报告], index0, help双智能体研究员写作家三智能体研究员写作家评审员 ) auto_approve st.checkbox( 自动批准, valueFalse, help自动批准研究结果跳过人工审核环节 ) max_reviews st.slider( 最大评审次数, min_value1, max_value5, value3, help最多允许修改的次数 ) # 主界面 col1, col2 st.columns([3, 1]) with col1: report_topic st.text_area( 报告主题, height120, placeholder请输入报告主题例如生成一份关于公司2026年第一季度销售情况的分析报告 ) if st.button( 生成报告, typeprimary, disablednot report_topic, use_container_widthTrue): st.info( 报告生成功能将在第9天课程中实现) st.info(当前使用模拟数据演示界面效果) # 模拟生成过程 progress_bar st.progress(0) status_text st.empty() stages [正在分析需求..., 正在检索信息..., 正在整理研究要点..., 正在生成报告...] for i, stage in enumerate(stages): status_text.text(stage) for j in range(25): time.sleep(0.02) progress_bar.progress(i * 25 j 1) status_text.text(✅ 报告生成完成) progress_bar.empty() # 显示模拟报告 st.success(✅ 报告生成成功) st.markdown( # 2026年第一季度销售情况分析报告 ## 一、摘要 本报告总结了公司2026年第一季度的销售情况分析了市场趋势和竞品动态并提出了第二季度的销售建议。第一季度总销售额达到1.2亿元同比增长25%超出预期目标。 ## 二、核心发现 1. 华东地区表现最佳贡献了40%的销售额 2. 新产品A系列销量超出预期成为新的增长点 3. 线上渠道销售额占比首次超过线下渠道 ## 三、结论与建议 1. 加大对华东地区的市场投入 2. 继续推广新产品A系列 3. 加强线上渠道建设 ) with col2: st.subheader(生成状态) st.info(等待生成报告) st.subheader(历史报告) st.write(暂无历史报告)2.4.2 代码生成页面创建frontend/pages/3_code.pyimport streamlit as st st.title( 智能代码生成) st.divider() # 侧边栏配置 with st.sidebar: st.subheader(代码配置) language st.selectbox( 编程语言, [Python, JavaScript, Java, Go], index0 ) include_tests st.checkbox( 生成测试用例, valueTrue ) auto_test st.checkbox( 自动执行测试, valueTrue ) # 主界面 col1, col2 st.columns([1, 1]) with col1: st.subheader(需求描述) requirement st.text_area( 请详细描述你需要的功能, height200, placeholder例如实现一个计算器函数支持加减乘除运算并处理异常情况 ) if st.button( 生成代码, typeprimary, disablednot requirement, use_container_widthTrue): st.info( 代码生成功能将在第10天课程中实现) st.info(当前使用模拟数据演示界面效果) # 模拟生成过程 progress_bar st.progress(0) status_text st.empty() stages [正在分析需求..., 正在生成PRD..., 正在编写代码..., 正在执行测试...] for i, stage in enumerate(stages): status_text.text(stage) for j in range(25): time.sleep(0.02) progress_bar.progress(i * 25 j 1) status_text.text(✅ 代码生成完成) progress_bar.empty() with col2: st.subheader(生成结果) st.code( def calculator(expression: str) - str: \\\ 简单计算器函数支持加减乘除运算 Args: expression: 数学表达式字符串例如 2 3 * 4 Returns: 计算结果字符串 \\\ try: # 使用eval计算表达式注意生产环境中需要添加安全限制 result eval(expression) return f计算结果{expression} {result} except ZeroDivisionError: return 错误除数不能为零 except SyntaxError: return 错误表达式语法错误 except Exception as e: return f错误{str(e)} # 测试用例 def test_calculator(): assert calculator(2 3) 计算结果2 3 5 assert calculator(10 - 5) 计算结果10 - 5 5 assert calculator(4 * 6) 计算结果4 * 6 24 assert calculator(8 / 2) 计算结果8 / 2 4.0 assert calculator(8 / 0) 错误除数不能为零 print(所有测试通过) , languagepython)2.4.3 系统管理页面创建frontend/pages/4_admin.pyimport streamlit as st import os st.title(⚙️ 系统管理) st.divider() # 标签页 tab1, tab2, tab3 st.tabs([系统状态, 知识库管理, 日志查看]) with tab1: st.subheader(系统状态概览) col1, col2, col3, col4 st.columns(4) with col1: st.metric(总会话数, 128, 12) st.metric(今日活跃用户, 15, 3) with col2: st.metric(知识库文档数, 32, 5) st.metric(总向量数, 1,245, 120) with col3: st.metric(模型调用次数, 1,567, 234) st.metric(平均响应时间, 2.3s, -0.5s) with col4: st.metric(任务成功率, 98.7%, 0.3%) st.metric(系统运行时间, 7天12小时) st.divider() st.subheader(服务状态) st.success(✅ RAG服务正常运行) st.success(✅ Agent服务正常运行) st.success(✅ 向量数据库正常连接) st.success(✅ LLM服务正常连接) with tab2: st.subheader(知识库管理) # 文件上传 uploaded_files st.file_uploader( 上传文档到知识库, type[txt, md, pdf, docx, xlsx], accept_multiple_filesTrue, help支持上传txt、md、pdf、docx、xlsx格式的文档 ) if uploaded_files: if st.button(添加到知识库, typeprimary): with st.spinner(正在处理文档...): # 这里将在后续课程中集成实际的文档处理功能 for file in uploaded_files: st.success(f✅ 已添加{file.name}) st.divider() st.subheader(知识库文档列表) st.dataframe( [ {文件名: 员工手册.pdf, 大小: 2.3MB, 上传时间: 2026-05-20, 状态: 已处理}, {文件名: 产品介绍.md, 大小: 156KB, 上传时间: 2026-05-22, 状态: 已处理}, {文件名: 销售数据.xlsx, 大小: 456KB, 上传时间: 2026-05-25, 状态: 已处理} ], use_container_widthTrue, hide_indexTrue ) with tab3: st.subheader(系统日志) log_level st.selectbox( 日志级别, [INFO, WARNING, ERROR, DEBUG], index0 ) if st.button(刷新日志, use_container_widthTrue): st.code( 2026-05-27 10:23:45 - INFO - 系统启动成功 2026-05-27 10:23:46 - INFO - RAG服务初始化完成 2026-05-27 10:23:46 - INFO - Agent服务初始化完成 2026-05-27 10:24:12 - INFO - 用户admin发起聊天请求 2026-05-27 10:24:15 - INFO - 回答生成完成耗时3.2秒 2026-05-27 10:25:32 - INFO - 用户admin发起Agent请求 2026-05-27 10:25:38 - INFO - Agent调用工具calculator 2026-05-27 10:25:40 - INFO - 工具执行完成返回结果 2026-05-27 10:25:42 - INFO - 回答生成完成耗时10.1秒 )三、测试与验证3.1 启动前端服务在项目根目录执行以下命令启动 Streamlit 服务streamlit run frontend/app.py