从零到一：打造一个支持 RAG 的智能聊天应用

RAG（检索增强生成）的核心思路是：先把知识切分、向量化存起来，问答时先检索最相关的片段，再让大模型基于这些片段作答，从而缓解"幻觉"与"知识过时"。本文用一个真实可跑的 RAG 工程，串起从混合检索 → 嵌入 → 生成的完整链路，所有代码均取自实际仓库。

为什么需要混合检索

纯向量检索擅长"语义相似"，但对精确关键词（专有名词、型号、报错码）容易漏召回。混合检索 = 向量语义召回 + BM25 关键词召回，两路分数融合后再交给大模型，召回率与准确率都更稳。

本文涉及两套互补的实现：

• rag-platform：PostgreSQL + pgvector 存向量，自研 jieba-BM25 做二级融合，适合"可控、可 SQL 审计"的生产场景。
• langchain-llm-toolkit：基于 LangChain 的 RAGSystem，向量库支持 FAISS（本地）/ Qdrant，融合策略支持 RRF / weighted / score，适合快速搭管线。

整体架构

用户问题
   │
   ▼
┌──────────────┐   向量召回      ┌──────────────────┐
│  Query 预处理 │──────────────▶│ pgvector / FAISS  │
│ (同义扩展等)  │              │  cosine / L2     │
└──────┬───────┘               └──────────────────┘
       │ BM25 关键词召回
       ▼
┌──────────────┐   融合 (加权/RRF)   ┌──────────────┐
│  Hybrid 融合  │──────────────────▶│  Top-K 上下文 │
└──────┬───────┘                    └──────┬───────┘
       │                                    │
       ▼                                    ▼
┌──────────────┐                   ┌──────────────────┐
│  提示组装      │◀──── 上下文 ──────│  RAGPromptBuilder │
│ (RAG_QA 模板) │                   └──────────────────┘
└──────┬───────┘
       ▼
   LLM 生成回答

一、rag-platform：基于 pgvector 的混合检索

1. 配置与权重

权重、Top-K、Embedding 后端都收敛到 Settings，混合检索的 vector_weight / bm25_weight 在此定义：

# app/config.py（节选）
class Settings:
    # LLM（OpenAI 兼容接口）
    openai_api_key: str = _env("OPENAI_API_KEY", "")
    openai_base_url: str = _env("OPENAI_BASE_URL", "https://api.openai.com/v1")
    llm_model: str = _env("LLM_MODEL", "gpt-3.5-turbo")

    # Embedding 双后端：ollama 优先，sentence-transformers 兜底
    embed_backend: str = _env("EMBED_BACKEND", "ollama")
    embed_model: str = _env("EMBED_MODEL", "snowflake-arctic-embed2")
    embed_dim: int = int(_env("EMBED_DIM", "1024"))

    # 混合检索权重
    vector_weight: float = float(_env("VECTOR_WEIGHT", "0.6"))
    bm25_weight: float = float(_env("BM25_WEIGHT", "0.4"))
    top_k: int = int(_env("TOP_K", "4"))

    database_url: str = _env("DATABASE_URL", "postgresql+psycopg://rag:rag@localhost:5432/rag")
    redis_url: str = _env("REDIS_URL", "redis://localhost:6379/0")

settings = Settings()

2. 数据模型：向量存进 pgvector 列

用 SQLModel 定义，Chunk 的 embedding 直接是 pgvector 的 Vector(EMBED_DIM) 列：

# app/models.py（节选）
from pgvector.sqlalchemy import Vector
from sqlalchemy import Column
from sqlmodel import Field, SQLModel

class Document(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    content: str
    status: str = Field(default="active")

class Chunk(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    doc_id: int = Field(foreign_key="document.id", ondelete="CASCADE", index=True)
    text: str
    # pgvector 向量列；维度在建表后固定，改维度需重建表
    embedding: list[float] | None = Field(
        default=None, sa_column=Column(Vector(settings.embed_dim))
    )

class QAPair(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    question: str
    answer: str
    status: str = Field(default="active")

建表时先确保 pgvector 扩展存在（SQLite 不支持则跳过）：

# app/db.py（节选）
def init_db() -> None:
    if settings.database_url.startswith("postgresql"):
        with engine.connect() as conn:
            conn.exec_driver_sql("CREATE EXTENSION IF NOT EXISTS vector")
            conn.commit()
    SQLModel.metadata.create_all(engine)

3. 自研 BM25（jieba 分词）

纯向量方案只走语义，这里额外补一路 Okapi BM25 做关键词召回。实现是教科书式的 TF-IDF + 文档长度归一化：

# app/bm25.py（节选）
import math
from collections import defaultdict
import jieba

class BM25:
    def __init__(self, docs: list[str], k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.docs = [self._tokenize(d) for d in docs]
        self.N = len(self.docs)
        self.avgdl = sum(len(d) for d in self.docs) / self.N if self.N else 0.0
        self.f = []
        self.df = defaultdict(int)
        for doc in self.docs:
            tf: dict[str, int] = defaultdict(int)
            for w in doc:
                tf[w] += 1
            self.f.append(dict(tf))
            for w in tf:
                self.df[w] += 1

    @staticmethod
    def _tokenize(text: str) -> list[str]:
        return jieba.lcut(text)

    def score(self, query: str) -> list[float]:
        q_tokens = self._tokenize(query)
        scores = []
        for i, doc in enumerate(self.docs):
            dl = len(doc)
            score = 0.0
            for w in q_tokens:
                if w not in self.f[i]:
                    continue
                idf = math.log(1 + (self.N - self.df[w] + 0.5) / (self.df[w] + 0.5))
                tf = self.f[i][w]
                score += idf * (tf * (self.k1 + 1)) / (
                    tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
                )
            scores.append(score)
        return scores

4. 混合检索融合核心

向量召回走 pgvector 的 cosine_distance 在 SQL 层完成，BM25 做二级打分，两路各自 min-max 归一化后按权重求和，取 Top-K：

# app/retriever.py（节选）
def _minmax(values: list[float]) -> list[float]:
    if not values:
        return []
    lo, hi = min(values), max(values)
    if hi - lo < 1e-9:
        return [0.0] * len(values)
    return [(v - lo) / (hi - lo) for v in values]

def hybrid_search(query: str, session, top_k: int | None = None) -> list[dict]:
    top_k = top_k or settings.top_k
    q_vec = embed_one(query).tolist()

    # 1) pgvector 向量召回 top_k*3 候选（语义余弦）
    stmt = (
        select(
            Chunk.id, Chunk.text,
            (1 - Chunk.embedding.cosine_distance(q_vec)).label("vec_score"),
        )
        .order_by(Chunk.embedding.cosine_distance(q_vec))
        .limit(top_k * 3)
    )
    rows = session.exec(stmt).all()
    if not rows:
        return []

    texts = [r.text for r in rows]
    vec_scores = [float(r.vec_score) for r in rows]

    # 2) BM25 二级融合（关键词命中增强）
    bm25 = BM25(texts)
    bm25_n = _minmax(bm25.score(query))
    vec_n = _minmax(vec_scores)
    fused = [settings.vector_weight * v + settings.bm25_weight * b
             for v, b in zip(vec_n, bm25_n)]

    order = sorted(range(len(rows)), key=lambda i: fused[i], reverse=True)[:top_k]
    ids = [rows[i].id for i in order]
    return [{"id": ids[i], "text": texts[i], "score": fused[i]} for i in order]

5. Embedding 双后端

Embedding 用"薄门面 + 双后端"：主路径走 Ollama（仅 requests，零重依赖），失败再延迟 import sentence-transformers（带 torch 的重依赖），保证主路径启动快：

# app/embed/__init__.py（节选）
from app.embed.core import embed, embed_one
__all__ = ["embed", "embed_one"]

# app/embed/st_backend.py（节选，兜底后端）
from sentence_transformers import SentenceTransformer

class STBackend(EmbedBackend):
    def __init__(self, model: str) -> None:
        self._model = SentenceTransformer(model)

    def encode(self, texts: list[str]) -> np.ndarray:
        vecs = self._model.encode(texts, normalize_embeddings=True, convert_to_numpy=True)
        return np.asarray(vecs, dtype="float32")

二、langchain-llm-toolkit：RAG 管线

另一套实现用 LangChain 把"检索 → 融合 → 提示 → 生成"封装成 RAGSystem，向量库支持 FAISS（本地）与 Qdrant。

1. 混合检索（weighted / RRF / score）

retrieve_hybrid 同时做语义召回与 BM25 召回，按 bm25_weight 加权融合：

# src/langchain_llm_toolkit/rag.py（节选）
def retrieve_hybrid(self, query: str, k: int = 5, bm25_weight: float = 0.3):
    if not self.vector_store:
        raise ValueError("Vector store not initialized")

    semantic_docs = self.retrieve_documents(query, k=k * 2)
    if self.bm25 is None:
        return semantic_docs[:k]

    bm25_results = self.bm25.search(query, k=k * 2)

    sem_scores, bm25_scores = {}, {}
    for i, doc in enumerate(semantic_docs):
        sem_scores[doc.page_content[:200]] = 1.0 - (i / max(len(semantic_docs), 1))
    max_bm25 = max((s for _, s in bm25_results), default=1.0)
    for doc, score in bm25_results:
        if max_bm25 > 0:
            bm25_scores[doc.page_content[:200]] = score / max_bm25

    all_docs = {}
    for doc, _ in bm25_results:
        all_docs[doc.page_content[:200]] = (doc, 0)
    for doc in semantic_docs:
        if doc.page_content[:200] not in all_docs:
            all_docs[doc.page_content[:200]] = (doc, 0)

    sem_w = 1.0 - bm25_weight
    for key, (doc, _) in all_docs.items():
        score = bm25_weight * bm25_scores.get(key, 0) + sem_w * sem_scores.get(key, 0)
        all_docs[key] = (doc, score)

    sorted_docs = sorted(all_docs.values(), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in sorted_docs[:k]]

更进一步的 HybridRetriever 默认用 RRF（Reciprocal Rank Fusion）——只依赖排名、不依赖绝对分数，对两路分数尺度不一致更鲁棒：

# src/langchain_llm_toolkit/hybrid_retriever.py（节选）
class HybridRetriever:
    def __init__(self, fusion_method: str = "rrf", keyword_weight: float = 0.3, rrf_k: int = 60):
        self.fusion_method = fusion_method
        self.keyword_weight = keyword_weight
        self.rrf_k = rrf_k
        self.bm25 = BM25()

    def _rrf_fusion(self, keyword_results, semantic_results, k):
        doc_scores = {}
        for rank, (doc, _) in enumerate(keyword_results):
            doc_id = self._get_doc_id(doc)
            doc_scores[doc_id] = (doc, 1 / (self.rrf_k + rank + 1))
        for rank, (doc, _) in enumerate(semantic_results):
            doc_id = self._get_doc_id(doc)
            rrf_score = 1 / (self.rrf_k + rank + 1)
            if doc_id in doc_scores:
                d, s = doc_scores[doc_id]
                doc_scores[doc_id] = (d, s + rrf_score)
            else:
                doc_scores[doc_id] = (doc, rrf_score)
        return sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)[:k]

2. 提示组装

上下文拼进 RAG_QA 模板，并约束"只基于上下文、无信息就明说"，是抑制幻觉的关键一步：

# src/langchain_llm_toolkit/prompt_templates.py（节选）
class PromptTemplateType(Enum):
    RAG_QA = "rag_qa"
    RAG_SUMMARY = "rag_summary"
    # ...

class RAGPromptBuilder:
    def build_qa_prompt(self, query, documents, max_context_length=4000) -> str:
        context_parts = []
        current_length = 0
        for i, doc in enumerate(documents):
            content = doc.page_content if hasattr(doc, "page_content") else str(doc)
            if current_length + len(content) > max_context_length:
                break
            context_parts.append(f"[文档 {i + 1}]\n{content}\n")
            current_length += len(content)
        context = "\n".join(context_parts)
        return self.template_manager.render(PromptTemplateType.RAG_QA, context=context, query=query)

RAG_QA 模板的要点：

你是一个专业的知识库问答助手。请基于以下上下文回答问题。

## 上下文文档
{context}

## 用户问题
{query}

## 回答要求
1. 准确性：只基于上下文回答，不要编造信息
2. 完整性：尽可能详细回答
3. 引用：必要时标注信息来源
4. 诚实性：如果上下文中没有相关信息，请明确说明"根据现有知识库，我无法回答"

三、串起一次完整问答

1. 检索：hybrid_search(query) 或 RAGSystem.retrieve_hybrid(query)，拿到 Top-K 片段。
2. 组装：RAGPromptBuilder.build_qa_prompt(query, docs) 拼出带上下文的提示。
3. 生成：把提示交给 LLMIntegration.generate(prompt)，得到最终回答。
4. 兜底：QAPair 表里的标准问答对可"直答"，命中即返，不必过 LLM。

四、部署与优化建议

• 向量库选型：原型用 FAISS / SQLite 最快；需要并发与持久化用 pgvector / Qdrant。
• Embedding 后端：本地 Ollama 省 API 成本、数据不出域；注意维度一旦建表就固定，改维度要重建。
• 融合策略：两路分数尺度差异大时优先 RRF；可控调权时用 weighted。
• 上下文长度：max_context_length 截断 + 引用标注，平衡成本与可追溯性。
• 缓存：相同 query 的结果可缓存，降低重复检索 / 生成开销。

总结

RAG 不是"接个向量库 + 调个大模型"就完事——混合检索决定了召回质量，提示约束决定了回答可信度。本文两套实现（可控的 pgvector+BM25、灵活的 LangChain 管线）都围绕这两点展开，且代码均来自可运行仓库，可直接对照改造。

项目地址

• rag-platform：github.com/erishen/rag-platform
• langchain-llm-toolkit：github.com/erishen/langchain-llm-toolkit

相关阅读

• 《用 shadcn 风格组件库统一多端 UI》
• 《Monorepo + Next.js 14 + FastAPI 全栈实践》

源码导航

完整工程分两个仓库：erishen/rag-platform（检索与索引内核）与 erishen/langchain-llm-toolkit（LangChain RAG 封装）。

检索内核（rag-platform）

文件	作用
app/bm25.py	BM25 关键词检索实现
app/db.py	向量/文档存储层
app/config.py	配置（模型 / 嵌入 / 库路径）
app/models.py	数据模型
app/embed/st_backend.py	嵌入后端（Sentence-Transformers 等）

LangChain 封装（langchain-llm-toolkit）

文件	作用
rag_query_web.py	RAG 查询 Web 入口
src/langchain_llm_toolkit/hybrid_retriever.py	BM25 + 语义混合检索器
src/langchain_llm_toolkit/llm_integration.py	LLM 调用、重试、缓存、限流
src/langchain_llm_toolkit/rag.py	完整 RAG 系统

常见问题