1. 项目概述为什么一个LLM SaaS后端需要Celery和pg-vector我做AI应用开发快五年了从最早的FlaskSQLite小工具到后来带用户体系的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的brokerRedis和result backendRedis提供了可靠的任务队列和状态追踪。当用户上传文件后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下载这个文件。这个设计有三大好处解耦FastAPI服务和文件处理服务物理隔离一个挂了不影响另一个安全FastAPI不接触原始文件二进制流避免了恶意PDF触发的任意代码执行风险虽然概率低但生产环境必须考虑弹性你可以根据负载动态扩缩Celery Worker数量而FastAPI API服务器保持稳定。这个设计的代价是增加了一次网络IOWorker下载文件但相比阻塞式处理带来的服务不可用这点延迟完全可以接受。而且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 \ --restartunless-stopped \ --memory512m \ --memory-swap1g \ -e REDIS_PASSWORDyour_strong_password \ -v /path/to/redis/data:/data \ redis:7-alpine \ redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}--restartunless-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_passwordredis:6379/0Docker Compose内网通信或redis://:your_strong_password127.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 --appmain.celery worker --concurrency1 --loglevelDEBUG只是入门。生产级配置如下# Linux/Mac (推荐) celery -A celery_worker.celery \ worker \ --loglevelINFO \ --logfile/var/log/celery/worker.log \ --pidfile/var/run/celery/worker.pid \ --concurrency4 \ --max-tasks-per-child1000 \ --time-limit300 \ --soft-time-limit240 # Windows (开发用) celery -A celery_worker.celery \ worker \ --loglevelINFO \ --logfilecelery_worker.log \ --concurrency1 \ --poolsolo \ --time-limit300 \ --soft-time-limit240--concurrency4表示启动4个子进程。这个值不是越大越好要根据你的CPU核心数和任务类型调整。CPU密集型如PDF解析建议设为CPU核心数-1IO密集型如大量HTTP请求可以设更高。--max-tasks-per-child1000强制Worker处理1000个任务后自杀重启防止内存泄漏累积。--time-limit300和--soft-time-limit240硬性超时5分钟软性超时4分钟。超过软超时Celery会抛出SoftTimeLimitExceeded异常让你有机会优雅清理资源超过硬超时进程被SIGKILL强制杀死。这对防止一个坏任务拖垮整个Worker至关重要。--poolsolo这是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_hnswHNSWHierarchical 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端点是整个流程的入口。它的精妙之处在于把三个看似无关的操作用原子性方式串在一起权限校验Depends(AuthBearer())确保只有登录用户才能上传安全存储upload_file_storage()将文件存入Supabase Storage并生成一个带用户ID前缀的唯一路径{user_id}/filename.pdf既隔离了用户数据又避免了文件名冲突异步解耦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_code400, detailOnly PDF, TXT, and Markdown files are supported. ) # 2. 构造安全文件路径 filename_with_user_id f{current_user.id}/{uploadFile.filename} logger.info(fUser {current_user.id} uploading file: {filename_with_user_id}) # 3. 读取文件内容注意大文件需流式处理此处简化 try: file_content await uploadFile.read() if len(file_content) 0: raise HTTPException(status_code400, detailEmpty file uploaded.) if len(file_content) 50 * 1024 * 1024: # 50MB限制 raise HTTPException(status_code413, detailFile size exceeds 50MB limit.) except Exception as e: logger.error(fError reading file {uploadFile.filename}: {e}) raise HTTPException(status_code500, detailFailed to read file content.) # 4. 存储到Supabase try: fileInStorage upload_file_storage(file_content, filename_with_user_id) logger.info(fFile {fileInStorage} uploaded successfully to Supabase Storage) except Exception as e: logger.error(fFailed to upload {uploadFile.filename} to storage: {e}) if The resource already exists in str(e): raise HTTPException( status_code409, detailfFile {uploadFile.filename} already exists in your storage. ) else: raise HTTPException( status_code500, detailFailed to upload file to storage. Please try again. ) # 5. 触发后台任务 try: task process_file.delay( file_namefilename_with_user_id, file_original_nameuploadFile.filename, user_idstr(current_user.id), ) logger.info(fTask {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(fFailed 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_code500, detailFailed 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_passwordredis:6379/0) celery.conf.result_backend os.getenv(CELERY_RESULT_BACKEND, redis://:your_passwordredis:6379/0) celery.conf.task_serializer json celery.conf.result_serializer json celery.conf.accept_content [json] celery.conf.timezone UTC celery.task(nameprocess_file, bindTrue, max_retries3, default_retry_delay60) 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( filetmp_file_path, file_original_namefile_original_name, user_iduser_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_infoTrue) # 如果是网络错误或临时故障重试 if ConnectionError in str(exc) or timeout in str(exc).lower(): raise self.retry(excexc, countdown2 ** 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})bindTrue让任务函数能访问self.request.id方便日志追踪max_retries3, default_retry_delay60自动重试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(fPyPDFLoader succeeded for {file_path}) return loader except Exception as e: logger.debug(fPyPDFLoader 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(fPDFMinerLoader succeeded for {file_path}) return loader except Exception as e: logger.debug(fPDFMinerLoader failed for {file_path}: {e}) # 最后 resort to UnstructuredPDFLoader需要unstructured服务 logger.warning(fAll 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(fUnsupported 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(fLoading {file_original_name} with {loader.__class__.__name__}) documents loader.load() logger.info(fLoaded {len(documents)} pages from {file_original_name}) if not documents: raise ValueError(fNo content extracted from {file_original_name}) # 2. 文本分割关键chunk_size和overlap要根据模型调整 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # OpenAI text-embedding-ada-002的推荐值 chunk_overlap50, # 重叠确保语义连贯 length_functionlen, separators[\n\n, \n, , ], ) split_docs text_splitter.split_documents(documents) logger.info(fSplit 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_keyos.getenv(OPENAI_API_KEY), modeltext-embedding-ada-002, ) supabase_client get_supabase_client() vector_store SupabaseVectorStore( clientsupabase_client, embeddingembeddings, table_namevectors ) logger.info(fStarting embedding and storage for {len(split_docs)} chunks) # 批量添加提高效率 ids vector_store.add_documents(split_docs) logger.info(fSuccessfully 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(fLinked {len(ids)} vectors to user {user_id}) elapsed time.time() - start_time logger.info(fFile processing completed in {elapsed:.2f}s) return fProcessed {len(split_docs)} chunks from {file_original_name} except Exception as e: logger.error(fError processing {file_original_name}: {e}, exc_infoTrue) raise finally: # 清理临时文件如果存在 if os.path.exists(file): try: os.remove(file) logger.debug(fTemporary file {file} cleaned up) except OSError: pass关键避坑点PDF加载器选择PyPDFLoader最快但对扫描PDF或加密PDF无效PDFMinerLoader更鲁棒UnstructuredPDFLoader功能最强但需额外服务。代码实现了自动fallback确保不因单个PDF失败而中断整个任务。Chunk Size500字符是text-embedding-ada-002的黄金值。太大语义模糊太小信息碎片化。chunk_overlap50是经验值确保相邻块有上下文。批量操作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也显示空。排查步骤检查Broker URL在Celery Worker启动后立刻执行celery -A celery_worker.celery inspect ping。如果返回Error: No nodes replied within time constraint.说明Worker根本没连上Redis。检查CELERY_BROKER_URL是否拼写错误密码是否正确端口是否被防火墙拦截。检查序列化如果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(nameprocess_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_nametest.txt, file_original_nametest.txt, user_idtest-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秒。诊断方法看执行计划在psql中运行EXPLAIN (ANALYZE, BUFFERS) SELECT ...。如果输出里有Seq Scan on vectors说明HNSW索引根本没用上。检查索引状态SELECT * FROM pg_indexes WHERE tablename vectors;确认idx_vectors_embedding_hnsw存在且状态正常。检查向量维度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万向量m16, ef_construction64是平衡点。强制使用索引在查询中加/* IndexScan(vectors idx_vectors_embedding_hnsw) */PostgreSQL 14支持排除优化器误判。5.3 文件处理“卡在中间”内存溢出与大文件处理现象上传一个200页PDFCelery Worker进程内存飙升到2GB然后被系统OOM Killer杀死。根本原因UnstructuredPDFLoader或PyPDFLoader在加载大PDF时会将整个文档的文本和元数据加载到内存没有流式处理。解决方案预检查文件大小在upload_file路由里await uploadFile.read()前先检查uploadFile.size对超过50MB的文件直接拒绝。使用流式加载器改用PyMuPDFLoaderfitz库它支持逐页加载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_contenttext, metadata{page: page_num}) docs.append(doc) return docs增加Worker内存限制在Docker Compose中为Celery Worker服务添加mem_limit: 2g并设置--max-memory-per-child1g强制内存超限时重启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的读取权限。解决步骤进入Supabase控制台 → Storage → Buckets →quivr→ Policies。创建一个新Policy名称如allow_authenticated_readSQL如下(bucket_id quivr) AND (auth.role() authenticated)在Database的storage.objects表上同样创建Policy确保SELECT权限对authenticated角色开放。实操心得Supabase的权限模型是“最小权限原则”的典范。我曾因在Storage Policy里写了auth.uid() some-user-id
