目录1 前端交互设计2 服务接口设计2.1 接口代码实现3 历史对话记录管理1 前端交互设计本节介绍如何基于 FastAPI 搭建后端服务提供页面访问和查询接口。检索 Web 聊天界面chat.html的核心结构与数据流转逻辑该页面直接决定了后端接口的设计规范。1页面核心组件顶栏 (Topbar)展示服务连接状态与流式开关。对话区 (Chat)用户气泡展示提问内容。系统气泡集成文本答案与处理进度折叠面板实时展示后台检索/重排/生成的执行状态。输入区 (Composer)支持快捷发送与多行输入。2数据交互闭环前端与后端的全双工交互流程如下会话初始化加载页面时生成或读取session_id作为用户唯一标识。提交任务点击发送 - POST/query携带问题与流式标记 - 获取session_id确认任务已接收。建立长连接 (SSE)立即通过EventSource监听/stream/{session_id}建立实时通信管道。事件驱动更新ready: 连接握手成功。progress:实时更新进度条如正在检索… - 检索完成。delta:流式逐字输出打字机效果。final: 接收完整答案与引用源。基于此逻辑后端需提供/query任务提交与/stream事件推送两个核心接口。2 服务接口设计本节详细定义查询服务的所有对外接口包括页面访问、任务提交、流式推送及历史管理。1页面访问接口路径:/chat.html(GET)功能: 返回前端聊天界面。响应: HTML 静态页面。2检索查询接口路径:/query(POST)功能: 接收用户提问并启动后台处理图逻辑。参数:{query:万用表怎么测量电压,session_id:可选未传则后台自动生成,is_stream:true// 是否启用流式推送}响应 (is_stream: true):{message:结果正在处理中...,session_id:xxx-uuid}响应 (is_stream: false):{message:处理完成,session_id:xxx,answer:回答内容...,done_list:[]}3流式获取接口 (SSE)路径:/stream/{session_id}(GET)功能: 建立 SSE 长连接实时推送任务进度与生成文本。推送数据格式 (JSON):progress 事件:{done_list: [节点A, 节点B], running_list: [节点C]}delta 事件:{text: 生成的增量字符}final 事件:{answer: 完整最终答案}error 事件:{error: 错误详情}4会话历史查询路径:/history/{session_id}(GET)功能: 从 MongoDB 中获取当前会话的历史聊天记录。参数:limit(可选默认50条)响应:{session_id:xxx,items:[{_id:str(r.get(_id))ifr.get(_id)is not Noneelse,session_id:r.get(session_id,),role:r.get(role,),text:r.get(text,),rewritten_query:r.get(rewritten_query,),item_names:r.get(item_names,[]),ts:r.get(ts)}]}5清空会话历史路径:/history/{session_id}(DELETE)功能: 删除 MongoDB 中该会话的所有记录。响应:{ message: History cleared, deleted_count: 10 }6健康检查接口路径:/health(GET)功能: 检查服务存活状态。响应:{ ok: True }2.1 接口代码实现1导入查询页面将资料chat.html添加到app/query_process/page文件夹中3前后端交互说明知识库查询服务的流式响应流程 涉及到四个核心模块的协同工作Web 服务层 ( query_service.py ): 负责接收请求、建立 SSE 连接。SSE 工具层 ( sse_utils.py ): 负责管理消息队列、打包和推送事件。任务状态层 ( task_utils.py ): 负责记录每个节点的执行进度并自动触发 SSE 推送。图节点执行层 ( query_process/ ): 实际的业务逻辑节点如检索、Rerank、生成答案它们通过更新状态来驱动进度条。核心流程时序图2定义和实现api接口服务在app/query_process/api目录下创建query_service.py我们将代码拆解为以下几个部分进行实现。首先引入 FastAPI、Pydantic 以及项目内部的工具类。frompathlibimportPathimportuuidimportuvicornfromfastapiimportFastAPI,BackgroundTasks,HTTPException,Requestfromfastapi.responsesimportFileResponse,StreamingResponsefrompydanticimportBaseModel,Fieldfromstarlette.middleware.corsimportCORSMiddlewarefromapp.core.loggerimportloggerfromapp.utils.task_utilsimport*fromapp.utils.sse_utilsimportcreate_sse_queue,SSEEvent,sse_generatorfromapp.clients.mongo_history_utilsimport*fromapp.query_process.agent.main_graphimportquery_app# 定义fastapi对象appFastAPI(titlequery service,description掌柜智库查询服务)# 跨域配置app.add_middleware(CORSMiddleware,allow_origins[*],allow_methods[*],allow_headers[*],)# 返回chat.html页面app.get(/chat.html)asyncdefchat():current_dir_parent_pathPath(__file__).absolute().parent.parent chat_html_pathcurrent_dir_parent_path/page/chat.htmlifnotchat_html_path.exists():logger.error(f页面不存在{chat_html_path})raiseHTTPException(status_code404,detailf没有查询到页面地址为{chat_html_path})logger.info(成功加载 chat.html 页面)returnFileResponse(chat_html_path)3定义数据模型 (Pydantic)定义前端请求的数据结构确保参数类型安全并增加兼容字段。# 定义接口接收的数据结构classQueryRequest(BaseModel):查询请求数据结构query:strField(...,description查询内容)session_id:strField(None,description会话ID)is_stream:boolField(False,description是否流式返回)4实现核心查询逻辑这是最关键的逻辑部分包含后台任务处理函数run_query_graph和 API 接口/query。# 核心查询处理函数defrun_query_graph(session_id:str,user_query:str,is_stream:boolTrue):logger.info(f[{session_id}] 开始执行查询流程流式模式{is_stream})default_state{original_query:user_query,session_id:session_id,is_stream:is_stream}try:query_app.invoke(default_state)update_task_status(session_id,TASK_STATUS_COMPLETED,is_stream)logger.info(f[{session_id}] 查询流程执行完成)exceptExceptionase:logger.exception(f[{session_id}] 查询流程异常{str(e)})update_task_status(session_id,TASK_STATUS_FAILED,is_stream)ifis_stream:push_to_session(session_id,SSEEvent.ERROR,{error:str(e)})# 查询接口app.post(/query)asyncdefquery(background_tasks:BackgroundTasks,request:QueryRequest): 1 解析参数 2 更新任务状态 3 调用处理流程图 4 返回结果 user_queryrequest.query session_idrequest.session_idifrequest.session_idelsestr(uuid.uuid4())is_streamrequest.is_streamifis_stream:create_sse_queue(session_id)logger.info(f[{session_id}] 已创建 SSE 消息队列)update_task_status(session_id,TASK_STATUS_PROCESSING,is_stream)logger.info(f[{session_id}] 任务开始处理查询内容{user_query})ifis_stream:background_tasks.add_task(run_query_graph,session_id,user_query,is_stream)logger.info(f[{session_id}] 流式任务已提交至后台执行)return{message:结果正在处理中...,session_id:session_id}else:run_query_graph(session_id,user_query,is_stream)answerget_task_result(session_id,answer,)logger.info(f[{session_id}] 非流式查询处理完成)return{message:处理完成,session_id:session_id,answer:answer,done_list:[]}# SSE 流式推送接口app.get(/stream/{session_id})asyncdefstream(session_id:str,request:Request):logger.info(f[{session_id}] 客户端已建立 SSE 流式连接)returnStreamingResponse(sse_generator(session_id,request),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,X-Accel-Buffering:no})5健康检查接口# 健康检查app.get(/health)asyncdefhealth():服务健康检查logger.info(健康检查接口调用成功)return{ok:True}6启动查询服务if__name____main__:uvicorn.run(app,host127.0.0.1,port8001)3 历史对话记录管理上下文连续保留关键信息支持追问、补充条件与多轮推理不断档。指代消歧如果只考虑本次的问题往往不足以让大模型理解用户的上下文含义尤其是一些指代比如“这个设备”“它”等等。交互确认为明确问题的一些信息Agent 可能会反问用户一些问题经过几次确认后才能准确解答后续问题比如商品的信号。识别用户长期记录对话可以记住用户偏好最终形成用户画像。咱们项目这里没有使用 LangChain 的 Checkpointer 方式管理会话而是自定义了一套基于 MongoDB 的持久化管理策略。这样做的有很多好处更加灵活自主可以根据条件范围查询对话可以给对话自定义格式管理查询写入内容。但是相对来说需要做的开发工作也比较多。MongoDB 是一种文档数据库特别适用于存储海量数据结构简单关系简单的数据。性能和单表容量上都比 MySQL/PostgreSQL 等传统关系型数据库要好。虽比不上 Redis 这种内存数据库但在持久化能力和检索功能又比内存数据库强上很多。缺点是不适合保持有复杂关联关系查询复杂的场景。所以对于这种结构简单、查询简单但数据庞大的对话信息特别合适。尤其是 MongoDB 的存储基本单元就是一份JSON文档与对话也是非常契合。