香港服务器部署Qdrant向量数据库：LangChain RAG企业知识库语义搜索完整方案

传统关键词搜索找不到同义词——搜索「服务器宕机」不会返回包含「实例崩溃」的文档。向量搜索将文本转换为高维数值向量，语义相近的内容在向量空间中距离更近，因此能实现真正的语义理解。Qdrant 是目前性能最优秀的开源向量数据库，结合 LangChain 的 RAG 流水线，可以让企业内部知识库像 ChatGPT 一样被智能检索和问答。

一、RAG 架构原理

<code">
文档入库流程：
原始文档（PDF/Word/网页）
    → 文本分割（Chunking，每块约500 tokens）
    → Embedding 模型转化为向量（1536维浮点数组）
    → 存入 Qdrant 向量数据库

查询流程：
用户问题："香港服务器如何开启BBR？"
    → Embedding 转化为向量
    → Qdrant 向量相似度搜索（cosine similarity）
    → 返回最相关的 Top-K 文档片段
    → 将片段 + 问题拼接为 Prompt
    → LLM 生成最终答案

二、部署 Qdrant

<code"># Docker 单节点部署（推荐中小规模）
mkdir -p /opt/qdrant/storage

docker run -d \
  --name qdrant \
  --restart unless-stopped \
  -p 127.0.0.1:6333:6333 \
  -p 127.0.0.1:6334:6334 \
  -v /opt/qdrant/storage:/qdrant/storage \
  -e QDRANT__SERVICE__API_KEY=your_qdrant_api_key \
  qdrant/qdrant:latest

# 验证服务
curl http://localhost:6333/

# 访问 Web UI（Qdrant Dashboard）
# http://服务器IP:6333/dashboard
# 通过 Nginx 反代并加认证后对外暴露

<code"># Nginx 反向代理（HTTPS + API Key 认证）
# /etc/nginx/sites-available/qdrant.yourdomain.com
server {
    listen 443 ssl http2;
    server_name qdrant.yourdomain.com;

    ssl_certificate /etc/letsencrypt/live/qdrant.yourdomain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/qdrant.yourdomain.com/privkey.pem;

    location / {
        # 验证 API Key
        if ($http_api_key != "your_qdrant_api_key") {
            return 401 "Unauthorized";
        }
        proxy_pass http://127.0.0.1:6333;
        proxy_set_header Host $host;
    }
}

三、Python 环境准备

<code"># 安装依赖
pip install qdrant-client langchain langchain-openai \
  langchain-community openai tiktoken \
  pymupdf python-docx sentence-transformers

# 或使用本地 Embedding 模型（不需要 OpenAI API）
pip install sentence-transformers

四、文档入库（Building the Index）

<code">from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_community.document_loaders import PyMuPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
import uuid, os

# ── 初始化 Qdrant 客户端 ──
qdrant = QdrantClient(
    url="http://localhost:6333",
    api_key="your_qdrant_api_key"
)

# ── 选择 Embedding 模型 ──
# 方案A：OpenAI（效果最好，需要 API 费用）
embedder = OpenAIEmbeddings(model="text-embedding-3-small")
VECTOR_DIM = 1536

# 方案B：本地模型（免费，效果略差）
# embedder = HuggingFaceEmbeddings(
#     model_name="BAAI/bge-m3",  # 中英文双语，效果好
#     model_kwargs={"device": "cpu"}
# )
# VECTOR_DIM = 1024

# ── 创建 Collection（向量表）──
COLLECTION_NAME = "company_knowledge"

if not qdrant.collection_exists(COLLECTION_NAME):
    qdrant.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(
            size=VECTOR_DIM,
            distance=Distance.COSINE  # 余弦相似度（语义搜索最常用）
        ),
    )
    print(f"Collection {COLLECTION_NAME} 创建成功")

def index_documents(docs_dir: str, collection_name: str):
    """将目录下所有文档入库"""
    # 1. 加载文档
    loader = DirectoryLoader(
        docs_dir,
        glob="**/*.pdf",
        loader_cls=PyMuPDFLoader,
        show_progress=True
    )
    raw_docs = loader.load()
    print(f"加载了 {len(raw_docs)} 个文档")

    # 2. 文本分割（Chunking）
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,       # 每块约500字符
        chunk_overlap=50,     # 相邻块重叠50字符（保持上下文连贯）
        separators=["\n\n", "\n", "。", "！", "？", " "]
    )
    chunks = splitter.split_documents(raw_docs)
    print(f"分割为 {len(chunks)} 个片段")

    # 3. 向量化并入库（批量处理）
    batch_size = 50
    points = []

    for i, chunk in enumerate(chunks):
        embedding = embedder.embed_query(chunk.page_content)
        points.append(PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={
                "text": chunk.page_content,
                "source": chunk.metadata.get("source", ""),
                "page": chunk.metadata.get("page", 0),
            }
        ))

        # 每50条批量插入
        if len(points) >= batch_size:
            qdrant.upsert(collection_name=collection_name, points=points)
            points = []
            print(f"已入库 {i+1}/{len(chunks)} 片段")

    # 插入剩余
    if points:
        qdrant.upsert(collection_name=collection_name, points=points)

    print(f"入库完成！Collection 中共 {qdrant.count(collection_name).count} 条向量")

# 执行入库
index_documents("/opt/knowledge-base/docs/", COLLECTION_NAME)

五、语义搜索与 RAG 问答

<code">from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def semantic_search(query: str, top_k: int = 5) -> list[dict]:
    """语义搜索——返回最相关的文档片段"""
    query_vector = embedder.embed_query(query)

    results = qdrant.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_vector,
        limit=top_k,
        score_threshold=0.6,  # 只返回相似度 > 0.6 的结果
        with_payload=True
    )

    return [
        {
            "text": r.payload["text"],
            "source": r.payload["source"],
            "score": round(r.score, 3)
        }
        for r in results
    ]

RAG_PROMPT = ChatPromptTemplate.from_template("""
你是企业知识库助手。请根据以下参考资料回答用户问题。

参考资料：
{context}

用户问题：{question}

回答要求：
- 基于参考资料回答，不要编造信息
- 如果参考资料中没有相关信息，明确说明"知识库中暂无相关信息"
- 引用具体来源（文件名+页码）
""")

def rag_answer(question: str) -> dict:
    """RAG 问答——检索相关文档后由 LLM 生成答案"""
    # 1. 检索相关片段
    chunks = semantic_search(question, top_k=5)

    if not chunks:
        return {"answer": "知识库中未找到相关信息", "sources": []}

    # 2. 构建上下文
    context = "\n\n".join([
        f"[来源：{c['source']} | 相似度：{c['score']}]\n{c['text']}"
        for c in chunks
    ])

    # 3. LLM 生成答案
    prompt = RAG_PROMPT.format_messages(
        context=context,
        question=question
    )
    response = llm.invoke(prompt)

    return {
        "answer": response.content,
        "sources": [c["source"] for c in chunks],
        "retrieved_chunks": len(chunks)
    }

# 使用示例
result = rag_answer("香港服务器如何配置 CN2 GIA 线路？")
print(result["answer"])
print("参考来源：", result["sources"])

六、混合检索（Hybrid Search）提升精度

<code">from qdrant_client.models import SparseVectorParams, SparseVector, NamedSparseVector

# 混合检索 = 语义向量搜索 + BM25 关键词搜索
# 两种结果 RRF（Reciprocal Rank Fusion）融合，效果比单一方式提升 15～25%

# 在 Collection 中添加稀疏向量支持
# qdrant.create_collection(
#     collection_name=COLLECTION_NAME,
#     vectors_config={
#         "dense": VectorParams(size=VECTOR_DIM, distance=Distance.COSINE),
#     },
#     sparse_vectors_config={
#         "sparse": SparseVectorParams(index=SparseIndexParams(on_disk=False))
#     }
# )

def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    """混合检索（语义 + 关键词）"""
    # 密集向量（语义搜索）
    dense_vector = embedder.embed_query(query)

    # 执行混合搜索
    results = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=[
            # 语义搜索召回 20 个
            {"query": dense_vector, "using": "dense", "limit": 20},
        ],
        query=dense_vector,   # 最终用密集向量重排序
        limit=top_k,
        with_payload=True
    )
    return [{"text": r.payload["text"], "score": r.score} for r in results.points]

七、多租户隔离（企业场景）

<code">
# 方案：用 payload filter 实现同一 Collection 多租户隔离
# 每个文档入库时附加 tenant_id

def index_for_tenant(docs, tenant_id: str):
    """为特定租户入库文档"""
    for chunk in docs:
        embedding = embedder.embed_query(chunk.page_content)
        qdrant.upsert(
            collection_name=COLLECTION_NAME,
            points=[PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload={
                    "text": chunk.page_content,
                    "tenant_id": tenant_id,   # 关键：租户标识
                    "source": chunk.metadata["source"],
                }
            )]
        )

def search_for_tenant(query: str, tenant_id: str, top_k: int = 5):
    """只搜索特定租户的文档"""
    from qdrant_client.models import Filter, FieldCondition, MatchValue

    query_vector = embedder.embed_query(query)
    results = qdrant.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_vector,
        query_filter=Filter(    # 过滤条件：只匹配指定租户
            must=[FieldCondition(
                key="tenant_id",
                match=MatchValue(value=tenant_id)
            )]
        ),
        limit=top_k,
        with_payload=True
    )
    return results

八、FastAPI 封装为 REST 接口

<code">from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="企业知识库 API")

class QueryRequest(BaseModel):
    question: str
    tenant_id: str = "default"
    top_k: int = 5

class QueryResponse(BaseModel):
    answer: str
    sources: list[str]
    confidence: float

@app.post("/api/search", response_model=QueryResponse)
async def search(req: QueryRequest):
    try:
        result = rag_answer(req.question)
        return QueryResponse(
            answer=result["answer"],
            sources=result["sources"],
            confidence=0.9 if result["retrieved_chunks"] >= 3 else 0.6
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    count = qdrant.count(COLLECTION_NAME).count
    return {"status": "ok", "indexed_chunks": count}

# 运行：uvicorn main:app --host 0.0.0.0 --port 8080

九、总结

Qdrant + LangChain 构建的 RAG 企业知识库，让员工用自然语言提问就能精准检索内部文档，准确率远超传统关键词搜索。香港服务器部署 Qdrant 后，接入本地 Ollama 模型可以实现全链路数据不出境，是金融、法律、医疗等对数据合规要求严格的行业的最优解。4核8G 服务器可以支撑约 100 万个向量（约 2000 份 PDF 文档）的实时检索。