基于 Python + LangChain + React 的 AI 流式对话与历史存储实战

前言

在大语言模型（LLM）飞速发展的今天，将 AI 对话能力集成到自己的应用中已经成为非常普遍的需求。然而，一个生产级别的 AI 聊天应用不仅仅是调用 API 那么简单——流式输出让用户体验更自然，对话历史持久化确保数据不丢失，多会话管理让用户可以在不同话题间自由切换。

本文将基于一个完整的实战项目，带你从零搭建一个包含以下功能的 AI 聊天系统：

• 后端：Python + Flask + LangChain，对接通义千问（Qwen）大模型，支持 SSE 流式输出
• 前端：React + Zustand 状态管理，实现类似 DeepSeek 的对话界面
• 数据库：MySQL 存储对话历史，支持多会话、重命名、置顶等管理功能

一、项目架构概览

├── test-project/ # 后端（Python Flask） │ ├── server.py # 主服务，所有 API 端点 │ ├── chat_qwen.py # 通义千问模型封装 │ ├── chat_history.py # 对话历史数据库操作 │ ├── rag_service.py # RAG 知识检索服务 │ ├── config.py # 数据库配置 │ ├── init_chat_history.sql # 建表 SQL │ └── .env # 环境变量（API Key） │ └── my-react-app/ # 前端（React） ├── src/view/ │ ├── pages/chat/ # 聊天页面组件 │ │ ├── Chat.jsx │ │ ├── Chat.module.scss │ │ └── components/ │ │ ├── ChatSidebar.jsx # 侧边栏（对话列表） │ │ ├── ChatMessages.jsx # 消息气泡区域 │ │ └── ChatInput.jsx # 输入框 │ ├── store/chatStore.js # Zustand 状态管理 │ └── services/chatService.js # API 服务（SSE + REST） ├── package.json └── server.js # Node.js 辅助服务

二、数据库设计

2.1 建表 SQL

我们需要两张表：chat_conversations（对话表）和chat_messages（消息表）。

-- 聊天历史相关表 -- 在 pytosql 数据库中执行 CREATE TABLE IF NOT EXISTS chat_conversations ( id VARCHAR(36) PRIMARY KEY COMMENT '会话UUID', title VARCHAR(200) DEFAULT '新对话' COMMENT '会话标题', use_rag TINYINT DEFAULT 0 COMMENT '是否使用RAG', pinned TINYINT DEFAULT 0 COMMENT '是否置顶', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS chat_messages ( id INT AUTO_INCREMENT PRIMARY KEY, conversation_id VARCHAR(36) NOT NULL COMMENT '关联会话ID', role ENUM('user', 'assistant', 'system') NOT NULL COMMENT '消息角色', content TEXT NOT NULL COMMENT '消息内容', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (conversation_id) REFERENCES chat_conversations(id) ON DELETE CASCADE, INDEX idx_msg_conv (conversation_id) );

2.2 表结构说明

为什么用 UUID 而不是自增 ID？UUID 在分布式场景下更安全，且前端可以先生成 ID 再与后端同步，避免 ID 冲突。

三、后端实现

3.1 环境准备

pip install flask flask-cors langchain-openai langchain-community pymysql python-dotenv faiss-cpu

3.2 数据库配置（config.py）

DB_CONFIG = { "host": "localhost", "port": 3306, "user": "root", "password": "your_password", "database": "pytosql", "charset": "utf8mb4", }

3.3 对话历史管理模块（chat_history.py）

这个模块封装了所有与对话和消息相关的数据库操作：

import uuid import pymysql from config import DB_CONFIG def _get_conn(): return pymysql.connect(**DB_CONFIG) def create_conversation(use_rag=False): """创建新对话，返回对话 ID""" conv_id = str(uuid.uuid4()) conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "INSERT INTO chat_conversations (id, use_rag) VALUES (%s, %s)", (conv_id, 1 if use_rag else 0) ) conn.commit() return conv_id finally: conn.close() def list_conversations(): """获取所有对话列表，置顶优先，再按更新时间倒序""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "SELECT id, title, use_rag, pinned, created_at, updated_at " "FROM chat_conversations ORDER BY pinned DESC, updated_at DESC" ) rows = cursor.fetchall() return [ { "id": row[0], "title": row[1], "useRag": bool(row[2]), "pinned": bool(row[3]), "createdAt": str(row[4]) if row[4] else "", "updatedAt": str(row[5]) if row[5] else "", } for row in rows ] finally: conn.close() def get_messages(conv_id): """获取某个对话的所有消息""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "SELECT id, role, content, created_at " "FROM chat_messages WHERE conversation_id = %s ORDER BY id", (conv_id,) ) rows = cursor.fetchall() return [ {"id": row[0], "role": row[1], "content": row[2], "createdAt": str(row[3]) if row[3] else ""} for row in rows ] finally: conn.close() def save_message(conv_id, role, content): """保存一条消息并更新对话时间""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "INSERT INTO chat_messages (conversation_id, role, content) " "VALUES (%s, %s, %s)", (conv_id, role, content) ) cursor.execute( "UPDATE chat_conversations SET updated_at = NOW() WHERE id = %s", (conv_id,) ) conn.commit() finally: conn.close() def update_title(conv_id, title): """更新对话标题""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "UPDATE chat_conversations SET title = %s WHERE id = %s", (title, conv_id) ) conn.commit() finally: conn.close() def toggle_pin(conv_id): """切换置顶状态""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "UPDATE chat_conversations SET pinned = IF(pinned = 1, 0, 1) " "WHERE id = %s", (conv_id,) ) conn.commit() finally: conn.close() def delete_conversation(conv_id): """删除对话及其所有消息（级联删除）""" conn = _get_conn() try: cursor = conn.cursor() cursor.execute("DELETE FROM chat_conversations WHERE id = %s", (conv_id,)) conn.commit() finally: conn.close() def get_or_create_conversation(conv_id=None, use_rag=False): """获取已有对话或创建新对话""" if conv_id: conn = _get_conn() try: cursor = conn.cursor() cursor.execute( "SELECT id FROM chat_conversations WHERE id = %s", (conv_id,) ) if cursor.fetchone(): return conv_id finally: conn.close() return create_conversation(use_rag=use_rag)

3.4 Flask 主服务（server.py）

3.4.1 基础配置与模型初始化

import os import uuid import json from dotenv import load_dotenv from flask import Flask, request, jsonify, send_from_directory, Response, stream_with_context from flask_cors import CORS from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from chat_history import ( create_conversation, list_conversations, get_messages, save_message, update_title, delete_conversation, get_or_create_conversation, toggle_pin ) load_dotenv() app = Flask(__name__, static_folder="static") CORS(app) # 允许跨域请求 SYSTEM_PROMPT = "你是一个有用的AI助手，请用中文回答问题。" def get_chat_model(): """创建通义千问聊天模型""" return ChatOpenAI( model="qwen-plus", openai_api_key=os.getenv("DASHSCOPE_API_KEY"), openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", )

关键点：通义千问提供了 OpenAI 兼容的 API 接口，所以可以直接使用 LangChain 的ChatOpenAI类，只需修改openai_api_base指向阿里云的 DashScope 端点即可。

3.4.2 对话管理 API

# 创建新对话 @app.route("/api/conversations", methods=["POST"]) def new_conversation(): data = request.json or {} use_rag = data.get("use_rag", False) try: conv_id = create_conversation(use_rag=use_rag) return jsonify({"code": 200, "data": {"id": conv_id}}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # 获取所有对话列表 @app.route("/api/conversations", methods=["GET"]) def get_conversations(): try: return jsonify({"code": 200, "data": list_conversations()}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # 获取某个对话的消息列表 @app.route("/api/conversations/<conv_id>/messages", methods=["GET"]) def get_conv_messages(conv_id): try: return jsonify({"code": 200, "data": get_messages(conv_id)}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # 删除对话 @app.route("/api/conversations/<conv_id>", methods=["DELETE"]) def remove_conversation(conv_id): try: delete_conversation(conv_id) return jsonify({"code": 200, "msg": "删除成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # 重命名对话 @app.route("/api/conversations/<conv_id>/rename", methods=["PUT"]) def rename_conversation(conv_id): data = request.json title = data.get("title", "").strip() if not title: return jsonify({"code": 400, "msg": "标题不能为空"}) try: update_title(conv_id, title) return jsonify({"code": 200, "msg": "重命名成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)}) # 切换置顶 @app.route("/api/conversations/<conv_id>/pin", methods=["PUT"]) def pin_conversation(conv_id): try: toggle_pin(conv_id) return jsonify({"code": 200, "msg": "操作成功"}) except Exception as e: return jsonify({"code": 500, "msg": str(e)})

3.4.3 SSE 流式聊天接口（核心）

这是整个后端最核心的部分，实现了 Server-Sent Events（SSE）流式输出：

@app.route("/api/chat/stream", methods=["POST"]) def chat_stream(): data = request.json user_message = data.get("message", "").strip() session_id = data.get("session_id") use_rag = data.get("use_rag", False) if not user_message: return jsonify({"error": "消息不能为空"}), 400 def generate(): full_reply = "" conv_id = None try: # 获取或创建对话 conv_id = get_or_create_conversation(session_id, use_rag=use_rag) # 保存用户消息到数据库 save_message(conv_id, "user", user_message) # 加载已有消息历史 existing = get_messages(conv_id) # 第一轮对话：自动生成标题 if len(existing) <= 1: title = user_message[:20] + ("..." if len(user_message) > 20 else "") update_title(conv_id, title) # 构建 LangChain 消息历史 history = [SystemMessage(content=SYSTEM_PROMPT)] for msg in existing: if msg["role"] == "user": history.append(HumanMessage(content=msg["content"])) elif msg["role"] == "assistant": history.append(AIMessage(content=msg["content"])) history.append(HumanMessage(content=user_message)) # 调用 LLM 流式生成 llm = get_chat_model() for chunk in llm.stream(history): token = chunk.content if token: full_reply += token # 注意：必须 yield bytes 类型！ yield ( f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n" ).encode("utf-8") # 保存 AI 回复到数据库 if full_reply: save_message(conv_id, "assistant", full_reply) # 发送完成信号 yield ( f"data: {json.dumps({'done': True, 'session_id': conv_id}, ensure_ascii=False)}\n\n" ).encode("utf-8") except Exception as e: import traceback traceback.print_exc() yield ( f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n" ).encode("utf-8") resp = Response( stream_with_context(generate()), mimetype="text/event-stream", ) resp.headers["Cache-Control"] = "no-cache" resp.headers["X-Accel-Buffering"] = "no" resp.headers["Connection"] = "keep-alive" return resp

SSE 实现的几个关键踩坑点

3.4.4 启动服务

if __name__ == "__main__": app.run(debug=True, port=5000, use_reloader=False)

注意：use_reloader=False很重要。Flask 的 reloader 会在独立进程中重启应用，导致 SSE 流式连接中断。

四、前端实现

4.1 安装依赖

npm install antd @ant-design/x@^1.0.5 react-markdown zustand react-router-dom sass

注意：@ant-design/x要安装 v1.x 版本，v2.x 需要 antd v6。

4.2 状态管理（chatStore.js）

使用 Zustand 管理全局聊天状态，搭配persist中间件持久化关键配置：

import { create } from 'zustand'; import { persist } from 'zustand/middleware'; export const useChatStore = create( persist( (set, get) => ({ conversations: [], // 对话列表 currentConversationId: null, // 当前对话 ID messages: [], // 当前对话的消息 isStreaming: false, // 是否正在流式输出 streamingContent: '', // 流式输出中的内容 useRag: false, // RAG 模式开关 sidebarCollapsed: false, // 侧边栏折叠状态 // 切换对话 selectConversation: (id) => set({ currentConversationId: id, messages: [], streamingContent: '', isStreaming: false, }), // 流式输出控制 startStreaming: () => set({ isStreaming: true, streamingContent: '' }), appendStreamContent: (token) => set((state) => ({ streamingContent: state.streamingContent + token, })), finishStreaming: () => set((state) => ({ isStreaming: false, streamingContent: '', messages: [...state.messages, { id: Date.now(), role: 'assistant', content: state.streamingContent, createdAt: new Date().toISOString(), }], })), cancelStreaming: () => set((state) => { const partial = state.streamingContent; return partial ? { isStreaming: false, streamingContent: '', messages: [...state.messages, { id: Date.now(), role: 'assistant', content: partial, createdAt: new Date().toISOString() }] } : { isStreaming: false, streamingContent: '' }; }), toggleSidebar: () => set((s) => ({ sidebarCollapsed: !s.sidebarCollapsed })), }), { name: 'chat-storage', // 只持久化配置项，不持久化消息（消息从后端数据库加载） partialize: (state) => ({ currentConversationId: state.currentConversationId, useRag: state.useRag, }), } ) );

4.3 API 服务层（chatService.js）

封装了所有与后端交互的逻辑，包括 SSE 流式请求：

const BASE_URL = process.env.REACT_APP_CHAT_API_URL || 'http://localhost:5000'; export class ChatService { // REST 接口封装 async getConversations() { /* GET /api/conversations */ } async getMessages(id) { /* GET /api/conversations/:id/messages */ } async createConversation(useRag = false) { /* POST /api/conversations */ } async deleteConversation(id) { /* DELETE /api/conversations/:id */ } async renameConversation(id, title) { /* PUT /api/conversations/:id/rename */ } async togglePin(id) { /* PUT /api/conversations/:id/pin */ } /** * SSE 流式聊天 —— 核心方法 * 使用 fetch + ReadableStream 手动解析 SSE 数据 */ streamChat(message, sessionId, useRag, callbacks) { const { onToken, onDone, onError } = callbacks; const controller = new AbortController(); fetch(`${BASE_URL}/api/chat/stream`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message, session_id: sessionId || null, use_rag: useRag, }), signal: controller.signal, }) .then(async (response) => { if (!response.ok) throw new Error(`HTTP ${response.status}`); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // 按 SSE 双换行分割数据帧 const parts = buffer.split('\n\n'); buffer = parts.pop(); // 保留不完整的尾部数据 for (const part of parts) { for (const line of part.split('\n')) { if (!line.startsWith('data: ')) continue; const raw = line.slice(6); try { const parsed = JSON.parse(raw); if (parsed.error) { onError(new Error(parsed.error)); return; } if (parsed.token) onToken(parsed.token); if (parsed.done) { onDone(parsed.session_id); return; } } catch { /* 忽略解析失败的行 */ } } } } onDone(); }) .catch((err) => { if (err.name !== 'AbortError') onError(err); }); return controller; // 返回控制器，可用于中断请求 } } export const chatService = new ChatService();

为什么不使用EventSource？EventSource只支持 GET 请求，而我们的流式接口是 POST，所以使用fetch+ReadableStream手动解析 SSE 格式。这也是业界通用做法。

4.4 聊天页面组件

4.4.1 主页面（Chat.jsx）

import React, { useEffect } from 'react'; import { useChatStore } from '../../store/chatStore'; import { chatService } from '../../services/chatService'; import ChatSidebar from './components/ChatSidebar'; import ChatMessages from './components/ChatMessages'; import ChatInput from './components/ChatInput'; import styles from './Chat.module.scss'; export default function Chat() { const { currentConversationId, sidebarCollapsed, setMessages, toggleSidebar } = useChatStore(); // 切换对话时从数据库加载历史消息 useEffect(() => { if (!currentConversationId) { setMessages([]); return; } chatService.getMessages(currentConversationId) .then((msgs) => setMessages(msgs.map((m) => ({ id: m.id, role: m.role, content: m.content, createdAt: m.createdAt, })))) .catch(() => setMessages([])); }, [currentConversationId]); return ( <div className={styles.chatPage}> {!sidebarCollapsed && <ChatSidebar />} {sidebarCollapsed && ( <div className={styles.collapsedBar} onClick={toggleSidebar}>展开侧栏</div> )} <div className={styles.mainArea}> <ChatMessages /> <ChatInput /> </div> </div> ); }

4.4.2 侧边栏（ChatSidebar.jsx）

侧边栏包含对话列表、新建对话按钮、搜索弹窗、重命名/置顶/删除菜单：

// 对话项组件 —— 悬停显示操作菜单 function ConversationItem({ conv, isActive, onSelect, onRefresh }) { const [renaming, setRenaming] = useState(false); const menuItems = { items: [ { key: 'rename', icon: <EditOutlined />, label: '重命名' }, { key: 'pin', icon: conv.pinned ? <PushpinFilled /> : <PushpinOutlined />, label: conv.pinned ? '取消置顶' : '置顶' }, { type: 'divider' }, { key: 'delete', icon: <DeleteOutlined />, label: '删除', danger: true }, ], onClick: ({ key, domEvent }) => { domEvent.stopPropagation(); // 处理各操作... }, }; return ( <div className={`${styles.conversationItem} ${isActive ? styles.active : ''}`} onClick={() => !renaming && onSelect(conv.id)}> {conv.pinned && <PushpinFilled style={{ color: '#1677ff', fontSize: 12 }} />} {renaming ? ( <Input size="small" value={newTitle} onPressEnter={handleRename} onBlur={() => setRenaming(false)} autoFocus /> ) : ( <span className={styles.conversationTitle}>{conv.title}</span> )} <Dropdown menu={menuItems} trigger={['click']}> <EllipsisOutlined className={styles.ellipsisIcon} onClick={(e) => e.stopPropagation()} /> </Dropdown> </div> ); }

对话列表按时间自动分组（今天 / 昨天 / 7天内 / 30天内 / 更早），置顶对话始终在最前面。

4.4.3 消息展示（ChatMessages.jsx）

export default function ChatMessages() { const { messages, isStreaming, streamingContent } = useChatStore(); // 无消息时显示欢迎页 if (messages.length === 0 && !isStreaming && !streamingContent) { return ( <div className={styles.welcome}> <RobotOutlined className={styles.icon} /> <h2>AI Chat 助手</h2> <p>点击「开启新对话」或直接输入消息开始聊天</p> </div> ); } return ( <div className={styles.messagesArea}> <div className={styles.messageList}> {/* 已完成的消息 */} {messages.map((msg) => ( <div key={msg.id} className={`${styles.messageItem} ${styles[msg.role]}`}> <Avatar icon={msg.role === 'user' ? <UserOutlined /> : <RobotOutlined />} /> <div className={`${styles.messageContent} ${styles[msg.role]}`}> {msg.role === 'assistant' ? <ReactMarkdown>{msg.content}</ReactMarkdown> : msg.content} </div> </div> ))} {/* 流式输出中的消息（带闪烁光标） */} {isStreaming && streamingContent && ( <div className={`${styles.messageItem} ${styles.assistant}`}> <Avatar icon={<RobotOutlined />} /> <div className={`${styles.messageContent} ${styles.assistant}`}> <ReactMarkdown>{streamingContent}</ReactMarkdown> <span className={styles.streamingCursor} /> </div> </div> )} {/* 思考中状态 */} {isStreaming && !streamingContent && <div>思考中...</div>} </div> </div> ); }

4.4.4 输入框（ChatInput.jsx）

export default function ChatInput() { const [inputValue, setInputValue] = useState(''); const abortRef = useRef(null); const handleSend = useCallback(() => { const text = inputValue.trim(); if (!text || isStreaming) return; // 更新 UI setInputValue(''); addMessage({ id: Date.now(), role: 'user', content: text, createdAt: new Date().toISOString() }); startStreaming(); // 发起 SSE 流式请求 abortRef.current = chatService.streamChat( text, currentConversationId, useRag, { onToken: (token) => appendStreamContent(token), onDone: (sessionId) => { finishStreaming(); if (sessionId) setCurrentConversationId(sessionId); // 刷新侧边栏对话列表 chatService.getConversations() .then(list => useChatStore.getState().setConversations(list)); }, onError: (err) => { message.error(err.message || '请求失败'); cancelStreaming(); }, } ); }, [inputValue, isStreaming, currentConversationId, useRag]); // Enter 发送，Shift+Enter 换行 const handleKeyDown = (e) => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); handleSend(); } }; return ( <div className={styles.inputArea}> <textarea value={inputValue} onChange={handleChange} onKeyDown={handleKeyDown} placeholder="输入消息，Enter 发送，Shift+Enter 换行..." /> {isStreaming ? <Button danger icon={<StopOutlined />} onClick={handleStop} /> : <Button type="primary" icon={<SendOutlined />} onClick={handleSend} disabled={!inputValue.trim()} /> } </div> ); }

4.5 路由配置

聊天页面使用独立路由（不嵌套在 Layout 内），因为它有自己全屏的侧边栏布局：

const router = createBrowserRouter([ { path: '/', element: <Layout />, children: [ { path: '/', element: <Home /> }, { path: '/count', element: <Count /> }, // ... 其他页面 ] }, { path: '/chat', element: <Chat /> // 独立路由，全屏布局 } ]);

五、完整数据流

整个系统的数据流如下：

用户输入消息 ↓ ChatInput.handleSend() ↓ chatService.streamChat() ──── HTTP POST ────→ Flask /api/chat/stream ↓ ↓ onToken(token) ←──── SSE data ────── llm.stream(history) 逐 token 输出 ↓ ↓ appendStreamContent() save_message() 保存到 MySQL ↓ ChatMessages 实时渲染（带闪烁光标） ↓ onDone() → finishStreaming() ↓ messages 数组更新，流式光标消失

六、启动与测试

6.1 初始化数据库

6.2 启动后端

cd test-project python server.py # 聊天机器人服务启动: http://127.0.0.1:5000

6.3 启动前端

cd my-react-app npm start # 访问 http://localhost:3000/chat

6.4 验证功能

1. 输入消息并发送 → AI 流式逐 token 回答
2. 点击「开启新对话」→ 创建空白对话
3. 侧边栏按时间分组展示历史对话
4. 悬停对话项 → 显示菜单（重命名 / 置顶 / 删除）
5. 刷新页面 → 历史消息从数据库恢复

七、总结

本文实现了一个完整的 AI 流式聊天系统，涵盖了以下技术要点：

• SSE 流式输出：Flask + LangChain 的llm.stream()实现逐 token 推送，前端fetch + ReadableStream手动解析 SSE 数据帧
• 对话持久化：MySQL 存储对话和消息，支持多会话管理、历史恢复
• 前端状态管理：Zustand 的persist中间件持久化用户配置，消息数据从后端按需加载
• 对话管理：重命名、置顶、删除、时间分组，提供完整的会话管理体验

这个架构可以进一步扩展：添加用户认证、接入更多 LLM 模型、支持图片/文件上传、实现 RAG 知识库检索等。希望本文能为你的 AI 应用开发提供参考。

注意⚠：本文已分享完整流程及主要代码，完整代码可后台私信联系！！！