记录导入模块三个接口（html 返回、文件上传、状态查询）的实现，梳理查询模块 8 个节点的图结构与并发 state 报错，并整理 SSE 基本用法及 task_util 与 sse_util 的关系。

本篇要点

  • 导入模块三个接口：html 页面返回、文件上传、状态查询
  • 查询模块图结构定义：8 个节点的读 / 写字段规划
  • 并发节点 state 增量更新的报错原因与规避方式
  • SSE 技术的基本使用与前端三种交互方案

内容回顾

1、上一天主题（Day06）

  • node_bge_embedding：item_name + content 拼接成向量，批量处理，8192 token 上限、6 个一批
  • node_import_milvus：创建 schema 与 index，先删后插写入 Milvus
  • FastAPI 使用回顾：路径、参数（路径 / 查询 / json / 文件）、响应（json / 文件 / 流式）、异步 BackgroundTasks

今天内容

1、导入模块的三个接口实现

1.1 三个接口的约定

导入模块一共提供三个对外的 HTTP 接口，接口定义如下。

接口 1：返回 html 页面

1
2
3
4
url:     /import/html
method:  get
参数:    没有
响应:    import.html 文件

接口 2：文件上传

1
2
3
4
5
6
7
8
9
url:     /upload
method:  post
参数:    files = 文件列表
响应:
{
   "code": 200 / 500,
   "message": "文件上传成功/失败",
   "task_ids": ["任务id"]
}

接口 3：状态查询

1
2
3
4
5
6
7
8
9
10
11
url:     /status/{task_id}
method:  get
参数:    path -> task_id
响应:
{
   "code": 200,
   "task_id": "uuid-1",
   "status": "processing",
   "done_list": ["node_entry", "node_pdf_to_md", "node_md_img"],
   "running_list": ["node_document_split"]
}

1.2 准备工作：定义两个响应 Schema

1、定义上传文件响应的 JSON 数据格式

1
2
3
4
class UploadResponseSchema(BaseModel):
    code: int = 200
    message: str | None = None
    task_ids: list[str]

2、定义一个响应状态 JSON 数据格式

1
2
3
4
5
6
class StatusResponseSchema(BaseModel):
    code: int = 200          # 业务状态
    task_id: str             # task_id
    status: str              # 当前 task_id 解析文件的整体状态：正在进行 / 已经完成 / 失败
    done_list: list[str]     # 本次 task_id 对应已经完成的节点
    running_list: list[str]  # 本次 task_id 正在进行的节点

1.3 实现接口

第一步：创建导入接口文件

1
/http/import_server.py

第二步：粘贴一些导入文件。

第三步：html 页面返回接口。

第四步：文件上传接口。

先定义 api 接口函数：

1
2
3
@app.post(...)
def 函数(backgroundtask, files: list[UploadFile]):
    ...

函数内部按四个步骤实现：

  1. 准备三个参数
1
2
3
task_id = str(uuid.uuid4())
local_dir = output / datetime.now().strftime("%Y%m%d") / task_id
local_file_path = local_dir / file.filename
  1. 将文件存储到本地磁盘
1
2
with local_file_path.open("wb") as out_file:
    shutil.copyfileobj(file.file, out_file)
  1. 异步调用执行函数
1
backgroundtask.add_task(函数名, task_id, local_file_path, local_dir)
  1. 返回结果
1
return xxResponse(code=..., message=..., task_ids=[...])

再定义真正执行图的函数：

1
2
3
4
5
def xx(task_id, local_file_path, local_dir):
    try:
        state = (task_id, local_file_path, local_dir)
        state = import_graph_app.invoke(state)  # 没必要使用 stream(state)
    ...

第五步：状态查询接口。

根据 task_utils 获取任务状态、正在进行的以及已经完成的任务列表，装到响应数据格式里即可。

注意：

  1. 要修改异步图执行！添加 update_task_status 更新状态。
  2. 文件上传也算一个节点：存储之前 add_running，存储完成 add_done，节点名固定，与 task_utils 的 map key 对比。

2、查询模块图结构定义

查询模块的图一共规划了 8 个节点，先回顾每个节点的作用、读取与写入的 state 字段。

2.1 用户提问

用户提问是整个流程的起点。

2.2 意图识别和问题重写节点

  • 意图识别：由「问题 + 历史聊天记录」确定 item_name，结果分三种：
    1. 明确的 item_name（向量数据库搜索相似度分高于 0.75）
    2. 不确定的 item_name（相似度分没有达到 0.75，落在 0.75 - 0.6 之间）
    3. 完全不确定 item_name（没有高于 0.6 分）
  • 问题重写：把「他好用么？」改成「烫金机好用么？」——口语化、没有主语的问题要改成标准的问题，利用模型 + 历史聊天记录完成。

该节点读写字段：

  • 读 original_query：用户原始问题，例如「他好吃么？？」
  • 读 history：历史对话记录，用于上下文理解（history 历史记录来自数据库）
  • 读 session_id：会话追踪。一个客户端一个 session，每次问题不并发！
  • 写 item_names：识别出的商品名称列表（确认的实体名称）
  • 写 rewritten_query：改写后的精准问题
  • 写 answer：如果无法确认或需要反问，直接写入答案并提前结束流程
1
2
3
answer 示例：
  本次提问没有关联的主体! 请明确后再提问!
  本次提问关联的主体是: 1, 2, 3, 4 嘛? 请您确认!

2.3 问题向量搜索

重写后的问题 -> 向量数据库。

  • 读 rewritten_query：改写后的问题进行向量检索
  • 读 item_names：限定检索范围（只在这些商品内搜索）
  • 读 session_id：任务追踪
  • 写 embedding_chunks：普通向量检索回来的 Top-K 切片

2.4 假设性问题向量搜索

重写后的问题 -> LLM -> 润色出一个初版本的答案，再用「问题 + 答案」去向量数据库搜索。

  • 读 rewritten_query：生成假设性答案的原材料
  • 读 item_names：限定检索范围
  • 读 session_id：任务追踪
  • 写 hyde_embedding_chunks：HyDE 策略检索回来的切片

2.5 mcp 网络信息搜索

问题 -> mcp[模型上下文协议] -> bailian 提供网络搜索工具 -> 网络数据搜索。

  • 读 rewritten_query：网络搜索的查询词
  • 读 session_id：任务追踪
  • 写 web_search_docs：联网搜索返回的文档列表

2.6 同源数据融合

问题搜索和假设性问题搜索各 10 条 -> 向量数据库 10 条 -> 用 rrf 算法融合成 10 条。

  • 读 embedding_chunks：向量检索结果
  • 读 hyde_embedding_chunks：HyDE 检索结果
  • 读 session_id：任务追踪
  • 写 rrf_chunks：RRF 融合排序后的统一列表

2.7 非同源进行模型打分 + 动态 topk 算法

rrf 10 条 + mcp 10 条 -> reranker 打分 20 条 -> 动态 topk => topk 数据。

  • 读 rrf_chunks：待重排的文档列表
  • 读 web_search_docs：网络搜索结果
  • 读 rewritten_query：用于计算相关性得分
  • 读 session_id：任务追踪
  • 写 reranked_docs：重排序后的最终 Top-K 文档

2.8 进行数据的格式化处理

模型进行润色 + 图片的处理 + 前端的响应。

  • 读 reranked_docs：作为 LLM 回答的参考依据
  • 读 rewritten_query：用户问题
  • 读 history：历史对话
  • 读 session_id / is_stream：控制输出方式
  • 写 answer：最终生成的答案
  • 写 image_urls：从引用文档中提取的图片链接

2.9 并发节点的 state 增量更新问题

在图节点测试时出现如下报错：

1
2
langgraph.errors.InvalidUpdateError: At key 'session_id':
Can receive only one value per step. Use an Annotated key to handle multiple values.

报错原因：图节点对 state 进行增量更新时出现冲突。

注意：并发节点，每个节点只能返回自己节点修改的 key！并发节点不能返回相同的 key！！

举例说明并发场景：

1
2
3
问题搜索   -> session_id
假设搜索   -> session_id   -> graph state 合并 -> node_rrf
网络搜索   -> session_id

三个并发节点都返回了 session_id 这个相同的 key，就会触发上面的 InvalidUpdateError。

3、SSE 技术基本使用

3.1 SSE 的作用

SSE 是一种服务端单向向客户端（前端）推送数据的技术，分两部分：

  • 第一部分：后端 -> 推数据，StreamingResponse
  • 第二部分：前端 -> 接数据，EventSource（js）

3.2 SSE 的数据格式

1
2
event:[空格]事件类型(自定义)\n
data:[空格]数据 普通字符串 json字符串\n\n   [空行 -> 一个 sse 的数据包]

后端写法：

1
2
3
4
StreamingResponse(生成器函数(), ...)
# 生成器函数中：
yield "event: completed\n"
yield "data: json字符串\n\n"

3.3 前后端交互的三种方案

方案 1：基础交互款（没有事件，只有数据，就一个接口）

1
2
# 后端
StreamingResponse(生成器函数(), ...)
1
2
3
4
5
// 前端
es = new EventSource("http://后端的流式响应接口地址")
es.onmessage = function (event) {
    // event.data 逻辑代码修改页面
}

方案 2：进阶交互款（有事件）

1
2
3
4
# 后端
StreamingResponse(生成器函数(), ...)
yield "event: completed\n"
yield "data: json字符串\n\n"
1
2
3
// 前端
es = new EventSource("http://后端的流式响应接口地址")
es.addEventListener("事件类型", (e) => { e.data })

方案 3：模拟项目返回结构

1
2
3
4
后端：StreamingResponse(生成器函数(), ...) -> 循环 + yield -> 取值 -> 队列
      （异步对象 get 阻塞，没有值就等）[提问流式响应接口]
      模拟触发查询接口 -> 将数据添加到队列 [提问]
前端：...

3.4 项目提问和推送数据的交互流程

接口 1：提问接口

1
2
3
4
前端 => session_id / query => 问题接口:
   1. 接收参数
   2. 调用异步函数执行查询图对象
   3. 快速响应结果

接口 2：流式响应接口

1
2
前端 => /stream/{session_id} => 流式接口:
   获取 session_id 对应的结果 => 流式响应推送

推送是怎么做的？摒弃了生成器中「循环 + 定时休眠」的方式，改成生成器中「循环 + 阻塞获取队列」的函数。

1
2
3
4
2. 调用异步函数执行查询图对象 -> node node node
   add_running / done_task(session_id, 节点名, is_stream=True)
   session_id, 节点名 -> 字典 {session_id, []} -> session_id -> [节点名]
   is_stream = True: session_id -> [] -> 推送到队列中 Queue

3.5 理解 task_util 与 sse_util 的作用和关联

task_util：

1
2
3
4
task_util -> add_running_task / add_done_task / update_task_status(session_id, "节点名/任务状态", is_stream=False)
    -> is_stream=False -> 三个字段存储了对应的状态 -> 导入模块状态检索
    -> is_stream=True  -> task_push_queue(session_id) -> process, {done_list, run.., status:..}
                       -> 将数据推送到「队列」中

可以用一个比喻来理解：task_util 类似「外卖员将外卖放在货架」——负责把数据搬到货架旁边。

1
2
============= 以上就是外卖员的工作 =====================
============= 小区的物业保安(汤臣一品) =================

sse_util：

  1. 把 task_push_queue(session_id)（来自 task_util）真正地放在货架上，queue == push_to_session
  2. 属性：_session_stream: Dict[str, queue.Queue] = {session_id, 队列}
  3. 方法：
    • get_sse_queue(session_id: str)：根据 session_id 获取队列（不需要我们调用）
    • push_to_session(session_id, event, data)：stream_queue.put({"event": event, "data": data})
    • remove_sse_queue(session_id: str)：清空队列！每次提问执行之前先清空！
    • create_sse_queue(session_id: str)：创建队列，每次提问都需要创建一次队列！[每次调用]
  4. 封装了生成器方法：
1
sse_generator(session_id) -> 1. session_id 获取队列  2. while True  3. get  4. _sse_pack

今日小结：

  • 导入模块的三个接口
  • 查询 state / 节点 / 图结构 / 测试 -> 并发节点 -> state 报错 -> 增量
  • sse 的使用：1. 基本使用 2. 带有事件 3. 项目结构
  • 理解 task_util / sse_util 的关系和作用