开始学习
课程 › 0基础Agent开发课 / 生产化部署 / FastAPI构建AI服务
— 16 min read

FastAPI构建AI服务

RAG 系统调通之后，下一步是将其包装成服务，提供 HTTP 接口供前端调用或与其他系统集成。

FastAPI 构建 AI 服务：Java 开发者的 Python 后端入门

RAG 系统调通之后，下一步是将其包装成服务，提供 HTTP 接口供前端调用或与其他系统集成。

1.1 为什么 AI 应用选 FastAPI：不选 Flask，不选 Django

FastAPI AI服务架构图
FastAPI AI 服务完整请求链路——客户端→路由→中间件→业务逻辑→LLM API→异步响应

这个问题值得认真回答，因为有 Python 经验的人往往先想到 Flask。

Flask 的问题：Flask 是同步框架。LLM 调用通常需要 2-30 秒，在 Flask 里一个请求占着线程等 LLM 返回，期间无法处理其他请求。10 个用户同时请求，9 个人在等。你可以用 Gunicorn 多进程，但每个进程都要加载一次 LLM 模型（8GB），内存立刻爆。

Django 的问题：Django 是为内容管理系统设计的，带了 ORM、模板引擎、Admin 后台……如果你只是要暴露几个 AI 接口，这些东西全是多余的负担。

FastAPI 的优势：

  • 原生 async：async def 函数在等待 LLM 响应时自动让出控制权处理其他请求，同等资源下并发能力强 5-10 倍
  • Pydantic 数据校验：请求参数自动校验，错误信息清晰，减少大量防御性代码
  • 自动生成 OpenAPI 文档：/docs 直接出 Swagger UI，不需要手写一行文档
  • 性能：接近 Go 和 Node.js，比 Flask 快 2-5 倍

对于 AI 服务这个场景，FastAPI 的选择几乎没有争议。以下开始实战。

1.2 FastAPI 是什么

FastAPI 是一个基于 Python 类型注解的现代 Web 框架。它的设计哲学很简单：类型注解既是代码的文档，也是运行时的校验逻辑，还是自动生成 OpenAPI 文档的依据。

三个核心特点：

性能高。 FastAPI 基于 Starlette（异步 Web 框架）和 Pydantic（数据校验），性能接近 Go 和 Node.js，远超 Flask 和 Django。这对 AI 服务来说很重要，因为 LLM 调用本身很慢，框架层面不能再拖后腿。

自动生成文档。 启动服务后访问 /docs，Swagger UI 就出来了，所有接口、参数类型、响应格式一目了然，完全不需要手写文档。

原生支持 async。 LLM 调用是 IO 密集型操作，等待远程 API 返回的时候 CPU 是闲着的，async 可以在等待期间处理其他请求，吞吐量大幅提升。

1.3 和 Spring Boot 对比

维度 Spring Boot FastAPI
启动时间 秒级甚至十几秒 毫秒级
代码量 多（注解、配置类） 少
类型安全 编译期 运行时（Pydantic）
OpenAPI 文档 需要引入 SpringDoc 内置，自动生成
生态完整性 非常完整 相对较薄
适合场景 企业级复杂业务 AI 服务、中小型 API
Async 支持 有，但不是默认 原生设计

Spring Boot 生态确实更完整，但对于 AI 服务这个场景，FastAPI 的轻量和 async 优势更突出。不需要 JPA，不需要 Spring Security 那套复杂配置，需要的是快速把 LLM 能力暴露成接口。

1.4 基本用法

先安装：

bash 
pip install fastapi uvicorn pydantic

最简单的路由定义：

python 
from fastapi import FastAPI

app = FastAPI(title="AI服务", version="1.0.0")

@app.get("/health")
def health_check():
    return {"status": "ok"}

启动：

bash 
uvicorn main:app --reload --port 8000

然后访问 http://localhost:8000/docs，Swagger 文档就出来了。

1.5 请求体和响应模型

FastAPI 的请求校验完全依赖 Pydantic Model，写法和 Java 的 DTO 类很像：

python 
from pydantic import BaseModel, Field
from typing import Optional

class QuestionRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=500, description="用户问题")
    session_id: Optional[str] = Field(None, description="会话ID，用于多轮对话")
    top_k: int = Field(3, ge=1, le=10, description="检索文档数量")

class QuestionResponse(BaseModel):
    answer: str
    sources: list[str]
    tokens_used: int

定义了 Model 之后，FastAPI 会自动校验入参，字段缺失或类型错误会直接返回 422，不需要写任何校验代码。

路由接收 Pydantic Model：

python 
@app.post("/ask", response_model=QuestionResponse)
def ask_question(request: QuestionRequest):
    # FastAPI 自动校验 request，类型安全
    answer = rag_chain.invoke(request.question)
    return QuestionResponse(
        answer=answer,
        sources=["doc1.pdf", "doc2.pdf"],
        tokens_used=500
    )

1.6 依赖注入

Spring Boot 里依赖注入是核心，@Autowired 满天飞。FastAPI 的依赖注入机制更简洁，用 Depends() 函数：

python 
from fastapi import Depends, HTTPException, Header

def verify_api_key(x_api_key: str = Header(...)):
    """验证 API Key 的依赖函数"""
    if x_api_key != "your-secret-key":
        raise HTTPException(status_code=401, detail="无效的 API Key")
    return x_api_key

@app.post("/ask", response_model=QuestionResponse)
def ask_question(
    request: QuestionRequest,
    api_key: str = Depends(verify_api_key)  # 注入依赖
):
    # 只有验证通过才会执行这里
    ...

依赖可以嵌套，可以有返回值，也可以用于数据库连接管理、缓存等场景。

1.7 异步路由：处理 LLM 调用的正确方式

LLM 调用是 IO 密集型操作：发出 HTTP 请求后，等待 OpenAI 或其他 LLM 服务返回，这个等待过程可能要 5-10 秒。在等待期间，Python 进程什么都不干，无法处理其他请求。

用 async def 定义异步路由，配合 await 处理 IO 操作：

python 
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

@app.post("/ask", response_model=QuestionResponse)
async def ask_question(request: QuestionRequest):
    # await 让出控制权，等待 LLM 返回期间可以处理其他请求
    response = await llm.ainvoke([HumanMessage(content=request.question)])
    return QuestionResponse(
        answer=response.content,
        sources=[],
        tokens_used=response.response_metadata.get("token_usage", {}).get("total_tokens", 0)
    )

注意：ainvoke 是 LangChain 提供的异步版本，对应同步的 invoke。大多数 LangChain 组件都有 async 版本。

1.8 中间件

中间件用于在请求到达路由之前、响应返回之前做统一处理。

CORS（Cross-Origin Resource Sharing，跨域资源共享，浏览器的安全机制，默认禁止网页向不同域名的服务器发请求，需要服务器明确允许才能通过）配置（前端跨域必须）：

python 
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://your-frontend.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

请求日志中间件：

python 
import time
import logging
from fastapi import Request

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} "
        f"duration={duration:.3f}s"
    )
    return response

1.9 完整的 AI 服务示例

把以上内容串起来，一个可以实际运行的 RAG 服务：

python 
# main.py
import os
import time
import logging
from contextlib import asynccontextmanager
from typing import Optional

from fastapi import FastAPI, Depends, HTTPException, Header, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ——— 全局变量，启动时初始化 ———
rag_chain = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用启动/关闭时的生命周期管理"""
    global rag_chain
    logger.info("初始化RAG链...")

    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=embeddings
    )

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    rag_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        return_source_documents=True
    )
    logger.info("RAG链初始化完成")

    yield  # 服务运行期间

    logger.info("服务关闭，清理资源")

app = FastAPI(
    title="RAG知识库服务",
    version="1.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    logger.info(f"{request.method} {request.url.path} {response.status_code} {time.time()-start:.2f}s")
    return response


# ——— 数据模型 ———
class QuestionRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=500)
    top_k: int = Field(3, ge=1, le=10)

class Source(BaseModel):
    content: str
    source: str

class QuestionResponse(BaseModel):
    answer: str
    sources: list[Source]
    latency_ms: float


# ——— 依赖 ———
API_KEY = os.getenv("API_KEY", "dev-key-123")

def verify_key(x_api_key: str = Header(...)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="无效的API Key")


# ——— 路由 ———
@app.get("/health")
def health():
    return {"status": "ok", "rag_ready": rag_chain is not None}

@app.post("/ask", response_model=QuestionResponse)
async def ask(
    request: QuestionRequest,
    _: None = Depends(verify_key)
):
    if rag_chain is None:
        raise HTTPException(status_code=503, detail="RAG链未初始化")

    start = time.time()

    result = await rag_chain.ainvoke({"query": request.question})

    sources = [
        Source(
            content=doc.page_content[:200],
            source=doc.metadata.get("source", "unknown")
        )
        for doc in result.get("source_documents", [])
    ]

    return QuestionResponse(
        answer=result["result"],
        sources=sources,
        latency_ms=(time.time() - start) * 1000
    )

运行：

bash 
export OPENAI_API_KEY="sk-..."
export API_KEY="your-secret-key"
uvicorn main:app --reload --port 8000

访问 http://localhost:8000/docs 测试接口。

1.10 生产部署

开发时用 --reload 模式，生产绝对不能这样。

生产部署用 gunicorn（WSGI/ASGI 服务器，负责管理多个 worker 进程，提高并发处理能力）管理进程，uvicorn 作为 worker：

bash 
pip install gunicorn

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --access-logfile -

workers 数量通常设为 CPU 核数×2+1。AI 服务因为大量 IO 等待，可以适当多一点。

--timeout 120 很重要，LLM 调用可能要十几秒，默认 30 秒超时会直接杀掉请求。

1.11 常见问题：同步函数里调用异步代码

在同步路由函数里直接 await 会报 SyntaxError，用 asyncio.run() 会报 RuntimeError: This event loop is already running。

python 
# 错误：在同步函数里直接 await
@app.post("/ask")
def ask_question(request: QuestionRequest):
    result = await llm.ainvoke(...)  # SyntaxError

# 错误：在已有事件循环中调用 asyncio.run()
@app.post("/ask")
def ask_question(request: QuestionRequest):
    result = asyncio.run(llm.ainvoke(...))  # RuntimeError

正确做法有两种：

  1. 把路由改成 async def（推荐）
  2. 如果必须用同步函数，使用同步版本的方法（invoke 而非 ainvoke），FastAPI 会把同步路由放到线程池执行
python 
# 正确：同步路由用同步方法
@app.post("/ask")
def ask_question(request: QuestionRequest):
    result = llm.invoke(...)  # 同步方法，FastAPI 会在线程池里跑
    return result

1.12 小结

FastAPI 对 Java 开发者来说上手成本不高。路由注解很像 Spring MVC，Pydantic Model 很像 Java 的 DTO，依赖注入的思想也是一样的，只是语法更简洁。

最大的思维转变是 async 编程模型。LLM 调用一定要用 async，不然并发性能会很差。

RAG 系统做好之后，用 FastAPI 包一层，加上认证、日志、CORS，就能构建一个生产可用的 AI 服务。

←上一节生产化部署导读-从Demo到生产的距离下一节流式输出SSE实现→
本页目录