在构建 AI 对话应用时，"如何让 AI 记住上下文"和"如何管理对话历史"是两个绕不开的核心问题。本文将带你从零实现一个具备短时记忆、流式输出、自动标题生成、支持历史对话查询的完整对话系统，技术栈基于 LangChain create_agent（Langgraph） + 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)	对话推理 + 标题生成
记忆管理	LangGraph SqliteSaver	自动保存/恢复对话上下文
流式输出	FastAPI SSE (EventSourceResponse)	逐 token 推送给客户端
标题管理	自建 thread_meta 表 + 结构化输出	为每个新对话自动生成简短标题
日志系统	Loguru	结构化日志，自动轮转压缩
数据库	SQLite	轻量级持久化，单文件部署

---

二、环境准备与依赖安装

pip install langchain-deepseek langchain langgraph fastapi uvicorn pydantic loguru

提示： 需要配置 DEEPSEEK_API_KEY 环境变量，或在代码中通过其他方式注入。

---

三、核心代码解析

3.1 日志系统 —— Loguru 配置

良好的日志是排查问题的第一道防线。我们使用 Loguru 实现结构化日志：

from pathlib import Path
from loguru import logger

base_dir = Path(__file__).parent
log_dir = base_dir / "logs/log05.log"

format = (
    "<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,
    format=format,
    level="INFO",
    rotation="10 MB",        # 日志文件达到 10MB 时自动切割
    retention="7 days",      # 保留最近 7 天的日志
    compression="zip",       # 旧日志压缩为 zip
    encoding="utf-8",        # 中文不乱码
    enqueue=True,            # 多线程安全写入
)

关键参数解读：

• rotation="10 MB"：日志文件达到 10MB 自动轮转，防止单文件过大
• enqueue=True：使用队列写入，在 FastAPI 多线程/异步环境下保证线程安全
• compression="zip"：自动压缩旧日志，节省磁盘空间

---

3.2 SQLite 连接与 Checkpoint 初始化

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

db_dir = base_dir / "db"
connection = sqlite3.connect(str(db_dir / "test.db"), check_same_thread=False)
connection.row_factory = sqlite3.Row  # 查询结果可按列名访问

checkpointer = SqliteSaver(connection)
checkpointer.setup()  # 自动创建 checkpoint 所需的表结构

注意事项：

• check_same_thread=False：FastAPI 在多线程环境下调用 SQLite，必须关闭线程检查
• connection.row_factory = sqlite3.Row：让查询结果支持 dict(row) 转换，方便后续处理
• checkpointer.setup()：LangGraph 会自动创建 checkpoints、checkpoint_blobs、checkpoint_writes 等表

---

3.3 创建 LangGraph Agent

from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent

agent = create_agent(
    model=ChatDeepSeek(model="deepseek-v4-flash"),
    system_prompt="""您是一个智慧小助手""",
    tools=[],
    middleware=[],
    checkpointer=checkpointer
)

核心概念 —— checkpointer 的作用：

checkpointer 是 LangGraph 的"记忆中枢"。每次对话结束后，它会自动将完整的对话状态（messages 列表）序列化存入 SQLite。下次请求时只需传入相同的 thread_id，Agent 就能自动恢复上下文，无需手动管理消息历史。

---

3.4 自动标题生成 —— 结构化输出

为新对话自动生成简短标题，方便用户在历史列表中快速定位：

from pydantic import BaseModel, Field

class TitleOut(BaseModel):
    title: str = Field(description="标题")

# 使用独立 LLM 实例，关闭思考模式，提高速度
title_llm = ChatDeepSeek(
    model="deepseek-v4-flash",
    temperature=0,                              # 温度为0，保证稳定输出
    extra_body={"thinking": {"type": "disabled"}}  # 关闭深度思考，加速响应
)
structured_model = title_llm.with_structured_output(TitleOut)

def create_title(user_message: str):
    messages = [
        ("system", "根据用户的提问,总结一个简短精辟的对话标题(8个字以内,不要引号)"),
        ("human", user_message)
    ]
    response = structured_model.invoke(messages)
    logger.info(response.title)
    return response.title

设计亮点：

1. with_structured_output：利用 Pydantic 模型约束 LLM 输出格式，确保返回的是结构化的 TitleOut 对象，而不是自由文本
2. temperature=0：标题生成不需要创意，低温保证结果稳定
3. thinking: disabled：关闭 DeepSeek 的深度思考模式，大幅降低延迟（标题生成不需要复杂推理）

---

3.5 标题的数据库管理

我们需要单独建表存储对话标题，因为 LangGraph 的 checkpoint 表结构是内部的，不适合直接查询业务数据。

创建表结构

CREATE TABLE IF NOT EXISTS thread_meta (
    thread_id  TEXT PRIMARY KEY,
    title      TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL
);

插入标题（带并发防护）

def insert_title(thread_id: str, title: str, created_at: datetime) -> dict:
    cursor = None
    try:
        cursor = connection.cursor()
        cursor.execute("BEGIN IMMEDIATE")  # 立即获取写锁，防止并发重复
        cursor.execute("SELECT 1 FROM thread_meta WHERE thread_id = ?", (thread_id,))
        
        if cursor.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"}
    except sqlite3.Error as e:
        logger.error(f"标题写入失败 thread_id={thread_id}, error={e}")
        connection.rollback()
        return {"code": 500, "msg": "db error"}
    finally:
        if cursor:
            cursor.close()

为什么用 BEGIN IMMEDIATE？

在并发场景下，多个请求可能同时为同一个 thread_id 生成标题。BEGIN IMMEDIATE 在事务开始时就获取写锁，配合 SELECT ... INSERT 的检查逻辑，实现乐观锁 + 悲观锁的双重防护，避免重复插入。

查询标题列表

def query_title() -> dict:
    cursor = None
    try:
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM thread_meta ORDER BY created_at DESC")
        rows = cursor.fetchall()
        data = [dict(row) for row in rows]
        return {"code": 200, "msg": data}
    except sqlite3.Error as e:
        logger.error(f"查询标题列表失败, error={e}")
        return {"code": 500, "msg": "db error"}
    finally:
        if cursor:
            cursor.close()

---

3.6 流式对话接口 —— SSE 实现

这是整个系统的核心接口：

from fastapi import FastAPI
from fastapi.sse import ServerSentEvent, EventSourceResponse

class ChatIn(BaseModel):
    user_id: str
    thread_id: str
    question: str

@app.post('/chat/stream', response_class=EventSourceResponse)
def chat(data: ChatIn):
    thread_id = data.user_id + data.thread_id
    config = {"configurable": {"thread_id": thread_id}}
    
    # 快速判断是否为新线程（只查数据库，不调 LLM）
    is_new = checkpointer.get_tuple(config) is None
    logger.info(f"收到请求 user_id={data.user_id}, is_new={is_new}")
    
    response = agent.stream_events(
        {"messages": [{"role": "user", "content": data.question}]},
        config=config,
        version="v3"
    )
    
    for msg in response.messages:
        for delta in msg.text:
            yield ServerSentEvent(data=delta, event="token")
    
    yield ServerSentEvent(data="DONE", event="done")
    
    # 流式输出结束后，如果是新对话，生成标题
    if is_new:
        title = create_title(data.question)
        insert_title(thread_id, title, datetime.now())

流程图解：

客户端请求
    │
    ▼
┌──────────────────┐
│ 拼接 thread_id    │  user_id + thread_id → 唯一会话标识
│ (user_id+thread_id)│
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│ 判断是否新对话    │  checkpointer.get_tuple(config) is None
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│ 流式推理 + SSE推送 │  agent.stream_events → 逐 token yield
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│ 推送 DONE 信号    │  yield ServerSentEvent(data="DONE")
└────────┬─────────┘
         │
    ┌────┴────┐
    │ is_new? │
    ├───是────┤
    │         ▼
    │  ┌────────────┐    ┌──────────────┐
    │  │ 生成对话标题 │───▶│ 写入 thread_meta│
    │  └────────────┘    └──────────────┘
    └───否────┘
              │
              ▼
            结束

关键设计点：

1. thread_id 拼接策略：user_id + thread_id 确保不同用户的同名会话不会冲突
2. 新对话判断：通过 checkpointer.get_tuple(config) 检查是否已有 checkpoint，比查自己的表更准确（因为标题插入可能有延迟）
3. 标题生成时机：在 yield DONE 之后执行，不阻塞流式输出，用户体验更好
4. stream_events + version="v3"：使用 LangGraph 的事件流接口，v3 是最新版本协议

---

3.7 对话历史查询接口

标题列表

@app.get('/chat/history/list')
def chat_history_list():
    return query_title()

单条对话历史

from langchain_core.messages import HumanMessage, AIMessage

@app.get('/chat/history/{thread_id}')
def chat_history(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    checkpoint_tuple = checkpointer.get_tuple(config)
    
    if checkpoint_tuple is None:
        return {"code": 404, "msg": "thread not found"}
    
    messages = checkpoint_tuple.checkpoint['channel_values']['messages']
    history_msg = []
    
    for msg in messages:
        if isinstance(msg, HumanMessage):
            history_msg.append({"role": "user", "content": msg.content})
        elif isinstance(msg, AIMessage):
            reasoning_parts = []
            text_parts = []
            
            # DeepSeek 可能返回 reasoning + text 混合内容
            if isinstance(msg.content, list):
                for delta in msg.content:
                    if delta['type'] == 'reasoning':
                        reasoning_parts.append(delta['reasoning'])
                    elif delta['type'] == 'text':
                        text_parts.append(delta['text'])
            else:
                text_parts.append(msg.content)
            
            entry = {"role": "assistant"}
            if reasoning_parts:
                entry["reasoning"] = "".join(reasoning_parts)
            if text_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 流：

const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
        user_id: 'user_001',
        thread_id: 'session_001',
        question: '你好，介绍一下你自己'
    })
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    const text = decoder.decode(value);
    // 解析 SSE 数据
    const lines = text.split('\n');
    for (const line of lines) {
        if (line.startsWith('data:')) {
            const data = line.slice(5).trim();
            if (data === 'DONE') {
                console.log('对话结束');
            } else {
                // 追加到页面
                appendToUI(data);
            }
        }
    }
}

---

六、完整启动方式

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("05shortmemorytitle:app", host="127.0.0.1", port=8000, reload=True)

python 05shortmemorytitle.py
# 服务启动后访问: http://127.0.0.1:8000/docs 查看 API 文档

---

七、关键设计总结

7.1 记忆机制：为什么用 Checkpoint 而非手动管理消息？

方案	优点	缺点
手动拼接 messages	完全可控	需要自行存储/查询/截断，代码量大
LangGraph Checkpoint	自动持久化/恢复，支持线程隔离	依赖 LangGraph 生态

本方案选择 Checkpoint，核心优势在于零代码管理上下文——只需传入 thread_id，Agent 自动处理一切。

7.2 标题生成：为什么不阻塞流式输出？

❌ 错误做法：先生成标题 → 再流式输出（用户等待时间长）
✅ 正确做法：先流式输出 → 完成后异步生成标题（用户感知零延迟）

7.3 并发安全：SQLite 的线程安全处理

# 1. 连接层面
sqlite3.connect(..., check_same_thread=False)

# 2. 事务层面
cursor.execute("BEGIN IMMEDIATE")  # 写锁提前获取

# 3. 日志层面
logger.add(..., enqueue=True)  # 队列写入，线程安全

---