记录 task_util 与 sse_util 的作用和 SSE 事件类型，实现查询 web api 三个接口并修复三处 bug，整理 MongoDB 安装/CRUD/索引与 pymongo 历史记录管理，最后解读 node_item_name_confirm 节点。

本篇要点

  • task_util 与 sse_util 的作用和关联，以及 SSE 六种事件类型
  • 查询 web api 接口实现（html / 健康检查 / 提问），流式与非流式的区别
  • 三处 bug 修复：final 提前推送、节点状态缓存、清空 task
  • MongoDB 安装、CRUD 命令与索引原理
  • pymongo 封装与历史聊天记录管理，node_item_name_confirm 节点

内容回顾

1、上一天主题（Day07）

  • 导入模块三个接口：html 页面返回、文件上传、状态查询
  • 查询模块 8 个节点的图结构定义，以及并发节点 state 增量更新报错
  • SSE 基本使用：数据格式、前端 EventSource 三种交互方案、task_util 与 sse_util 的关系

今天内容

1、task_util 和 sse_util 作用说明与 SSE 事件类型

1.1 作用说明

task_util：

1
2
3
4
task_util 主要是存储节点和任务状态!
第一级存储当前类的 dict[task_id, []] -> 导入模块
第二级存储时候(第三个参数 is_stream = True)
    task_push_queue(task_id) -> [push_to_session(sse)] -> sse 的队列中, 供 sse 响应(查询模块)

sse_util：

1
2
sse_util 主要存储数据, 内有 dict{session_id: queue},
提供一个生成器方法, 响应 yield 数据..

1.2 事件类型

1
2
3
4
5
6
7
8
9
10
11
class SSEEvent:
    READY    = "ready"       # 连接建立：后端的流式响应和前端 sse 技术已经建立连接! 前端可以等待接收数据! [不需要管]
    PROGRESS = "progress"    # 任务节点进度：add_done/running, update_task_status -> progress 三个状态一起包, 推送到前端
    DELTA    = "delta"       # LLM 流式输出增量：流式响应的时候, 返回大模型反馈的增量字符串!
                             #   我们想用这个事件类型给前端返回! (不对) 我们无需直接响应!
                             #   我们只需要把数据推送到队列中..
    FINAL    = "final"       # 最终完整答案：流式的时候, 需要调用, 最终完整推送字符串和图片链接, 供前端显示...
                             #   push_to_session(session_id, final, {answer: 回答字符串, image_urls: [图片地址]})
                             #   es.close(); 关闭当前的流式链接!! 后续所有推送都是失效的!
    ERROR    = "error"       # 错误信息
    CLOSE    = "__close__"   # 关闭连接信号

注意： 除了 ready、progress（add）、__close__ 不需要自己推送，剩下的 delta、final、error 需要自己调用 push_to_session 推到 sse 队列中即可。

1
push_to_session(session_id, "delta", {"delta": 增量字符串})  # 即 data

2、查询 web api 接口实现

2.1 准备前置依赖导入

  1. 将 chat.html 页面导入到 html 文件夹
  2. 实现 html 返回接口

2.2 健康状态检查

1
2
3
4
5
6
7
8
9
10
11
12
13
# 健康检查的响应 json
class HealthResponseSchema(BaseModel):
    code: int = 200
    message: str = None


@app.get("/health")
def health():
    logger.info(f"完成健康检查!{datetime.datetime.now()}")
    return HealthResponseSchema(
        code=200,
        message=f"完成健康检查!{datetime.datetime.now()}"
    )

2.3 提问接口

  • 流式（异步，立即返回，后续异步执行，走 sse）
1
2
if is_stream:
    backgroundtasks.add_task(xx)
  • 非流式（同步，不会立即返回，等待图执行完毕，一起返回）
1
xx()

本质都是执行图！执行的方式不一样！

实现步骤：

  1. 获取执行的 state 参数 (session_id, query, is_stream)
  2. 定义一个执行查询图的方法（普通方法，session_id, query, is_stream）-> state
1
2
3
4
5
步骤1: try + state 生成 + 图执行 + return ...
步骤2: 考虑异步流式问题, 是否将状态添加到队列
       update_task_status(session_id, 状态, is_stream)
       if is_stream:
           create_sse_queue(session_id)
  1. 判断是同步还是异步
1
2
3
4
5
6
是异步:
    backgroundtasks.add_task(xx)
    return {异步结果}
是同步:
    state = xx()
    return {同步结果}

2.4 bug 修复

bug 1：流式返回的时候，任务状态完成了还是在进行中。

bug 2：流式返回的时候，有两个节点显示没有完成。

1
2
3
4
5
6
7
8
9
10
11
# 执行完毕以后 [final - sse - is_stream = True]
if is_stream:
    push_to_session(
        session_id,
        SSEEvent.FINAL,
        {
            "answer": result_state.get('answer'),
            "status": "completed",
            "image_urls": result_state.get('image_urls', [])
        }
    )

bug 3：第二次开始提问，节点起始状态就是完成，不真实！

节点状态被缓存了——用的是上一次的结果。不应该这样：

1
2
3
sse -> sse_utils -> queue (每次都是新的) -> task_utils 三个字典 (是旧的) 8 个已完成节点!
异步执行 -> 第一个节点 -> add_running_list(session_id) -> push_to_queue(session_id)
        -> session_id -> 三个字典获取数据 -> queue

修复方式：每次请求都做两件事。

1
2
1. 清空 queue
2. 清空 session_id 对应的 task 的字典
1
2
3
4
5
# 清空 task 字典! 不管是流式还是非流式都需要!!!
clear_task(session_id)
# 如果是流式, 我们可以提前创建队列! session_id : queue
if is_stream:
    create_sse_queue(session_id)

3、MongoDB 介绍、安装与基本使用

3.1 介绍 MongoDB

既快（内存）又稳（结构化数据）。

3.2 安装

1
2
3
4
5
6
7
docker pull mongo:latest
docker run -d --name mongo -p 27017:27017 mongo:latest
docker ps | grep mongo
docker stop mongo
docker start mongo
docker restart mongo
docker exec -it mongo mongosh   # == cmd - mysql -uroot -pxxx == navicat sqlyog

安装 mongo 的客户端：

1
https://www.mongodb.com/try/download/compass

3.3 常用命令（CRUD）

参考文档：https://www.mongodb.com/zh-cn/docs/manual/tutorial/insert-documents/

库和表操作：

1
2
3
4
5
6
7
8
// 创建和切换库：有就切换，没有就创建
use 库名称

// 表创建和切换：有就切换，没有就创建
db.表名

// 删库跑路
db.dropDatabase()

插入数据：

1
2
db.表名.insertOne({ BSON key: "" / 111 / true })
db.表名.insertMany([...])

查询数据：

操作符 含义
$eq 如果这些值相等，则返回 true
$gt 如果第一个值大于第二个值，则返回 true
$gte 如果第一个值大于等于第二个值，则返回 true
$lt 如果第一个值小于第二个值，则返回 true
$lte 如果第一个值小于等于第二个值，则返回 true
$ne 如果值不相等，则返回 true 
1
2
3
4
5
6
db.表名.find()                           // 查询全部
db.表名.find({key: 值})                   // key = 值 || {key: {特殊符号: 值}} -> 单个参数
db.表名.find({name: "张三", age: 18})      // 多个参数, 默认是 and
db.表名.find({$or: [{name: "张三"}, {age: 18}]})  // 或
db.表名.find({key: x, key: x, $or: [{}, {}]})     // and + or 组合
db.表名.find().sort({key: 1 / -1}).skip(number).limit(number)

更新数据：

1
2
// updateOne 更新符合的第一条 / updateMany 更新符合的所有({条件}, {$set: {key: 新的值}})
db.users.updateOne({name: "李四"}, {$set: {age: 33, city: "Dongdaihe"}})

删除数据：

1
db.users.deleteOne / deleteMany({ name: "王五" })

3.4 索引

单列索引：常用于查询的条件列，设置索引可以提高查询效率。

1
2
3
db.<collection>.createIndex({ <field>: 1 / -1 })
// 1  索引的顺序就正序, 从小到大
// -1 索引的顺序就倒序, 从大到小

选 1 还是 -1 呢？随便！凭感觉！

复合索引：用于条件查询 + 排序场景。

1
db.<collection>.createIndex({<field1>: <sortOrder>, <field2>: <sortOrder>})

例如：查询 session_id 对应的历史会话前 20 条。

1
db.xxxx.find({session_id: session_id}).sort({ts: -1}).limit(20)

为什么复合索引适合「条件查询 + 排序」场景？因为复合索引根据第一列进行排序，第二列在第一列的基础上再进行排序。也就是说，第二列以后不具备单独提高查询效率的能力了，只有在满足第一列的条件下，才能快速定位到结果。

1
2
session_id = session_id  -> 在所有的数据中快速对应!        -> 复合索引的第一列!!! -> 正序 | 倒序
sort(ts, -1)             -> 快速将数据倒序排序, 全部导出(不是筛选) -> 复合索引的第二列!!! -> sort 排序的方式

面试题：为什么有索引就能提高查询效率？为啥不把所有的字段都设置索引？

索引是一种额外的数据结构！脱离原始数据存储顺序，是更高效搜索的数据（红黑树等，logN）。索引需要占有磁盘或者内存空间！不常用的字段设置了，会浪费空间。

4、使用 MongoDB 进行历史聊天记录管理

4.1 pymongo 的基本使用

1
uv add pymongo
1
2
3
4
5
6
7
8
9
10
11
12
from pymongo import MongoClient

uri = "mongodb://localhost:27017/"
client = MongoClient(uri)

# use test_database
database = client["test_database"]

# db.test_collection.
collection = database["test_collection"]

# crud：命令相同! 例如: insertOne -> insert_one ...

4.2 封装一个 mongo_client（解读代码）

封装后对外提供库、集合以及索引的创建，以及三个核心方法：

  • save_chat_message(, message_id)：保存或者更新 -> 保存数据的主键 id
  • get_recent_messages(session_id, limit) -> list[dict]
  • clear_history(session_id) -> 清空的数量

封装类转移到 infra 目录，粘贴即可。

4.3 历史聊天记录的 BSON 数据格式

1
2
3
4
5
6
7
session_id
role
text
rewritten_query
item_names
image_urls
ts
  • role = user：提问，text -> 原始的问题 original_query
  • role = assistant：回答，text -> answer 回答内容
  • image_urls：为了回答提供的图片

后续想使用历史记录作为上下文以及 item_name 识别的依据。

注意：item_names 必须有值！必须是有效的问题或者回答！

4.4 存储历史对话信息

  • 第一个节点 -> role = user 的问题记录
  • 最后一个节点 -> role = assistant 的回答记录

4.5 完善 server api 接口

  1. 查询历史对话记录
  2. 清空历史对话记录（session_id）
1
2
3
4
5
6
7
@app.delete("/history/{session_id}")
def clear_history(session_id: str):
    deleted_count = history_repository.clear_session(session_id=session_id)
    return HistoryClearResponseSchema(
        message=f"session_id:{session_id}历史聊天记录已经清空!",
        deleted_count=deleted_count
    )
1
items.reverse() -> chat.html 677

5、node_item_name_confirm 节点

5.1 作用

识别 item_names（提问 / 历史上下文），并把 original 重写成 rewritten...。有可能 item_name 识别不到（1. 分低；2. 完全没识别）。

5.2 参数

使用（读取）：

  • session_id -> 获取用户的聊天记录
  • original_query -> 原始问题
  • is_stream

生成（写入）：

  • item_names
  • rewritten_query
  • answer（可能）

5.3 实现步骤

步骤 1：获取并且校验参数（state）-> session_id / original_query

1
2
3
4
session_id, original_query = get_data_and_validates(state)
# 获取数据
# 非空校验
# 返回结果

步骤 2：获取（有效的）历史聊天记录，并且拼接成 history_text 提示词

1
2
3
4
5
6
7
8
9
10
11
12
13
get_history_messages_and_context(session_id) -> history_text
# 获取 mongodb client
# 根据 session_id 获取最近聊天记录(10)
# 列表推导式推到有效的 messages [item_names]
# 将有效的 messages 拼接成 history_text
#   {
#     id, session_id, 序号: 1 / 2,
#     role [提问还是回答] text [user -> 原始问题 -> rewritten_query;
#                               assistant -> 回答 -> text[:60]],
#     rewritten_query, item_names => [主体: x, x, x],
#     image_urls, ts
#   } -> item_name 识别和问题重写
return history_text

步骤 3：调用模型识别 item_names 和 rewritten_query -> json（item_names 不一定准，大模型）

1
2
3
4
5
6
7
dict = call_llm_item_name_and_rewritten(history_text, original_query)
# 获取模型对象
# 加载提示词字符串
# 封装提示词对象
# 包装 chains -> StrOutput... JsonOutput ->
# dict = 执行链获取字典
return dict

面试题：怎么确保大语言模型返回的数据是 JSON 格式？

  1. 提示词强调返回数据格式，提供返回数据示例
  2. 模型的 json 模式设置
  3. 做好返回值的参数校验 JsonOutputParser ``` json -> json dict
  4. 返回字典 key 的校验：item_names 无 -> []；rewritten_query 无 -> original_query

步骤 4：item_names 进行向量检索（item_name）对应的 collection -> [item_names milvus 分高 / 分低]

步骤 5：置信度判断 milvus -> 分 -> 可信的列表 | 不可信的列表

步骤 6： 没有识别 -> answer；识别了但分不高 -> answer -> state answer…；识别出来且分高 -> state [item_names / rewritten_query]

步骤 7：保存本次提问的历史对话（item_names / rewritten_query），并相应 return。

5.4 本周总结

  1. 理解 document 切割的四个核心原则 [标题 / 长了短切 / 短了合并 600 400 1000 50]
  2. item_name 识别 -> 封装提示词 -> 调用模型 -> 准备 milvus 集合（相似度 / 索引类型 / field 数据类型）-> 数据存储
  3. chunks 批量向量化 -> 提高生成向量的效率 -> 批量处理 -> bge-m3 -> 8192 -> 1000 -> 6 一批
  4. chunks 的入口处理
  5. fastapi 的基本使用（路径设计 / 参数接收[路径 / 查询 / json / 文件] / 响应数据[json / 文件 / 流式] / 异步操作 back…）
  6. 导入模块对应接口（html / 上传文件 -> 接收文件 | 存储 | 异步调用 | 返回结果 / 轮询获取状态 task_util）
  7. 查询的状态 / 节点 / 图结构 [并发节点 state 问题 || add_done/running_list(1,2,3 is_stream)]
  8. sse 技术的使用 [后端的流式 + 前端 EventSource：1. 只有 data；2. event + data；3. 项目结构：触发 -> 存储队列 -> 流式]
  9. 正式介绍 task_utils | sse_utils 作用和区别 / 理解项目的事件类型 process、delta、final、error
  10. 查询模块的四个接口 -> 基本使用
  11. mongodb 的基本使用：库、表、crud 以及索引！以及 pymongo 的使用
  12. 实现了历史记录的两个接口：查询 | 清空 -> 前端页面添加一个
  13. item_name_confirm：校验、加载 history、调用模型