14 · FastAPI 服务化：把 RAG 机器人变成 HTTP API

本片目标：让前端网页/其他系统能调用你的 RAG 机器人。用 FastAPI 暴露 REST API：问答接口 + 流式接口 + 会话接口。 新增规定性：13（服务契约：HTTP 路由、请求/响应模型、同步 vs 异步路由） 数据字典：FastAPI 路由声明、Pydantic 请求/响应模型、uvicorn 启动。 进程线程模型：本片核心——同步 def 路由丢进线程池 vs 异步 async def 路由在事件循环。 网络模型：HTTP/1.1 REST；流式接口走 SSE（EventSource 协议）。

1. 上集回顾

第 13 篇把 langchain_core 整个地基翻了一遍——你知道了哪些模块是你的日常武器。但地基再清楚，机器人还跑在你的本地进程里，外部世界够不着。还差最后一步：怎么让外部世界调用你的机器人？

前端网页怎么问？——浏览器不能 import 你的 Python 函数；
其他微服务怎么问？——它们发 HTTP 请求。

方案：把 RAG 链包成一个 HTTP API——别人 POST /chat 发问题，你返回答案。这是生产系统的标准接口形态。

2. 数据字典：FastAPI 三件套

2.1 路由（route）

from fastapi import FastAPI
app = FastAPI(title="新人培训手册问答机器人")

@app.post("/chat")          # POST /chat 接口
async def chat(req: ChatRequest):   # 参数类型 = 请求模型
    ...

元素	说明
`@app.post("/chat")`	定义 POST 方法的路由
`async def` vs `def`	关键：见第 4 节线程模型
参数 `req: ChatRequest`	FastAPI 自动解析 JSON 请求体 → Pydantic 模型

2.2 请求/响应模型（Pydantic）

from pydantic import BaseModel

class ChatRequest(BaseModel):
    question: str          # 用户问题
    session_id: str = "default"   # 会话 id（第 06 篇记忆）

class ChatResponse(BaseModel):
    answer: str            # 答案
    sources: list[str] = []  # 引用来源（第 15 篇填充）

为什么用 Pydantic：FastAPI 自动做请求体校验（缺字段 422）、自动生成 OpenAPI 文档（/docs 白送）。

2.3 启动（uvicorn）

uvicorn 14_api:app --reload --port 8000
# 14_api = 文件名, app = FastAPI 实例
# --reload 开发热重载；生产去掉

3. 关键方法/装饰器

元素	签名	说明
`FastAPI(title=...)`	构造	应用实例
`@app.post("/path")`	装饰器	注册路由
`uvicorn.run(app, host, port)`	或命令行	启动服务器
`/docs`	内置	自动生成的 API 文档页面

4. 进程线程模型（本片最重要）

FastAPI 有两种路由写法，性能差异巨大：

4.1 同步路由：`def chat(...)`（错误示范）

@app.post("/chat")
def chat(req: ChatRequest):      # ← 同步 def
    result = rag_chain.invoke(req.question)   # 阻塞 1~3s
    return {"answer": result}

uvicorn 收到请求 → 丢进 anyio 线程池（默认 ~40 线程）
  线程池线程 → rag_chain.invoke() 阻塞 1~3s → 占死这个线程
100 并发 → 40 个线程全被占 → 剩下 60 个请求排队！

问题：每个同步请求占一个线程等网络。线程是稀缺资源，高并发下很快耗尽。

4.2 异步路由：`async def chat(...)`（生产正解）

@app.post("/chat")
async def chat(req: ChatRequest):      # ← async def
    result = await rag_chain.ainvoke(req.question)   # 不阻塞！
    return {"answer": result}

uvicorn 收到请求 → 事件循环调度协程
  await rag_chain.ainvoke() → 发出 HTTP 请求后挂起，事件循环去处理别的请求
100 并发 → 只有 1 个事件循环线程 + 100 个协程在等 → 不占线程！

为什么：ainvoke 走 httpx.AsyncClient，等待响应时释放线程（第 12 篇网络模型）。异步路由可以支撑几百上千并发，同步路由几十就满了。

4.3 桥接：同步链 + 异步路由

如果链是同步的（比如第 09 篇的 rag_chain.invoke），异步路由里要桥接：

from langchain_core.runnables.config import run_in_executor

@app.post("/chat")
async def chat(req: ChatRequest):
    result = await run_in_executor(None, rag_chain.invoke, req.question)
    return {"answer": result}

run_in_executor（第 12 篇的桥）把同步调用丢进线程池，事件循环等 Future——LangChain 内部自己也用这个机制。

5. 网络模型

5.1 普通问答接口

前端 → POST http://localhost:8000/chat
        Body: {"question": "转正需要什么条件？", "session_id": "user_1"}
  ← uvicorn(FastAPI) → RAG 链 → 智谱 API
前端 ← 200 {"answer": "转正需试用期满、至少 60 篇日总结、答辩 75 分以上...", "sources": []}

5.2 流式接口（SSE，打字机）

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def gen():
        async for chunk in rag_chain.astream(req.question):
            yield f"data: {chunk}\n\n"    # SSE 协议：每块一行 data:
    return StreamingResponse(gen(), media_type="text/event-stream")

前端用 EventSource/fetch(stream:true) 订阅
服务端每生成一块 → data: 内容\n\n → 前端逐字渲染

SSE 与第 04 篇的联系：模型的流式（SSE）→ LangChain astream → 你的服务再转成 SSE 给前端。两层流式叠加，中间是你的服务做桥。

6. 验证：跑起来

配套代码 code/14_api.py：一个完整的问答服务（3 个接口）：

POST /chat：普通问答（异步路由 + 记忆会话）
POST /chat/stream：流式问答（SSE）
GET /health：健康检查

启动：

cd enterprise-rag-course\code
uvicorn 14_api:app --port 8000

另开终端测试：

# ① 普通问答
curl -X POST http://localhost:8000/chat -H "Content-Type: application/json" -d '{\"question\":\"转正需要什么条件？\",\"session_id\":\"u1\"}'

# ② 流式（打字机）
curl -N -X POST http://localhost:8000/chat/stream -H "Content-Type: application/json" -d '{\"question\":\"日总结什么时候发？\"}'

# ③ 健康检查
curl http://localhost:8000/health

预期输出：

① → {"answer":"根据手册，转正需试用期满、累计至少 60 篇日总结、答辩平均分 75 分以上。","sources":[]}
② → data: 日总结
    data: 于
    data: 每天...
③ → {"status":"ok"}

浏览器打开 http://localhost:8000/docs 能看到自动生成的 API 文档。

7. 边界

同步 def 路由别用于模型调用——除非并发极低。生产一律 async def + ainvoke。
SSE 是单向推送——适合流式文本；双向交互要 WebSocket（超出本系列范围）。
--reload 只用于开发——生产用 uvicorn --workers N 多进程（但内存里记忆/Chroma 每进程一份，注意第 15 篇的持久化）。
CORS：前端跨域要配 CORSMiddleware（第 15 篇）。

8. 未完待续

API 有了，但离”生产级”还差最后一公里：

配置散在代码里（模型名、k 值、路径）——要配置化；
向量库索引怎么自动更新（文档改了要重新入库）——要索引管线；
出错怎么办、怎么溯源（答案来自哪块）——要错误处理 + 引用溯源；
会话记忆持久化（重启不丢）——要存储层。

第 15 篇把这一切拼成完整的后端项目——本系列最终交付。 → 15 · 完整后端项目

RAG 教程

小林学 LangChain

14 · FastAPI 服务化：把 RAG 机器人变成 HTTP API

14 · FastAPI 服务化：把 RAG 机器人变成 HTTP API

1. 上集回顾

2. 数据字典：FastAPI 三件套

2.1 路由（route）

2.2 请求/响应模型（Pydantic）

2.3 启动（uvicorn）

3. 关键方法/装饰器

4. 进程线程模型（本片最重要）

4.1 同步路由：`def chat(...)`（错误示范）

4.2 异步路由：`async def chat(...)`（生产正解）

4.3 桥接：同步链 + 异步路由

5. 网络模型

5.1 普通问答接口

5.2 流式接口（SSE，打字机）

6. 验证：跑起来

7. 边界

推荐资料（延伸阅读）

8. 未完待续

​14 · FastAPI 服务化：把 RAG 机器人变成 HTTP API

​1. 上集回顾

​2. 数据字典：FastAPI 三件套

​2.1 路由（route）

​2.2 请求/响应模型（Pydantic）

​2.3 启动（uvicorn）

​3. 关键方法/装饰器

​4. 进程线程模型（本片最重要）

​4.1 同步路由：def chat(...)（错误示范）

​4.2 异步路由：async def chat(...)（生产正解）

​4.3 桥接：同步链 + 异步路由

​5. 网络模型

​5.1 普通问答接口

​5.2 流式接口（SSE，打字机）

​6. 验证：跑起来

​7. 边界

​推荐资料（延伸阅读）

​8. 未完待续

14 · FastAPI 服务化：把 RAG 机器人变成 HTTP API

1. 上集回顾

2. 数据字典：FastAPI 三件套

2.1 路由（route）

2.2 请求/响应模型（Pydantic）

2.3 启动（uvicorn）

3. 关键方法/装饰器

4. 进程线程模型（本片最重要）

4.1 同步路由：`def chat(...)`（错误示范）

4.2 异步路由：`async def chat(...)`（生产正解）

4.3 桥接：同步链 + 异步路由

5. 网络模型

5.1 普通问答接口

5.2 流式接口（SSE，打字机）

6. 验证：跑起来

7. 边界

推荐资料（延伸阅读）

8. 未完待续