零构建企业级 RAG 问答ai产品：架构、调优与可观测性实践

# hybrid_retriever.py
from langchain_community.vectorstores import Milvus
from langchain_community.retrievers import BM25Retriever
from langchain.schema import BaseRetriever, Document
from typing import List
import numpy as np

class HybridRetriever(BaseRetriever):
    """向量检索与 BM25 加权融合"""
    def __init__(self, vector_store, bm25_retriever, weight_v=0.6, weight_b=0.4, top_k=20):
        self.vector_store = vector_store
        self.bm25 = bm25_retriever
        self.weight_v = weight_v
        self.weight_b = weight_b
        self.top_k = top_k

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 向量检索
        vec_docs = self.vector_store.similarity_search_with_score(query, k=self.top_k)
        # BM25 检索
        bm_docs = self.bm25.get_relevant_documents(query)[:self.top_k]
        
        # 合并打分（归一化后加权求和）
        all_docs = {doc.page_content: doc for doc, _ in vec_docs}
        scores = {}
        for doc, score in vec_docs:
            scores[doc.page_content] = self.weight_v * (1 - score)  # 距离转相似度
        for doc in bm_docs:
            if doc.page_content in scores:
                scores[doc.page_content] += self.weight_b * 0.8  # 固定 BM25 权重，可改进
            else:
                scores[doc.page_content] = self.weight_b * 0.5
        
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [Document(page_content=doc, metadata=all_docs[doc].metadata) for doc, _ in sorted_docs[:5]]

# reranker.py
from sentence_transformers import CrossEncoder
from langchain.retrievers.document_compressors import LLMChainExtractor

class Reranker:
    def __init__(self, model_name="BAAI/bge-reranker-v2-m3"):
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, docs: List[Document], top_n=3) -> List[Document]:
        pairs = [[query, doc.page_content] for doc in docs]
        scores = self.model.predict(pairs)
        sorted_idx = np.argsort(scores)[::-1][:top_n]
        return [docs[i] for i in sorted_idx]

# generator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

PROMPT_TEMPLATE = """
你是一个企业制度问答助手。根据以下文档片段回答问题。
必须返回 JSON 格式：{{"answer": "...", "confidence": 0.0-1.0, "source_ids": [...]}}
若文档无法回答，请设置 confidence < 0.3 并说明未知。

问题：{query}
文档：{context}
"""

class Generator:
    def __init__(self, model_name="qwen2.5-72b-instruct", api_key=None):
        self.llm = ChatOpenAI(model=model_name, api_key=api_key, temperature=0.1)
        self.parser = JsonOutputParser()

    def generate(self, query: str, docs: List[Document]):
        context = "\n\n".join([f"[{i}] {doc.page_content}" for i, doc in enumerate(docs)])
        prompt = ChatPromptTemplate.from_template(PROMPT_TEMPLATE)
        chain = prompt | self.llm | self.parser
        result = chain.invoke({"query": query, "context": context})
        # 简单事实校验：若答案中提及金额/日期，检查是否在文档中出现
        # 此处省略具体正则校验逻辑
        return result

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

app = FastAPI()
FastAPIInstrumentor.instrument_app(app)
tracer = trace.get_tracer(__name__)

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

@app.post("/ask")
async def ask(request: QueryRequest):
    with tracer.start_as_current_span("rag_query") as span:
        span.set_attribute("query", request.question)
        # 1. 意图识别（简单示例，实际可用轻量分类器）
        if "对比" in request.question or "区别" in request.question:
            # 调整检索策略，增加召回数
            pass
        
        # 2. 混合检索
        docs = hybrid_retriever.invoke(request.question)
        # 3. 重排序
        top_docs = reranker.rerank(request.question, docs, top_n=3)
        # 4. 生成
        result = generator.generate(request.question, top_docs)
        span.set_attribute("confidence", result["confidence"])
        return result

# evaluator.py
import pandas as pd
from rouge_score import rouge_scorer
from bert_score import score as bert_score

def evaluate(test_set_path: str):
    df = pd.read_csv(test_set_path)  # 包含 question, ground_truth, expected_doc_ids
    results = []
    for _, row in df.iterrows():
        pred = rag_pipeline(row["question"])
        # 检索命中率
        hit = any(doc.metadata["id"] in row["expected_doc_ids"] for doc in top_docs)
        # 生成质量
        rouge = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=True)
        rouge_l = rouge.score(row["ground_truth"], pred["answer"])["rougeL"].fmeasure
        # BERTScore（更鲁棒）
        P, R, F1 = bert_score([pred["answer"]], [row["ground_truth"]], lang="zh")
        results.append({"hit": hit, "rouge_l": rouge_l, "bert_f1": F1.item()})
    return pd.DataFrame(results).mean()

# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

services:
  app:
    build: .
    environment:
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"