> ## Documentation Index
> Fetch the complete documentation index at: https://www.yuan111.asia/doc/llms.txt
> Use this file to discover all available pages before exploring further.

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

# 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）

```python theme={null}
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）

```python theme={null}
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）

```powershell theme={null}
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(...)`（错误示范）

```python theme={null}
@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(...)`（生产正解）

```python theme={null}
@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`），异步路由里要桥接：

```python theme={null}
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，打字机）

```python theme={null}
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`：健康检查

**启动**：

```powershell theme={null}
cd enterprise-rag-course\code
uvicorn 14_api:app --port 8000
```

**另开终端测试**：

```powershell theme={null}
# ① 普通问答
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 篇）。

***

## 推荐资料（延伸阅读）

* [FastAPI 官方文档](https://fastapi.tiangolo.com/) —— 路由 / 依赖 / 生命周期权威参考
* [FastAPI 官方文档 · 流式响应](https://fastapi.tiangolo.com/advanced/custom-response/) —— StreamingResponse 与 SSE
* [LangChain 官方文档 · 流式](https://docs.langchain.com/oss/python/langchain/streaming) —— 服务端 astream 的官方姿势

***

## 8. 未完待续

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

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

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

→ [15 · 完整后端项目](/doc/doc/enterprise-rag-course/15-完整后端项目)