在构建 AI 对话应用时如何让 AI 记住上下文和如何管理对话历史是两个绕不开的核心问题。本文将带你从零实现一个具备短时记忆、流式输出、自动标题生成、支持历史对话查询的完整对话系统技术栈基于 LangChain create_agentLanggraph DeepSeek FastAPI SQLite。一、系统架构总览先来看整体架构做到心中有数┌─────────────┐ ┌──────────────────────────────────────────────┐ │ 前端/客户端 │────▶│ FastAPI 服务 │ └─────────────┘ │ │ │ ┌─────────────┐ ┌─────────────────────┐ │ │ │ /chat/stream│───▶│ LangGraph Agent │ │ │ │ (SSE 流式) │ │ (DeepSeek 模型) │ │ │ └──────┬──────┘ └────────┬────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────────────┐ │ │ │ 标题生成 LLM │ │ SqliteSaver 记忆 │ │ │ │ (Structured) │ │ (SQLite 持久化) │ │ │ └──────┬──────┘ └────────┬────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────────────────┐ │ │ │ SQLite (test.db) │ │ │ │ ┌──────────────┐ ┌────────────────┐ │ │ │ │ │ thread_meta │ │ checkpoint 表 │ │ │ │ │ │ (标题管理) │ │ (对话记忆) │ │ │ │ │ └──────────────┘ └────────────────┘ │ │ │ └──────────────────────────────────────┘ │ └──────────────────────────────────────────────┘核心设计思路模块技术选型作用AI 模型DeepSeek (deepseek-v4-flash)对话推理 标题生成记忆管理LangGraphSqliteSaver自动保存/恢复对话上下文流式输出FastAPI SSE (EventSourceResponse)逐 token 推送给客户端标题管理自建thread_meta表 结构化输出为每个新对话自动生成简短标题日志系统Loguru结构化日志自动轮转压缩数据库SQLite轻量级持久化单文件部署二、环境准备与依赖安装pipinstalllangchain-deepseek langchain langgraph fastapi uvicorn pydantic loguru提示需要配置DEEPSEEK_API_KEY环境变量或在代码中通过其他方式注入。三、核心代码解析3.1 日志系统 —— Loguru 配置良好的日志是排查问题的第一道防线。我们使用 Loguru 实现结构化日志frompathlibimportPathfromloguruimportlogger base_dirPath(__file__).parent log_dirbase_dir/logs/log05.logformat(green{time:YYYY-MM-DD HH:mm:ss.SSS}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level)logger.add(log_dir,formatformat,levelINFO,rotation10 MB,# 日志文件达到 10MB 时自动切割retention7 days,# 保留最近 7 天的日志compressionzip,# 旧日志压缩为 zipencodingutf-8,# 中文不乱码enqueueTrue,# 多线程安全写入)关键参数解读rotation10 MB日志文件达到 10MB 自动轮转防止单文件过大enqueueTrue使用队列写入在 FastAPI 多线程/异步环境下保证线程安全compressionzip自动压缩旧日志节省磁盘空间3.2 SQLite 连接与 Checkpoint 初始化importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver db_dirbase_dir/dbconnectionsqlite3.connect(str(db_dir/test.db),check_same_threadFalse)connection.row_factorysqlite3.Row# 查询结果可按列名访问checkpointerSqliteSaver(connection)checkpointer.setup()# 自动创建 checkpoint 所需的表结构注意事项check_same_threadFalseFastAPI 在多线程环境下调用 SQLite必须关闭线程检查connection.row_factory sqlite3.Row让查询结果支持dict(row)转换方便后续处理checkpointer.setup()LangGraph 会自动创建checkpoints、checkpoint_blobs、checkpoint_writes等表3.3 创建 LangGraph Agentfromlangchain_deepseekimportChatDeepSeekfromlangchain.agentsimportcreate_agent agentcreate_agent(modelChatDeepSeek(modeldeepseek-v4-flash),system_prompt您是一个智慧小助手,tools[],middleware[],checkpointercheckpointer)核心概念 ——checkpointer的作用checkpointer是 LangGraph 的记忆中枢。每次对话结束后它会自动将完整的对话状态messages 列表序列化存入 SQLite。下次请求时只需传入相同的thread_idAgent 就能自动恢复上下文无需手动管理消息历史。3.4 自动标题生成 —— 结构化输出为新对话自动生成简短标题方便用户在历史列表中快速定位frompydanticimportBaseModel,FieldclassTitleOut(BaseModel):title:strField(description标题)# 使用独立 LLM 实例关闭思考模式提高速度title_llmChatDeepSeek(modeldeepseek-v4-flash,temperature0,# 温度为0保证稳定输出extra_body{thinking:{type:disabled}}# 关闭深度思考加速响应)structured_modeltitle_llm.with_structured_output(TitleOut)defcreate_title(user_message:str):messages[(system,根据用户的提问,总结一个简短精辟的对话标题(8个字以内,不要引号)),(human,user_message)]responsestructured_model.invoke(messages)logger.info(response.title)returnresponse.title设计亮点with_structured_output利用 Pydantic 模型约束 LLM 输出格式确保返回的是结构化的TitleOut对象而不是自由文本temperature0标题生成不需要创意低温保证结果稳定thinking: disabled关闭 DeepSeek 的深度思考模式大幅降低延迟标题生成不需要复杂推理3.5 标题的数据库管理我们需要单独建表存储对话标题因为 LangGraph 的 checkpoint 表结构是内部的不适合直接查询业务数据。创建表结构CREATETABLEIFNOTEXISTSthread_meta(thread_idTEXTPRIMARYKEY,titleTEXTNOTNULL,created_atTIMESTAMPNOTNULL);插入标题带并发防护definsert_title(thread_id:str,title:str,created_at:datetime)-dict:cursorNonetry:cursorconnection.cursor()cursor.execute(BEGIN IMMEDIATE)# 立即获取写锁防止并发重复cursor.execute(SELECT 1 FROM thread_meta WHERE thread_id ?,(thread_id,))ifcursor.fetchone():cursor.execute(ROLLBACK)logger.warning(f标题已存在 thread_id{thread_id})return{code:200,msg:thread_id already exists}cursor.execute(INSERT INTO thread_meta (thread_id, title, created_at) VALUES (?, ?, ?),(thread_id,title,created_at),)connection.commit()logger.info(f标题写入成功 thread_id{thread_id}, title{title})return{code:200,msg:ok}exceptsqlite3.Errorase:logger.error(f标题写入失败 thread_id{thread_id}, error{e})connection.rollback()return{code:500,msg:db error}finally:ifcursor:cursor.close()为什么用BEGIN IMMEDIATE在并发场景下多个请求可能同时为同一个thread_id生成标题。BEGIN IMMEDIATE在事务开始时就获取写锁配合SELECT ... INSERT的检查逻辑实现乐观锁 悲观锁的双重防护避免重复插入。查询标题列表defquery_title()-dict:cursorNonetry:cursorconnection.cursor()cursor.execute(SELECT * FROM thread_meta ORDER BY created_at DESC)rowscursor.fetchall()data[dict(row)forrowinrows]return{code:200,msg:data}exceptsqlite3.Errorase:logger.error(f查询标题列表失败, error{e})return{code:500,msg:db error}finally:ifcursor:cursor.close()3.6 流式对话接口 —— SSE 实现这是整个系统的核心接口fromfastapiimportFastAPIfromfastapi.sseimportServerSentEvent,EventSourceResponseclassChatIn(BaseModel):user_id:strthread_id:strquestion:strapp.post(/chat/stream,response_classEventSourceResponse)defchat(data:ChatIn):thread_iddata.user_iddata.thread_id config{configurable:{thread_id:thread_id}}# 快速判断是否为新线程只查数据库不调 LLMis_newcheckpointer.get_tuple(config)isNonelogger.info(f收到请求 user_id{data.user_id}, is_new{is_new})responseagent.stream_events({messages:[{role:user,content:data.question}]},configconfig,versionv3)formsginresponse.messages:fordeltainmsg.text:yieldServerSentEvent(datadelta,eventtoken)yieldServerSentEvent(dataDONE,eventdone)# 流式输出结束后如果是新对话生成标题ifis_new:titlecreate_title(data.question)insert_title(thread_id,title,datetime.now())流程图解客户端请求 │ ▼ ┌──────────────────┐ │ 拼接 thread_id │ user_id thread_id → 唯一会话标识 │ (user_idthread_id)│ └────────┬─────────┘ │ ▼ ┌──────────────────┐ │ 判断是否新对话 │ checkpointer.get_tuple(config) is None └────────┬─────────┘ │ ▼ ┌──────────────────┐ │ 流式推理 SSE推送 │ agent.stream_events → 逐 token yield └────────┬─────────┘ │ ▼ ┌──────────────────┐ │ 推送 DONE 信号 │ yield ServerSentEvent(dataDONE) └────────┬─────────┘ │ ┌────┴────┐ │ is_new? │ ├───是────┤ │ ▼ │ ┌────────────┐ ┌──────────────┐ │ │ 生成对话标题 │───▶│ 写入 thread_meta│ │ └────────────┘ └──────────────┘ └───否────┘ │ ▼ 结束关键设计点thread_id拼接策略user_id thread_id确保不同用户的同名会话不会冲突新对话判断通过checkpointer.get_tuple(config)检查是否已有 checkpoint比查自己的表更准确因为标题插入可能有延迟标题生成时机在yield DONE之后执行不阻塞流式输出用户体验更好stream_eventsversionv3使用 LangGraph 的事件流接口v3 是最新版本协议3.7 对话历史查询接口标题列表app.get(/chat/history/list)defchat_history_list():returnquery_title()单条对话历史fromlangchain_core.messagesimportHumanMessage,AIMessageapp.get(/chat/history/{thread_id})defchat_history(thread_id:str):config{configurable:{thread_id:thread_id}}checkpoint_tuplecheckpointer.get_tuple(config)ifcheckpoint_tupleisNone:return{code:404,msg:thread not found}messagescheckpoint_tuple.checkpoint[channel_values][messages]history_msg[]formsginmessages:ifisinstance(msg,HumanMessage):history_msg.append({role:user,content:msg.content})elifisinstance(msg,AIMessage):reasoning_parts[]text_parts[]# DeepSeek 可能返回 reasoning text 混合内容ifisinstance(msg.content,list):fordeltainmsg.content:ifdelta[type]reasoning:reasoning_parts.append(delta[reasoning])elifdelta[type]text:text_parts.append(delta[text])else:text_parts.append(msg.content)entry{role:assistant}ifreasoning_parts:entry[reasoning].join(reasoning_parts)iftext_parts:entry[content].join(text_parts)history_msg.append(entry)return{code:200,msg:history_msg}AI 消息解析的特殊处理DeepSeek 开启思考模式后返回的AIMessage.content是一个列表包含reasoning思考过程和text最终回答两种类型的片段。我们需要分别提取并拼接# content 结构示例[{type:reasoning,reasoning:让我想想...用户问的是...},{type:text,text:你好很高兴为你服务。},]最终输出格式{role:assistant,reasoning:让我想想...用户问的是...,content:你好很高兴为你服务。}四、API 接口一览方法路径功能入参返回POST/chat/stream流式对话{user_id, thread_id, question}SSE 流token 事件 done 事件GET/chat/history/list对话标题列表无{code, msg: [{thread_id, title, created_at}]}GET/chat/history/{thread_id}单条对话历史thread_id路径参数{code, msg: [{role, content, reasoning?}]}五、SSE 前端对接示例前端使用EventSource或fetch消费 SSE 流constresponseawaitfetch(/chat/stream,{method:POST,headers:{Content-Type:application/json},body:JSON.stringify({user_id:user_001,thread_id:session_001,question:你好介绍一下你自己})});constreaderresponse.body.getReader();constdecodernewTextDecoder();while(true){const{done,value}awaitreader.read();if(done)break;consttextdecoder.decode(value);// 解析 SSE 数据constlinestext.split(\n);for(constlineoflines){if(line.startsWith(data:)){constdataline.slice(5).trim();if(dataDONE){console.log(对话结束);}else{// 追加到页面appendToUI(data);}}}}六、完整启动方式if__name____main__:importuvicorn uvicorn.run(05shortmemorytitle:app,host127.0.0.1,port8000,reloadTrue)python 05shortmemorytitle.py# 服务启动后访问: http://127.0.0.1:8000/docs 查看 API 文档七、关键设计总结7.1 记忆机制为什么用 Checkpoint 而非手动管理消息方案优点缺点手动拼接 messages完全可控需要自行存储/查询/截断代码量大LangGraph Checkpoint自动持久化/恢复支持线程隔离依赖 LangGraph 生态本方案选择 Checkpoint核心优势在于零代码管理上下文——只需传入thread_idAgent 自动处理一切。7.2 标题生成为什么不阻塞流式输出❌ 错误做法先生成标题 → 再流式输出用户等待时间长 ✅ 正确做法先流式输出 → 完成后异步生成标题用户感知零延迟7.3 并发安全SQLite 的线程安全处理# 1. 连接层面sqlite3.connect(...,check_same_threadFalse)# 2. 事务层面cursor.execute(BEGIN IMMEDIATE)# 写锁提前获取# 3. 日志层面logger.add(...,enqueueTrue)# 队列写入线程安全