ai_pipeline/Demo1.md

216 lines
5.6 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

下面我给你分别写 **最小可用、可复制进项目的 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
* 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”**