cfc/cfc-langgraph/app/rag/retriever.py

from langchain_chroma import Chroma
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever
from langchain_openai import ChatOpenAI
from .embeddings import get_embeddings
from app.config import settings
from app.tools.java_client import JavaClient
from typing import Optional
import logging

logger = logging.getLogger(__name__)


class RagRetriever:
    """升级版混合检索器: 向量 + BM25 + LLM 压缩重排序"""

    def __init__(self, collection_name: str = "cfc_knowledge"):
        embeddings = get_embeddings()
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=embeddings,
            persist_directory=settings.chroma_db_path,
        )
        self.java_client = JavaClient()
        self._bm25_retriever: Optional[BM25Retriever] = None
        self._bm25_texts: list[str] = []

    async def initialize(self):
        """从 Java 侧拉取知识库, 构建 BM25 索引"""
        try:
            articles = await self.java_client.get_published_articles()
            self._bm25_texts = [
                f"{a['title']} {a['summary']} {a.get('tags', '')}"
                for a in articles
            ]
            if self._bm25_texts:
                self._bm25_retriever = BM25Retriever.from_texts(
                    self._bm25_texts,
                    metadatas=articles,
                )
                logger.info("BM25 索引就绪: %d 条", len(self._bm25_texts))
        except Exception as e:
            logger.warning("BM25 初始化失败: %s", e)

    async def retrieve(
        self,
        query: str,
        filters: Optional[dict] = None,
        k: int = 5,
        use_compression: bool = True,
    ) -> list[dict]:
        """混合检索 + 可选 LLM 压缩重排序"""
        retrievers = []

        # 1. 向量检索 (多取一些方便后续 ensemble 排序)
        vector_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": k * 2, "filter": filters},
        )
        retrievers.append(vector_retriever)

        # 2. BM25 关键词检索
        if self._bm25_retriever:
            bm25_k = self._bm25_retriever.k
            self._bm25_retriever.k = k * 2
            retrievers.append(self._bm25_retriever)
            self._bm25_retriever.k = bm25_k

        if len(retrievers) == 1:
            docs = await retrievers[0].ainvoke(query)
            ensemble = retrievers[0]
        else:
            ensemble = EnsembleRetriever(
                retrievers=retrievers,
                weights=[0.6, 0.4],
            )
            docs = await ensemble.ainvoke(query)

        # 3. LLM 压缩 (剔除不相关内容)
        if use_compression and docs:
            llm = ChatOpenAI(
                model=settings.llm_model,
                api_key=settings.llm_api_key,
                base_url=settings.llm_base_url,
                temperature=0,
            )
            compressor = LLMChainExtractor.from_llm(llm)
            compression_retriever = ContextualCompressionRetriever(
                base_compressor=compressor,
                base_retriever=ensemble if len(retrievers) > 1 else retrievers[0],
            )
            docs = await compression_retriever.ainvoke(query)

        # 4. 格式化为统一输出 + 去重
        results = []
        seen = set()
        for doc in docs:
            content_hash = hash(doc.page_content[:100])
            if content_hash in seen:
                continue
            seen.add(content_hash)
            results.append({
                "content": doc.page_content,
                "metadata": doc.metadata,
                "score": doc.metadata.get("score", 0) if hasattr(doc, "metadata") else 0,
            })

        return results[:k]