From 9776f3c7913baaac14e7e51606675f723a346273 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E5=A6=82=E5=A8=81?= Date: Fri, 12 Dec 2025 17:59:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=89=E8=A3=85=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Demo1.md | 215 +++++++++++++++++++++++++++++ docker-compose.yml | 6 +- src/pipeline/config.py | 10 +- src/pipeline/core/qdrant_client.py | 0 4 files changed, 226 insertions(+), 5 deletions(-) create mode 100644 Demo1.md create mode 100644 src/pipeline/core/qdrant_client.py diff --git a/Demo1.md b/Demo1.md new file mode 100644 index 0000000..3ef91c8 --- /dev/null +++ b/Demo1.md @@ -0,0 +1,215 @@ +下面我给你分别写 **最小可用、可复制进项目的 Qdrant + Elasticsearch(通过 httpx)的简洁 async client**,专为 **RAG 系统 / 多知识库** 场景设计。 + +代码特点: + +* **极简、明确、可运行** +* **全部 async** +* 支持 **多知识库(多 collection / 多 index)** +* 备注清晰 +* 适合直接丢进你的项目(如 `app/vectorstore/`) + +--- + +# ✅ 一、Qdrant 最小异步客户端(AsyncQdrantClient) + +文件:`app/vectorstore/qdrant_client.py` + +```python +# 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。 + +```python +# 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 + +* BM25(text)解决 recall +* Vector(embedding)解决 semantic match +* 结果融合(RRF / weighted sum) + +你下一步只需要一个混合检索 pipeline,我可以给你写。 + +--- + +# 📌 四、如果你需要,我可以继续生成: + +1. **Qdrant + ES 混合检索 pipeline(RRF / 乘法融合 / 权重融合)** +2. `/hybrid_search` FastAPI 完整 API +3. ingest 文档(txt/docx/pdf),自动切片 + embedding + 写入 Qdrant & ES +4. KB 管理(创建/删除/重建/清空) +5. Docker Compose + 环境变量 + 部署脚本 + +只需告诉我: +👉 **“生成混合检索 pipeline 和 API”** diff --git a/docker-compose.yml b/docker-compose.yml index 0146a75..6afa178 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,3 @@ -version: "3.8" - services: elasticsearch: image: elasticsearch:9.2.2 @@ -12,7 +10,7 @@ services: ports: - "9210:9200" volumes: - - ./datas/es-data:/usr/share/elasticsearch/data + - ../datas/es-data:/usr/share/elasticsearch/data restart: unless-stopped qdrant: @@ -24,5 +22,5 @@ services: - "6333:6333" # HTTP - "6334:6334" # gRPC volumes: - - ./datas/qdrant-storage:/qdrant/storage + - ../datas/qdrant-storage:/qdrant/storage restart: unless-stopped \ No newline at end of file diff --git a/src/pipeline/config.py b/src/pipeline/config.py index 9f0685a..71d4aa2 100644 --- a/src/pipeline/config.py +++ b/src/pipeline/config.py @@ -6,6 +6,7 @@ import os load_dotenv() class Config(TypedDict): + logger_level: str version: str port: int host: str @@ -15,7 +16,10 @@ class Config(TypedDict): embedding_api_key: str embedding_api_host: str embedding_model: str - logger_level: str + es_host: str + es_key: str + qdrant_host: str + qdrant_key: str def _read_config() -> Config: @@ -30,6 +34,10 @@ def _read_config() -> Config: "embedding_api_host": os.getenv("EMBEDDING_API_HOST"), "embedding_api_key": os.getenv("EMBEDDING_API_KEY"), "embedding_model": os.getenv("EMBEDDING_MODEL"), + "es_host": os.getenv("ES_HOST"), + "es_key": os.getenv("ES_KEY") or "", + "qdrant_host": os.getenv("QDRANT_HOST"), + "qdrant_host": os.getenv("QDRANT_KEY") or "", } config = _read_config() diff --git a/src/pipeline/core/qdrant_client.py b/src/pipeline/core/qdrant_client.py new file mode 100644 index 0000000..e69de29