LLM之RAG实战（五十）| FastAPI：构建基于LLM的WEB接口界面

FastAPI是WEB UI接口，随着LLM的蓬勃发展，FastAPI的生态也迎来了新的机遇。本文将围绕FastAPI、OpenAI的API以及FastCRUD，来创建一个个性化的电子邮件写作助手，以展示如何结合这些技术来构建强大的应用程序。

下面我们开始分步骤操作：

一、安装环境

首先，我们创建一个文件夹

mkdir email-assistant-api

进入到该文件夹

cd email-assistant-api

我们使用poetry来管理python包，首先需要先安装poetry

pip install poetry

进入到email-assistant-api文件夹中，并初始化

poetry init

写下所要求的内容（默认为空），然后按 Enter 键

对交互式依赖项键入 no，直到您获得如下内容：

然后只需按 enter 键确认生成。您应该注意到在您的文件夹中创建了一个 pyproject.toml 文件，这是 poetry 用来管理依赖项的文件。

让我们从添加依赖项开始

poetry add fastapi fastcrud sqlmodel openai aiosqlite greenlet python-jose bcrypt

现在你应该还注意到一个 poetry.lock 文件，这是 poetry 保存已安装包的实际版本的方式。

二、项目结构

对于 FastAPI 应用程序，我们有三个主要内容：模型、架构和端点。由于这个 API 很简单，我们可以像这样创建我们的结构：

email_assistant_api/│├── app/│   ├── __init__.py│   ├── main.py        # The main application file│   ├── routes.py      # Contains API route definitions and endpoint logic│   ├── database.py    # Database setup and session management│   ├── models.py      # SQLModel models for the application│   ├── crud.py        # CRUD operation implementations using FastCRUD│   ├── schemas.py     # Schemas for request and response models│   └── .env           # Environment variables│├── pyproject.toml     # Project configuration and dependencies├── README.md          # Provides an overview and documentation└── .gitignore         # Files to be ignored by version control

models.py 中定义我们的模型（数据库表的抽象）；
schemas.py用于验证和序列化数据；
routes.py定义端点；
database.py定义数据库相关信息；
crud.py定义与数据库交互的crud操作；
.env定义环境变量，比如API Key；
main.py定义 FastAPI 应用程序和 API 的入口点；

请注意，此结构适用于小型应用程序，但如果您想为大型应用程序提供更强大的模板，请参考：https://github.com/igorbenav/FastAPI-boilerplate

2.1 对数据库进行建模

对于数据库，我们有一个简单的模型

用户有一个用户名、一个名字、一个电子邮件，我们存储一个哈希密码。
电子邮件日志包含我们为输入定义的内容，以及与此日志关联的用户的时间戳、generated_email 和 ID。
用户可能有多个电子邮件日志。

models.py代码示例：

# app/models.pyfrom sqlmodel import SQLModel, Fieldfrom typing import Optionalclass User(SQLModel, table=True):    id: Optional[int] = Field(default=None, primary_key=True)    name: str = Field(..., min_length=2, max_length=30)    username: str = Field(..., min_length=2, max_length=20)    email: str    hashed_password: strclass EmailLog(SQLModel, table=True):    id: Optional[int] = Field(default=None, primary_key=True)    user_id: int = Field(foreign_key="user.id")    user_input: str    reply_to: Optional[str] = Field(default=None)    context: Optional[str] = Field(default=None)    length: Optional[int] = Field(default=None)    tone: str    generated_email: str    timestamp: str

为了与我们的数据库交互，我们将在 crud.py 中为每个模型实例化 FastCRUD[1]

# app/crud.pyfrom fastcrud import FastCRUDfrom .models import User, EmailLogcrud_user = FastCRUD(User)crud_email_log = FastCRUD(EmailLog)

2.2 创建schemas

schemas.py 中创建我们的 schemas

# app/schemas.pyfrom datetime import datetimefrom typing import Optionalfrom sqlmodel import SQLModel, Field# ------- user -------class UserCreate(SQLModel):    name: str    username: str    email: str    password: strclass UserRead(SQLModel):    id: int    name: str    username: str    email: strclass UserCreateInternal(SQLModel):    name: str    username: str    email: str    hashed_password: str# ------- email -------class EmailRequest(SQLModel):    user_input: str    reply_to: Optional[str] = None    context: Optional[str] = None    length: int = 120    tone: str = "formal"class EmailResponse(SQLModel):    generated_email: str# ------- email log -------class EmailLogCreate(SQLModel):    user_id: int    user_input: str    reply_to: Optional[str] = None    context: Optional[str] = None    length: Optional[int] = None    tone: Optional[str] = None    generated_email: str    timestamp: datetime = Field(      default_factory=lambda: datetime.now(UTC)    )class EmailLogRead(SQLModel):    user_id: int    user_input: str    reply_to: Optional[str]    context: Optional[str]    length: Optional[int]    tone: Optional[str]    generated_email: str    timestamp: datetime

要创建用户，我们要求提供姓名、用户名、电子邮件和密码（我们将存储哈希值）；
我们将默认长度设置为 120，默认tone设置为 “正式”；
我们自动生成 EmailLog 的时间戳

2.3 创建我们的应用程序并设置数据库

尽管我们已经有了模型和架构，但实际上我们既没有为终端节点提供服务的应用程序，也没有用于创建表的数据库。

下面看一下database.py：

# app/database.pyfrom sqlmodel import SQLModel, create_engine, AsyncSessionfrom sqlalchemy.ext.asyncio import create_async_enginefrom sqlalchemy.orm import sessionmakerDATABASE_URL = "sqlite+aiosqlite:///./emailassistant.db"engine = create_async_engine(DATABASE_URL, echo=True)async_session = sessionmaker(    engine, class_=AsyncSession, expire_on_commit=False)async def create_db_and_tables():    async with engine.begin() as conn:        await conn.run_sync(SQLModel.metadata.create_all)async def get_session() -> AsyncSession:    async with async_session() as session:        yield session

在这里，我们连接到了一个 SQLite 数据库，创建了一个函数来创建我们的数据库和表，以及一个允许我们与该数据库交互的会话。

现在让我们最终创建我们的 FastAPI 应用程序：

# app/main.pyfrom fastapi import FastAPIfrom .database import create_db_and_tablesasync def lifespan(app):    await create_db_and_tables()    yieldapp = FastAPI(lifespan=lifespan)

我们定义lifespan，以便在启动时创建 db 和 tables。

让我们运行一下代码来测试一下：

poetry run fastapi run

结果如下所示：

在浏览器登录如下地址：

127.0.0.1:8000/docs

可以看到如下界面：

可以在终端中按 Ctrl C 暂时关闭应用程序。

2.4 创建端点

接下来，创建端点来生成电子邮件。

首先在 .env 中输入 OpenAI API 密钥（这将被 .gitignore 忽略，并且不会出现在我们的存储库中）：

# app/.envOPENAI_API_KEY="my_openai_api_key"

将其写入 .gitignore 以确保不会提交此 API 密钥：

# .gitignore.env.venvenv/venv/ENV/env.bak/venv.bak/

现在从 .env 中获取 OpenAI API 密钥并实例化客户端：

# app/routes.pyimport osfrom starlette.config import Configfrom openai import OpenAIcurrent_file_dir = os.path.dirname(os.path.realpath(__file__))env_path = os.path.join(current_file_dir, ".env")config = Config(env_path)OPENAI_API_KEY = config("OPENAI_API_KEY")open_ai_client = OpenAI(api_key=OPENAI_API_KEY)

为电子邮件终端节点创建一个路由器，并实际为电子邮件创建终端节点：

我们将创建一个系统提示符，使输出适应我们想要的结果；
我们将创建一个基本提示，用于格式化传递的信息；
然后我们将此信息传递给 OpenAI 客户端；
最后，我们将在数据库中创建一个日志条目并返回生成的电子邮件

# app/routes.py...from openai import OpenAIfrom fastapi import APIRouter, Depends, HTTPExceptionfrom .schemas import EmailRequest, EmailResponsefrom .database import get_session...# ------- email -------email_router = APIRouter()@email_router.post("/", response_model=EmailResponse)async def generate_email(    request: EmailRequest,     db: AsyncSession = Depends(get_session)):    try:        system_prompt = f"""        You are a helpful email assistant.         You get a prompt to write an email,        you reply with the email and nothing else.        """                prompt = f"""        Write an email based on the following input:        - User Input: {request.user_input}        - Reply To: {request.reply_to if request.reply_to else 'N/A'}        - Context: {request.context if request.context else 'N/A'}        - Length: {request.length if request.length else 'N/A'} characters        - Tone: {request.tone if request.tone else 'N/A'}        """                response = await open_ai_client.chat.completions.create(            model="gpt-3.5-turbo",            messages=[                {"role": "system", "content": system_prompt},                {"role": "user", "content": prompt},            ],            max_tokens=request.length        )                generated_email = response.choices[0].message['content'].strip()        log_entry = EmailLogCreate(            user_id=request.user_id,            user_input=request.user_input,            reply_to=request.reply_to,            context=request.context,            length=request.length,            tone=request.tone,            generated_email=generated_email,        )        await crud_email_logs.create(db, log_entry)        return EmailResponse(generated_email=generated_email)    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))

现在定义 app/main.py ，将这个电子邮件路由器包含到我们的 FastAPI 应用程序中：

# app/main.pyfrom fastapi import FastAPIfrom .database import create_db_and_tablesfrom .routes import email_routerasync def lifespan(app):    await create_db_and_tables()    yieldapp = FastAPI(lifespan=lifespan)app.include_router(email_router, prefix="/generate", tags=["Email"])

再次保存并运行 FastAPI 应用程序（127.0.0.1：8000/docs），会看到如下界面：

点击这个新创建的 post 端点，传递一些信息并单击 execute

可以得到如下内容：

结果是我们所希望的回应，但是目前还无法通过查看日志来判断系统是否正常工作，因此让我们也创建电子邮件日志端点：

# app/routes.py...from fastapi import APIRouter, Depends, HTTPExceptionfrom sqlalchemy.ext.asyncio.session import AsyncSessionfrom .schemas import EmailLogCreate, EmailLogRead...# ------- email log -------log_router = APIRouter()@log_router.get("/")async def read_logs(db: AsyncSession = Depends(get_session)):    logs = await crud_email_logs.get_multi(db)    return logs@log_router.get("/{log_id}", response_model=EmailLogRead)async def read_log(log_id: int, db: AsyncSession = Depends(get_session)):    log = await crud_email_logs.get(db, id=log_id)    if not log:        raise HTTPException(status_code=404, detail="Log not found")    return log

我们可以按其 ID 查看多个日志或一个日志。让我们也将这个路由器包含在我们的应用程序中：

# app/main.pyfrom fastapi import FastAPIfrom .database import create_db_and_tablesfrom .routes import email_router, log_routerasync def lifespan(app):    await create_db_and_tables()    yieldapp = FastAPI(lifespan=lifespan)app.include_router(email_router, prefix="/generate", tags=["Email"])app.include_router(log_router, prefix="/logs", tags=["Logs"])

三、用户功能、身份验证和安全性

现在，让我们添加实际的用户创建功能。首先在终端上运行：

openssl rand -hex 32

然后将结果写入 .env 作为SECRET_KEY

# app/.envOPENAI_API_KEY="my_openai_api_key"SECRET_KEY="my_secret_key"

首先创建一个文件 helper.py 并将以下代码粘贴到其中：

# app/helper.pyimport osfrom datetime import UTC, datetime, timedeltafrom typing import Any, Annotatedimport bcryptfrom jose import JWTError, jwtfrom fastapi import Depends, HTTPExceptionfrom fastapi.security import OAuth2PasswordBearerfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqlmodel import SQLModelfrom starlette.config import Configfrom .database import get_sessionfrom .crud import crud_userscurrent_file_dir = os.path.dirname(os.path.realpath(__file__))env_path = os.path.join(current_file_dir, ".env")config = Config(env_path)# Security settingsSECRET_KEY = config("SECRET_KEY")oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/login")# Token modelsclass Token(SQLModel):    access_token: str    token_type: strclass TokenData(SQLModel):    username_or_email: str# Utility functionsasync def verify_password(plain_password: str, hashed_password: str) -> bool:    """Verify a plain password against a hashed password."""    return bcrypt.checkpw(plain_password.encode(), hashed_password.encode())def get_password_hash(password: str) -> str:    """Hash a password."""    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()async def create_access_token(    data: dict[str, Any],     expires_delta: timedelta | None = None) -> str:    """Create a JWT access token."""    to_encode = data.copy()    if expires_delta:        expire = datetime.now(UTC).replace(tzinfo=None) + expires_delta    else:        expire = datetime.now(UTC).replace(tzinfo=None) + timedelta(minutes=15)    to_encode.update({"exp": expire})    return jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")async def verify_token(token: str, db: AsyncSession) -> TokenData | None:    """Verify a JWT token and extract the user data."""    try:        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])        username_or_email: str = payload.get("sub")        if username_or_email is None:            return None        return TokenData(username_or_email=username_or_email)    except JWTError:        return Noneasync def authenticate_user(username_or_email: str, password: str, db: AsyncSession):    if "@" in username_or_email:        db_user: dict | None = await crud_users.get(db=db, email=username_or_email, is_deleted=False)    else:        db_user = await crud_users.get(db=db, username=username_or_email, is_deleted=False)    if not db_user:        return False    elif not await verify_password(password, db_user["hashed_password"]):        return False    return db_user# Dependencyasync def get_current_user(    token: Annotated[str, Depends(oauth2_scheme)],     db: Annotated[AsyncSession, Depends(get_session)]) -> dict[str, Any] | None:    """Get the current authenticated user."""    token_data = await verify_token(token, db)    if token_data is None:        raise HTTPException(status_code=401, detail="User not authenticated.")    if "@" in token_data.username_or_email:        user = await crud_users.get(            db=db, email=token_data.username_or_email, is_deleted=False        )    else:        user = await crud_users.get(            db=db, username=token_data.username_or_email, is_deleted=False        )    if user:        return user    raise HTTPException(status_code=401, detail="User not authenticated.")

verify_password：根据哈希密码验证普通密码。它用于检查用户提供的密码是否与存储的哈希密码匹配。
get_password_hash：在将用户提供的密码存储到数据库之前对其进行哈希处理。
create_access_token：用于为经过身份验证的用户生成 JWT 类型的令牌。
verify_token：验证 JWT 令牌并提取用户数据。
authenticate_user：负责根据用户的用户名或电子邮件和密码对用户进行身份验证。
get_current_user：是一种依赖项，它根据提供的令牌检索当前经过身份验证的用户。

现在，让我们使用这些实用程序函数来创建用户路由。

# app/routes.pyfrom datetime import timedeltafrom fastapi import APIRouter, Depends, HTTPExceptionfrom .database import get_sessionfrom .schemas import UserCreate, UserReadfrom .helper import (    get_password_hash,     authenticate_user,     create_access_token,     get_current_user,     Token)# ------- user -------user_router = APIRouter()@user_router.post("/register", response_model=UserRead)async def register_user(    user: UserCreate,     db: AsyncSession = Depends(get_session)):    hashed_password = get_password_hash(user.password)    user_data = user.dict()    user_data["hashed_password"] = hashed_password    del user_data["password"]        new_user = await crud_users.create(        db,         object=UserCreateInternal(**user_data)    )    return new_user@user_router.post("/login", response_model=Token)async def login_user(user: UserCreate, db: AsyncSession = Depends(get_session)):    db_user = await crud_users.get(db, email=user.email)    password_verified = await verify_password(      user.password, db_user.hashed_password    )    if not db_user or not password_verified:        raise HTTPException(status_code=400, detail="Invalid credentials")        access_token_expires = timedelta(minutes=30)    access_token = await create_access_token(        data={"sub": user["username"]},         expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}

并将路由器包含在我们的应用程序中：

# app/main.pyfrom fastapi import FastAPIfrom .database import create_db_and_tablesfrom .routes import user_router, email_router, log_routerasync def lifespan(app):    await create_db_and_tables()    yieldapp = FastAPI(lifespan=lifespan)app.include_router(user_router, prefix="/users", tags=["Users"])app.include_router(email_router, prefix="/generate", tags=["Email"])app.include_router(log_router, prefix="/logs", tags=["Logs"])

最后，让我们在 generate_email 端点中注入 get_current_user 依赖项，添加用户登录以生成电子邮件的需求，此外，还会自动将用户的 ID 存储在日志中：

# app/routes.py...@email_router.post("/", response_model=EmailResponse)async def generate_email(    request: EmailRequest,     db: AsyncSession = Depends(get_session),    current_user: dict = Depends(get_current_user)):    try:        prompt = f"""        Write an email based on the following input:        - User Input: {request.user_input}        - Reply To: {request.reply_to if request.reply_to else 'N/A'}        - Context: {request.context if request.context else 'N/A'}        - Length: {request.length if request.length else 'N/A'} characters        - Tone: {request.tone if request.tone else 'N/A'}        """                response = open_ai_client.chat.completions.create(            model="gpt-3.5-turbo",            messages=[                {"role": "system", "content": "You are a helpful email assistant."},                {"role": "user", "content": prompt}            ],            max_tokens=request.length        )        generated_email = response.choices[0].message.content        log_entry = EmailLogCreate(            user_id=current_user['id'],            user_input=request.user_input,            reply_to=request.reply_to,            context=request.context,            length=request.length,            tone=request.tone,            generated_email=generated_email,        )        await crud_email_logs.create(db, log_entry)        return EmailResponse(generated_email=generated_email)    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))

如果现在检查终端节点，您将在右侧看到一个小锁。

现在运行程序，需要进行身份验证，您可以通过单击锁并在此处传递有效的用户名和密码（您创建的用户）来完成：

现在，我们还将此依赖项设置为日志端点，此外，让我们使用 FastCRUD 仅过滤当前用户 ID 的日志。

我们可以通过注入 get_current_user 依赖项并将 user_id=current_user[“id”] 传递给 FastCRUD 来实现这一点（当前用户是 get_current_user 返回的）。

...# ------- email log -------log_router = APIRouter()@log_router.get("/")async def read_logs(    db: AsyncSession = Depends(get_session),    current_user: dict[str, Any] = Depends(get_current_user)):    logs = await crud_email_logs.get_multi(db, user_id=current_user["id"])    return logs@log_router.get("/{log_id}", response_model=EmailLogRead)async def read_log(    log_id: int,     db: AsyncSession = Depends(get_session),    current_user: dict[str, Any] = Depends(get_current_user)):    log = await crud_email_logs.get(db, id=log_id, user_id=current_user["id"])    if not log:        raise HTTPException(status_code=404, detail="Log not found")    return log

现在，我们实际上只能读取我们自己的日志，而且，只有在登录时才能读取。

最终的 routes 文件：

# app/routes.pyimport osfrom typing import Annotated, Anyfrom datetime import timedeltafrom fastapi import APIRouter, Depends, HTTPExceptionfrom fastapi.security import OAuth2PasswordRequestFormfrom sqlalchemy.ext.asyncio.session import AsyncSessionfrom starlette.config import Configfrom openai import OpenAIfrom .crud import crud_email_logs, crud_usersfrom .database import get_sessionfrom .schemas import (    EmailRequest,     EmailResponse,     EmailLogCreate,     EmailLogRead,    UserCreate,     UserRead,     UserCreateInternal, )from .helper import (    get_password_hash,     authenticate_user,     create_access_token,     get_current_user,     Token)current_file_dir = os.path.dirname(os.path.realpath(__file__))env_path = os.path.join(current_file_dir, ".env")config = Config(env_path)OPENAI_API_KEY = config("OPENAI_API_KEY")open_ai_client = OpenAI(api_key=OPENAI_API_KEY)# ------- user -------user_router = APIRouter()@user_router.post("/register", response_model=UserRead)async def register_user(    user: UserCreate,     db: AsyncSession = Depends(get_session)):    hashed_password = get_password_hash(user.password)    user_data = user.dict()    user_data["hashed_password"] = hashed_password    del user_data["password"]        new_user = await crud_users.create(        db,         object=UserCreateInternal(**user_data)    )    return new_user@user_router.post("/login", response_model=Token)async def login_user(    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],    db: AsyncSession = Depends(get_session)):    user = await authenticate_user(        username_or_email=form_data.username,         password=form_data.password,         db=db    )    if not user:        raise HTTPException(status_code=400, detail="Invalid credentials")        access_token_expires = timedelta(minutes=30)    access_token = await create_access_token(        data={"sub": user["username"]},         expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}# ------- email -------email_router = APIRouter()@email_router.post("/", response_model=EmailResponse)async def generate_email(    request: EmailRequest,     db: AsyncSession = Depends(get_session),    current_user: dict = Depends(get_current_user)):    try:        system_prompt = f"""        You are a helpful email assistant.         You get a prompt to write an email,        you reply with the email and nothing else.        """        prompt = f"""        Write an email based on the following input:        - User Input: {request.user_input}        - Reply To: {request.reply_to if request.reply_to else 'N/A'}        - Context: {request.context if request.context else 'N/A'}        - Length: {request.length if request.length else 'N/A'} characters        - Tone: {request.tone if request.tone else 'N/A'}        """                response = open_ai_client.chat.completions.create(            model="gpt-3.5-turbo",            messages=[                {"role": "system", "content": system_prompt},                {"role": "user", "content": prompt}            ],            max_tokens=request.length        )        generated_email = response.choices[0].message.content        log_entry = EmailLogCreate(            user_id=current_user['id'],            user_input=request.user_input,            reply_to=request.reply_to,            context=request.context,            length=request.length,            tone=request.tone,            generated_email=generated_email,        )        await crud_email_logs.create(db, log_entry)        return EmailResponse(generated_email=generated_email)    except Exception as e:        raise HTTPException(status_code=500, detail=str(e))# ------- email log -------log_router = APIRouter()@log_router.get("/")async def read_logs(    db: AsyncSession = Depends(get_session),    current_user: dict[str, Any] = Depends(get_current_user)):    logs = await crud_email_logs.get_multi(db, user_id=current_user["id"])    return logs@log_router.get("/{log_id}", response_model=EmailLogRead)async def read_log(    log_id: int,     db: AsyncSession = Depends(get_session),    current_user: dict[str, Any] = Depends(get_current_user)):    log = await crud_email_logs.get(db, id=log_id, user_id=current_user["id"])    if not log:        raise HTTPException(status_code=404, detail="Log not found")    return log