背景

一、RAG关键流程源码解读

# 知识工厂进行实例化KnowledgeFactory -> create() -> load() -> Document- knowledge - markdown - pdf - docx - txt - html - pptx - url - ...

class Knowledge(ABC):    def load(self) -> List[Document]:        """Load knowledge from data loader."""    @classmethod    def document_type(cls) -> Any:        """Get document type."""    def support_chunk_strategy(cls) -> List[ChunkStrategy]:        """Return supported chunk strategy."""        return [            ChunkStrategy.CHUNK_BY_SIZE,            ChunkStrategy.CHUNK_BY_PAGE,            ChunkStrategy.CHUNK_BY_PARAGRAPH,            ChunkStrategy.CHUNK_BY_MARKDOWN_HEADER,            ChunkStrategy.CHUNK_BY_SEPARATOR,        ]    @classmethod    def default_chunk_strategy(cls) -> ChunkStrategy:        """Return default chunk strategy.
        Returns:            ChunkStrategy: default chunk strategy        """        return ChunkStrategy.CHUNK_BY_SIZE

class ChunkManager:    """Manager for chunks."""
    def __init__(        self,        knowledge: Knowledge,        chunk_parameter: Optional[ChunkParameters] = None,        extractor: Optional[Extractor] = None,    ):        """Create a new ChunkManager with the given knowledge.
        Args:            knowledge: (Knowledge) Knowledge datasource.            chunk_parameter: (Optional[ChunkParameter]) Chunk parameter.            extractor: (Optional[Extractor]) Extractor to use for summarization.        """        self._knowledge = knowledge
        self._extractor = extractor        self._chunk_parameters = chunk_parameter or ChunkParameters()        self._chunk_strategy = (            chunk_parameter.chunk_strategy            if chunk_parameter and chunk_parameter.chunk_strategy            else self._knowledge.default_chunk_strategy().name        )        self._text_splitter = self._chunk_parameters.text_splitter        self._splitter_type = self._chunk_parameters.splitter_type

class ChunkStrategy(Enum):    """Chunk Strategy Enum."""
    CHUNK_BY_SIZE: _STRATEGY_ENUM_TYPE = (        RecursiveCharacterTextSplitter,        [            {                "param_name": "chunk_size",                "param_type": "int",                "default_value": 512,                "description": "The size of the data chunks used in processing.",            },            {                "param_name": "chunk_overlap",                "param_type": "int",                "default_value": 50,                "description": "The amount of overlap between adjacent data chunks.",            },        ],        "chunk size",        "split document by chunk size",    )    CHUNK_BY_PAGE: _STRATEGY_ENUM_TYPE = (        PageTextSplitter,        [],        "page",        "split document by page",    )    CHUNK_BY_PARAGRAPH: _STRATEGY_ENUM_TYPE = (        ParagraphTextSplitter,        [            {                "param_name": "separator",                "param_type": "string",                "default_value": "\n",                "description": "paragraph separator",            }        ],        "paragraph",        "split document by paragraph",    )    CHUNK_BY_SEPARATOR: _STRATEGY_ENUM_TYPE = (        SeparatorTextSplitter,        [            {                "param_name": "separator",                "param_type": "string",                "default_value": "\n",                "description": "chunk separator",            },            {                "param_name": "enable_merge",                "param_type": "boolean",                "default_value": False,                "description": "Whether to merge according to the chunk_size after "                "splitting by the separator.",            },        ],        "separator",        "split document by separator",    )    CHUNK_BY_MARKDOWN_HEADER: _STRATEGY_ENUM_TYPE = (        MarkdownHeaderTextSplitter,        [],        "markdown header",        "split document by markdown header",    )

@abstractmethod    def embed_documents(self, texts: List[str]) -> List[List[float]]:        """Embed search docs."""
    @abstractmethod    def embed_query(self, text: str) -> List[float]:        """Embed query text."""
    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:        """Asynchronous Embed search docs."""        return await asyncio.get_running_loop().run_in_executor(            None, self.embed_documents, texts        )
    async def aembed_query(self, text: str) -> List[float]:        """Asynchronous Embed query text."""        return await asyncio.get_running_loop().run_in_executor(            None, self.embed_query, text        )

    # EMBEDDING_MODEL=proxy_openai# proxy_openai_proxy_server_url=https://api.openai.com/v1# proxy_openai_proxy_api_key={your-openai-sk}# proxy_openai_proxy_backend=text-embedding-ada-002

## qwen embedding model, See dbgpt/model/parameter.py# EMBEDDING_MODEL=proxy_tongyi# proxy_tongyi_proxy_backend=text-embedding-v1# proxy_tongyi_proxy_api_key={your-api-key}
## qianfan embedding model, See dbgpt/model/parameter.py#EMBEDDING_MODEL=proxy_qianfan#proxy_qianfan_proxy_backend=bge-large-zh#proxy_qianfan_proxy_api_key={your-api-key}#proxy_qianfan_proxy_api_secret={your-secret-key}

class TripletExtractor(LLMExtractor):    """TripletExtractor class."""
    def __init__(self, llm_client: LLMClient, model_name: str):        """Initialize the TripletExtractor."""        super().__init__(llm_client, model_name, TRIPLET_EXTRACT_PT)
TRIPLET_EXTRACT_PT = (    "Some text is provided below. Given the text, "    "extract up to knowledge triplets as more as possible "    "in the form of (subject, predicate, object).n"    "Avoid stopwords. The subject, predicate, object can not be none.n"    "---------------------n"    "Example:n"    "Text: Alice is Bob's mother.n"    "Triplets:n(Alice, is mother of, Bob)n"    "Text: Alice has 2 apples.n"    "Triplets:n(Alice, has 2, apple)n"    "Text: Alice was given 1 apple by Bob.n"    "Triplets:(Bob, gives 1 apple, Bob)n"    "Text: Alice was pushed by Bob.n"    "Triplets:(Bob, pushes, Alice)n"    "Text: Bob's mother Alice has 2 apples.n"    "Triplets:n(Alice, is mother of, Bob)n(Alice, has 2, apple)n"    "Text: A Big monkey climbed up the tall fruit tree and picked 3 peaches.n"    "Triplets:n(monkey, climbed up, fruit tree)n(monkey, picked 3, peach)n"    "Text: Alice has 2 apples, she gives 1 to Bob.n"    "Triplets:n"    "(Alice, has 2, apple)n(Alice, gives 1 apple, Bob)n"    "Text: Philz is a coffee shop founded in Berkeley in 1982.n"    "Triplets:n"    "(Philz, is, coffee shop)n(Philz, founded in, Berkeley)n"    "(Philz, founded in, 1982)n"    "---------------------n"    "Text: {text}n"    "Triplets:n")

- VectorStoreBase    - ChromaStore    - MilvusStore    - OceanbaseStore    - ElasticsearchStore    - PGVectorStore
class VectorStoreBase(IndexStoreBase, ABC):    """Vector store base class."""
    @abstractmethod    def load_document(self, chunks: List[Chunk]) -> List[str]:        """Load document in index database."""    @abstractmethod    async def aload_document(self, chunks: List[Chunk]) -> List[str]:        """Load document in index database."""            @abstractmethod    def similar_search_with_scores(        self,        text,        topk,        score_threshold: float,        filters: Optional[MetadataFilters] = None,    ) -> List[Chunk]:        """Similar search with scores in index database."""    def similar_search(        self, text: str, topk: int, filters: Optional[MetadataFilters] = None    ) -> List[Chunk]:        return self.similar_search_with_scores(text, topk, 1.0, filters)

- GraphStoreBase    - TuGraphStore    - Neo4jStore
def insert_triplet(self, subj: str, rel: str, obj: str) -> None:    """Add triplet."""    ...TL;DR...    subj_query = f"MERGE (n1:{self._node_label} {{id:'{subj}'}})"    obj_query = f"MERGE (n1:{self._node_label} {{id:'{obj}'}})"    rel_query = (        f"MERGE (n1:{self._node_label} {{id:'{subj}'}})"        f"-[r:{self._edge_label} {{id:'{rel}'}}]->"        f"(n2:{self._node_label} {{id:'{obj}'}})"    )    self.conn.run(query=subj_query)    self.conn.run(query=obj_query)    self.conn.run(query=rel_query)

{            "analysis": {"analyzer": {"default": {"type": "standard"}}},            "similarity": {                "custom_bm25": {                    "type": "BM25",                    "k1": self._k1,                    "b": self._b,                }            },        }        self._es_mappings = {            "properties": {                "content": {                    "type": "text",                    "similarity": "custom_bm25",                },                "metadata": {                    "type": "keyword",                },            }        }

- FullTextStoreBase    - ElasticDocumentStore    - OpenSearchStore

class EmbeddingRetriever(BaseRetriever):    """Embedding retriever."""
    def __init__(        self,        index_store: IndexStoreBase,        top_k: int = 4,        query_rewrite: Optional[QueryRewrite] = None,        rerank: Optional[Ranker] = None,        retrieve_strategy: Optional[RetrieverStrategy] = RetrieverStrategy.EMBEDDING,    ):

    async def _aretrieve_with_score(        self,        query: str,        score_threshold: float,        filters: Optional[MetadataFilters] = None,    ) -> List[Chunk]:        """Retrieve knowledge chunks with score.
        Args:            query (str): query text            score_threshold (float): score threshold            filters: metadata filters.        Return:            List[Chunk]: list of chunks with score        """        queries = [query]        new_queries = await self._query_rewrite.rewrite(                    origin_query=query, context=context, nums=1                )                queries.extend(new_queries)        candidates_with_score = [                self._similarity_search_with_score(                    query, score_threshold, filters, root_tracer.get_current_span_id()                )                for query in queries            ]            ...
        new_candidates_with_score = await self._rerank.arank(                new_candidates_with_score, query            )            return new_candidates_with_score

class FilterCondition(str, Enum):    """Vector Store Meta data filter conditions."""    AND = "and"    OR = "or"class MetadataFilter(BaseModel):    """Meta data filter."""
    key: str = Field(        ...,        description="The key of metadata to filter.",    )    operator: FilterOperator = Field(        default=FilterOperator.EQ,        description="The operator of metadata filter.",    )    value: Union[str, int, float, List[str], List[int], List[float]] = Field(        ...,        description="The value of metadata to filter.",    )

KEYWORD_EXTRACT_PT = (    "A question is provided below. Given the question, extract up to "    "keywords from the text. Focus on extracting the keywords that we can use "    "to best lookup answers to the question.n"    "Generate as more as possible synonyms or alias of the keywords "    "considering possible cases of capitalization, pluralization, "    "common expressions, etc.n"    "Avoid stopwords.n"    "Provide the keywords and synonyms in comma-separated format."    "Formatted keywords and synonyms text should be separated by a semicolon.n"    "---------------------n"    "Example:n"    "Text: Alice is Bob's mother.n"    "Keywords:nAlice,mother,Bob;mummyn"    "Text: Philz is a coffee shop founded in Berkeley in 1982.n"    "Keywords:nPhilz,coffee shop,Berkeley,1982;coffee bar,coffee housen"    "---------------------n"    "Text: {text}n"    "Keywords:n")
def explore(    self,    subs: List[str],    direct: Direction = Direction.BOTH,    depth: Optional[int] = None,    fan: Optional[int] = None,    limit: Optional[int] = None,) -> Graph:    """Explore on graph."""

   def _similarity_search(        self, query, filters: Optional[MetadataFilters] = None    ) -> List[Chunk]:        """Similar search."""        table_chunks = self._table_vector_store_connector.similar_search_with_scores(            query, self._top_k, 0, filters        )
        not_sep_chunks = [            chunk for chunk in table_chunks if not chunk.metadata.get("separated")        ]        separated_chunks = [            chunk for chunk in table_chunks if chunk.metadata.get("separated")        ]        if not separated_chunks:            return not_sep_chunks
        # Create tasks list        tasks = [            lambda c=chunk: self._retrieve_field(c, query) for chunk in separated_chunks        ]        # Run tasks concurrently        separated_result = run_tasks(tasks, concurrency_limit=3)
        # Combine and return results        return not_sep_chunks + separated_result

二、知识加工，知识检索优化思路

async def aretrieve(    self, query: str, filters: Optional[MetadataFilters] = None) -> List[Chunk]:    """Retrieve knowledge chunks.
        Args:            query (str): async query text.            filters: (Optional[MetadataFilters]) metadata filters.
        Returns:            List[Chunk]: list of chunks        """    return await self._aretrieve(query, filters)

class RetrieverChain(BaseRetriever):    """Retriever chain class."""
    def __init__(        self,        retrievers: Optional[List[BaseRetriever]] = None,        executor: Optional[Executor] = None,    ):        """Create retriever chain instance."""        self._retrievers = retrievers or []        self._executor = executor or ThreadPoolExecutor()            for retriever in self._retrievers:            candidates_with_scores = await retriever.aretrieve_with_scores(                query=query, score_threshold=score_threshold, filters=filters            )            if candidates_with_scores:                return candidates_with_scores

## Rerank model#RERANK_MODEL=bce-reranker-base#### If you not set RERANK_MODEL_PATH, DB-GPT will read the model path from EMBEDDING_MODEL_CONFIG based on the RERANK_MODEL.#RERANK_MODEL_PATH=/Users/chenketing/Desktop/project/DB-GPT-NEW/DB-GPT/models/bce-reranker-base_v1#### The number of rerank results to return#RERANK_TOP_K=5

score = 0.0for q in queries:    if d in result(q):        score += 1.0 / ( k + rank( result(q), d ) )return score
# where# k is a ranking constant# q is a query in the set of queries# d is a document in the result set of q# result(q) is the result set of q# rank( result(q), d ) is d's rank within the result(q) starting from 1

基于以下给出的已知信息, 准守规范约束，专业、简要回答用户的问题.规范约束:    1.如果已知信息包含的图片、链接、表格、代码块等特殊markdown标签格式的信息，确保在答案中包含原文这些图片、链接、表格和代码标签，不要丢弃不要修改，如:图片格式：![image.png](xxx), 链接格式:[xxx](xxx), 表格格式:|xxx|xxx|xxx|, 代码格式:```xxx```.    2.如果无法从提供的内容中获取答案, 请说: "知识库中提供的内容不足以回答此问题" 禁止胡乱编造.    3.回答的时候最好按照1.2.3.点进行总结, 并以markdwon格式显示.

EvaluationMetric├── LLMEvaluationMetric│   ├── AnswerRelevancyMetric├── RetrieverEvaluationMetric│   ├── RetrieverSimilarityMetric│   ├── RetrieverMRRMetric│   └── RetrieverHitRateMetric

三、RAG落地案例分享

四、总结