官方网站建设有限公司,网络应用程序设计报告,主机屋 wordpress,网络市场调研的步骤FastAPI是WEB UI接口#xff0c;随着LLM的蓬勃发展#xff0c;FastAPI的生态也迎来了新的机遇。本文将围绕FastAPI、OpenAI的API以及FastCRUD#xff0c;来创建一个个性化的电子邮件写作助手#xff0c;以展示如何结合这些技术来构建强大的应用程序。
下面我们开始分步骤操… 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 Keymain.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, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) name: str Field(..., min_length2, max_length30) username: str Field(..., min_length2, max_length20) email: str hashed_password: strclass EmailLog(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) user_id: int Field(foreign_keyuser.id) user_input: str reply_to: Optional[str] Field(defaultNone) context: Optional[str] Field(defaultNone) length: Optional[int] Field(defaultNone) 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 formalclass 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_factorylambda: 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 sqliteaiosqlite:///./emailassistant.dbengine create_async_engine(DATABASE_URL, echoTrue)async_session sessionmaker( engine, class_AsyncSession, expire_on_commitFalse)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(lifespanlifespan)
我们定义lifespan以便在启动时创建 db 和 tables。
让我们运行一下代码来测试一下
poetry run fastapi run
结果如下所示 在浏览器登录如下地址
127.0.0.1:8000/docs
可以看到如下界面 可以在终端中按 Ctrl C 暂时关闭应用程序。
2.4 创建端点 接下来创建端点来生成电子邮件。 首先在 .env 中输入 OpenAI API 密钥这将被 .gitignore 忽略并且不会出现在我们的存储库中
# app/.envOPENAI_API_KEYmy_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_keyOPENAI_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_modelEmailResponse)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( modelgpt-3.5-turbo, messages[ {role: system, content: system_prompt}, {role: user, content: prompt}, ], max_tokensrequest.length ) generated_email response.choices[0].message[content].strip() log_entry EmailLogCreate( user_idrequest.user_id, user_inputrequest.user_input, reply_torequest.reply_to, contextrequest.context, lengthrequest.length, tonerequest.tone, generated_emailgenerated_email, ) await crud_email_logs.create(db, log_entry) return EmailResponse(generated_emailgenerated_email) except Exception as e: raise HTTPException(status_code500, detailstr(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(lifespanlifespan)app.include_router(email_router, prefix/generate, tags[Email]) 再次保存并运行 FastAPI 应用程序 127.0.0.18000/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 logslog_router.get(/{log_id}, response_modelEmailLogRead)async def read_log(log_id: int, db: AsyncSession Depends(get_session)): log await crud_email_logs.get(db, idlog_id) if not log: raise HTTPException(status_code404, detailLog 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(lifespanlifespan)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_KEYmy_openai_api_keySECRET_KEYmy_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(tzinfoNone) expires_delta else: expire datetime.now(UTC).replace(tzinfoNone) timedelta(minutes15) to_encode.update({exp: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithmHS256)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_emailusername_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(dbdb, emailusername_or_email, is_deletedFalse) else: db_user await crud_users.get(dbdb, usernameusername_or_email, is_deletedFalse) 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_code401, detailUser not authenticated.) if in token_data.username_or_email: user await crud_users.get( dbdb, emailtoken_data.username_or_email, is_deletedFalse ) else: user await crud_users.get( dbdb, usernametoken_data.username_or_email, is_deletedFalse ) if user: return user raise HTTPException(status_code401, detailUser 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_modelUserRead)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, objectUserCreateInternal(**user_data) ) return new_useruser_router.post(/login, response_modelToken)async def login_user(user: UserCreate, db: AsyncSession Depends(get_session)): db_user await crud_users.get(db, emailuser.email) password_verified await verify_password( user.password, db_user.hashed_password ) if not db_user or not password_verified: raise HTTPException(status_code400, detailInvalid credentials) access_token_expires timedelta(minutes30) access_token await create_access_token( data{sub: user[username]}, expires_deltaaccess_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(lifespanlifespan)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_modelEmailResponse)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( modelgpt-3.5-turbo, messages[ {role: system, content: You are a helpful email assistant.}, {role: user, content: prompt} ], max_tokensrequest.length ) generated_email response.choices[0].message.content log_entry EmailLogCreate( user_idcurrent_user[id], user_inputrequest.user_input, reply_torequest.reply_to, contextrequest.context, lengthrequest.length, tonerequest.tone, generated_emailgenerated_email, ) await crud_email_logs.create(db, log_entry) return EmailResponse(generated_emailgenerated_email) except Exception as e: raise HTTPException(status_code500, detailstr(e))
如果现在检查终端节点您将在右侧看到一个小锁。 现在运行程序需要进行身份验证您可以通过单击锁并在此处传递有效的用户名和密码您创建的用户来完成 现在我们还将此依赖项设置为日志端点此外让我们使用 FastCRUD 仅过滤当前用户 ID 的日志。 我们可以通过注入 get_current_user 依赖项并将 user_idcurrent_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_idcurrent_user[id]) return logslog_router.get(/{log_id}, response_modelEmailLogRead)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, idlog_id, user_idcurrent_user[id]) if not log: raise HTTPException(status_code404, detailLog 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_keyOPENAI_API_KEY)# ------- user -------user_router APIRouter()user_router.post(/register, response_modelUserRead)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, objectUserCreateInternal(**user_data) ) return new_useruser_router.post(/login, response_modelToken)async def login_user( form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: AsyncSession Depends(get_session)): user await authenticate_user( username_or_emailform_data.username, passwordform_data.password, dbdb ) if not user: raise HTTPException(status_code400, detailInvalid credentials) access_token_expires timedelta(minutes30) access_token await create_access_token( data{sub: user[username]}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer}# ------- email -------email_router APIRouter()email_router.post(/, response_modelEmailResponse)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( modelgpt-3.5-turbo, messages[ {role: system, content: system_prompt}, {role: user, content: prompt} ], max_tokensrequest.length ) generated_email response.choices[0].message.content log_entry EmailLogCreate( user_idcurrent_user[id], user_inputrequest.user_input, reply_torequest.reply_to, contextrequest.context, lengthrequest.length, tonerequest.tone, generated_emailgenerated_email, ) await crud_email_logs.create(db, log_entry) return EmailResponse(generated_emailgenerated_email) except Exception as e: raise HTTPException(status_code500, detailstr(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_idcurrent_user[id]) return logslog_router.get(/{log_id}, response_modelEmailLogRead)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, idlog_id, user_idcurrent_user[id]) if not log: raise HTTPException(status_code404, detailLog not found) return log
参考文献
[1] https://github.com/igorbenav/fastcrud?tabreadme-ov-file
