先看最终效果：

这是一个基于 RAG 范式 的 Text-to-SQL 智能体：用户用自然语言提问，系统自动召回相关表结构、指标定义和字段取值，生成并执行 SQL，流式返回查询结果。换句话说，不懂 SQL 也没关系，用大白话问就行，系统帮你翻译成 SQL 再查数据库。

下面从零开始，一步步搭出来。

1. 初始化项目

# uv 是 Python 的包管理器，用法和 pnpm 很像（文末有详细介绍）uv init n2sql-agentcd n2sql-agent# 试试能不能跑通uv run main.py# 终端输出 "Hello from n2sql-agent!" 说明初始化成功

此时目录结构如下：

.├── README.md├── main.py├── pyproject.toml└── uv.lock

2. 添加前端页面

前端代码直接从仓库里拷贝 frontend/ 目录到项目根目录即可：

frontend/├── index.html├── package.json├── pnpm-lock.yaml├── vite.config.js├── public/│ └── vite.svg└── src/├── App.vue# 主页面：聊天对话框 + SSE 进度展示 + 结果表格├── main.js├── style.css└── assets/└── vue.svg

拷贝完后，安装依赖并启动：

cd frontendpnpm installpnpm dev

浏览器打开 http://localhost:5173 就能看到页面了。不过现在发消息不会有响应——后端接口还没启动呢。

3. FastAPI 启动后端接口

回到项目根目录，先安装 FastAPI：

uv add "fastapi[standard]"# fastapi[standard] 包含 uvicorn 服务器和 fastapi 命令行工具，一步到位

然后把 main.py 改成下面这样——从最简单的 hello 接口开始：

from fastapi import FastAPIapp = FastAPI(title="n2sql-agent")@app.get("/hello")async def hello():return {"msg": "Hello FastAPI + uv"}

启动后端：

uv run fastapi dev main.py# 默认跑在 http://localhost:8000

验证一下：

curl http://127.0.0.1:8000/hello# 返回 {"msg":"Hello FastAPI + uv"} —— 接口通了！

4. 添加 SSE 流式查询接口

前端请求的是 POST /api/query，而且是 SSE 流式模式。我们需要在 main.py 里加上这个接口，顺便加上 CORS 跨域支持（前端 5173 端口请求 8000 端口需要它）。

把 main.py 完整替换为：

from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddleware# 创建 FastAPI 应用实例app = FastAPI(title="n2sql-agent")# 配置 CORS 中间件，开发阶段全放通app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)# ---------- 基础接口 ----------@app.get("/hello")async def hello():"""健康检查 / 测试接口"""return {"msg": "Hello FastAPI + uv"}# ---------- SSE 流式接口 ----------from fastapi.responses import StreamingResponseimport jsonimport asyncioasync def sse_stream():"""模拟 NLP-to-SQL 全流程进度推送的 SSE 生成器"""# 真实流程顺序：# 抽取关键词 → 召回字段/指标/值 → 合并召回信息 →# 过滤指标/表 → 添加上下文 → 生成 SQL → 验证 SQL → 执行 SQLtexts = [{"type": "progress", "step": "抽取关键词", "status": "running"},{"type": "progress", "step": "抽取关键词", "status": "success"},{"type": "progress", "step": "召回字段", "status": "running"},{"type": "progress", "step": "召回指标", "status": "running"},{"type": "progress", "step": "召回值", "status": "running"},{"type": "progress", "step": "召回值", "status": "success"},{"type": "progress", "step": "召回字段", "status": "success"},{"type": "progress", "step": "召回指标", "status": "success"},{"type": "progress", "step": "合并召回信息", "status": "running"},{"type": "progress", "step": "合并召回信息", "status": "success"},{"type": "progress", "step": "过滤指标", "status": "running"},{"type": "progress", "step": "过滤表", "status": "running"},{"type": "progress", "step": "过滤指标", "status": "success"},{"type": "progress", "step": "过滤表", "status": "success"},{"type": "progress", "step": "添加额外上下文", "status": "running"},{"type": "progress", "step": "添加额外上下文", "status": "success"},{"type": "progress", "step": "生成 SQL", "status": "running"},{"type": "progress", "step": "生成 SQL", "status": "success"},{"type": "progress", "step": "验证 SQL", "status": "running"},{"type": "progress", "step": "验证 SQL", "status": "success"},{"type": "progress", "step": "执行 SQL", "status": "running"},{"type": "result","data": [{"gender": "男", "sales_amount": 135370.5},{"gender": "女", "sales_amount": 143789.0},],},{"type": "progress", "step": "执行 SQL", "status": "success"},]# 逐条推送 SSE 事件，每条间隔 200ms 模拟异步处理延迟for text in texts:yield f"data: {json.dumps(text, ensure_ascii=False)}nn"await asyncio.sleep(0.2)@app.post("/api/query")async def query():"""自然语言查询入口，以 SSE 流式返回处理进度和最终结果"""return StreamingResponse(sse_stream(),media_type="text/event-stream",headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},)

保存后重启后端，用 curl 试一下 SSE 接口：

curl -N -X POST http://127.0.0.1:8000/api/query -H "Content-Type: application/json" -d '{"query":"测试"}'# 一行一行地输出进度和结果 —— 流式接口跑通了！

5. 前后端联调，看最终效果

确保后端跑在 8000 端口，前端跑在 5173 端口。然后打开浏览器访问 http://localhost:5173，在对话框里输入任意问题（比如"男女性别销售额分别是多少"），点击发送。

你会看到：

1. 步骤条依次亮起：抽取关键词 → 召回字段 → … → 执行 SQL
2. 最后弹出一张结果表格

目前数据是写死的 mock 数据，但整个前后端 SSE 流式交互的骨架已经搭好了。

架构小结

浏览器 (localhost:5173)││POST /api/query { query: "..." }▼Vite 袋里 → FastAPI (localhost:8000)││SSE 流式推送├─ data: {"type":"progress","step":"抽取关键词","status":"running"}├─ data: {"type":"progress","step":"抽取关键词","status":"success"}├─ ...└─ data: {"type":"result","data":[...]}│▼前端实时更新步骤条 + 最终表格

• 前端：一个对话框，用 fetch + ReadableStream 消费 SSE 流
• 后端：一个 POST /api/query 接口，用 StreamingResponse + async generator 逐条推送事件
• 一次请求，持续响应：这就是 SSE 的核心价值

科普名词：UV — Python 世界的 pnpm

前端从 npm 到 yarn 再到 pnpm，Python 世界也经历了类似的进化：最早用 pip + venv 手动管理，繁琐且容易踩坑；如今有了 uv，体验和 pnpm 高度一致。

uv 和 pnpm 在设计哲学上非常相似——所有项目下载的包统一缓存在机器全局位置，再通过硬链接/符号链接引入项目，避免重复下载、节省磁盘空间。

命令速查对照表

关键细节

1. 开发依赖 vs 生产依赖

# 生产依赖：程序运行必需，部署时安装uv add "fastapi[standard]" httpx# 开发依赖：仅本地开发、测试、格式化，--no-dev 时不安装uv add --dev ruff pytest mypy

2. 带扩展特性 extras（pkg[extra1,extra2]）

[] 表示安装包的同时，附带一组可选子依赖。比如 fastapi[standard] 会额外安装 uvicorn、http 等标准依赖，省去逐个添加的麻烦。书写时必须用引号包裹：

uv add "fastapi[standard]"uv add "httpx[http2,socks]"

自动写入 pyproject.toml 的效果：

dependencies = ["fastapi[standard]",]

3. 版本约束

# 固定版本uv add "uvicorn==0.30.0"# 兼容版本：>=0.30 且 <0.31uv add "uvicorn~=0.30.0"# 最低版本uv add "uvicorn>=0.28"

4. 全局工具（替代 pipx）

uv tool install 会为每个工具创建独立的隔离环境，全局可用但不污染项目虚拟环境。

uv tool install "fastapi[standard]" # 安装uv tool upgrade fastapi # 更新uv tool uninstall fastapi # 删除uv tool list# 列出全部

典型工作流

# 1. 初始化项目uv init# 2. 添加生产依赖uv add "fastapi[standard]" uvicorn# 3. 添加开发依赖uv add --dev ruff pytest# 4. 移除不需要的依赖uv remove httpxuv remove --dev pytest# 5. 同步依赖（拉取别人代码后 / 服务器部署）uv sync# 安装全部依赖uv sync --no-dev # 仅安装生产依赖（部署用）# 6. 在项目虚拟环境中执行命令（无需手动 source/activate）uv run fastapi dev main.py# 7. 全局安装 CLI 工具（安装后可省略 uv run 前缀）uv tool install "fastapi[standard]"fastapi dev main.py# 全局可用

科普名词 fastAPI

FastAPI 中文官网

一句话理解

FastAPI 是一个 Python 后端 Web 框架，用来写接口（API）。如果你写过 Node.js 的 Express，可以把它理解为"带类型校验 + 自动文档 + 异步原生支持的 Python 版 Express"，但比 Express 开箱即用的东西多得多。

最小可运行代码（感受一下长什么样）

from fastapi import FastAPIapp = FastAPI()@app.get("/hello")async def hello():return {"msg": "Hello FastAPI"}# 终端运行：fastapi dev main.py# 浏览器打开 http://localhost:8000/hello 就能看到 {"msg": "Hello FastAPI"}

不需要手动调 res.json()，直接 return 一个字典就行，框架自动帮你转 JSON。

vs Express 核心差异速览

一、基础概念（写接口的第一步）

1. App 实例

app = FastAPI(title="我的项目")

整个项目的入口对象，管理路由、中间件、生命周期、全局配置。等价于 Express 的 express()。

2. 路由 —— 各种姿势接收参数

# 路径参数：写在 URL 里@app.get("/user/{user_id}")async def get_user(user_id: int):return {"user_id": user_id}# 查询参数：跟在 ? 后面，直接声明函数参数即可@app.get("/search")async def search(q: str, page: int = 1):return {"query": q, "page": page}# POST JSON 请求体：用 Pydantic 模型接收from pydantic import BaseModelclass CreateUserReq(BaseModel):name: strage: int@app.post("/user")async def create_user(body: CreateUserReq):return {"name": body.name, "age": body.age}

3. 路径操作函数

被 @app.get/post 装饰的函数就是接口处理函数。优先写 async def，因为大部分后端工作都是 IO 密集型（数据库查询、调外部 API、SSE 流式推送）。

4. 自动序列化

直接 return 字典或 Pydantic 对象，FastAPI 自动转 JSON，不用像 Express 那样手动 res.json()。

5. 自动 API 文档

启动服务后直接访问：

• http://localhost:8000/docs → Swagger UI（可交互调试）
• http://localhost:8000/redoc → 更美观的文档页

零配置，纯靠代码和类型注解自动生成。

二、请求与数据校验（日常最高频）

FastAPI 用 Pydantic 做数据校验，类似于 TypeScript 生态里的 Zod 或 Joi。

from pydantic import BaseModel, Fieldclass QueryReq(BaseModel):question: str = Field(..., description="用户输入的自然语言问题")limit: int = Field(10, ge=1, le=100, description="返回条数")@app.post("/api/query")async def query(body: QueryReq):# body.question 和 body.limit 已经被自动校验过了# 如果前端传了 limit=999，框架直接返回 422 错误，不会进入业务逻辑return {"question": body.question, "limit": body.limit}

• Body：POST 请求的 JSON 体
• Query：URL 查询参数 ?key=value
• Path：路径参数 /item/{id}
• Header / Cookie：按需声明即可，语法一致

三、三大进阶概念（从能用到写好）

1. 中间件 Middleware —— 全局拦截

中间件夹在"客户端 ↔ 业务路由"之间，所有请求都会经过它。

from fastapi import FastAPI, Requestimport timeapp = FastAPI()@app.middleware("http")async def add_process_time(request: Request, call_next):start = time.time()response = await call_next(request)# 放行给下一个中间件/路由process_time = time.time() - startresponse.headers["X-Process-Time"] = str(process_time)return response

适用场景：请求耗时统计、全局跨域 CORS、统一异常处理、注入 traceId。

2. 依赖注入 Depends —— FastAPI 标志性能力

Express 没有原生依赖注入，而 FastAPI 把它做成了核心卖点。

from fastapi import Depends# 定义一个可复用的依赖函数async def get_current_user(token: str):# 实际项目里查数据库或解析 JWTreturn {"user_id": 1, "name": "小明"}@app.get("/profile")async def profile(user: dict = Depends(get_current_user)):# 框架自动调 get_current_user，结果注入到 userreturn {"user": user}@app.get("/orders")async def orders(user: dict = Depends(get_current_user)):# 复用同一个校验逻辑，不需要在每个接口里写重复代码return {"user": user, "orders": [...]}

• 按需引入：哪个接口需要就加 Depends(xxx)，不像中间件全局强制
• 支持嵌套：A 依赖 B，B 依赖 C，框架自动解析
• 适用场景：获取登录用户、获取数据库会话、权限校验

中间件 vs Depends 的区别：中间件在"最外层"，所有请求必须经过；Depends 在"路由层"，按接口粒度选择性地注入。

3. lifespan 生命周期 —— 启动时加载、关闭时释放

替代老旧框架的 startup / shutdown 事件，用 Python 的 asynccontextmanager 实现：

from contextlib import asynccontextmanagerfrom fastapi import FastAPI@asynccontextmanagerasync def lifespan(app: FastAPI):# 服务启动时执行（yield 之前）print("正在加载模型...")app.state.model = load_my_model() # 大模型只加载一次print("模型加载完毕，开始接受请求")yield# 服务关闭时执行（yield 之后）print("正在释放资源...")app.state.model = Noneapp = FastAPI(lifespan=lifespan)

适用场景：加载大模型到内存、创建数据库连接池、启动时预热缓存。

四、新手学习路线（照着走，不迷路）

相关文档链接

• 流式响应 StreamingResponse
• 生命周期 lifespan
• 中间件
• 依赖注入 Depends

科普名词 curl

一句话理解

curl 是一个命令行 HTTP 客户端，作用等价于浏览器地址栏输入 URL 回车，或者 Postman 发请求。后端开发调试接口时，curl 是最快、最轻量的工具——不需要打开任何软件，终端里一行命令就能验证接口通不通。

基本语法

curl [选项] <URL>

常用选项速查

一、基础 GET 请求

# 最简单的 GETcurl http://127.0.0.1:8000/hello# 带查询参数（⚠️ 地址含 & 必须用引号包裹，否则 shell 会误解）curl "http://127.0.0.1:8000/hello?name=小明&age=18"

二、调试利器：看请求/响应细节

# -i：响应中包含 HTTP 状态码和 Headerscurl -i http://127.0.0.1:8000/hello# 输出示例：# HTTP/1.1 200 OK# content-type: application/json# ...# {"msg": "Hello FastAPI"}# -v：连请求头、握手过程一起显示（调试神器，出问题先上 -v）curl -v http://127.0.0.1:8000/hello

三、POST 请求（FastAPI 最常用场景）

发送 JSON 请求体

curl -X POST http://127.0.0.1:8000/api/query -H "Content-Type: application/json" -d '{"question": "男女性别销售额分别是多少", "limit": 10}'

发送表单数据（x-www-form-urlencoded）

curl -X POST http://127.0.0.1:8000/login -d "username=admin&password=123456"

四、携带认证信息

# Header 中带 Token（最普遍的鉴权方式）curl http://127.0.0.1:8000/api/private -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..."# 携带 Cookiecurl http://127.0.0.1:8000/api/user --cookie "sessionId=abc123; uid=1001"

五、流式接口调试（SSE）

# -N 是关键：禁用 curl 的输出缓冲，否则流式数据会憋住不显示curl -N -X POST http://127.0.0.1:8000/api/query -H "Content-Type: application/json" -d '{"question": "男女销售额对比"}'

不加 -N 时，curl 会把 SSE 事件攒到缓冲区，等连接结束才一次性输出——看起来就像卡住了。

六、保存响应到文件

# -o：输出到指定文件curl http://127.0.0.1:8000/hello -o result.json# -O（大写）：按 URL 中的文件名保存curl -O http://127.0.0.1:8000/download/report.csv

新手常见踩坑

科普名词 SSE

一句话理解

SSE（Server-Sent Events，服务端推送事件）是一种让服务器"主动、持续"向浏览器推送数据的技术。想象一下：普通 HTTP 请求是你问一句服务器答一句，SSE 是你问一句，服务器一直断断续续回答你，直到事情办完。

SSE vs 其他方案：什么时候用它？

SSE 协议格式（核心，必须看懂）

SSE 的数据格式极其简单，就一条规则：

data: 一行 JSON[空行]

每个事件必须以 data: 开头，以两个换行符 nn 结束。看后端 main.py 中的实际输出：

data: {"type": "progress", "step": "抽取关键词", "status": "running"}data: {"type": "progress", "step": "抽取关键词", "status": "success"}data: {"type": "result", "data": [{"gender": "男", "sales_amount": 135370.5}]}data: [DONE]

• 每两行是一个"事件块"，前端逐个接收
• data: [DONE] 不是协议要求，是我们约定的结束信号

后端：FastAPI 怎么写 SSE 接口

核心三步：

1. 写一个 async def 生成器函数，用 yield 产出一行行 data: {json}nn
2. 用 StreamingResponse 包裹生成器，设置 media_type="text/event-stream"
3. 头里加缓存控制和长连接标记

以下是你项目 main.py 中的实际代码（简化版）：

from fastapi.responses import StreamingResponseimport jsonimport asyncioasync def sse_stream():"""异步生成器：逐条产出 SSE 事件。每 yield 一次，框架自动把这段文本推送给前端。"""steps = ["抽取关键词", "召回字段", "生成 SQL", "验证 SQL", "执行 SQL"]for step in steps:# 推送 running 状态yield f"data: {json.dumps({'type': 'progress', 'step': step, 'status': 'running'}, ensure_ascii=False)}nn"await asyncio.sleep(0.3)# 推送 success 状态yield f"data: {json.dumps({'type': 'progress', 'step': step, 'status': 'success'}, ensure_ascii=False)}nn"await asyncio.sleep(0.1)# 推送最终结果result = {"type": "result", "data": [{"gender": "男", "sales": 135370.5}]}yield f"data: {json.dumps(result, ensure_ascii=False)}nn"# 结束信号yield "data: [DONE]nn"@app.post("/api/query")async def query():return StreamingResponse(sse_stream(),media_type="text/event-stream",headers={"Cache-Control": "no-cache", # 告诉浏览器别缓存"Connection": "keep-alive",# 长连接},)

关键点说明：

前端：用原生 fetch 消费 SSE（项目实际代码）

不走 EventSource API（它只支持 GET），而用 fetch + ReadableStream 手动解析，这是处理 POST SSE 的标准做法。

以下是你项目 frontend/src/App.vue 中的实际代码（带注释精讲）：

// 1. 用 POST 发请求，拿到 Response 对象const response = await fetch(API_URL, {method: "POST",headers: { "Content-Type": "application/json" },body: JSON.stringify({ query: q }),});if (!response.body) throw new Error("服务器未返回流");// 2. 从 response.body 获取 ReadableStream 读取器const reader = response.body.getReader();const decoder = new TextDecoder("utf-8");let buffer = "";// 缓冲区：TCP 流不是按行到达的，需要拼装// 3. 循环读流，直到 donewhile (true) {const { value, done } = await reader.read();if (done) break;// 流结束// 4. 把收到的字节解码为文本，追加到缓冲区buffer += decoder.decode(value, { stream: true });// 5. 按 nn 切分出完整的事件块const events = buffer.split("nn");buffer = events.pop();// 最后一段可能不完整，留着下次拼// 6. 逐个处理事件for (const evt of events) {const line = evt.trim();if (!line.startsWith("data:")) continue;// 不是 data 行就跳过// 7. 解析 JSONconst data = JSON.parse(line.replace(/^data:s*/, ""));// 8. 按 type 分类处理if (data.type === "progress") {// 更新进度条 / 步骤列表console.log(`${data.step}: ${data.status}`);} else if (data.type === "result") {// 渲染最终的表格console.table(data.data);}}}

前端关键点拆解：

数据流全景图（从前端发问到拿到结果）

用户输入 → fetch POST → 后端 sse_stream()│├─ yield "running" ──→ 前端更新进度条├─ yield "success" ──→ 前端标记完成├─ ...├─ yield "result"──→ 前端渲染表格└─ yield "[DONE]"──→ 前端知道流结束

新手常见踩坑