Compare commits
2 Commits
dface2ba34
...
8132b1d515
| Author | SHA1 | Date |
|---|---|---|
|
|
8132b1d515 | |
|
|
9776f3c791 |
|
|
@ -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”**
|
||||
|
|
@ -1,19 +1,29 @@
|
|||
version: "3.8"
|
||||
networks:
|
||||
ai-pipeline:
|
||||
driver: bridge
|
||||
|
||||
volumes:
|
||||
es-data:
|
||||
qdrant-storage:
|
||||
|
||||
services:
|
||||
elasticsearch:
|
||||
image: elasticsearch:9.2.2
|
||||
image: docker.elastic.co/elasticsearch/elasticsearch:9.2.2
|
||||
container_name: elasticsearch
|
||||
user: "1000:1000"
|
||||
environment:
|
||||
- discovery.type=single-node # 单节点模式
|
||||
- xpack.security.enabled=true # 开启安全(才可以设密码)
|
||||
- ELASTIC_PASSWORD=12345 # 设置密码(重要)
|
||||
- ES_JAVA_OPTS=-Xms1g -Xmx1g
|
||||
- xpack.security.enabled=true # 开启安全
|
||||
- ELASTIC_PASSWORD=12345 # 设置密码
|
||||
# - xpack.security.http.ssl.enabled=false # 关闭 HTTPS
|
||||
# - ES_JAVA_OPTS=-Xms1g -Xmx1g # JVM 初始堆内存大小,1GB,最大堆内存大小,1GB
|
||||
ports:
|
||||
- "9210:9200"
|
||||
volumes:
|
||||
- ./datas/es-data:/usr/share/elasticsearch/data
|
||||
- es-data:/usr/share/elasticsearch/data
|
||||
restart: unless-stopped
|
||||
networks:
|
||||
- ai-pipeline
|
||||
|
||||
qdrant:
|
||||
image: qdrant/qdrant:latest
|
||||
|
|
@ -24,5 +34,7 @@ services:
|
|||
- "6333:6333" # HTTP
|
||||
- "6334:6334" # gRPC
|
||||
volumes:
|
||||
- ./datas/qdrant-storage:/qdrant/storage
|
||||
restart: unless-stopped
|
||||
- qdrant-storage:/qdrant/storage
|
||||
restart: unless-stopped
|
||||
networks:
|
||||
- ai-pipeline
|
||||
|
|
|
|||
|
|
@ -1,11 +1,13 @@
|
|||
from mimetypes import init
|
||||
from typing import TypedDict
|
||||
from dotenv import load_dotenv
|
||||
from src.pipeline.core.utils import logger
|
||||
import os
|
||||
|
||||
load_dotenv()
|
||||
|
||||
class Config(TypedDict):
|
||||
logger_level: str
|
||||
version: str
|
||||
port: int
|
||||
host: str
|
||||
|
|
@ -15,7 +17,13 @@ class Config(TypedDict):
|
|||
embedding_api_key: str
|
||||
embedding_api_host: str
|
||||
embedding_model: str
|
||||
logger_level: str
|
||||
es_host: str
|
||||
es_port: int
|
||||
es_user: str
|
||||
es_password: str
|
||||
qdrant_host: str
|
||||
qdrant_port: int
|
||||
qdrant_api_key: str
|
||||
|
||||
|
||||
def _read_config() -> Config:
|
||||
|
|
@ -30,6 +38,14 @@ 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_port": int(os.getenv("ES_PORT")),
|
||||
"es_host": os.getenv("ES_USER") or "elastic",
|
||||
"es_password": os.getenv("ES_PASSWORD") or "",
|
||||
"qdrant_host": os.getenv("QDRANT_HOST"),
|
||||
"qdrant_port": int(os.getenv("QDRANT_PORT")),
|
||||
"qdrant_api_key": os.getenv("QDRANT_API_KEY") or "",
|
||||
}
|
||||
|
||||
config = _read_config()
|
||||
logger.debug("创建全局: config")
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
import httpx
|
||||
from pipeline.config import config
|
||||
|
||||
class AsyncES:
|
||||
def __init__(self):
|
||||
self.base = f"{config['es_host']}:{config['es_port']}"
|
||||
self.auth = (config["es_user"], config['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_client = AsyncES
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
from qdrant_client import AsyncQdrantClient
|
||||
from qdrant_client.models import (
|
||||
VectorParams,
|
||||
Distance,
|
||||
PointStruct,
|
||||
)
|
||||
from pipeline.config import config
|
||||
|
||||
|
||||
class AsyncQD:
|
||||
def __init__(self):
|
||||
self.client = AsyncQdrantClient(
|
||||
host=config["qdrant_host"],
|
||||
port=config["qdrant_port"],
|
||||
api_key=config["qdrant_api_key"],
|
||||
)
|
||||
|
||||
async def ensure_collection(self, name: str, dim: int):
|
||||
"""
|
||||
如果 collection 不存在则创建。
|
||||
适用于多知识库:每个知识库一个 collection。
|
||||
"""
|
||||
try:
|
||||
await self.client.get_collection(name)
|
||||
except Exception:
|
||||
await self.client.recreate_collection(
|
||||
collection_name=name,
|
||||
vectors_config=VectorParams(size=dim, distance=Distance.COSINE),
|
||||
)
|
||||
|
||||
async def upsert_vectors(self, 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 self.client.upsert(collection_name=collection, points=points)
|
||||
|
||||
async def vector_search(self, collection: str, query_vector: list, top_k: int = 5):
|
||||
"""
|
||||
查询向量相似度
|
||||
"""
|
||||
result = await self.client.search(
|
||||
collection_name=collection,
|
||||
query_vector=query_vector,
|
||||
limit=top_k,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
qd_client = AsyncQD()
|
||||
Loading…
Reference in New Issue