# 16 - 电商问数：查询接口实现与依赖组装 --- **本章课程目标：** - 把第 15 章的假流式接口替换成真实问数工作流。 - 使用 FastAPI 依赖注入组装 Repository、Session、Client 和 Service。 - 使用 `lifespan` 管理 Qdrant、ES、MySQL、Embedding 等应用级资源。 **学习建议：** 这一章开始把 API 和问数工作流接起来。读的时候沿着一条请求走：HTTP 进入路由，路由拿到 `QueryService`，`QueryService` 调用 LangGraph，执行进度再通过 SSE 返回前端。依赖函数看起来多，但本质是在按层次组装底层对象；别把装配代码误读成业务逻辑。 **对应代码分支：** `16-api-query-service` --- 第 15 章已经验证了 `StreamingResponse + SSE` 可以正常工作，但当时返回的是 `fake_streamer()` 模拟数据。本章要把它换成真实的问数智能体。 最终调用链路如下： ```text 前端 / Apifox -> POST /api/query -> query_router.py 接收请求 -> Depends(get_query_service) 获取 QueryService -> QueryService.query(...) -> graph.astream(...) -> StreamingResponse 按 SSE 格式持续返回 ``` 完成这一章后，问数智能体就不再只是命令行里的内部脚本，而是一个可以通过 HTTP 调用的后端接口。 --- ## 1、API 代码目录结构 这一阶段 API 代码主要涉及下面几个文件： ```text shopkeeper-agent/ ├─ main.py # FastAPI 入口脚本，负责创建应用并注册路由 └─ app/ ├─ api/ │ ├─ routers/ │ │ └─ query_router.py # 查询接口路由，接收请求并返回流式响应 │ ├─ schemas/ │ │ └─ query_schema.py # 查询接口请求体结构 │ ├─ dependencies.py # 查询接口依赖项，负责组装 QueryService │ └─ lifespan.py # FastAPI 生命周期事件，负责初始化和关闭外部客户端 └─ services/ └─ query_service.py # 查询接口核心业务逻辑，负责调用问数工作流 ``` 可以先把它们分成五层： | 层次 | 文件 | 主要职责 | | -------- | ------------------ | --------------------------------------------------------- | | 入口层 | `main.py` | 创建 FastAPI 应用，注册生命周期和路由 | | HTTP 层 | `query_router.py` | 定义 `/api/query`，接收请求，返回流式响应 | | 业务层 | `query_service.py` | 创建 `State` / `Context`，调用 LangGraph，并包装 SSE 消息 | | 依赖层 | `dependencies.py` | 组装 Service、Repository、Session、Client | | 生命周期 | `lifespan.py` | 应用启动时初始化客户端，应用关闭时释放连接 | 这几个文件的关系可以记成一句话： ```text main 挂路由 router 接请求 service 调工作流 dependencies 组装对象 lifespan 管理应用级资源 ``` --- ## 2、入口和路由：让 HTTP 请求先进来 这一部分只解决一个问题：外部请求怎样进入我们的 Python 代码。 ### 2.1 main.py：创建应用并挂载路由 项目对应文件路径：`shopkeeper-agent/main.py` ```python from fastapi import FastAPI from app.api.lifespan import lifespan from app.api.routers.query_router import query_router # lifespan 交给 FastAPI 管理，用于在服务启动和关闭时统一初始化与释放外部客户端 app = FastAPI(lifespan=lifespan) # 把查询路由注册进应用；没有挂载时，/docs 和真实 HTTP 请求都访问不到该接口 app.include_router(query_router) ``` 这里最关键的是两行： ```python app = FastAPI(lifespan=lifespan) app.include_router(query_router) ``` `lifespan` 负责服务启动和关闭时的资源管理，`include_router` 负责把 `query_router.py` 中定义的接口挂到 FastAPI 应用上。 如果忘了 `app.include_router(query_router)`，即使 `query_router.py` 里写了 `/api/query`，FastAPI 应用也不知道这个接口存在，`/docs` 页面里自然也看不到它。 ### 2.2 query_router.py：路由层只做 HTTP 相关的事 项目对应文件路径：`shopkeeper-agent/app/api/routers/query_router.py` ```python from typing import Annotated from fastapi import APIRouter, Depends from starlette.responses import StreamingResponse from app.api.dependencies import get_query_service from app.api.schemas.query_schema import QuerySchema from app.services.query_service import QueryService # 当前模块只维护查询相关接口，避免后续所有 API 都挤在 main.py 中 query_router = APIRouter() @query_router.post("/api/query") async def query_handler( # FastAPI 会把前端 JSON 请求体自动解析成 QuerySchema query: QuerySchema, # FastAPI 会调用 get_query_service，递归组装它依赖的仓储和客户端 query_service: Annotated[QueryService, Depends(get_query_service)], ): return StreamingResponse( # query.query 是用户问题字符串；QueryService.query 返回异步生成器 query_service.query(query.query), media_type="text/event-stream", ) ``` 路由层只做三件事： ```text 1. 接收 QuerySchema 请求体 2. 通过 Depends 拿到 QueryService 3. 把 QueryService.query(...) 交给 StreamingResponse ``` 它不直接创建 Qdrant、ES、MySQL Repository，也不直接执行图节点。路由层越薄，后续接口越容易维护。 本节最重要的是这一行： ```python query_service: Annotated[QueryService, Depends(get_query_service)] ``` 它表达的是：当前接口需要一个 `QueryService`，这个对象由 `get_query_service()` 提供。路由只声明“我需要什么”，至于怎么创建，交给 `dependencies.py`。 --- ## 3、QueryService：把一次请求变成一次图执行 `QueryService` 是这一章的核心。它把 HTTP 层传入的自然语言问题，转换成一次 LangGraph 工作流执行。 如果不抽出 `QueryService`，路由函数里就会堆满这些逻辑： ```text 解析请求体 创建 DataAgentState 创建 DataAgentContext 准备 Repository 和 Client 调用 graph.astream(...) 把 chunk 包成 SSE 处理异常 ``` 这会让路由层既懂 HTTP，又懂工作流，又懂底层依赖，边界很快就乱了。抽出 `QueryService` 后，分工会清楚很多： ```text query_router.py -> 负责 HTTP 层：接请求、返响应 query_service.py -> 负责业务层：调问数智能体、组织流式输出 ``` ### 3.1 从测试脚本迁移到 API 前面测试 LangGraph 工作流时，核心代码大致是： ```python state = DataAgentState(query="统计华北地区的销售总额") context = DataAgentContext(...) async for chunk in graph.astream( input=state, context=context, stream_mode="custom", ): print(chunk) ``` 迁移到 API 后，变化主要有三处： | 测试脚本 | API 接口版本 | | --------------------------- | ---------------------------------- | | 用户问题写死在代码里 | 使用请求体传入的 `query` | | 上下文在测试脚本里手动创建 | 由 `QueryService` 接收依赖后创建 | | `print(chunk)` 输出到控制台 | `yield data: ...\n\n` 流式写给前端 | ### 3.2 QueryService 核心代码 项目对应文件路径：`shopkeeper-agent/app/services/query_service.py` ```python import json from langchain_huggingface import HuggingFaceEndpointEmbeddings from app.agent.context import DataAgentContext from app.agent.graph import graph from app.agent.state import DataAgentState from app.repositories.es.value_es_repository import ValueESRepository from app.repositories.mysql.dw.dw_mysql_repository import DWMySQLRepository from app.repositories.mysql.meta.meta_mysql_repository import MetaMySQLRepository from app.repositories.qdrant.column_qdrant_repository import ColumnQdrantRepository from app.repositories.qdrant.metric_qdrant_repository import MetricQdrantRepository class QueryService: def __init__( self, meta_mysql_repository: MetaMySQLRepository, embedding_client: HuggingFaceEndpointEmbeddings, dw_mysql_repository: DWMySQLRepository, column_qdrant_repository: ColumnQdrantRepository, metric_qdrant_repository: MetricQdrantRepository, value_es_repository: ValueESRepository, ): # MySQL 仓储分别负责元数据补全和真实数仓环境信息读取 self.meta_mysql_repository = meta_mysql_repository self.dw_mysql_repository = dw_mysql_repository # 召回链路依赖的向量检索、Embedding 和全文检索能力由依赖层注入 self.embedding_client = embedding_client self.column_qdrant_repository = column_qdrant_repository self.metric_qdrant_repository = metric_qdrant_repository self.value_es_repository = value_es_repository async def query(self, query: str): # State 只放会被图节点读写和合并的业务数据，外部工具对象不塞进 State state = DataAgentState(query=query) # Context 保存本次图执行需要复用的外部依赖，节点通过 runtime.context 读取 context = DataAgentContext( column_qdrant_repository=self.column_qdrant_repository, embedding_client=self.embedding_client, metric_qdrant_repository=self.metric_qdrant_repository, value_es_repository=self.value_es_repository, meta_mysql_repository=self.meta_mysql_repository, dw_mysql_repository=self.dw_mysql_repository, ) try: # stream_mode="custom" 对应节点内部 writer(...) 写出的进度消息 async for chunk in graph.astream( input=state, context=context, stream_mode="custom", ): # SSE 要求每条消息以 data: 开头，并以两个换行符结束 # ensure_ascii=False 保留中文进度文案，default=str 兜底处理日期等非 JSON 类型 yield f"data: {json.dumps(chunk, ensure_ascii=False, default=str)}\n\n" except Exception as e: # 流式接口已经开始返回后不能再改 HTTP 状态码，因此把异常也包装成一条 SSE 消息 error = {"type": "error", "message": str(e)} yield f"data: {json.dumps(error, ensure_ascii=False, default=str)}\n\n" ``` 这段代码里最重要的不是 `json.dumps`，而是两个对象：`state` 和 `context`。 ### 3.3 state 和 context 怎么区分 `state` 保存的是本次任务会不断变化的业务数据，例如： ```text query keywords retrieved_column_infos retrieved_metric_infos table_infos metric_infos sql result ``` `context` 保存的是节点运行时需要使用的外部能力，例如： ```text Embedding Client ColumnQdrantRepository MetricQdrantRepository ValueESRepository MetaMySQLRepository DWMySQLRepository ``` 可以这样记： ```text state：任务数据，图执行过程中会变 context：工具资源，节点执行时拿来用 ``` 不要把 Repository、Client 这类对象塞进 `state`。它们不是业务中间结果，而是节点执行时需要调用的外部能力，更适合放在 `context` 里。 ### 3.4 为什么异常也要包装成 SSE 普通接口出错时，可以直接返回 500 状态码。但流式接口不太一样：一旦 `StreamingResponse` 开始往外写数据，HTTP 响应头通常已经发送出去了，后面就不能再随便改状态码。 所以当前版本先把异常包装成一条 SSE 消息： ```python error = {"type": "error", "message": str(e)} yield f"data: {json.dumps(error, ensure_ascii=False, default=str)}\n\n" ``` 前端拿到这条消息后，可以按 `type=error` 展示错误状态。下一章会继续完善节点级异常处理、前后端联调和 `request_id` 日志追踪。 --- ## 4、dependencies.py：把对象创建交给依赖层 现在路由里已经声明了： ```python query_service: Annotated[QueryService, Depends(get_query_service)] ``` 这意味着项目必须提供 `get_query_service()`。 如果直接在路由里创建 `QueryService`，路由就会知道太多底层细节： ```text Qdrant 客户端怎么取 ES 客户端怎么取 MySQL Session 怎么创建和释放 Repository 怎么实例化 QueryService 需要哪些参数 ``` 这些不是 HTTP 层该关心的事。所以本项目把依赖组装统一放到： ```text shopkeeper-agent/app/api/dependencies.py ``` ### 4.1 先看最终依赖树 本章最终要组装的是 `QueryService`，依赖树大致如下： ```text get_query_service -> get_meta_mysql_repository -> get_meta_session -> get_embedding_client -> get_dw_mysql_repository -> get_dw_session -> get_column_qdrant_repository -> get_metric_qdrant_repository -> get_value_es_repository ``` 再往底层，这些依赖会使用生命周期中初始化好的客户端管理器： ```text embedding_client_manager.client qdrant_client_manager.client es_client_manager.client meta_mysql_client_manager.session_factory dw_mysql_client_manager.session_factory ``` 所以 `dependencies.py` 和 `lifespan.py` 是一组配合关系： ```text lifespan.py -> 应用启动时初始化客户端管理器 dependencies.py -> 每次请求中取出客户端或 Session，组装 Repository 和 QueryService ``` ### 4.2 MySQL Session 是请求级资源 项目中的 MySQL Session 用带 `yield` 的依赖项管理： ```python async def get_meta_session(): """创建一次请求内使用的元数据库 Session""" # yield 之后的清理逻辑由 async with 负责，FastAPI 会在请求结束后继续执行退出流程 async with meta_mysql_client_manager.session_factory() as meta_session: yield meta_session async def get_dw_session(): """创建一次请求内使用的数仓 Session""" async with dw_mysql_client_manager.session_factory() as dw_session: yield dw_session ``` 执行顺序可以理解成： ```text 请求需要 Session -> 创建 Session -> yield 给 Repository 使用 -> 请求结束 -> 退出 async with，释放 Session ``` 这里不要把 Session 做成全局对象。数据库 Session 通常属于一次请求的工作单元，而客户端管理器、连接池这类才适合放到应用生命周期里。 ### 4.3 Repository 和 Client 怎么组装 有了 Session，就可以创建 MySQL Repository： ```python async def get_meta_mysql_repository( session: Annotated[AsyncSession, Depends(get_meta_session)], ) -> MetaMySQLRepository: """基于请求级 Session 创建元数据仓储""" return MetaMySQLRepository(session) async def get_dw_mysql_repository( session: Annotated[AsyncSession, Depends(get_dw_session)], ) -> DWMySQLRepository: """基于请求级 Session 创建数仓仓储""" return DWMySQLRepository(session) ``` Qdrant 和 ES Repository 使用的是应用启动阶段已经初始化好的客户端： ```python async def get_column_qdrant_repository() -> ColumnQdrantRepository: """创建字段向量检索仓储""" return ColumnQdrantRepository(qdrant_client_manager.client) async def get_metric_qdrant_repository() -> MetricQdrantRepository: """创建指标向量检索仓储""" return MetricQdrantRepository(qdrant_client_manager.client) async def get_value_es_repository() -> ValueESRepository: """创建字段取值全文检索仓储""" return ValueESRepository(es_client_manager.client) ``` Embedding 客户端也是同样的思路： ```python async def get_embedding_client() -> HuggingFaceEndpointEmbeddings: """获取应用启动阶段初始化好的 Embedding 客户端""" return embedding_client_manager.client ``` ### 4.4 最后组装 QueryService 所有依赖最终收拢到 `get_query_service()`： ```python async def get_query_service( meta_mysql_repository: Annotated[ MetaMySQLRepository, Depends(get_meta_mysql_repository) ], embedding_client: Annotated[ HuggingFaceEndpointEmbeddings, Depends(get_embedding_client) ], dw_mysql_repository: Annotated[DWMySQLRepository, Depends(get_dw_mysql_repository)], column_qdrant_repository: Annotated[ ColumnQdrantRepository, Depends(get_column_qdrant_repository) ], metric_qdrant_repository: Annotated[ MetricQdrantRepository, Depends(get_metric_qdrant_repository) ], value_es_repository: Annotated[ValueESRepository, Depends(get_value_es_repository)], ) -> QueryService: """组装一次查询所需的业务服务""" # QueryService 只接收已经创建好的依赖对象，不关心这些对象具体来自哪里 return QueryService( meta_mysql_repository=meta_mysql_repository, embedding_client=embedding_client, dw_mysql_repository=dw_mysql_repository, column_qdrant_repository=column_qdrant_repository, metric_qdrant_repository=metric_qdrant_repository, value_es_repository=value_es_repository, ) ``` 这就是 FastAPI 子依赖的价值：我们只声明依赖关系，FastAPI 会自动从叶子节点往上解析整棵依赖树。 --- ## 5、lifespan.py：在应用启动时准备外部客户端 `dependencies.py` 能拿到 `qdrant_client_manager.client`、`es_client_manager.client`，前提是这些 manager 已经初始化。 所以需要在应用启动阶段执行： ```python qdrant_client_manager.init() embedding_client_manager.init() es_client_manager.init() meta_mysql_client_manager.init() dw_mysql_client_manager.init() ``` 项目对应文件路径：`shopkeeper-agent/app/api/lifespan.py` ```python from contextlib import asynccontextmanager from fastapi import FastAPI from app.clients.embedding_client_manager import embedding_client_manager from app.clients.es_client_manager import es_client_manager from app.clients.mysql_client_manager import ( dw_mysql_client_manager, meta_mysql_client_manager, ) from app.clients.qdrant_client_manager import qdrant_client_manager @asynccontextmanager async def lifespan(app: FastAPI): """管理应用启动和关闭两个阶段的外部资源""" # 启动阶段：先建立各类外部服务客户端，后续依赖函数会从 manager 中取已初始化对象 qdrant_client_manager.init() embedding_client_manager.init() es_client_manager.init() meta_mysql_client_manager.init() dw_mysql_client_manager.init() # yield 之前是启动逻辑，yield 之后是关闭逻辑；中间阶段由 FastAPI 正常处理请求 yield # 关闭阶段：统一释放外部连接，避免进程退出前留下未关闭的网络连接 await qdrant_client_manager.close() await es_client_manager.close() await meta_mysql_client_manager.close() await dw_mysql_client_manager.close() ``` 按 `yield` 拆开看： ```text yield 前： FastAPI 应用启动时执行 初始化 Qdrant、Embedding、ES、MySQL 客户端管理器 yield 处： 应用进入运行状态 开始接收请求 yield 后： 应用关闭前执行 释放外部客户端连接 ``` 注意这里没有关闭 `embedding_client_manager`，因为当前项目代码里只对 Qdrant、ES、MySQL manager 提供了异步关闭逻辑。文档跟着实际代码走，不额外虚构关闭方法。 --- ## 6、启动后端并测试真实查询接口 在 `shopkeeper-agent` 项目根目录启动后端： ```bash uv run fastapi dev main.py ``` 启动前要确认 MySQL、Qdrant、Elasticsearch 等外部服务可用，Embedding 相关配置也已经准备好。 用 Apifox 测试： ```text POST http://127.0.0.1:8000/api/query Content-Type: application/json ``` 请求体示例： ```json { "query": "统计华北地区销售额" } ``` 如果一切正常，接口不会一次性返回完整结果，而是持续返回多段 SSE 消息。前端或 Apifox 能看到问数智能体执行过程中的进度输出。 ![API 代码目录结构](./images/16/16-6-1-1.png) --- **本章小结：** 本章把 `/api/query` 从协议验证推进到了真实业务执行。 ```text main.py -> 创建 FastAPI 应用 -> 注册 lifespan 和 query_router query_router.py -> 接收 QuerySchema -> 通过 Depends 获取 QueryService -> 用 StreamingResponse 返回 query_service.query(...) QueryService -> 创建 DataAgentState -> 创建 DataAgentContext -> 调用 graph.astream(..., stream_mode="custom") -> 把 chunk 转成 JSON 并包装成 SSE dependencies.py -> 组装 Session、Repository、Client、Service lifespan.py -> 应用启动时初始化客户端 -> 应用关闭时释放客户端 ``` 这一章完成后，后端查询接口已经能跑真实问数流程。下一章要继续处理交付级问题：节点失败时如何停止工作流、前端如何稳定消费消息，以及并发请求时如何通过 `request_id` 追踪日志。