17-前后端联调与日志追踪

17 - 电商问数：前后端联调与日志追踪

---

本章课程目标：

• 统一后端给前端的 SSE 消息协议：progress、result、error。
• 理解节点异常为什么不能被吞掉，以及什么时候应该继续抛出异常。
• 在 QueryService 中把节点消息统一序列化成 SSE 文本。
• 完成前后端联调，让页面能展示进度、结果和错误。
• 使用 request_id、ContextVar 和 loguru 追踪并发请求日志。

学习建议： 这一章用“出错时怎么查”来读会更有效。先看前后端消息协议，再看节点异常如何被捕获，QueryService 如何兜底，最后看 request_id 怎样把一次请求的日志串起来。联调不是只看页面能不能显示结果，更要确认失败时前端、后端和日志都能给出有用线索。

对应代码分支： 17-api-integration-logging

官方文档参考：

• FastAPI 中间件：https://fastapi.org.cn/tutorial/middleware/
• Python contextvars：https://docs.python.org/zh-cn/3/library/contextvars.html
• loguru 官方文档：https://loguru.readthedocs.io/

---

上一章已经完成真实查询接口，但真正联调时，还会遇到三个问题：

• 节点失败后，后续节点还要不要执行？
• 前端如何知道某一步是 running、success 还是 error？
• 多个请求并发时，日志如何区分是哪一次请求？

本章就是围绕这三个问题收尾。

---

1、先统一后端给前端的消息协议

前端要稳定渲染页面，后端返回的数据结构就不能随意变化。本项目约定三类消息：

类型	用途	示例
progress	节点执行进度	某节点开始、成功、失败
result	最终查询结果	SQL 查询结果列表
error	全局错误信息	异常 message

这三类消息会通过同一条 SSE 流返回给前端。

1.1 progress：节点进度

节点刚开始执行时：

{
  "type": "progress",
  "step": "抽取关键词",
  "status": "running"
}

节点执行成功时：

{
  "type": "progress",
  "step": "抽取关键词",
  "status": "success"
}

节点执行失败时：

{
  "type": "progress",
  "step": "抽取关键词",
  "status": "error"
}

前端拿到这些消息后，就可以把步骤展示成不同状态：

running -> 执行中
success -> 已完成
error   -> 已失败

1.2 result：最终查询结果

最终查询结果由 run_sql 节点输出：

{
  "type": "result",
  "data": [{ "region_name": "华北", "amount": 41099.5 }]
}

这里的 data 通常是 list[dict]，对应数据库查出来的多行结果。每一行结果是一个字典。

1.3 error：错误详情

如果图执行过程中发生系统异常，需要返回：

{
  "type": "error",
  "message": "division by zero"
}

注意，progress.status=error 和 type=error 不是重复：

• progress error：告诉前端哪一步失败了
• error message：告诉前端失败原因是什么

一个负责界面状态，一个负责错误详情。前端拿到这两类消息后，既能把失败节点标红，也能展示具体报错原因。

---

2、节点异常处理：失败后必须停止错误链路

问数智能体由多个节点组成。如果某个节点失败，后续节点通常不应该继续执行。

例如：

字段召回失败
  -> 后续生成 SQL 的上下文不完整

生成 SQL 失败
  -> 后续校验 SQL 没有意义

执行 SQL 失败
  -> 必须把错误返回给前端

所以节点内部捕获异常时，不能只是打印日志然后继续。推荐流程是：

节点开始
  -> writer progress running

节点正常结束
  -> writer progress success
  -> return state 更新

节点发生异常
  -> logger.error(...)
  -> writer progress error
  -> raise 异常
  -> graph.astream(...) 停止
  -> QueryService 捕获异常
  -> SSE 返回 type=error

关键点只有一个：节点捕获异常后，还要继续 raise。

如果不 raise，LangGraph 会认为当前节点已经处理完成，然后继续执行后续节点。这不符合问数系统的预期。

2.1 普通节点的推荐写法

以“抽取关键词”节点为例，外层结构应该尽量统一：

async def extract_keywords(state: DataAgentState, runtime: Runtime[DataAgentContext]):
    step = "抽取关键词"
    writer = runtime.stream_writer

    # 进入节点时，先告诉前端当前步骤开始执行
    writer({"type": "progress", "step": step, "status": "running"})

    try:
        query = state["query"]
        keywords = extract(query)

        # 业务执行成功后，先写 success，再返回 state 更新
        writer({"type": "progress", "step": step, "status": "success"})
        logger.info(f"抽取关键词成功：{keywords}")
        return {"keywords": keywords}
    except Exception as e:
        # 失败时记录日志，通知前端当前节点失败，然后继续抛出异常中断图执行
        logger.error(f"{step} failed: {e}")
        writer({"type": "progress", "step": step, "status": "error"})
        raise

这个结构有三个好处：

1. 前端能立即知道当前节点开始执行。
2. 成功和失败都会有明确状态。
3. 异常继续抛出，图执行会停止，最终由 QueryService 统一返回错误消息。

实际项目中每个节点的业务逻辑不同，但外层异常处理结构应该尽量一致。这样前端消费协议会稳定，后端排查也更容易。

2.2 validate_sql 是特殊节点

validate_sql 和普通节点不完全一样。

它内部用数据库执行：

explain <generated_sql>

如果 SQL 语法或字段有问题，数据库会报错。但这个错误在业务语义上不一定代表“节点失败”。因为本项目本来就设计了后续的 correct_sql 节点，用来根据错误信息修正 SQL。

因此 validate_sql 的处理方式是：

EXPLAIN 成功
  -> 返回 {"error": None}
  -> 条件边进入 run_sql

EXPLAIN 失败
  -> 返回 {"error": "具体错误信息"}
  -> 条件边进入 correct_sql

也就是说，validate_sql 捕获 SQL 校验错误后，不一定要 raise。它要把错误写进 state["error"]，交给图的条件边判断是否需要校正 SQL。

这个点很重要：系统异常要中断，业务分支要进入图的条件流转。

2.3 run_sql 输出最终 result

前面多数节点只输出进度，最终结果应该由执行 SQL 的节点输出。

项目中 run_sql 的核心结构如下：

async def run_sql(state: DataAgentState, runtime: Runtime[DataAgentContext]):
    writer = runtime.stream_writer
    step = "执行SQL"

    writer({"type": "progress", "step": step, "status": "running"})

    try:
        # 这里拿到的可能是 generate_sql 直接通过校验的 SQL，也可能是 correct_sql 覆盖后的 SQL
        sql = state["sql"]
        dw_mysql_repository = runtime.context["dw_mysql_repository"]

        # 真实数据库访问统一封装在仓储层，节点只负责从状态取 SQL 并触发执行
        result = await dw_mysql_repository.run(sql)

        logger.info(f"SQL执行结果：{result}")
        writer({"type": "progress", "step": step, "status": "success"})
        writer({"type": "result", "data": result})
    except Exception as e:
        logger.error(f"{step} failed: {e}")
        writer({"type": "progress", "step": step, "status": "error"})
        raise

这样前端会先看到“执行 SQL 成功”，再收到最终查询结果。

---

3、QueryService：统一序列化和兜底错误

节点通过 writer(...) 写出的不是普通字符串，而是结构化字典，例如：

writer({"type": "progress", "step": step, "status": "running"})
writer({"type": "result", "data": result})

但是 SSE 最终写给浏览器的必须是文本。所以 QueryService 要负责把 Python 字典转成 JSON 字符串，再包成 SSE 格式。

转换过程如下：

节点 writer(...) 写出 Python 字典
  -> graph.astream(...) 产出 chunk
  -> QueryService 把 chunk 序列化成 JSON 字符串
  -> 包成 data: ...\n\n
  -> StreamingResponse 返回给前端

项目对应文件路径：shopkeeper-agent/app/services/query_service.py

try:
    async for chunk in graph.astream(
        input=state,
        context=context,
        stream_mode="custom",
    ):
        yield f"data: {json.dumps(chunk, ensure_ascii=False, default=str)}\n\n"
except Exception as e:
    error = {"type": "error", "message": str(e)}
    yield f"data: {json.dumps(error, ensure_ascii=False, default=str)}\n\n"

这段逻辑负责两件事。

第一，正常节点消息要统一序列化：

json.dumps(chunk, ensure_ascii=False, default=str)

其中两个参数很关键：

参数	作用
ensure_ascii=False	保留中文，避免把“抽取关键词”转成 Unicode 转义
default=str	遇到 Decimal、日期等 JSON 不认识的类型时，转成字符串兜底

第二，异常也要按同一套协议返回：

节点抛出异常
  -> graph.astream(...) 中断
  -> QueryService 捕获异常
  -> 构造 {"type": "error", "message": "..."}
  -> json.dumps(...)
  -> data: ...\n\n
  -> 返回前端

这里不能简单理解成“后端报错就返回 500”。因为流式响应一旦开始写出，HTTP 响应头通常已经发给客户端了，后面很难再临时改状态码。更合适的方式是把错误也包装成一条 SSE 消息：

{
  "type": "error",
  "message": "division by zero"
}

这就是 API 层的最后兜底：节点负责告诉前端“哪一步失败”，QueryService 负责告诉前端“失败原因是什么”。

---

4、测试接口与前后端联调

这一节先用 Apifox 验证后端，再启动前端页面联调。顺序不要反过来：先确认后端消息协议稳定，前端展示才好排查。

4.1 用 Apifox 测试正常和异常场景

后端启动：

uv run fastapi dev main.py

请求：

POST http://127.0.0.1:8000/api/query
Content-Type: application/json

请求体：

{
  "query": "统计华北地区的销售总额"
}

4.2 启动前端项目

当后端接口已经能通过 Apifox 返回稳定 SSE 消息后，就可以进入前后端联调。

前端项目位于 shopkeeper-agent/frontend，使用 React + Vite + Tailwind CSS 编写，包管理器使用 pnpm。

在前端项目根目录执行：

cd frontend
pnpm install
pnpm dev

启动后根据终端输出访问页面。开发环境下，前端默认通过 Vite 代理访问后端：

/api -> http://127.0.0.1:8000

Vite 会把它转发到后端真实接口：

POST http://127.0.0.1:8000/api/query

如果需要修改后端地址，可以复制环境变量示例文件：cp .env.example .env

大功告成~~~~~~

---

5、request_id：让并发日志能查得清楚

本地单次测试时，日志通常很好读。但真实服务里可能同时有多个请求在跑：

请求 A：统计华北地区销售总额
请求 B：查询黄金会员客单价
请求 C：统计数码品类 GMV

这些请求的日志会交错输出：

抽取关键词成功
召回字段信息成功
抽取关键词成功
召回指标信息成功
生成SQL
召回字段信息成功

如果没有请求标识，很难判断某一行日志属于哪个用户的问题。

所以每个请求进入后端时，都应该生成一个唯一 request_id。同一次请求链路里的所有日志都带上同一个 request_id，排查时就能按这个 ID 过滤完整链路。

5.1 用中间件生成 request_id

中间件适合处理“所有请求都要做”的逻辑。生成 request_id 正好属于这类需求。

完整思路是：

请求进入 FastAPI
  -> 中间件生成 uuid
  -> 写入 ContextVar
  -> call_next(request) 继续执行路由
  -> 后续日志都能读取当前请求的 request_id

项目对应文件路径：shopkeeper-agent/main.py

import uuid

from fastapi import FastAPI, Request

from app.api.lifespan import lifespan
from app.api.routers.query_router import query_router
from app.core.context import request_id_ctx_var

app = FastAPI(lifespan=lifespan)

app.include_router(query_router)


@app.middleware("http")
async def add_request_id(request: Request, call_next):
    # 请求被处理之前，先为当前请求生成一个唯一 ID
    request_id = uuid.uuid4()

    # 写入当前异步上下文，后续业务日志会从这里取 request_id
    request_id_ctx_var.set(request_id)

    # 继续执行后续路由处理逻辑
    response = await call_next(request)

    # 请求被处理之后，返回响应
    return response

uuid.uuid4() 生成的是 UUID 对象，loguru 输出时会自动转成可读字符串。也可以显式写成 str(uuid.uuid4())，日志展示和文本过滤会更直接。

5.2 ContextVar：并发请求下保存请求上下文

如果用普通全局变量保存 request_id，并发请求会互相覆盖。

先简单区分两个概念：线程和协程。

线程可以理解成操作系统调度的执行单元。一个程序里可以有多个线程，它们可能并行执行，也可能交替执行。

协程可以理解成更轻量的异步任务。它通常运行在线程内部，遇到 await、网络 IO、数据库 IO 这类等待操作时，会主动让出执行权，让其他协程继续执行。

FastAPI 的异步接口通常就是用协程来处理请求的。多个请求同时进来时，它们会在不同的异步上下文中交替执行。

再看普通全局变量为什么不行。假设我们这样保存当前请求 ID：

current_request_id = None

请求 A 进来，把它设置成：

current_request_id = "A"

请求 B 又进来，把它改成：

current_request_id = "B"

那么请求 A 后续再打印日志时，读到的就可能已经不是 "A"，而是被请求 B 覆盖后的 "B"。这就是并发场景下普通全局变量的问题：所有请求共用同一份值。

如果有 Java 或后端开发经验，可以把 ContextVar 类比成 ThreadLocal。

ThreadLocal：
  一个全局变量
  每个线程各自保存一份值
  适合多线程程序

ContextVar：
  一个全局变量
  每个异步上下文各自保存一份值
  适合 asyncio / FastAPI 这类异步程序

在 FastAPI 中，可以简单理解为：每个请求都有自己的异步上下文。请求 A 设置的 request_id，不会覆盖请求 B 的 request_id。

项目对应文件路径：shopkeeper-agent/app/core/context.py

"""
请求上下文

用于保存和传递一次请求或一次脚本执行过程中的上下文变量，
当前先维护 request_id，后续日志模块会从这里读取 request_id 并注入到每条日志中
"""

from contextvars import ContextVar

request_id_ctx_var: ContextVar[str] = ContextVar("request_id", default="1")

为什么要设置默认值 "1"？

因为项目里不只有 HTTP 请求会打印日志，还有命令行脚本，例如构建元数据知识库的脚本。脚本执行时不会经过 FastAPI 中间件，也就没有人提前 set(request_id)。如果没有默认值，日志模块读取时可能报错。设置默认值后，脚本场景也能正常打印日志。

5.3 loguru：让日志自动带上 request_id

项目使用 loguru 管理日志。我们不希望在每个 logger.info(...) 里手动拼 request_id，那样太繁琐，也容易漏。

更好的做法是在日志模块统一注入。

项目对应文件路径：shopkeeper-agent/app/core/log.py

from loguru import logger

from app.core.context import request_id_ctx_var

# 日志格式统一展示时间、级别、request_id 和调用位置，便于排查链路问题
log_format = (
    "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
    "<level>{level: <8}</level> | "
    "<magenta>request_id - {extra[request_id]}</magenta> | "
    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
    "<level>{message}</level>"
)


def inject_request_id(record):
    """把上下文中的 request_id 注入到每条日志的 extra 字段"""
    # 从当前异步上下文中读取 request_id
    request_id = request_id_ctx_var.get()

    # extra 是 loguru 预留给自定义字段的位置
    record["extra"]["request_id"] = request_id


# 移除 Loguru 默认的输出目标，避免和项目自定义配置重复打印
logger.remove()

# 生成带 request_id 注入能力的 logger，后续业务代码统一使用这个实例
logger = logger.patch(inject_request_id)

这段代码有三个重点：

inject_request_id(record)
  -> 每条日志输出前执行
  -> 从 ContextVar 读取当前请求的 request_id
  -> 写入 record["extra"]["request_id"]
  -> 日志格式通过 {extra[request_id]} 展示出来

这样业务代码仍然只需要写：

logger.info("抽取关键词成功")

输出时就会自动带上：

request_id - 550e8400-e29b-41d4-a716-446655440000

5.4 测试 request_id 是否生效

我们可以打开两个前端页面，同时发起两个查询：

页面 A：统计华北地区销售总额
页面 B：统计各地区销售总额

观察后端日志，应该能看到两组不同的 request_id。

接着复制其中一个 request_id 去日志里搜索，应该只会命中同一次请求相关的日志：

抽取关键词
召回字段信息
召回指标信息
生成SQL
执行SQL

这说明并发请求虽然日志交错输出，但我们已经能够按请求维度把链路重新串起来。

---

本章小结：

本章完成了 API 部分的交付级收尾：

异常处理：
  -> 节点失败后标记 progress error
  -> 继续 raise 异常
  -> QueryService 统一返回 type=error

前后端联调：
  -> 前端调用 /api/query
  -> 后端 SSE 持续返回 progress / result / error
  -> 页面实时展示进度和结果

日志追踪：
  -> 中间件生成 request_id
  -> ContextVar 保存请求上下文
  -> loguru 自动注入 request_id
  -> 并发请求可按 request_id 排查

到这里，「电商问数」的 API 部分就形成了完整闭环：前端可以发起自然语言查询，后端可以流式执行问数智能体，异常可以及时反馈，日志也能按请求链路追踪。

---

项目总结

到这一章为止，「电商问数」项目的主体内容就完整收尾了。回头看整套项目，它不是一个简单的“把用户问题丢给大模型生成 SQL”的 Demo，而是一套围绕企业问数场景搭起来的最小闭环系统。

这里的“最小”不是简陋，而是最少且必要：为了让读者从入门走到进阶，项目只保留了一条问数系统必须具备的主链路，没有一开始就把所有生产级功能都塞进来。它覆盖了 MySQL、Qdrant、Elasticsearch、Embedding、FastAPI、LangGraph 等关键技术栈，也把这些技术放进了一个能跑通的业务场景里。

这条主线其实很清楚：先理解电商数仓里的表、字段和指标，再把这些信息构建成元数据知识库；用户提问时，系统先召回相关字段、指标和字段取值，再把这些信息整理成 SQL 生成上下文；SQL 生成后，还要经过校验、校正和执行；最后再通过 FastAPI、SSE 和前端页面把结果交付给用户。

1、这个项目最终完成了什么

从用户视角看，系统最终提供的是一个自然语言问数能力：

用户输入“统计华北地区的销售总额”，系统会理解问题、检索相关元数据、生成 SQL、查询数仓，并把执行进度、查询结果或异常信息返回给前端。

从工程视角看，项目已经形成了一条完整链路：前端页面调用 FastAPI 查询接口，接口通过 QueryService 触发 LangGraph 问数工作流，工作流再访问元数据知识库、向量检索、全文检索和 MySQL 数仓。执行过程通过 SSE 持续返回 progress、result、error，后端日志则通过 request_id 串起同一次请求。

这条链路里，每一层都有明确分工：

层次	主要内容
数据层	MySQL 元数据库、模拟数仓、表字段和指标配置
检索层	Qdrant 字段/指标向量检索，Elasticsearch 字段取值全文检索
智能体层	LangGraph 节点编排，完成关键词抽取、召回、过滤、SQL 闭环
服务层	QueryService 组装 State、Context 并调用图工作流
API 层	FastAPI 路由、依赖注入、生命周期管理、SSE 流式响应
可观测层	统一异常协议、request_id、ContextVar、loguru 日志追踪
前端层	消费 SSE 消息，实时展示节点进度、最终结果和错误信息

2、这套教程学到的不只是代码

这套项目最重要的价值，不是某一个函数怎么写，而是把一个智能体系统拆成了可以落地的工程模块。

前半部分解决“数据和知识从哪里来”：从数仓基础、项目结构、配置管理开始，到 MySQL、Qdrant、Elasticsearch、Embedding 接入，再到元数据知识库和字段、指标、字段取值索引构建。

中间部分解决“智能体如何一步步问数”：用 LangGraph 搭起工作流骨架，完成关键词抽取、多路召回、召回信息合并、表和指标过滤、额外上下文补全，以及 SQL 生成、校验、校正和执行。

最后部分解决“怎么把能力交付出去”：用 FastAPI 暴露查询接口，用 StreamingResponse 和 SSE 返回流式消息，用 QueryService 组装真实工作流，再通过异常处理、前后端联调和 request_id 日志追踪，把这条能力变成一个可以调试、可以排查的应用。

这三段合在一起，就构成了一套从入门到进阶都值得拆解的智能体项目。它不是停留在 Notebook 或命令行里的模型调用示例，也不是一上来就追求“大而全”的生产系统，而是先把最关键的工程骨架跑通。

如果按照企业级生产环境的标准来看，一个真正上线给多人长期使用的问数系统，还会涉及更多能力。比如权限、缓存、评测、审计、监控、多轮记忆、租户隔离、任务队列、灰度发布等。

但这些能力如果全部放进第一个实战项目里，学习成本会非常高，主线反而容易被淹没。所以「电商问数」更适合作为入门到进阶阶段的完整项目：它把最核心、最必要的链路讲透，让你知道一个企业级智能体应用应该怎样拆层、怎样组织代码、怎样把模型能力接进真实业务系统。

本项目重点放在自然语言问数主链路、元数据知识库、检索增强、LangGraph 多节点工作流、SQL 闭环、API 流式响应、异常处理和日志追踪上。记忆系统、用户权限、多租户、复杂评测、生产部署治理和更完整的监控告警体系暂时不展开，不是因为它们不重要，而是因为这些能力更适合放到后续其他实战项目里继续实现。这样每个项目都有清晰主题，学习路径也更稳。

3、几个值得记住的工程经验

第一，不要让大模型凭空生成 SQL。

模型真正需要的是被整理过的上下文：可用表、字段含义、指标口径、字段示例值、日期信息和数据库环境。前面的元数据知识库、召回和过滤，都是在帮模型减少“瞎猜”的空间。

第二，智能体节点要职责清晰。

extract_keywords 只负责抽关键词，召回节点只负责找候选信息，过滤节点只负责裁剪上下文，generate_sql 只负责生成 SQL，validate_sql 只负责校验，run_sql 只负责执行。节点职责越清楚，工作流越容易调试和扩展。

第三，状态数据和运行时工具要分开。

DataAgentState 保存本次任务不断变化的业务数据，例如关键词、召回结果、SQL 和查询结果。DataAgentContext 保存运行时工具，例如 Repository、Embedding Client、Qdrant、ES、MySQL 访问能力。这个分层能让图节点更干净。

第四，接口交付时要考虑用户体验。

问数链路不是瞬间完成的，用户需要看到系统正在做什么。所以项目使用 SSE 流式返回 progress、result、error，让前端能实时展示执行进度和最终结果。

第五，异常和日志不是最后才补的边角料。

节点失败后要停止错误链路，前端要知道哪一步失败，后端要能通过 request_id 找回同一次请求的完整日志。没有这些能力，系统一旦进入联调或多人使用，就会很难排查问题。

4、后续可以继续扩展什么

当前项目已经完成了主链路，但如果要继续往生产级方向推进，还可以考虑这些方向：

方向	可以继续补充的能力
权限控制	不同用户只能查询自己有权限的数据范围
SQL 安全	增加只读限制、黑名单关键字、查询超时和结果行数限制
多轮问数	支持基于上一次结果继续追问，引入会话记忆理解上下文
结果解释	不只返回表格，还解释 SQL、指标口径和查询结果含义
缓存性能	对元数据召回、Embedding、热点问题结果做缓存
评测体系	构建问题集，评估 SQL 正确率、召回质量和端到端成功率
部署运维	容器化部署，接入监控、告警、链路追踪和完整日志平台

这些扩展都不是推翻当前架构，而是在当前架构上继续加能力。也就是说，这套项目已经打好了一个可以继续生长的底座。后续如果在其他实战项目中实现记忆、多轮对话、权限、评测或生产部署，也可以把这里的分层思想继续带过去。

5、最后再压缩成一句话

「电商问数」最终完成的是一条从自然语言到数据结果的工程闭环：用户提出问题，系统召回元数据并构建上下文，然后生成和校验 SQL，查询数仓，再通过流式接口返回给前端，并用日志追踪把整条链路串起来。

如果你能把这条链路讲清楚、跑通、改得动，就已经不只是“会调用大模型”，而是开始具备设计和实现企业级智能体应用的能力了。

至于记忆、权限、评测、部署治理这些更进一步的生产级能力，不急着在这一个项目里全部吃完。先把这一条最少且必要的主链路打牢，后面的实战项目再一层一层往上加。