如何实现参加RAG比赛但进不了复赛的总结

今天写这篇文章主要就是总结一下我使用的一些基本方法，虽然肯定比不上前十的大佬们的操作，但对于常规RAG实现来说也是够用的。这次的考题是给了一堆HTML的知识文档，基于这些文档来进行知识问答。这些文档是企业内部的运维相关文档，里面的内容我都看不太懂，包括有些题目我人工也没找到正确答案，主要还是依赖RAG基本实现和LLM的能力来进行解答。

知识库处理

首先第一步是对官方提供的这些HTML文件进行处理，它的根目录下有个xml文件，类似于目录的效果，我也是基于这个目录来进行文件夹的遍历的。当然我觉得如果直接遍历文件夹的每个HTML文件应该也不是不行，只是HTML文件里面包含了很多类似目录一样的页面，这些对于构建我的向量库来说作用不是很大，但如果构建知识图谱的话，我觉得还是很有用的，但我对知识图谱是在约等于一无所知，就放弃了这些数据。

HTML文件的分段处理有很多种方法，在 langchain 里面就有很多用于分段的工具，比如直接分割HTML的HTMLHeaderTextSplitter ，有递归分割的 RecursiveCharacterTextSplitter ，针对Markdown 文件的 MarkdownHeaderTextSplitter 等。我这里是将HTML处理成Markdown后使用 Markdown 的分割器进行分割的。但我这里做了一些特殊处理：

1. HTML转Markdown之前的特殊处理

首先，观察HTML里面的代码，找到适合作为标题的标签对应的class，将这些元素的标签转换成h1 和 h2 这种一二级标题的常规HTML标签。在原HTML中，用的不是这种标签，会导致我转换Markdown的时候丢失标题的标记。

此外，找到HTML中表格相关标签，因为我使用的html2text库进行的html转换，并不能很好的处理表格，因此我这里是对于表格标签放置了特殊标记，然后转换的过程中对于特殊标记进行了转换。实现代码如下：

def html_to_markdown(dst_url):  
    try:  
        with open(dst_url, 'r', encoding='utf-8') as f:  
            html_content = f.read()  
    except UnicodeDecodeError:  
        with open(dst_url, 'r', encoding='gb2312') as f:  
            html_content = f.read()  
    # 解析HTML内容  
    soup_root = BeautifulSoup(html_content, 'html.parser')  
  
    body = soup_root.find('body')  
    soup = BeautifulSoup(str(body), 'html.parser')  
  
    # 根据class属性修改HTML结构  
    for element in soup.find_all(class_=["title", "topictitle"]):  
        if "topictitle" in element.get("class", []):  
            element.name = "h1"  # 将class为title的标签转换为<h1>  
        elif "title" in element.get("class", []):  
            element.name = "h2"  # 将class为topictitle的标签转换为<h1>  
  
    # 处理表格部分  
    markdown_tables = {}  
    table_id = 0  
    for table in soup.find_all('table'):  
        markdown_table = convert_table_to_markdown(table)  
        placeholder = f"[[TABLE_{table_id}]]"  
        markdown_tables[placeholder] = markdown_table  
        table.replace_with(soup.new_string(placeholder))  
        table_id += 1  
  
    # 使用html2text处理剩余的HTML内容  
    h = html2text.HTML2Text()  
    h.ignore_links = True  
    markdown = h.handle(str(soup))  
  
    # 替换占位符为转换后的Markdown表格  
    for placeholder, markdown_table in markdown_tables.items():  
        markdown = markdown.replace(placeholder, markdown_table)  
  
    # 移除多余的空行  
    markdown = '\n'.join([line for line in markdown.split('\n') if line.strip() != ''])  
  
    return markdown  
  
  
def convert_table_to_markdown(table):  
    rows = table.find_all('tr')  
    markdown = []  
  
    for row in rows:  
        cols = row.find_all(['th', 'td'])  
        col_text = [col.get_text(strip=True) for col in cols]  
        markdown.append('| ' + ' | '.join(col_text) + ' |')  
  
    # 添加表头分隔符  
    if len(rows) > 1:  
        header_cols = rows[0].find_all(['th', 'td'])  
        header_separator = '| ' + ' | '.join(['---'] * len(header_cols)) + ' |'  
        markdown.insert(1, header_separator)  
  
    # 将表格内容用换行符连接起来  
    return '\n'.join(markdown) + '\n\n'  # 添加两个换行符

2. 关于分段

上面说了我使用的分割器是 MarkdownHeaderTextSplitter ，这种可以会把段落内容和标题放不同的字段返回，标题是作为拆分后的元数据存在的，这个切分方式相对于直接使用 RecursiveCharacterTextSplitter 的好处在于能知道标题和段落的关系，对于我数据存储的时候，可以标记每个段落对应的标题是什么，同样的如果构建知识图谱，我想这也是一种必要的手段。

第二点要注意的跟上面预处理html类似，就是要额外处理表格部分。因为我们知道，一个段落是不能太长，否则会影响搜索以及作为背景知识给LLM的时候很容易超长，那么一般都会设置一个最大值，我这里针对段落大于最大值的会特殊处理，会逐行读取，避免超过最大值的情况。普通段落倒是好说，表格如果也这么处理就会导致表格数据割裂，因为后面的数据都没有表头了，自然就变成了脏数据。因此这里遇到表格的时候，我会将一个大表格拆分成N个小表格，但每个小表格还是会保留表头，这样如果搜索到了表格的数据，至少会是一个完整的表格形式。我想这个小技巧应该还是很实用的。

3. 知识库数据模型

知识库的存储方案我使用的是ElasticSearch，很多做Java的同学应该对他都不陌生，一些需要搜索引擎的需求都会使用到它。那么把它用到RAG的知识库搜索和存储上自然也是合适的。而且它也是支持向量检索的，只需要设置一下分数计算函数，就可以让返回的score变成向量的余弦相似度，非常方便。

ES的数据建模我设置了挺多的元数据字段，原本是期望这些元数据能帮我提升搜索的准确性，但后来还是没有用上，但我觉得是我使用的方式不对，因此我也贴一下：

class DataModel:  
    def __init__(self, root: str, name: str, content: str, url: str, doctype: str, catalogs: [str], keywords: [str],  
                 vector: [float], titles: [str] = [], parent: str = '', seg_index: int = 0):  
        # 根目录名称
        self.root = root  
        # 文档名称（不是文件名称）  
        self.name = name  
        # 文本内容  
        self.content = content  
        # 文档路径
        self.url = url  
        # 文档类型  
        self.doctype = doctype  
        # 目录，从上至下  
        self.catalogs = catalogs  
        # 关键词，目录也作为关键词存在  
        self.keywords = keywords  
        # 向量  
        self.vector = vector  
        # 标题，从1级到2级  
        self.titles = titles  
        # 父标题  
        self.parent = parent  
        # 在这个标题下的段落序号  
        self.seg_index = seg_index

初版实现

基于上面的处理过的知识库就已经可以实现RAG了，将query向量化并进行向量匹配即可，此外本次试题里面是标注了每个问题是来自于哪个根目录，因此可以es搜索的时候额外加上根目录筛选的条件，对于搜索范围是小了很多的。

def search_by_vector(query_vector, root_value, top_n=10):  
    query = {  
        "size": top_n,  
        "query": {  
            "bool": {  
                "must": [  
                    {  
                        "script_score": {  
                            "query": {  
                                "match_all": {}  
                            },  
                            "script": {  
                                "source": "(cosineSimilarity(params.query_vector, 'vector') + 1.0) / 2",  
                                "params": {  
                                    "query_vector": query_vector  
                                }  
                            }                        }                    }                ],  
                "filter": [  
                    {  
                        "term": {  
                            "root.keyword": root_value  
                        }  
                    }                ]  
            }  
        }    }  
    response = es.search(index=index_name, body=query)  
    return response

es中要想使用使用相似度作为分数，可以在搜索语句里面加上script，将相似度的计算方式放进去，同样为了使得分数都是大于0，公式使用（similarity+1）/ 2 即可。这种Indexing->Retrieval ->Query 的方式就是最常见的RAG流程，可以称作Naive RAG.

检索优化

基于上面这个策略的到的结果只是堪堪及格，这倒是很合理的结果，虽然我一开始以为及格都难，可能是分段策略还凑合，所以结果没那么差。如果想自己本地部署一个RAG系统，我是觉得只要做好分段这一块，就能解决六成以上的问题了，因为如果要采取一些其他优化策略，势必会对硬件资源以及响应时间有很大影响，有时候没有这个必要。

那么现在如果想再提升效果，有什么简单又快速的方法呢？

我这里首先想到的还是检索优化，因为分段以及搜索使用向量检索的原因，如果一个段落的上下文实际上和当前段落是紧密相关的，但跟query的相似度又不高，很容易导致一些关键信息的丢失，尤其是我看到有些文档实际上是针对某个概念的解释，那如果只是检索到其中一个段落，很容易丢失关键信息。现在大家看到的上面的DataModel这个类实际上一开始我是没有设计seg_index这个字段的。是在这次检索优化中使用的，当检索到某个段落的时候，会带上它前面的n段和后面的m段，我实际使用的n=m=1。并且为了防止结果的list里面存在重复的段落，要记得进行段落的去重，不然很容易背景知识的list里面实际上都是重复的内容。

def retrieve(query: str, document: str, top_n=10):  
    vec = embedding.embedding(query)  
    kg = es.search_by_vector(vec, document, top_n=top_n)  
    hits = kg['hits']['hits']  
    # 找到对应的上下文  
    combines = []  
    for hit in hits:  
        _id = hit['_id']  
        source = hit['_source']  
        score = hit['_score']  
        url = source['url']  
        hit_content = source['content']  
        seg_index = source['seg_index']  
        parent = source['parent']  
        current_hit = {'id': _id, 'content': hit_content}  
        # print(f'{url}, {score}, {hit_content}')  
  
        if seg_index == 0:  
            query_index = [1]  
        else:  
            query_index = [seg_index - 1, seg_index + 1]  
  
        context = []  
        for index in query_index:  
            results = es.search_documents(url, parent, index)  
            # 相邻结果  
            if len(results['hits']['hits']) == 0:  
                continue  
  
            near_hit = results['hits']['hits'][0]  
            near_id = near_hit['_id']  
            near_content = near_hit['_source']['content']  
  
            content = {'id': near_id, 'content': near_content}  
            context.append(content)  
  
        if len(context) > 0:  
            if len(query_index) > 1:  
                if len(context) == 1:  
                    combine = [context[0], current_hit]  
                else:  
                    combine = [context[0], current_hit, context[1]]  
            else:  
                combine = [current_hit, context[0]]  
            combines.append(combine)  
        else:  
            combines.append([current_hit])  
    # 合并重复段落  
    distinct_results = merge_combinations(combines)  
    distinct_contents = ["\n".join(item['content'] for item in sublist) for sublist in distinct_results]  
    return distinct_contents  
  
  
def merge_combinations(combines):  
    def find_combination_with_id(combinations, target_id):  
        for combination in combinations:  
            if any(item['id'] == target_id for item in combination):  
                return combination  
        return None  
  
    merged_combinations = []  
  
    for combination in combines:  
        current_combination = []  
        for item in combination:  
            existing_combination = find_combination_with_id(merged_combinations, item['id'])  
            if existing_combination:  
                # 合并当前组合中的元素到已存在的组合中  
                existing_combination.extend(x for x in combination if x not in existing_combination)  
                break  
        else:  
            # 如果没有找到包含当前id的组合，则添加新的组合  
            merged_combinations.append(combination)  
  
    return merged_combinations

基于这个策略，结果会比前面的版本好不少，这也是我最终分数的策略，听起来挺好笑的，这是我开赛第一周就拿到的结果，当时还排名靠前，但后面两周做的所有优化反而还不如这个Naive RAG 的版本，但比赛这玩意就是不进则退，等到最后比赛排名就很垃圾了。虽然心有不甘，但谁让自己太菜了呢。

Query 优化

接下来讲讲我都做了哪些优化，虽然没做好，但思想应该是对的，只是策略使用的方式不对，所以理论不等于实践，要想真的学习好还是要多进行实践，哪怕失败了，这些失败的经验对自己的成长也是有帮助的。因此我也给大家分享一下这些策略。

首先是Query 方面的修改，常见的策略有很多，比如Query扩写改写、HyDE、问题拆解、提取关键词进行查询等。我这里尝试了HyDE, 问题拆解以及提取关键词的策略。对于改写Query的策略为什么不使用，因为我觉得一般来说是提问比较不精确的时候可以使用，但这次的题目都是比较明确的问题，因此没必要进行改写或者扩写。

问题拆解

对于问题拆解，其实就是将问题拆分成多个子问题，比如张三在24年的奥林匹克数学竞赛上有没有超过李四？ 可以拆解成：

24年奥林匹克数学竞赛成绩排名
张三比赛中成绩排名是多少
李四比赛中成绩排名是多少

通过综合这几个问题的查询结果可以得到最终张三是否超过了李四。但也有一些无法拆解的问题，比如张三的数学成绩是多少？那么其实只需要搜索张三的分数即可，这种情况的子查询就等于它原本的问题。对于问题拆解，可以直接使用LLM来进行拆分，我使用的prompt如下：

你是一名顶级运维工程师，可以针对用户的输入问题生成多个子查询问题，每个问题独立一行输出。  
首先你需要判断用户是否真的问了多个问题，如果没有，你就原样输出用户问题；  
如果用户真的询问了多个问题，请你拆解成多个子问题。  
  
重要提示：  
- 不要添加任何解释和文本。

这个拆分的效果还有待斟酌，有时候会拆出一些奇奇怪怪的问题，可以通过LLM的反思等策略进行过滤。

HyDE(Hypothetical Document Embeddings)

我使用的prompt比较简单，忘了是从Langchain还是llamaIndex里面薅的了。

请写一段话回答问题  
尽量包含关键细节。  
  
{content}

这个策略使用下来的感受就是，它可能不太适合知识过于私有化的情况，就是你的问题和答案几乎不可能存在于互联网上的那种，全是公司特有名词的知识。总之如果想单独使用这个策略的话，效果会非常差，建议如果想用的话，要考虑结合其他策略来进行进一步的知识过滤，否则很容易降低效率和准确度。

关键词搜索

对于关键词搜索这个我尝试了两种方式，一种准确来讲不是关键词搜索，而是直接使用ES的全文搜索能力，ES是支持使用其他的分词器的，我使用的中文支持比较好的ik分词器。ES进行搜索的时候如果使用 match 的方式，就会对Query进行分词搜索而不是完全匹配，我的理解这是跟关键词检索有点类似，这也是我多路召回的其中一路。

另外一种方式就是使用大模型进行了关键词提取，无论是Query和段落都要进行提取，上面的 DataModel 也能看到我是留了关键词这个字段的，一开始预处理数据的时候，关键词就是目录名称，但这肯定是不够的，因此我用大模型针对每条数据又进行了新的关键词补充。

你是一名运维技术专家，能阅读并理解运维相关的技术文档，熟悉当前市面上的各种运维产品，对于常见的品牌如华为/中兴等的硬件设备都很熟悉。  
现在我将会给你发送一些运维文档的段落内容，你需要从段落中提取这段内容的关键词，并遵守以下规则：  
1. 多个关键词用英文逗号隔开，如 关键词1,关键词2,关键词3  
2. 关键词必须在原文中出现过，不可以随便臆造  
3. 允许出现某个关键词包含了另一个关键词的情况，举个例子：高等数学,数学。这两个关键词有包含关系，但允许同时出现。  
请务必按照规则给我提取关键词。

数据处理完之后，查询时先让LLM提取出Query的关键词，然后使用关键词进行匹配得到一些段落。在某些情况下，关键词可以召回一些向量相似度低但实际很重要的知识，因为embedding的模型使用的是通用模型，对于一些私有化知识的embedding效果并不一定那么好，而且向量相似度的高低并不完全等价于语义相似度，可能两句语义完全相反的内容但相似度却也很高。

搜索结果处理

除了重排序，我还尝试了另一个方式，就是利用大模型的反思来过滤文档，这个方法怎么说呢，我觉得我使用的方式大概率是错误的，即让模型来判断段落能否支撑它来进行问答：

我有一段关于运维的材料的文本，内容如下:  
  
{content}  
  
然后我现在需要根据上述文本内容回答一个问题如下：  
{question}  
  
你觉得依靠这些内容能回答这个问题么？如果能，回复是；如果不能，回复否。  
  
重要提示：  
 - 不要添加任何解释和文本。

我做的最错误的可能是对于每个段落都让它去判断了，因为有时候一个问题需要多个段落才能判断的，那么可能对于很多实际有价值的都会返回否。而对于能回答的段落，也没啥过滤的必要，这个策略使用的很失败，因此我最终版的代码也是完全没用上的。要利用反思来增强RAG效果，更有效的应该是使用类似SelfRAG这样的框架，而不是简单的让LLM去判断。

知识图谱

最后，我要提一下我觉得最最最重要的方式，就是结合知识图谱去做RAG，我本次没有成功实现出来，所以很难给出太多的分享，主要还是这方面几乎是小白，学习起来没那么快，但也觉得它一定是当前将RAG做到极致的最佳方式。

RAG最难的问题是什么？我觉得就是检索，无论是query改写、重排序、反思等，都是为了让LLM能排除掉错误信息，只拿到最精准的文档来进行问答。最麻烦的场景就是多跳问题，即问题的答案存在于多个文档或段落中，甚至你需要通过推理才能得到应当要查询哪些段落。

在llamaIndex里面我有找到使用LLM提取知识图谱的方式，但我尝试提取了里面的prompt然后使用LLM去创建知识图谱，效果并不好，而且不知道是不是段落太长，提取速度也比较慢，下面是我提取的prompt翻译后的中文版：

你是一个顶级算法工程师，旨在从结构化格式的文本中提取信息，以构建知识图谱。你的任务是从给定的文本中识别用户提示中请求的实体和关系。  
你必须生成包含JSON对象列表的输出。每个对象应具有以下键：“head”、“head_type”、“relation”、“tail”和“tail_type”。  
“head”键必须包含从提供的列表中提取的实体文本。  
“head_type”键必须包含提取的head实体的类型  
“relation”键必须包含head和tail之间关系的类型  
“tail”键必须表示提取实体的文本，该实体是关系的tail  
“tail_type”键必须包含提取的tail实体的类型  
  
尝试提取尽可能多的实体和关系。保持实体一致性：在提取实体时，确保一致性非常重要。如果一个实体（例如“John Doe”）在文本中多次提到，但使用不同的名称或代词（例如“Joe”、“他”），始终使用最完整的标识符来表示该实体。知识图谱应该是连贯且易于理解的，因此保持实体引用的一致性至关重要。  
重要提示：  
- 不要添加任何解释和文本。

蚂蚁和微软最近都开源了其Graph RAG的框架，感兴趣的可以去看看相关论文以及代码。

如何学习AI大模型？

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述