第十五阶段：Celery 异步任务队列（文档批量入库）

问答接口已经能用了，还剩最后一个问题：文档入库。13_build_qdrant.py 跑一次要读 PDF、调本地 embedding、写 Qdrant，几十页文档轻松跑好几分钟。如果做成 HTTP 接口让用户上传文档后干等，连接早就超时了。

解法：异步任务队列。接口只负责"接单"，把耗时的工作丢给后台慢慢跑，用户拿着任务 id 轮询进度就行。这就是本阶段的 Celery + Redis。

一、Celery 的三个角色

角色	是什么	本阶段用谁
broker	任务队列，接口把任务丢进来	Redis
worker	干活的进程，从队列里取任务执行	`celery -A tasks worker`
backend	存任务状态和结果，轮询时查它	Redis

broker 和 backend 都用 Redis，但用 db1，和问答缓存的 db0 错开，互不干扰。

流程：

text

用户          api                  Redis(db1)              worker
 │ POST /ingest │                        │                      │
 │ ───────────▶ │  delay() 丢进队列       │                      │
 │              │ ──────────────────────▶ │                      │
 │ ◀── task_id ─ │（立刻返回，不等做完）    │                      │
 │              │                        │ ◀── 取任务 ──         │
 │ GET /tasks/… │                        │        执行中…         │
 │ ───────────▶ │ ── 查 backend ────────▶ │                      │
 │ ◀─ PROGRESS ─ │                        │                      │
 │   …轮询…     │                        │                      │
 │ ◀─ SUCCESS ─ │                        │                      │

二、`celery_app.py`：5 行配置

python

import os
from celery import Celery

# broker（任务队列）和 backend（存任务结果）都用 Redis，
# 用 db1，与问答缓存的 db0 错开，互不干扰
celery_app = Celery(
    "agent_tasks",
    broker=os.getenv("CELERY_REDIS_URL", "redis://localhost:6379/1"),
    backend=os.getenv("CELERY_REDIS_URL", "redis://localhost:6379/1"),
)

celery_app.conf.update(
    task_track_started=True,  # 任务一开工就更新状态，轮询时能看到 STARTED
)

三、`tasks.py`：把 `13_build_qdrant.py` 搬进任务

原来脚本的逻辑原样平移，只加两处：update_state 上报进度，跑完后清掉 ask:* 问答缓存——不然新文档入库了，问出来还是缓存里的老答案。

python

import os
import redis
import qdrant_client
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.embeddings.openai_like import OpenAILikeEmbedding

from celery_app import celery_app

# worker 里也要能调本地 embedding，配置与 13_build_qdrant.py 一致
Settings.embed_model = OpenAILikeEmbedding(
    model_name="这里填你的 Embedding 模型名",
    api_base=os.getenv("EMBED_URL", "http://localhost:1234/v1"), api_key="lm-studio")


@celery_app.task(bind=True)
def ingest_documents(self, data_dir="data"):
    """批量文档入库：读 PDF → 算向量 → 写 Qdrant，耗时操作放后台慢慢跑"""
    # bind=True 让任务拿到 self，可以用 update_state 上报进度
    self.update_state(state="PROGRESS", meta={"step": "读取 PDF 文档"})
    docs = SimpleDirectoryReader(data_dir, required_exts=[".pdf"]).load_data()

    self.update_state(state="PROGRESS", meta={"step": f"计算向量并写入 Qdrant（{len(docs)} 页）"})
    client = qdrant_client.QdrantClient(url=os.getenv("QDRANT_URL", "http://localhost:6333"))
    vector_store = QdrantVectorStore(client=client, collection_name="kb")
    VectorStoreIndex.from_documents(
        docs,
        storage_context=StorageContext.from_defaults(vector_store=vector_store),
        transformations=[SentenceSplitter(chunk_size=512, chunk_overlap=50)],
        show_progress=False)

    # 入库后把旧的问答缓存清掉，否则新文档入库了，问出来还是缓存里的老答案
    cache = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"),
                                 decode_responses=True)
    cleared = 0
    for key in cache.scan_iter("ask:*"):
        cache.delete(key)
        cleared += 1

    return {"pages": len(docs), "vectors": client.count("kb").count, "cache_cleared": cleared}

四、`api.py`：接单接口 + 进度查询接口

python

from celery.result import AsyncResult
from celery_app import celery_app
from tasks import ingest_documents

class IngestRequest(BaseModel):
    data_dir: str = "data"  # 容器里挂载的文档目录

@app.post("/ingest")
def ingest(req: IngestRequest):
    # delay() 把任务丢进队列就立刻返回，不等它做完
    task = ingest_documents.delay(req.data_dir)
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
def task_status(task_id: str):
    # 轮询这个接口看任务进度：PENDING / PROGRESS / SUCCESS / FAILURE
    r = AsyncResult(task_id, app=celery_app)
    body = {"task_id": task_id, "status": r.state}
    if r.state == "PROGRESS":
        body["step"] = (r.info or {}).get("step")
    elif r.state == "SUCCESS":
        body["result"] = r.result
    elif r.state == "FAILURE":
        body["error"] = str(r.result)
    return body

五、Docker：加一个 worker 服务

worker 和 api 用同一个镜像，只是启动命令不同。docker-compose.yml 新增：

yaml

  worker:
    build: .                # 和 api 同一个镜像，启动命令不同
    env_file: .env
    environment:
      QDRANT_URL: http://qdrant:6333
      EMBED_URL: http://host.docker.internal:1234/v1   # 容器里访问本机 LM Studio
      REDIS_URL: redis://redis:6379/0
      CELERY_REDIS_URL: redis://redis:6379/1
    command: celery -A tasks worker --loglevel=info --concurrency=1
    volumes: ["./data:/app/data"]   # worker 要读到 data/ 下的 PDF
    depends_on: [qdrant, redis]

api 服务也要加 CELERY_REDIS_URL: redis://redis:6379/1——查任务状态要连 backend。

Dockerfile 多 COPY 两个文件，pip 安装加上 cache mount（requirements.txt 再变也只下载新增的包，不用全部重下）：

dockerfile

COPY api.py celery_app.py tasks.py .

dockerfile

# 完整 Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN --mount=type=cache,target=/root/.cache/pip \
    pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY api.py celery_app.py tasks.py .
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt 加两行：

text

celery[redis]
llama-index-readers-file

第二行是 PDF 解析插件，必须有：缺了它 SimpleDirectoryReader 不认识 PDF，会按纯文本读出一堆乱码——这个坑在第五阶段踩过（见踩坑记录），Docker 镜像里别再漏了。

六、启动与验证

bash

docker compose up -d --build

发起入库（把 PDF 先放进 data/）：

bash

curl -X POST http://localhost:8000/ingest -H "Content-Type: application/json" -d "{}"
# {"task_id": "a1b2c3...", "status": "queued"}

轮询进度：

bash

curl http://localhost:8000/tasks/a1b2c3...
# 进行中：{"task_id": "...", "status": "PROGRESS", "step": "计算向量并写入 Qdrant（42 页）"}
# 完成：  {"task_id": "...", "status": "SUCCESS",
#          "result": {"pages": 42, "vectors": 128, "cache_cleared": 5}}

入库完成后，再去问之前问过的问题：cached_hit 会是 false——缓存已经被清掉，问到的是新文档的内容。

七、三个坑

worker 连不上 LM Studio：容器里 localhost 是容器自己，不是你的电脑，所以 EMBED_URL 必须写 http://host.docker.internal:1234/v1。另外 LM Studio 里要打开局域网访问（Settings → Local Server → Serve on local network），否则容器连不进来。
Windows 本机跑 worker 卡住：Windows 上 Celery 默认的 prefork 进程池有兼容问题，启动命令加 --pool=solo：celery -A tasks worker --pool=solo --loglevel=info。
db 别混用：问答缓存 db0、Celery db1。如果 Celery 也用 db0，cache.scan_iter("ask:*") 清缓存时不会误删任务数据（反过来任务结果也不会污染问答缓存），分开最省心。

八、提交代码

bash

git add celery_app.py tasks.py api.py docker-compose.yml Dockerfile requirements.txt
git commit -m "15 Celery 异步入库"
git push

项目 4 小结

第十二阶段：FastAPI 把问答做成 HTTP 接口；
第十三阶段：Qdrant 替换 Chroma，向量存进专业数据库；
第十四阶段：Dockerfile + Compose，一键拉起 redis / qdrant / api；
第十五阶段：Redis 缓存热点问题 + Celery 异步批量入库，完整服务闭环。

四个项目走完：从调第一个 LLM 接口，到 RAG、Agent、多 Agent 协作，再到分布式部署——一个能真正上线的知识库问答服务已经在你电脑上跑起来了。

👉 附录：踩坑记录

第十五阶段：Celery 异步任务队列（文档批量入库） ​

一、Celery 的三个角色 ​

二、celery_app.py：5 行配置 ​

三、tasks.py：把 13_build_qdrant.py 搬进任务 ​

四、api.py：接单接口 + 进度查询接口 ​

五、Docker：加一个 worker 服务 ​

六、启动与验证 ​

七、三个坑 ​

八、提交代码 ​

项目 4 小结 ​

第十五阶段：Celery 异步任务队列（文档批量入库）

一、Celery 的三个角色

二、`celery_app.py`：5 行配置

三、`tasks.py`：把 `13_build_qdrant.py` 搬进任务

四、`api.py`：接单接口 + 进度查询接口

五、Docker：加一个 worker 服务

六、启动与验证

七、三个坑

八、提交代码

项目 4 小结