LLM Agent 与 RAG 实战教程

Appearance

第十五阶段：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 两个文件：

dockerfile
COPY api.py celery_app.py tasks.py .

requirements.txt 加一行：

text
celery[redis]

六、启动与验证 ​

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——缓存已经被清掉，问到的是新文档的内容。

七、三个坑 ​

  1. worker 连不上 LM Studio：容器里 localhost 是容器自己，不是你的电脑，所以 EMBED_URL 必须写 http://host.docker.internal:1234/v1。另外 LM Studio 里要打开局域网访问（Settings → Local Server → Serve on local network），否则容器连不进来。
  2. Windows 本机跑 worker 卡住：Windows 上 Celery 默认的 prefork 进程池有兼容问题，启动命令加 --pool=solo：celery -A tasks worker --pool=solo --loglevel=info。
  3. 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 协作，再到分布式部署——一个能真正上线的知识库问答服务已经在你电脑上跑起来了。

👉 附录：踩坑记录