FastAPI+Celery+pg-vector构建LLM SaaS生产级文件处理架构

1. 项目概述：为什么一个LLM SaaS后端需要Celery和pg-vector？

我做AI应用开发快五年了，从最早的Flask+SQLite小工具，到后来带用户体系的RAG服务，再到如今给客户搭整套SaaS平台，踩过的坑比写过的代码还多。今天聊的这个FastAPI模板第二部分，不是教你怎么“跑通一个demo”，而是解决一个所有真实LLM产品上线前都绕不开的硬骨头：如何让文件处理不卡死你的API响应，又能让向量检索稳如磐石。

关键词里反复出现的“Towards AI”和“Medium”，其实暗示了这个模板的出身——它来自一线团队在真实项目中反复打磨出的工程实践，不是学院派纸上谈兵。你可能已经试过直接在FastAPI路由里调用langchain.document_loaders.PDFLoader，结果用户上传一个50页PDF，前端转圈30秒，Nginx直接返回504 Gateway Timeout。或者更糟，多个用户同时上传，CPU飙到100%，整个服务雪崩。这就是为什么我们今天要拆解Celery和pg-vector的组合拳。

Celery不是“高级功能”，它是生存必需品。它把耗时操作（PDF解析、文本分块、OpenAI嵌入计算）从HTTP请求生命周期里彻底剥离，让FastAPI只干它最擅长的事：快速接收请求、校验权限、返回任务ID。而pg-vector也不是“时髦插件”，它是PostgreSQL原生支持的向量数据库方案，意味着你不用额外维护一个独立的向量库服务（比如Qdrant或Weaviate），所有数据、权限、备份、监控都还在你熟悉的PostgreSQL生态里。Supabase作为托管PostgreSQL服务，天然集成了pg-vector扩展，省去了自己编译安装的麻烦。这两者结合，构成了一个零外部依赖、易运维、可水平扩展的LLM数据处理底座。

这个模板特别适合三类人：一是正在从原型转向产品的创业团队，需要快速验证MVP又不想被基础设施拖垮；二是技术负责人，需要给团队提供一套清晰、可审计、符合生产规范的参考架构；三是资深开发者，想搞懂“为什么是Celery而不是asyncio？为什么选pg-vector而不是FAISS？”背后的工程权衡。接下来，我会像带新人一样，把每个配置项、每行关键代码、每次调试失败的原因，掰开揉碎讲清楚。这不是文档翻译，这是我在凌晨三点排查完Celery任务丢失后，写给自己的备忘录。

2. 核心设计与思路拆解：为什么选这套组合，而不是其他方案？

2.1 Celery vs. 纯AsyncIO：一个关于“阻塞”的残酷真相

很多新手看到“异步”就本能地想用async/await。我试过，而且栽得很惨。在file_handler函数里，UnstructuredPDFLoader.load()看着是同步调用，但它底层会启动一个subprocess去调用unstructured命令行工具，这个过程是完全阻塞事件循环的。你用asyncio.to_thread()包装它，效果也极其有限——因为PDF解析本身是CPU密集型任务，Python的GIL会让asyncio的协程调度器彻底失效。结果就是，你开了10个async worker，实际还是串行跑，吞吐量没提升，反而增加了线程切换开销。

Celery的prefork工作模式（默认）完美规避了这个问题。每个worker进程是独立的Python解释器实例，拥有自己的GIL，能真正并行处理CPU密集型任务。更重要的是，Celery的broker（Redis）和result backend（Redis）提供了可靠的任务队列和状态追踪。当用户上传文件后，FastAPI只需发一条消息到Redis，立刻返回{"task_id": "xxx"}，整个过程毫秒级完成。后续无论PDF解析花了2秒还是2分钟，都不影响API的可用性。这背后是成熟的消息中间件保障的至少一次（at-least-once）投递语义，比自己手写一个内存队列或数据库轮询靠谱太多。

提示：有人会问“那用RabbitMQ不行吗？”当然可以，但Redis对这个场景是降维打击。它既是broker又是result backend，部署简单（一个Docker容器搞定），内存操作快，且Supabase项目通常已用Redis做缓存，复用率高。除非你有超大规模、跨机房、强事务一致性要求，否则Redis是更务实的选择。

2.2 pg-vector vs. 独立向量数据库：为什么“少一个服务”就是核心竞争力

选择pg-vector，本质是选择了运维极简主义。想象一下，如果你用Qdrant：

• 你需要单独部署、配置、监控Qdrant服务；
• 你需要管理Qdrant的索引策略、分片、副本；
• 你需要在Supabase的PostgreSQL里存元数据（用户ID、文件名），在Qdrant里存向量，两者ID必须严格一致，一旦不一致就是数据黑洞；
• 你需要为Qdrant单独申请域名、配置HTTPS、设置防火墙规则。

而pg-vector，它就是一个PostgreSQL的扩展。CREATE EXTENSION vector;一行命令，你的vectors表就能存VECTOR(1536)类型字段。所有ACID事务、所有SQL查询能力（JOIN、WHERE、ORDER BY）、所有备份恢复机制（pg_dump）、所有连接池（pgbouncer）全部原样继承。user_vectors表通过外键关联auth.users和vectors，权限控制直接走PostgreSQL的GRANT/REVOKE，和你的用户认证体系无缝咬合。这省下的不是几行代码，而是未来半年的运维时间。

注意：pg-vector的性能瓶颈在于PostgreSQL本身。对于千万级向量，你需要合理设计索引（HNSW或IVFFlat）、调整maintenance_work_mem、优化查询语句。但它的学习曲线，远低于同时掌握PostgreSQL和另一个向量数据库的双倍复杂度。

2.3 Supabase Storage + Celery Worker：文件流转的“无感”设计

整个文件处理链路的设计，核心目标是让FastAPI永远不知道文件内容。用户上传文件，FastAPI只负责把它扔进Supabase Storage Bucket（一个S3兼容的对象存储），拿到一个URL，然后立刻触发Celery任务。Celery Worker在另一台机器（或容器）上，用Supabase Service Key去Storage Bucket下载这个文件。这个设计有三大好处：

1. 解耦：FastAPI服务和文件处理服务物理隔离，一个挂了不影响另一个；
2. 安全：FastAPI不接触原始文件二进制流，避免了恶意PDF触发的任意代码执行风险（虽然概率低，但生产环境必须考虑）；
3. 弹性：你可以根据负载动态扩缩Celery Worker数量，而FastAPI API服务器保持稳定。

这个设计的代价是增加了一次网络IO（Worker下载文件），但相比阻塞式处理带来的服务不可用，这点延迟完全可以接受。而且，Supabase Storage的CDN加速，让这个下载速度通常很快。

3. 核心细节解析与实操要点：从环境搭建到代码落地

3.1 Redis环境：不止是docker run那么简单

原文的Docker命令docker run --name redis -d -p 6379:6379 redis:latest，在开发环境够用，但在准生产环境，必须补上关键参数：

docker run --name redis \ -d \ -p 6379:6379 \ --restart=unless-stopped \ --memory=512m \ --memory-swap=1g \ -e REDIS_PASSWORD=your_strong_password \ -v /path/to/redis/data:/data \ redis:7-alpine \ redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}

• --restart=unless-stopped：确保Docker守护进程重启后，Redis自动拉起；
• --memory和--memory-swap：防止Redis吃光宿主机内存导致OOM Killer干掉它；
• -e REDIS_PASSWORD和--requirepass：绝对禁止裸奔Redis！哪怕本地开发，也要设密码，否则任何能连上6379端口的人都能FLUSHALL清空你的任务队列；
• -v /path/to/redis/data:/data：将RDB/AOF文件持久化到宿主机，避免容器删除后数据全丢；
• redis-server --appendonly yes：开启AOF持久化，比RDB更安全，能保证最多丢失1秒数据。

对应的环境变量，不能写死localhost，要改成redis://:your_strong_password@redis:6379/0（Docker Compose内网通信）或redis://:your_strong_password@127.0.0.1:6379/0（本地开发）。CELERY_BROKER_URL和CELERY_RESULT_BACKEND必须指向同一个Redis DB（这里是/0），否则任务发出去，Worker收不到。

实操心得：我第一次部署时，忘了加--requirepass，结果Celery Worker日志里全是ConnectionError: Error 111 connecting to localhost:6379. Connection refused.。查了两小时网络配置，最后发现是Redis根本没启动密码认证，Celery客户端却按密码模式连接。教训是：所有环境变量，必须和Redis服务端的配置严格匹配，宁可多配，不可少配。

3.2 Celery Worker配置：并发数、日志、Windows陷阱

原文的启动命令celery --app=main.celery worker --concurrency=1 --loglevel=DEBUG，只是入门。生产级配置如下：

# Linux/Mac (推荐) celery -A celery_worker.celery \ worker \ --loglevel=INFO \ --logfile=/var/log/celery/worker.log \ --pidfile=/var/run/celery/worker.pid \ --concurrency=4 \ --max-tasks-per-child=1000 \ --time-limit=300 \ --soft-time-limit=240 # Windows (开发用) celery -A celery_worker.celery \ worker \ --loglevel=INFO \ --logfile=celery_worker.log \ --concurrency=1 \ --pool=solo \ --time-limit=300 \ --soft-time-limit=240

• --concurrency=4：表示启动4个子进程。这个值不是越大越好，要根据你的CPU核心数和任务类型调整。CPU密集型（如PDF解析）建议设为CPU核心数-1；IO密集型（如大量HTTP请求）可以设更高。--max-tasks-per-child=1000强制Worker处理1000个任务后自杀重启，防止内存泄漏累积。
• --time-limit=300和--soft-time-limit=240：硬性超时5分钟，软性超时4分钟。超过软超时，Celery会抛出SoftTimeLimitExceeded异常，让你有机会优雅清理资源；超过硬超时，进程被SIGKILL强制杀死。这对防止一个坏任务拖垮整个Worker至关重要。
• --pool=solo：这是Windows的救命稻草。Windows没有fork()系统调用，无法创建真正的子进程，solo池使用主线程模拟，虽慢但能跑。切记：这只是开发调试用，生产环境必须用Linux容器。

注意：-A celery_worker.celery中的celery_worker.celery是模块路径，不是文件名。celery_worker.py必须在Python路径下可导入。如果报ModuleNotFoundError，检查你的PYTHONPATH或项目根目录是否在sys.path里。

3.3 Supabase Schema：不只是建表，更是数据契约

原文的SQL脚本，缺了最关键的约束和索引。一个健壮的vectors表应该这样：

-- 创建向量扩展（必须在所有表之前） CREATE EXTENSION IF NOT EXISTS vector; -- 创建vectors表，增加NOT NULL和索引 CREATE TABLE IF NOT EXISTS vectors ( id UUID DEFAULT gen_random_uuid() PRIMARY KEY, content TEXT NOT NULL, -- 不允许空内容 metadata JSONB NOT NULL DEFAULT '{}'::jsonb, -- 默认空JSON，避免NULL embedding VECTOR(1536) NOT NULL, -- 向量不能为空 created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW() ); -- 为向量字段创建HNSW索引（pg-vector 0.5+） CREATE INDEX IF NOT EXISTS idx_vectors_embedding_hnsw ON vectors USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64); -- 为metadata字段创建GIN索引，支持JSONB查询 CREATE INDEX IF NOT EXISTS idx_vectors_metadata_gin ON vectors USING GIN (metadata); -- 创建user_vectors表，增加唯一约束 CREATE TABLE IF NOT EXISTS user_vectors ( user_id UUID NOT NULL, vector_id UUID NOT NULL, PRIMARY KEY (user_id, vector_id), FOREIGN KEY (vector_id) REFERENCES vectors (id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES auth.users (id) ON DELETE CASCADE ); -- 为user_id字段创建索引，加速按用户查询 CREATE INDEX IF NOT EXISTS idx_user_vectors_user_id ON user_vectors (user_id);

• ON DELETE CASCADE：当vectors表某条记录被删，user_vectors里关联的记录自动删除，避免孤儿数据；
• idx_vectors_embedding_hnsw：HNSW（Hierarchical Navigable Small World）索引是pg-vector的高性能近似最近邻搜索算法，比旧版的IVFFlat更快更准；
• idx_vectors_metadata_gin：让你能高效执行WHERE metadata->>'file_name' = 'report.pdf'这样的查询；
• created_at/updated_at：所有生产表必备的时间戳，用于审计和数据生命周期管理。

实操心得：我曾因忘记建HNSW索引，在10万向量上做相似搜索，响应时间从200ms飙升到8秒。pg-vector的向量搜索性能，90%取决于索引质量。建好索引后，务必用EXPLAIN ANALYZE验证查询计划是否真的用了Index Scan using idx_vectors_embedding_hnsw。

4. 实操过程与核心环节实现：从代码到可运行服务

4.1 文件上传路由：权限、存储、任务触发的完整闭环

upload_routes.py里的/upload端点，是整个流程的入口。它的精妙之处在于把三个看似无关的操作，用原子性方式串在一起：

1. 权限校验：Depends(AuthBearer())确保只有登录用户才能上传；
2. 安全存储：upload_file_storage()将文件存入Supabase Storage，并生成一个带用户ID前缀的唯一路径（{user_id}/filename.pdf），既隔离了用户数据，又避免了文件名冲突；
3. 异步解耦：process_file.delay(...)将任务推入Redis队列，立即返回，绝不等待。

下面是经过加固的upload_file函数，补充了错误处理和日志：

@upload_router.post("/upload", dependencies=[Depends(AuthBearer())], tags=["Upload"]) async def upload_file( request: Request, uploadFile: UploadFile, current_user: UserIdentity = Depends(get_current_user), ): # 1. 基础校验 if not uploadFile.filename.lower().endswith(('.pdf', '.txt', '.md')): raise HTTPException( status_code=400, detail="Only PDF, TXT, and Markdown files are supported." ) # 2. 构造安全文件路径 filename_with_user_id = f"{current_user.id}/{uploadFile.filename}" logger.info(f"User {current_user.id} uploading file: {filename_with_user_id}") # 3. 读取文件内容（注意：大文件需流式处理，此处简化） try: file_content = await uploadFile.read() if len(file_content) == 0: raise HTTPException(status_code=400, detail="Empty file uploaded.") if len(file_content) > 50 * 1024 * 1024: # 50MB限制 raise HTTPException(status_code=413, detail="File size exceeds 50MB limit.") except Exception as e: logger.error(f"Error reading file {uploadFile.filename}: {e}") raise HTTPException(status_code=500, detail="Failed to read file content.") # 4. 存储到Supabase try: fileInStorage = upload_file_storage(file_content, filename_with_user_id) logger.info(f"File {fileInStorage} uploaded successfully to Supabase Storage") except Exception as e: logger.error(f"Failed to upload {uploadFile.filename} to storage: {e}") if "The resource already exists" in str(e): raise HTTPException( status_code=409, detail=f"File {uploadFile.filename} already exists in your storage." ) else: raise HTTPException( status_code=500, detail="Failed to upload file to storage. Please try again." ) # 5. 触发后台任务 try: task = process_file.delay( file_name=filename_with_user_id, file_original_name=uploadFile.filename, user_id=str(current_user.id), ) logger.info(f"Task {task.id} created for file {filename_with_user_id}") return JSONResponse({ "task_id": task.id, "message": "File upload accepted. Processing started.", "file_name": uploadFile.filename }) except Exception as e: logger.error(f"Failed to create Celery task for {filename_with_user_id}: {e}") # 清理已上传的文件，避免垃圾数据 try: supabase_client = get_supabase_client() supabase_client.storage.from_("quivr").remove([filename_with_user_id]) except: pass raise HTTPException( status_code=500, detail="Failed to initiate background processing. File has been cleaned up." )

关键细节：try/except块不仅捕获异常，还做了补偿操作（compensating action）。如果任务触发失败，代码会尝试删除刚刚上传的文件，保证数据一致性。这是分布式系统设计的核心思想：没有完美的成功，只有可控的失败。

4.2 Celery Worker任务：下载、处理、清理的黄金三角

celery_worker.py里的process_file函数，是整个数据管道的引擎。原文代码存在一个严重隐患：asyncio.new_event_loop()在Celery Worker的多进程环境下，可能引发事件循环冲突。正确的做法是，让所有异步操作都在一个统一的、由Celery管理的事件循环中进行。以下是重构后的版本：

# celery_worker.py import os import asyncio from celery import Celery from utils.process_file import get_supabase_client, file_handler # 配置Celery celery = Celery(__name__) celery.conf.broker_url = os.getenv("CELERY_BROKER_URL", "redis://:your_password@redis:6379/0") celery.conf.result_backend = os.getenv("CELERY_RESULT_BACKEND", "redis://:your_password@redis:6379/0") celery.conf.task_serializer = 'json' celery.conf.result_serializer = 'json' celery.conf.accept_content = ['json'] celery.conf.timezone = 'UTC' @celery.task(name="process_file", bind=True, max_retries=3, default_retry_delay=60) def process_file(self, file_name: str, file_original_name: str, user_id: str): """ Celery任务：处理上传的文件，生成向量并存入pg-vector :param self: Celery任务实例，用于重试 :param file_name: Supabase Storage中的完整路径 :param file_original_name: 用户原始文件名 :param user_id: 用户UUID """ tmp_file_path = None try: # 1. 下载文件到临时路径 supabase_client = get_supabase_client() tmp_file_path = f"/tmp/{file_name.replace('/', '_')}" # 更安全的文件名 logger.info(f"[Task {self.request.id}] Downloading {file_name} to {tmp_file_path}") res = supabase_client.storage.from_("quivr").download(file_name) with open(tmp_file_path, "wb+") as f: f.write(res) # 2. 调用异步处理器 logger.info(f"[Task {self.request.id}] Starting async file handler for {file_original_name}") # 使用Celery内置的asyncio支持 result = asyncio.run( file_handler( file=tmp_file_path, file_original_name=file_original_name, user_id=user_id ) ) logger.info(f"[Task {self.request.id}] File processing completed: {result}") return {"status": "success", "file": file_original_name, "task_id": self.request.id} except Exception as exc: # 3. 错误处理与重试 logger.error(f"[Task {self.request.id}] Failed to process {file_name}: {exc}", exc_info=True) # 如果是网络错误或临时故障，重试 if "ConnectionError" in str(exc) or "timeout" in str(exc).lower(): raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60) # 其他错误，记录后失败 raise finally: # 4. 清理临时文件（无论如何都要执行） if tmp_file_path and os.path.exists(tmp_file_path): try: os.remove(tmp_file_path) logger.info(f"[Task {self.request.id}] Temporary file {tmp_file_path} removed") except OSError as e: logger.warning(f"[Task {self.request.id}] Failed to remove {tmp_file_path}: {e}")

• bind=True：让任务函数能访问self.request.id，方便日志追踪；
• max_retries=3, default_retry_delay=60：自动重试3次，间隔1/2/4分钟，应对网络抖动；
• asyncio.run()：在当前进程的主线程中启动一个新的事件循环，安全可靠；
• finally块：确保临时文件100%被清理，即使任务崩溃也不会留下磁盘垃圾。

4.3 文件处理器：LangChain的深度定制与避坑指南

utils/process_file.py是真正的“大脑”。原文代码直接用了UnstructuredPDFLoader，但生产环境必须面对现实：PDF格式千奇百怪。有的加密，有的扫描成图，有的表格错乱。以下是经过实战检验的增强版file_handler：

# utils/process_file.py import os import time import logging from pathlib import Path from typing import List, Dict, Any from langchain.document_loaders import ( UnstructuredPDFLoader, PyPDFLoader, PDFMinerLoader, TextLoader, UnstructuredMarkdownLoader, ) from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import SupabaseVectorStore from langchain.schema import Document from supabase.client import Client from dotenv import load_dotenv from logger import get_logger load_dotenv() logger = get_logger(__name__) # 定义文件加载器映射 LOADER_MAP = { ".pdf": lambda file_path: _get_pdf_loader(file_path), ".txt": TextLoader, ".md": UnstructuredMarkdownLoader, } def _get_pdf_loader(file_path: str): """智能选择PDF加载器""" try: # 尝试PyPDFLoader（纯文本提取，快但可能失败） loader = PyPDFLoader(file_path) docs = loader.load() if len(docs) > 0 and len(docs[0].page_content.strip()) > 100: logger.debug(f"PyPDFLoader succeeded for {file_path}") return loader except Exception as e: logger.debug(f"PyPDFLoader failed for {file_path}: {e}") try: # 尝试PDFMinerLoader（更鲁棒，但慢） loader = PDFMinerLoader(file_path) docs = loader.load() if len(docs) > 0 and len(docs[0].page_content.strip()) > 100: logger.debug(f"PDFMinerLoader succeeded for {file_path}") return loader except Exception as e: logger.debug(f"PDFMinerLoader failed for {file_path}: {e}") # 最后 resort to UnstructuredPDFLoader（需要unstructured服务） logger.warning(f"All PDF loaders failed for {file_path}, falling back to UnstructuredPDFLoader") return UnstructuredPDFLoader async def file_handler(file: str, file_original_name: str, user_id: str) -> str: """主文件处理器：加载、分割、嵌入、存储""" start_time = time.time() file_ext = Path(file).suffix.lower() if file_ext not in LOADER_MAP: raise ValueError(f"Unsupported file type: {file_ext}") try: # 1. 加载文档 loader_class = LOADER_MAP[file_ext] if file_ext == ".pdf": loader = loader_class(file) else: loader = loader_class(file) logger.info(f"Loading {file_original_name} with {loader.__class__.__name__}") documents = loader.load() logger.info(f"Loaded {len(documents)} pages from {file_original_name}") if not documents: raise ValueError(f"No content extracted from {file_original_name}") # 2. 文本分割（关键：chunk_size和overlap要根据模型调整） text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # OpenAI text-embedding-ada-002的推荐值 chunk_overlap=50, # 重叠确保语义连贯 length_function=len, separators=["\n\n", "\n", " ", ""], ) split_docs = text_splitter.split_documents(documents) logger.info(f"Split into {len(split_docs)} chunks") # 3. 添加元数据 date_short = time.strftime("%Y%m%d") for doc in split_docs: doc.metadata.update({ "file_name": file_original_name, "date": date_short, "user_id": user_id, "source": file, }) # 4. 嵌入并存储（使用SupabaseVectorStore） embeddings = OpenAIEmbeddings( openai_api_key=os.getenv("OPENAI_API_KEY"), model="text-embedding-ada-002", ) supabase_client = get_supabase_client() vector_store = SupabaseVectorStore( client=supabase_client, embedding=embeddings, table_name="vectors" ) logger.info(f"Starting embedding and storage for {len(split_docs)} chunks") # 批量添加，提高效率 ids = vector_store.add_documents(split_docs) logger.info(f"Successfully stored {len(ids)} vectors") # 5. 关联用户（批量插入user_vectors） if ids: database = get_supabase_db() user_vector_data = [ {"user_id": user_id, "vector_id": str(id_)} for id_ in ids ] response = database.db.table("user_vectors").insert(user_vector_data).execute() logger.info(f"Linked {len(ids)} vectors to user {user_id}") elapsed = time.time() - start_time logger.info(f"File processing completed in {elapsed:.2f}s") return f"Processed {len(split_docs)} chunks from {file_original_name}" except Exception as e: logger.error(f"Error processing {file_original_name}: {e}", exc_info=True) raise finally: # 清理临时文件（如果存在） if os.path.exists(file): try: os.remove(file) logger.debug(f"Temporary file {file} cleaned up") except OSError: pass

关键避坑点：

• PDF加载器选择：PyPDFLoader最快，但对扫描PDF或加密PDF无效；PDFMinerLoader更鲁棒；UnstructuredPDFLoader功能最强但需额外服务。代码实现了自动fallback，确保不因单个PDF失败而中断整个任务。
• Chunk Size：500字符是text-embedding-ada-002的黄金值。太大，语义模糊；太小，信息碎片化。chunk_overlap=50是经验值，确保相邻块有上下文。
• 批量操作：vector_store.add_documents()比循环调用add_document()快10倍以上；user_vectors的批量插入同理。

5. 常见问题与排查技巧实录：那些让我熬夜的Bug和解决方案

5.1 Celery任务“消失不见”：Broker连接与序列化陷阱

现象：FastAPI返回了{"task_id": "xxx"}，但Celery Worker日志里完全没有收到这个任务，celery -A celery_worker.celery inspect active也显示空。

排查步骤：

1. 检查Broker URL：在Celery Worker启动后，立刻执行celery -A celery_worker.celery inspect ping。如果返回Error: No nodes replied within time constraint.，说明Worker根本没连上Redis。检查CELERY_BROKER_URL是否拼写错误，密码是否正确，端口是否被防火墙拦截。
2. 检查序列化：如果ping成功，但任务仍不出现，很可能是序列化问题。Celery默认用pickle，但pickle在不同Python版本间不兼容，且有安全风险。在celery_worker.py中强制指定：

celery.conf.task_serializer = 'json' celery.conf.result_serializer = 'json' celery.conf.accept_content = ['json']

并确保所有传递给delay()的参数都是JSON可序列化的（字符串、数字、字典、列表），绝不能传入open()返回的文件对象或数据库连接。 3.检查任务名：@celery.task(name="process_file")里的名字，必须和process_file.delay()调用时的模块路径完全一致。如果celery_worker.py不在Python路径里，delay()会找不到任务。

终极解决方案：在FastAPI中，添加一个调试端点，手动触发任务并打印详细日志：

@upload_router.get("/debug/task") def debug_task(): from celery_worker import process_file task = process_file.delay(file_name="test.txt", file_original_name="test.txt", user_id="test-id") return {"task_id": task.id, "state": task.state, "result": task.result}

5.2 pg-vector搜索“慢如蜗牛”：索引失效与查询优化

现象：SELECT * FROM vectors ORDER BY embedding <=> '[0.1,0.2,...]' LIMIT 5查询耗时超过1秒。

诊断方法：

1. 看执行计划：在psql中运行EXPLAIN (ANALYZE, BUFFERS) SELECT ...。如果输出里有Seq Scan on vectors，说明HNSW索引根本没用上。
2. 检查索引状态：SELECT * FROM pg_indexes WHERE tablename = 'vectors';确认idx_vectors_embedding_hnsw存在且状态正常。
3. 检查向量维度：SELECT pg_typeof(embedding) FROM vectors LIMIT 1;确保返回的是vector(1536)，而不是vector(768)或其他。维度不匹配，索引失效。

解决方案：

• 重建索引：DROP INDEX idx_vectors_embedding_hnsw; CREATE INDEX ...。有时索引损坏，重建即可。
• 调整HNSW参数：ef_construction越大，索引越准但构建越慢；m越大，查询越快但内存占用越高。对于10万向量，m=16, ef_construction=64是平衡点。
• 强制使用索引：在查询中加/*+ IndexScan(vectors idx_vectors_embedding_hnsw) */（PostgreSQL 14+支持），排除优化器误判。

5.3 文件处理“卡在中间”：内存溢出与大文件处理

现象：上传一个200页PDF，Celery Worker进程内存飙升到2GB，然后被系统OOM Killer杀死。

根本原因：UnstructuredPDFLoader或PyPDFLoader在加载大PDF时，会将整个文档的文本和元数据加载到内存，没有流式处理。

解决方案：

1. 预检查文件大小：在upload_file路由里，await uploadFile.read()前，先检查uploadFile.size，对超过50MB的文件直接拒绝。
2. 使用流式加载器：改用PyMuPDFLoader（fitz库），它支持逐页加载：

from pypdfium2 import PdfDocument def load_pdf_stream(file_path: str) -> List[Document]: docs = [] pdf = PdfDocument(file_path) for page_num in range(len(pdf)): page = pdf.get_page(page_num) text = page.get_textpage().get_text_range() doc = Document(page_content=text, metadata={"page": page_num}) docs.append(doc) return docs

3. 增加Worker内存限制：在Docker Compose中为Celery Worker服务添加mem_limit: 2g，并设置--max-memory-per-child=1g，强制内存超限时重启Worker。

5.4 Supabase权限“神隐”：Storage Bucket和Database的双重门禁

现象：supabase_client.storage.from_("quivr").download(file_name)返回403 Forbidden，但supabase_client.auth.sign_in_with_password()却成功。

原因：Supabase的Storage Bucket和Database权限是分开管理的。你给了API Key访问Database的权限，但没给Storage Bucket的读取权限。

解决步骤：

1. 进入Supabase控制台 → Storage → Buckets →quivr→ Policies。
2. 创建一个新Policy，名称如allow_authenticated_read，SQL如下：

(bucket_id = 'quivr') AND (auth.role() = 'authenticated')

3. 在Database的storage.objects表上，同样创建Policy，确保SELECT权限对authenticated角色开放。

实操心得：Supabase的权限模型是“最小权限原则”的典范。我曾因在Storage Policy里写了`auth.uid() = 'some-user-id'