此博客帖子是 LLM SaaS 系列中的 FastAPI + Supabase 模板的一部分，构建在第一部分（身份验证和文件上传）介绍的概念之上。
FastAPI模板用于LLM SaaS第1部分-身份验证和文件上传
FastAPI在Python开发人员中的日益流行凸显了它的简单性和本地Swagger UI支持…
pub.towardsai.net
大部分代码示例参考自Quivr。
Celery Worker和消息队列用于长时间运行的进程
下图展示了Celery工作者和消息队列在FastAPI生态系统中的协作方式。该过程始于FastAPI将任务发送到指定的代理（在本例中为Redis）。随后，Celery工作者在分布式任务队列中检索和处理这些任务，并将结果保存到结果后端（同样是Redis）。同时，FastAPI可以监视任务的状态和结果。如果需要，可以使用不同的实例来分别用作代理和结果后端，尽管示例中使用了单个Redis实例。
要启动开发过程，必须使用以下Docker命令启动一个Redis实例：
# 拉取最新的Redis镜像docker pull redis:latest# 运行一个Redis实例docker run --name redis -d -p 6379:6379 redis:latest在FastAPI项目中为代理和结果后端配置环境变量：
# 代理实例 - RedisCELERY_BROKER_URL=redis://localhost:6379/0# 结果后端 - RedisCELERY_RESULT_BACKEND=redis://localhost:6379/0在main.py中创建一个用于测试的虚拟任务：
from celery import Celeryimport time
celery = Celery( __name__, broker=os.getenv("CELERY_BROKER_URL"), backend=os.getenv("CELERY_RESULT_BACKEND"),)@Celery.taskdef test(): import time time.sleep(5) return "你好，我喜欢吃芹菜！"返回终端，然后输入以下 celery 命令（假设您已经在环境中安装了 celery。如果没有，请使用 pip install）
celery --app=main.celery worker --concurrency=1 --loglevel=DEBUG注意：如果您要在Windows计算机上测试脚本，您可能需要在命令中添加’ -P solo’ 来使其在本地环境中工作。 这在生产环境中不需要。
您将看到类似以下内容：
-------------- celery@xxxx v5.2.7 (dawn-chorus)--- ***** ----- -- ******* ---- Windows-10-10.0.22621-SP0 2023-11-20 07:03:38- *** --- * --- - ** ---------- [config]- ** ---------- .> app: main:0x22d23e16d70- ** ---------- .> transport: redis://localhost:6379/0- ** ---------- .> results: redis://localhost:6379/0- *** --- * --- .> concurrency: 4 (prefork)-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)--- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery现在，您可以在基于工作目录的另一个终端中使用Python REPL进行快速测试。
(.venv) PS C:\Users\xxx\backend> pythonPython 3.10.11 (tags/v3.10.11:7d4cc5a, Apr 5 2023, 00:38:17) [MSC v.1929 64 bit (AMD64)] on win32Type "help", "copyright", "credits" or "license" for more information.>>> from main import app, celery, test>>> test.delay()<AsyncResult: 2ba428c1-5d82-4f37-aa89-5cef76b7a6eb>回到你正在运行Celery worker的另一个终端，你应该能在Celery worker的终端中观察到任务的执行。
[2023-11-20 12:21:46,743: INFO/MainProcess] 任务 main.test[2ba428c1-5d82-4f37-aa89-5cef76b7a6eb] 已接收[2023-11-20 12:21:51,754: INFO/MainProcess] 任务 main.test[2ba428c1-5d82-4f37-aa89-5cef76b7a6eb] 成功完成，耗时 5.014999999999418 秒: '你好，我喜欢吃芹菜！'文件上传和向量数据存储（pg-vector 插件）
基于Celery测试，实际的用例涉及运行后台Celery任务以嵌入PDF文档并将其保存到向量数据存储中。该过程包括将文件上传到Supabase存储桶并触发一个Celery任务来下载和处理该文件以用于向量数据存储。
整个过程有些复杂。该文件首先将被上传到Supabase存储桶。然后我们会触发一个Celery任务来下载这个文件并将其处理为向量数据存储。该过程将涉及文档加载器将原始文件格式转换为原始文本，并涉及文本分割器将文本分割成块（由于向量数据存储中单个向量的大小限制）。此外，我们还将为任何特定的文本块添加元数据。最后，文本块将被嵌入到向量中并上传到Supabase向量数据存储（postgres pg-vector 插件）。
Supabase上的SQL表
首先，确保在Supabase上创建了两个表用于此演示：（更多sql脚本示例，可以参考https://github.com/StanGirard/quivr/tree/main/scripts）
-- 创建用户X向量表CREATE TABLE IF NOT EXISTS user_vectors ( user_id UUID, vector_id UUID, PRIMARY KEY (user_id, vector_id), FOREIGN KEY (vector_id) REFERENCES vectors (id), FOREIGN KEY (user_id) REFERENCES auth.users (id));-- 创建向量扩展CREATE EXTENSION IF NOT EXISTS vector;-- 创建向量表CREATE TABLE IF NOT EXISTS vectors ( id UUID DEFAULT uuid_generate_v4() PRIMARY KEY, content TEXT, metadata JSONB, embedding VECTOR(1536));定义路由和端点
在main.py中，添加一个名为’upload_router’的新路由器。
from routes.upload_routes import upload_routerapp.include_router(upload_router)创建一个名为’routes’的新目录，并创建一个名为’upload_routes.py’的文件
from fastapi.responses import JSONResponsefrom auth import AuthBearer, get_current_userfrom celery_worker import process_filefrom celery.result import AsyncResultfrom fastapi import APIRouter, Depends, HTTPException, Request, UploadFilefrom repository.files.upload_file import upload_file_storagefrom logger import get_loggerfrom models import UserIdentitylogger = get_logger(__name__)upload_router = APIRouter()@upload_router.get("/upload/healthz", tags=["Health"])async def healthz(): return {"status": "ok"}@upload_router.post("/upload", dependencies=[Depends(AuthBearer())], tags=["Upload"])async def upload_file( request: Request, uploadFile: UploadFile, current_user: UserIdentity = Depends(get_current_user),): file_content = await uploadFile.read() filename_with_user_id = str(current_user.id) + "/" + str(uploadFile.filename) logger.info(f"文件名为：{filename_with_user_id}") try: fileInStorage = upload_file_storage(file_content, filename_with_user_id) logger.info(f"文件 {fileInStorage} 上传成功") except Exception as e: if "该资源已存在" in str(e): raise HTTPException( status_code=403, detail=f"文件 {uploadFile.filename} 已存在于存储中。", ) else: raise HTTPException( status_code=500, detail="将文件上传到存储失败。" ) task = process_file.delay( file_name=filename_with_user_id, file_original_name=uploadFile.filename, user_id=current_user.id, ) return JSONResponse({"task_id": task.id})@upload_router.get("/upload/{task_id}", dependencies=[Depends(AuthBearer())], tags=["Upload"])def get_status(task_id: str): task_result = AsyncResult(task_id) result = { "task_id": task_id, "task_status": task_result.status } return JSONResponse(result)此脚本将在 ‘upload_routes.py’ 中定义两个端点，用于上传文件和检查任务状态。
您可以看到，在 /upload 中有一个名为 ‘process_file’ 的 celery 任务。现在让我们在 celery 中创建此任务。
Celery 工作器和任务
首先，在主目录中创建一个名为 ‘celery_worker.py’ 的文件。
import osfrom celery import Celeryimport asynciofrom utils.process_file import get_supabase_client,file_handlercelery = Celery( __name__, broker="redis://127.0.0.1:6379/0", backend="redis://127.0.0.1:6379/0")@celery.task(name="process_file")def process_file( file_name: str, file_original_name: str, user_id: str,): supabase_client = get_supabase_client() tmp_file_name = "tmp-file-"+file_name tmp_file_name = tmp_file_name.replace("/", "_") with open(tmp_file_name, "wb+") as file: res = supabase_client.storage.from_("quivr").download(file_name) file.write(res) loop = asyncio.new_event_loop() message = loop.run_until_complete( file_handler( file=tmp_file_name, user_id=user_id, file_original_name=file_original_name ) ) file.close os.remove(tmp_file_name)此 ‘process_file’ 任务（参考上面的流程图）将下载文件，使用 file_handler 处理文件，然后在完成后删除临时文件。
文件处理和嵌入
为了简单起见，您可以使用下面的 file_handler 脚本。此脚本具有一个工作器来完成所有嵌入。您还可以查看 Quivr 的代码库，他们有另一个共享任务，可以将嵌入分配给多个工作器。
# utils/process_file.py 处理上传的文件import osimport timefrom logger import get_loggerfrom repository.files.upload_file import DocumentSerializablefrom langchain.document_loaders import UnstructuredPDFLoaderfrom models.databases.supabase.supabase import SupabaseDBfrom supabase.client import Client, create_clientfrom langchain.vectorstores import SupabaseVectorStorefrom langchain.embeddings.openai import OpenAIEmbeddingsfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom dotenv import load_dotenvload_dotenv()logger = get_logger(__name__)def get_supabase_client() -> Client: supabase_client: Client = create_client( os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_SERVICE_KEY") ) return supabase_clientdef get_supabase_db() -> SupabaseDB: supabase_client = get_supabase_client() return SupabaseDB(supabase_client)def get_embeddings() -> OpenAIEmbeddings: embeddings = OpenAIEmbeddings( openai_api_key=os.getenv("OPENAI_API_KEY") ) # pyright: ignore reportPrivateUsage=none return embeddingsdef get_documents_vector_store() -> SupabaseVectorStore: # settings = BrainSettings() # pyright: ignore reportPrivateUsage=none embeddings = get_embeddings() supabase_client: Client = create_client( os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_SERVICE_KEY") ) documents_vector_store = SupabaseVectorStore( supabase_client, embeddings, table_name="vectors" ) return documents_vector_storedef create_vector(doc): documents_vector_store = get_documents_vector_store() try: sids = documents_vector_store.add_documents([doc]) if sids and len(sids) > 0: return sids except Exception as e: logger.error(f"Error creating vector for document: {e}") def create_user_vector(user_id, vector_id): database = get_supabase_db() response = ( database.db.table("user_vectors") .insert( { "user_id": str(user_id), "vector_id": str(vector_id), } ) .execute() ) return response.data def create_embedding_for_document(user_id, doc_with_metadata): doc = DocumentSerializable.from_json(doc_with_metadata) created_vector = create_vector(doc) created_vector_id = created_vector[0] # pyright: ignore reportPrivateUsage=none create_user_vector(user_id, created_vector_id) def compute_documents_from_pdf(file,loader): loader = loader(file) documents=[] documents = loader.load() # split the documents into chunks text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder( chunk_size=500, chunk_overlap=0 ) documents = text_splitter.split_documents(documents) return documents async def file_handler( file: str, file_original_name: str, user_id, loader_class=UnstructuredPDFLoader, #the loader class from Langchain): dateshort = time.strftime("%Y%m%d") documents = compute_documents_from_pdf(file,loader_class) for doc in documents: # pyright: ignore reportPrivateUsage=none metadata = { "file_name": file_original_name, "date": dateshort } doc_with_metadata = DocumentSerializable( page_content=doc.page_content, metadata=metadata ) create_embedding_for_document( user_id, doc_with_metadata.to_json() ) return "Hello, processing is done!"仅用于演示目的，这里只测试了pdf文件。更多文件格式可以参考Quivr的代码库，这里使用了File类来处理各种文件格式。
端对端测试
要测试这个，请同时启动Uvicorn服务器（用于FastAPI）和Celery服务器。
uvicorn main:app --reload
celery -A celery_worker worker --loglevel=info --logfile=celery.log --concurrency=1 -P solo— logfile（可选）：可以在工作目录中保存celery日志文件
— concurrency（可选）：设置要同时启动的工作者数量
— P solo：我需要这个来使celery在Windows笔记本电脑上运行。如果在Mac/Docker上运行，可能不需要这个。
这里是测试端点的代码片段。
