萬字長文講透RAG在實際落地場景中的最佳化

阿里妹導讀

本文主要圍繞DB-GPT應用開發框架如何在實際落地場景做RAG最佳化。

背景

在過去兩年中，檢索增強生成（RAG，Retrieval-Augmented Generation）技術逐漸成為提升智慧體的核心組成部分。透過結合檢索與生成的雙重能力，RAG能夠引入外部知識，從而為大模型在複雜場景中的應用提供更多可能性。但是在實際落地場景中，往往會存在檢索準確率低，噪音干擾多，召回完整性，專業性不夠，導致LLM幻覺嚴重的問題。本次分享會聚焦RAG在實際落地場景中的知識加工和檢索細節，如何去最佳化RAG Pineline鏈路，最終提升召回準確率。

快速搭建一個RAG智慧問答應用很簡單，但是需要在實際業務場景落地還需要做大量的工作。

本文將主要介紹圍繞DB-GPT應用開發框架(https://github.com/eosphoros-ai/DB-GPT)，如何在實際落地場景做RAG最佳化。

一、RAG關鍵流程原始碼解讀

主要講述在DB-GPT中，知識加工和RAG流程關鍵原始碼實現。

1.1 知識加工

知識載入 -> 知識切片 -> 資訊抽取 -> 知識加工(embedding/graph/keywords) -> 知識儲存

知識載入：透過知識工廠類將不同格式的非結構化文件進行例項化。

# 知識工廠進行例項化KnowledgeFactory -> create() -> load() -> Document- knowledge - markdown - pdf - docx - txt - html - pptx - url - ...

如何擴充套件：透過繼承Knowledge介面，實現load(),support_chunk_strategy(),default_chunk_strategy()等方法

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

知識切片

ChunkManager: 透過載入後的知識資料，根據使用者指定的分片策略和分片引數路由到對應的分片處理器進行分配。

classChunkManager:"""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_strategyif chunk_parameter and chunk_parameter.chunk_strategyelse self._knowledge.default_chunk_strategy().name ) self._text_splitter = self._chunk_parameters.text_splitter self._splitter_type = self._chunk_parameters.splitter_type

如何擴充套件：如果你想在介面上自定義一個新的分片策略

新增切片策略ChunkStrategy
新增Splitter實現邏輯

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", )

知識抽取，目前支援向量抽取，知識圖譜抽取，關鍵詞抽取。

向量抽取 -> embedding, 實現Embeddings介面

@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}

知識圖譜抽取 -> knowledge graph，透過利用大模型提取(實體,關係,實體)三元組結構。

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")

倒排索引抽取 -> keywords分詞

可以用es預設的分詞庫，也可以使用es的外掛模式自定義分詞

知識儲存

整個知識持久化統一實現了IndexStoreBase介面，目前提供了向量資料庫、圖資料庫、全文索引三類實現。

VectorStore，向量資料庫主要邏輯都在load_document()，包括索引schema建立，向量資料分批寫入等等。

- VectorStoreBase - ChromaStore - MilvusStore - OceanbaseStore - ElasticsearchStore - PGVectorStoreclass 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)

GraphStore ，具體的圖儲存提供了三元組寫入的實現，一般會呼叫具體的圖資料庫的查詢語言來完成。例如TuGraphStore會根據三元組生成具體的Cypher語句並執行。

圖儲存介面GraphStoreBase提供統一的圖儲存抽象，目前內建了MemoryGraphStore和TuGraphStore的實現，我們也提供Neo4j介面給開發者進行接入。

- GraphStoreBase - TuGraphStore - Neo4jStoredef 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)

FullTextStore: 透過構建es索引，透過es內建分詞演算法進行分詞，然後由es構建keyword->doc_id的倒排索引。

{"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", }, } }

目前提供的全文索引介面支援Elasticsearch，同時也定義了OpenSearch的介面

- FullTextStoreBase - ElasticDocumentStore - OpenSearchStore

1.2 知識檢索

question -> rewrite -> similarity_search -> rerank -> context_candidates

接下來是知識檢索，目前社群的檢索邏輯主要分為這幾步，如果你設定了查詢改寫引數，目前會透過大模型給你進行一輪問題改寫，然後會根據你的知識加工方式路由到對應的檢索器，如果你是透過向量進行加工的，那就會透過EmbeddingRetriever進行檢索，如果你構建方式是透過知識圖譜構建的，就會按照知識圖譜方式進行檢索，如果你設定了rerank模型，會給粗篩後的候選值進行精篩，讓候選值和使用者問題更有關聯。

EmbeddingRetriever

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 textscore_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

index_store: 具體的向量資料庫
top_k: 返回的具體候選chunk個數
query_rewrite：查詢改寫函式
rerank：重排序函式
query:原始查詢
score_threshold：得分，我們預設會把相似度得分小於閾值的上下文資訊給過濾掉
filters：Optional[MetadataFilters], 元資料資訊過濾器，主要是可以用來前置透過屬性資訊篩掉一些不匹配的候選資訊。

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.", )

Graph RAG

首先透過模型進行關鍵詞抽取，這裡可以透過傳統的nlp技術進行分詞，也可以透過大模型進行分詞，然後進行關鍵詞按照同義詞做擴充，找到關鍵詞的候選列表，最好根據關鍵詞候選列表呼叫explore方法召回區域性子圖。

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;mummy\n""Text: Philz is a coffee shop founded in Berkeley in 1982.\n""Keywords:\nPhilz,coffee shop,Berkeley,1982;coffee bar,coffee house\n""---------------------\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."""

DBSchemaRetriever 這部分是ChatData場景的schema-linking檢索

主要是透過schema-linking方式透過二階段相似度檢索，首先先找到最相關的表，然後再最相關的欄位資訊。

優點：這種二階段檢索也是為了解決社群反饋的大寬表體驗的問題。

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 ifnot chunk.metadata.get("separated") ] separated_chunks = [ chunk for chunk in table_chunks if chunk.metadata.get("separated") ]ifnot 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 andreturn resultsreturn not_sep_chunks + separated_result

table_vector_store_connector: 負責檢索最相關的表。
field_vector_store_connector: 負責檢索最相關的欄位。

二、知識加工，知識檢索最佳化思路

目前RAG智慧問答應用幾個痛點：

知識庫文件越來越多以後，檢索噪音大，召回準確率不高
召回不全，完整性不夠
召回和使用者問題意圖相關性不大
只能回答靜態資料，無法動態獲取知識，導致答疑應用比較呆，比較笨。

2.1知識處理最佳化

非結構化/半結構化/結構化資料的處理，準備決定著RAG應用的上限，因此首先需要在知識處理，索引階段做大量的細粒度的ETL工作，主要最佳化的思路方向：

非結構化 -> 結構化：有條理地組織知識資訊。
提取更加豐富的, 多元化的語義資訊。

2.1.1 知識載入

目的：需要對文件進行精確的解析，更多元化的識別到不同型別的資料。

最佳化建議：

建議將docx、txt或者其他文字事先處理為pdf或者markdown格式，這樣可以利用一些識別工具更好的提取文字中的各項內容。
提取文字中的表格資訊。
保留markdown和pdf的標題層級資訊，為接下來的層級關係樹等索引方式準備。
保留圖片連結，公式等資訊，也統一處理成markdown的格式。

2.1.2 切片Chunk儘量保持完整

目的：儲存上下文完整性和相關性，這直接關乎回覆準確率。

保持在大模型的上下文限制內，分塊保證輸入到LLMs的文字不會超過其token限制。

最佳化建議：

圖片 + 表格單獨抽取成Chunk，將表格和圖片標題保留到metadata元資料裡。
文件內容儘量按照標題層級或者Markdown Header進行拆分，儘可能保留chunk的完整性。
如果有自定義分隔符可以按照自定義分割符切分。

2.1.3 多元化的資訊抽取

除了對文件進行Embedding向量抽取外，其他多元化的資訊抽取能夠對文件進行資料增強，顯著提升RAG召回效果。

知識圖譜

優點：1. 解決NativeRAG的完整性缺失，依然存在幻覺問題，知識的準確性，包括知識邊界的完整性、知識結構和語義的清晰性，是對相似度檢索的能力的一種語義補充。
適用場景：適用於嚴謹的專業領域(醫療，運維等)，知識的準備需要受到約束的並且知識之間能夠明顯建立層級關係的。
如何實現：

1.依賴大模型提取(實體,關係,實體)三元組關係。
2. 依賴前期高質量，結構化的知識準備，清洗，抽取，透過業務規則透過手動或者自定義SOP流程構建知識圖譜。

Doc Tree

適用場景：解決了上下文完整性不足的問題，也能匹配時完全依據語義和關鍵詞，能夠減少噪音
如何實現：以標題層級構建chunk的樹形節點，形成一個多叉樹結構，每一層級節點只需要儲存文件標題，葉子節點儲存具體的文字內容。這樣利用樹的遍歷演算法，如果使用者問題命中相關非葉子標題節點，就可以將相關的子節點資料進行召回。這樣就不會存在chunk完整性缺失的問題。

這部分的Feature我們也會在明年年初放到社群裡面。

提取QA對，需要前置透過預定義或者模型抽取的方式提取QA對資訊

適用場景：

能夠在檢索中命中問題並直接進行召回，直接檢索到使用者想要的答案，適用於一些FAQ場景，召回完整性不夠的場景。

如何實現：

預定義:預先為每個chunk新增一些問題
模型抽取:透過給定一下上下文，讓模型進行QA對抽取

元資料抽取

如何實現：根據自身業務資料特點，提取資料的特徵進行保留，比如標籤，類別，時間，版本等元資料屬性。
適用場景：檢索時候能夠預先根據元資料屬性進行過濾掉大部分噪音。

總結提取

適用場景：解決這篇文章講了個啥，總結一下等全域性問題場景。
如何實現：透過mapreduce等方式分段抽取，透過模型為每段chunk提取摘要資訊。

2.1.4 知識處理工作流

目前DB-GPT知識庫提供了文件上傳 -> 解析 -> 切片 -> Embedding -> 知識圖譜三元組抽取 -> 向量資料庫儲存 -> 圖資料庫儲存等知識加工的能力，但是不具備對文件進行復雜的個性化的資訊抽取能力，因此希望透過構建知識加工工作流模版來完成複雜的，視覺化的，使用者可自定義的知識抽取，轉換，加工流程。

知識加工工作流：https://www.yuque.com/eosphoros/dbgpt-docs/vg2gsfyf3x9fuglf

2.2 RAG流程最佳化

RAG流程的最佳化我們又分為了靜態文件的RAG和動態資料獲取的RAG，目前大部分涉及到的RAG只覆蓋了非結構化的文件靜態資產，但是實際業務很多場景的問答是透過工具獲取動態資料 + 靜態知識資料共同回答的場景，不僅需要檢索到靜態的知識，同時需要RAG檢索到工具資產庫裡面工具資訊並執行獲取動態資料。

2.2.1靜態知識RAG最佳化

2.2.1.1 原始問題處理

目的：澄清使用者語義，將使用者的原始問題從模糊的，意圖不清晰的查詢最佳化為含義更豐富的一個可檢索的Query

原始問題分類，透過問題分類可以

LLM分類(LLMExtractor)
構建embedding+邏輯迴歸實現雙塔模型，text2nlu DB-GPT-Hub/src/dbgpt-hub-nlu/README.zh.md at main · eosphoros-ai/DB-GPT-Hub

tip:需要高質量的Embedding模型，推薦bge-v1.5-large

反問使用者，如果語義不清晰將問題再拋給使用者進行問題澄清，透過多輪互動

透過熱搜詞庫根據語義相關性給使用者推薦他想要的問題候選列表

槽位提取，目的是獲取使用者問題中的關鍵slot資訊，比如意圖，業務屬性等等

LLM提取(LLMExtractor)

問題改寫

熱搜詞庫進行改寫
多輪互動

2.2.1.2 元資料過濾

當我們把索引分成許多chunks並且都儲存在相同的知識空間裡面，檢索效率會成為問題。比如使用者問"浙江我武科技公司"相關資訊時，並不想召回其他公司的資訊。因此，如果可以透過公司名稱元資料屬性先進行過濾，就會大大提升效率和相關度。

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)

2.2.1.3 多策略混合召回

按照優先順序召回，分別為不同的檢索器定義優先順序，檢索到內容後立即返回

定義不同檢索，比如qa_retriever, doc_tree_retriever寫入到佇列裡面，透過佇列的先進先出的特性實現優先順序召回。

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

多知識索引/空間並行召回

透過知識的不同索引形式，透過並行召回方式獲取候選列表，保證召回完整性。

2.2.1.4 後置過濾

經過粗篩候選列表後，怎麼透過精篩過濾噪音呢

無關的候選分片剔除

時效性剔除
業務屬性不滿足剔除

topk去重
重排序僅僅靠粗篩的召回還不夠，這時候我們需要有一些策略來對檢索的結果做重排序，比如把組合相關度、匹配度等因素做一些重新調整，得到更符合我們業務場景的排序。因為在這一步之後，我們就會把結果送給LLM進行最終處理了，所以這一部分的結果很重要。

使用相關重排序模型進行精篩，可以使用開源的模型，也可以使用帶業務語義微調的模型。

## Rerank model#RERANK_MODEL=bce-reranker-base

#### If you notset 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

根據不同索引召回的內容進行業務RRF加權綜合打分剔除

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

2.2.1.5 顯示最佳化+兜底話術/話題引導

讓模型使用markdown的格式進行輸出

基於以下給出的已知資訊, 準守規範約束，專業、簡要回答使用者的問題.規範約束:

1.如果已知資訊包含的圖片、連結、表格、程式碼塊等特殊markdown標籤格式的資訊，確保在答案中包含原文這些圖片、連結、表格和程式碼標籤，不要丟棄不要修改，

如:圖片格式：![image.png](xxx), 連結格式:[xxx](xxx), 表格格式:|xxx|xxx|xxx|, 程式碼格式:```xxx```.

2.如果無法從提供的內容中獲取答案, 請說: "知識庫中提供的內容不足以回答此問題" 禁止胡亂編造.

3.回答的時候最好按照1.2.3.點進行總結, 並以markdwon格式顯示.

2.2.2 動態知識RAG最佳化

文件類知識是相對靜態的，無法回答個性化以及動態的資訊，需要依賴一些第三方平臺工具才可以回答，基於這種情況，我們需要一些動態RAG的方法，透過工具資產定義 -> 工具選擇 -> 工具校驗 -> 工具執行獲取動態資料

2.2.2.1 工具資產庫

構建企業領域工具資產庫，將散落到各個平臺的工具API，工具指令碼進行整合，進而提供智慧體端到端的使用能力。比如，除了靜態知識庫以外，我們可以透過匯入工具庫的方式進行工具的處理。

2.2.2.2 工具召回

工具召回沿用靜態知識的RAG召回的思路，再透過完整的工具執行生命週期來獲取工具執行結果。

槽位提取：透過傳統nlp獲取LLM將使用者問題進行解析，包括常用的業務型別，標籤，領域模型引數等等。
工具選擇：沿用靜態RAG的思路召回，主要有兩層，工具名召回和工具引數召回。

工具引數召回，和TableRAG思路類似，先召回表名，再召回欄位名。

引數填充：需要根據召回的工具引數定義，和槽位提取出來的引數進行match

可以程式碼進行填充，也可以讓模型進行填充。
最佳化思路：由於各個平臺工具的同樣的引數的引數名沒有統一，也不方便去治理，建議可以先進行一輪領域模型資料擴充，拿到整個領域模型後，需要的引數都會存在。

引數校驗

完整性校驗：進行引數個數完整性校驗
引數規則校驗：進行引數名型別，引數值，列舉等等規則校驗。

引數糾正/對齊，這部分主要是為了減少和使用者的互動次數，自動化完成使用者引數錯誤糾正，包括大小寫規則，列舉規則等等。eg:

2.2.3 RAG評測

在評估智慧問答流程時，需要單獨對召回相關性準確率以及模型問答的相關性進行評估，然後再綜合考慮，以判斷RAG流程在哪些方面仍需改進。

評價指標：

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

RAG召回指標(RetrieverEvaluationMetric)：

RetrieverHitRateMetric:命中率衡量的是RAG retriever召回出現在檢索結果前top-k個文件中的比例。
RetrieverMRRMetric: Mean Reciprocal Rank透過分析最相關文件在檢索結果裡的排名來計算每個查詢的準確性。更具體地說，它是所有查詢的相關文件排名倒數的平均值。例如，若最相關的文件排在第一位，其倒數排名為 1；排在第二位時，為 1/2；以此類推。
RetrieverSimilarityMetric: 相似度指標計算，計算召回內容與預測內容的相似度。

模型生成答案指標:

AnswerRelevancyMetric:智慧體答案相關性指標，透過智慧體答案與使用者提問的匹配程度。高相關性的答案不僅要求模型能夠理解使用者的問題，還要求其能夠生成與問題密切相關的答案。這直接影響到使用者的滿意度和模型的實用性。

RAG評測教程參考：

評估(Evaluation)https://www.yuque.com/eosphoros/dbgpt-docs/czgl7bsfclc1xsmh

三、RAG落地案例分享

3.1資料基礎設施領域的RAG

3.1.1運維智慧體背景

在資料基礎設施領域，有很多運維SRE，每天會接收到大量的告警，因此很多時間來需要響應應急事件，進而進行故障診斷，然後故障覆盤，進而進行經驗沉澱。另外一部分時間又需要響應使用者諮詢，需要他們用他們的知識以及三方平臺工具使用經驗進行答疑。

因此我們希望透過打造一個數據基礎設施的通用智慧體來解決告警診斷，答疑的這些問題。

3.1.2嚴謹專業的RAG

傳統的 RAG + Agent 技術可以解決通用的，確定性沒那麼高的，單步任務場景。但是面對資料基礎設施領域的專業場景，整個檢索過程必須是確定，專業和真實的，並且是需要一步一步推理的。

右邊是一個透過NativeRAG的一個泛泛而談的總結，可能對於一個普通的使用者，對專業的領域知識沒那麼瞭解時，可能是有用的資訊，但是這部分對於資料基礎設施領域的專業人士，就沒有什麼意義了。因此我們比較了通用的智慧體和資料基礎設施智慧體在RAG上面的區別：

通用的智慧體：傳統的RAG對知識的嚴謹和專業性要求沒那麼高，適用於客服，旅遊，平臺答疑機器人這樣的一些業務場景。
資料基礎設施智慧體：RAG流程是嚴謹和專業的，需要專屬的RAG工作流程，上下文包括(DB告警->根因定位->應急止血->故障恢復)，並且需要對專家沉澱的問答和應急經驗，進行結構化的抽取，建立層次關係。因此我們選擇知識圖譜來作為資料承載。