ai_pipeline/Demo1.md

5.6 KiB
Raw Blame History

下面我给你分别写 最小可用、可复制进项目的 Qdrant + Elasticsearch通过 httpx的简洁 async client,专为 RAG 系统 / 多知识库 场景设计。

代码特点:

  • 极简、明确、可运行
  • 全部 async
  • 支持 多知识库(多 collection / 多 index
  • 备注清晰
  • 适合直接丢进你的项目(如 app/vectorstore/

一、Qdrant 最小异步客户端AsyncQdrantClient

文件:app/vectorstore/qdrant_client.py

# app/vectorstore/qdrant_client.py
"""
Qdrant 异步客户端(适用于 RAG 向量检索)
- 支持多知识库(每个知识库一个 collection
- 简单、干净、无多余依赖
"""

from qdrant_client import AsyncQdrantClient
from qdrant_client.models import (
    VectorParams,
    Distance,
    PointStruct,
)
from app.config import settings


# 创建全局异步客户端
# 推荐FastAPI 单实例方式复用连接池
qdrant = AsyncQdrantClient(
    host=settings.QDRANT_HOST,
    port=settings.QDRANT_PORT,
    api_key=settings.QDRANT_API_KEY,
)


async def ensure_collection(name: str, dim: int):
    """
    如果 collection 不存在则创建。
    适用于多知识库:每个知识库一个 collection。
    """
    try:
        await qdrant.get_collection(name)
    except Exception:
        await qdrant.recreate_collection(
            collection_name=name,
            vectors_config=VectorParams(
                size=dim,
                distance=Distance.COSINE
            )
        )


async def upsert_vectors(collection: str, vectors: list, payloads: list):
    """
    写入向量id 自增或自行维护)
    - vectors: [[float...], ...]
    - payloads: [{"text": "...", "kb": "..."}]
    """
    points = [
        PointStruct(
            id=i,
            vector=vectors[i],
            payload=payloads[i]
        )
        for i in range(len(vectors))
    ]

    await qdrant.upsert(
        collection_name=collection,
        points=points
    )


async def vector_search(collection: str, query_vector: list, top_k: int = 5):
    """
    查询向量相似度
    """
    result = await qdrant.search(
        collection_name=collection,
        query_vector=query_vector,
        limit=top_k
    )
    return result

二、Elasticsearch使用 httpx最小 async client

文件:app/vectorstore/es_client.py

官方 elasticsearch-py 是同步阻塞,因此我们手写 httpx 版,更适合 FastAPI/Sanic。

# app/vectorstore/es_client.py
"""
Elasticsearch async client使用 httpx
- 用于 BM25 搜索
- 支持多知识库:每个知识库一个 index
"""

import httpx
from app.config import settings


class AsyncES:
    def __init__(self):
        self.base = f"http://{settings.ES_HOST}:{settings.ES_PORT}"
        self.auth = (settings.ES_USER, settings.ES_PASSWORD)

    async def create_index(self, index: str):
        """
        创建 index知识库
        """
        async with httpx.AsyncClient() as client:
            resp = await client.put(
                f"{self.base}/{index}",
                auth=self.auth,
                json={
                    "settings": {
                        "analysis": {
                            "analyzer": {
                                "default": {"type": "standard"}
                            }
                        }
                    },
                    "mappings": {
                        "properties": {
                            "text": {"type": "text"},
                            "kb": {"type": "keyword"},
                        }
                    }
                }
            )
            return resp.json()

    async def add_doc(self, index: str, text: str, kb: str):
        """
        写入文档(普通文本)
        """
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base}/{index}/_doc",
                auth=self.auth,
                json={"text": text, "kb": kb}
            )
        return resp.json()

    async def bm25_search(self, index: str, query: str, top_k: int = 10):
        """
        BM25 搜索
        """
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base}/{index}/_search",
                auth=self.auth,
                json={
                    "size": top_k,
                    "query": {
                        "match": {
                            "text": query
                        }
                    }
                }
            )
        return resp.json()
        

# 全局实例
es = AsyncES()

📌 三、为什么这两个 client 都适合 RAG / 多知识库?

Qdrant → 负责向量检索

  • 每个 knowledge_base = 一个 Qdrant collection
  • 完全异步
  • 适合你用 vLLM embedding

Elasticsearch → 负责 BM25 文本检索

  • 每个 knowledge_base = 一个 ES index
  • httpx 全异步,不阻塞 FastAPI/Sanic
  • 支持大文本搜索、排序强大

两者结合 = 最常见 RAG Hybrid Retrieval

  • BM25text解决 recall
  • Vectorembedding解决 semantic match
  • 结果融合RRF / weighted sum

你下一步只需要一个混合检索 pipeline我可以给你写。


📌 四、如果你需要,我可以继续生成:

  1. Qdrant + ES 混合检索 pipelineRRF / 乘法融合 / 权重融合)
  2. /hybrid_search FastAPI 完整 API
  3. ingest 文档txt/docx/pdf自动切片 + embedding + 写入 Qdrant & ES
  4. KB 管理(创建/删除/重建/清空)
  5. Docker Compose + 环境变量 + 部署脚本

只需告诉我: 👉 “生成混合检索 pipeline 和 API”