## 模块 4: 使用多路召回改进你的 RAG AI 应用程序

在模块 3 的开头，我们提到了 RAG 不仅仅可以使用 Embedding 向量作为我们的召回渠道。你可以使用如：

- BoW
- TF-IDF
- Word2Vec

等方法来生成文本向量，随后依旧计算语义相似度。
同时，你也可以引入图数据库，将文本抽取为图后，进行图搜索。换个词来说，GraphRAG。
当然，你也可以用下面我们叙述的方法，使用全文检索（Full-text Search）的方法进行第二路召回。这将给你的 RAG 以另一个角度进行召回，丰富召回的内容。

全文检索，是一类可以让你在大量文本中，快速通过关键词搜索到文本块的技术。常被用于搜索引擎、广告等业务中。一种常见全文检索的实现方式是使用倒排索引。它的核心流程是：

1. 在文本存储的时候，使用分词器将其分开成一个个的词。
2. 以词为 key，文本的 ID 为 value，建立索引
3. 在搜索关键词时，利用此索引，快速检索到文本的 ID 列表

幸运的是，我们不用自己实现这一套流程，在传统方法中，我们可以引入一个搜索引擎。ElasticSearch 就是一个常用的搜索引擎，我们可以使用它作为我们多路召回的底层存储。

### 步骤 1. 运行多路召回 RAG AI 应用程序的 Streamlit 服务

再次创建一个新的终端：

![90%](https://lab-static.pingcap.com/images/2025/7/24/51b77b5a782acbdfb63d2624ab21160896be7451.png)

然后运行新的多路召回 RAG AI 应用程序：

```bash
poetry run streamlit run improved_rag.py --server.port 8503
```

```output
Collecting usage statistics. To deactivate, set browser.gatherUsageStats to false.

You can now view your Streamlit app in your browser.

Local URL: http://localhost:8503
Network URL: http://172.18.0.7:8503
External URL: http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8503
```

### 步骤 2. 使用多路召回 RAG AI 应用程序的演示网页

- 打开您的 [RAG AI 应用程序](http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8503/)。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/3b6ce8c7141941151a9c7310b6b6c0318d7a1a97.png)

- 打开名为 `All Employees in the MySQL`, `All Employees ID and Embedding Mapping in the Milvus` 和 `All Employees in the Elasticsearch` 的文件夹，以查看当前 MySQL, Milvus 和 ElasticSearch 中的数据。由于我们在模块 2 中，在 MySQL 和 Milvus 中保存过数据，当前应在 ElasticSearch 中没有数据，而 MySQL 和 Milvus 存在数据。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/52c3905dc553522fdf12adc567dbf57add628338.png)

- 点击 `Reset` 删除数据。
- 随后可以看到 `All Employees in the MySQL` 和 `All Employees ID and Embedding Mapping in the Milvus` 的文件夹内为空。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/8f6fd05faeaba150c7b117baac22635b8df2622c.png)

- 点击 `Save Data` 保存数据。此时会保存数据到 MySQL，Mulvus 和 ElasticSearch。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/4a2eb8c44477f7528d78e95e153e6ba743f03e44.png)

- 点击 `Retrieval Only` 进行上下文检索。
- 输入问题：

    ```text
    Who is Cheese Wang at PingCAP?
    ```

- 点击 `Send` 按钮开始检索。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/62fd3d7c5e7d26f31cf16e808752d4fd4d9289c2.png)

- 这次的检索，我们使用了两个不同的召回途径，因此既有 Milvus 的 Embedding Vector 召回 ID，又有 ElasticSearch 的 score。ElasticSearch 默认使用基于 TF-IDF 的 BM25 算法计算相似度得分。你可以认为 score 越高，文本越相似。
- 接下来，将 `Select Mode` 选项更改为 `RAG Q&A`，然后点击 `Send`。

    ![90%](https://lab-static.pingcap.com/images/2025/7/24/80284f4bb440ca8bc343b6c0ca800cdce09c0542.png)

> [!NOTE]
> 在这个 Demo 中，多路召回与单路召回的信息相同，这是由于这个示例不够复杂所导致的。此处由于篇幅限制，不会给出证明，但你可以参考以下内容：
>
> - MicroSoft: [Doing RAG? Vector search is \*not\* enough](https://techcommunity.microsoft.com/blog/azuredevcommunityblog/doing-rag-vector-search-is-not-enough/4161073)
> - DataStax: [How to Implement Hybrid Search in RAG Pipelines for LLMs](https://www.datastax.com/guides/hybrid-search-rag-pipelines) （此篇文章提出，与传统的仅向量检索相比，混合搜索的答案精度提高了最多 35%）

### 步骤 3. 多路召回 RAG AI 应用程序代码阅读

让我们回到 [VS Code](${LAB:HTTPS_REVERSE_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage-openai"|urlencode} "VS Code") 并点击代码文件 `improved_rag.py`。

1. 与上一个 RAG 应用相比，脚本的导入部分最主要的变化是增加了 `elasticsearch` 库，用于与 Elasticsearch 服务进行交互。

    ```python
    import os
    import pandas as pd
    from pydantic_core import MultiHostUrl
    from sqlalchemy import func
    from sqlmodel import Field, SQLModel, Session, create_engine, select, delete
    from pymilvus import MilvusClient, DataType
    from elasticsearch import Elasticsearch

    from litellm import completion, embedding
    import streamlit as st

    from typing import List

    from common import LLM_RESPONSE_STYLE, RAG_PROMPT_TEMPLATE
    ```

2. 在配置和初始化部分，除了原有的 MySQL (RDB) 和 Milvus (向量数据库) 的客户端外，新增了 Elasticsearch 客户端的初始化。同时，定义了 Elasticsearch 的索引名称 `ES_INDEX_NAME` 和映射（mapping），该映射指定了存入 Elasticsearch 的文档结构，包括 `id`, `name`, `intro` 等字段及其类型。

    ```python
    # Elasticsearch
    es = Elasticsearch(f"http://{os.getenv('ELASTICSEARCH_HOST')}:{os.getenv('ELASTICSEARCH_PORT')}")
    ES_INDEX_NAME = "employee_id_mapping"
    es_mapping = {
        "mappings": {
            "properties": {
                "id": {"type": "integer"},
                "name": {"type": "text"},
                "intro": {"type": "text"}
            }
        }
    }
    ```

3. 此代码块确保在 Elasticsearch 中存在所需的索引。如果索引不存在，它会使用预定义的 `es_mapping` 创建一个新的索引。

    ```python
    if not es.indices.exists(index=ES_INDEX_NAME):
        es.indices.create(index=ES_INDEX_NAME, body=es_mapping)
    ```

4. 数据库操作的逻辑也进行了扩展。
    - **Reset 按钮**：除了清空 MySQL 和 Milvus 的数据外，现在还会删除并重建 Elasticsearch 中的索引，确保环境的完全重置。
    - **Save Data 按钮**：在将数据存入 MySQL 和 Milvus 之后，会遍历 `employees` 列表，并将每条员工信息作为一个文档索引到 Elasticsearch 中。这使得员工的简介（`intro`）内容可以被全文检索。

    ```python
    if left.button("Reset", type="primary"):
        left.write("Resetting...")
        with Session(engine) as session:
            session.exec(delete(Employee))
            session.commit()

        # Delete the employee id mapping data from Milvus
        milvus_client.delete(
            collection_name=milvus_collection_name,
            ids=[employee.id for employee in employees]
        )

        if es.indices.exists(index=ES_INDEX_NAME):
            es.indices.delete(index=ES_INDEX_NAME)
            es.indices.create(index=ES_INDEX_NAME, body=es_mapping)

        st.rerun()

    if right.button("Save Data"):
        right.write("Saving...")

        # Prepare RDB data
        with Session(engine) as session:
            count = session.exec(select(func.count()).select_from(Employee)).one()

            if count == 0:
                employee_intro_embedding = [
                    {
                        "id": employee.id,
                        "embedding": get_embedding(employee.intro)
                    } for employee in employees
                ]
                session.add_all(employees)
                session.commit()

                # Insert the employee id mapping data into Milvus
                milvus_client.insert(
                    collection_name=milvus_collection_name,
                    data=employee_intro_embedding
                )

                for employee in employees:
                    es.index(
                        index=ES_INDEX_NAME,
                        id=employee.id,
                        document={"id": employee.id, "name": employee.name, "intro": employee.intro}
                    )

                st.rerun()
            else:
                st.error("Data already exists")
    ```

5. 为了展示多路召回的效果，UI 上增加了一个新的可折叠部分，用于显示 Elasticsearch 中存储的所有员工数据。

    ```python
    with st.expander("📁 All Employees in the Elasticsearch", expanded=False):
        response = es.search(index=ES_INDEX_NAME, body={"query": {"match_all": {}}})
        employees = [{"id": item["_id"], "name": item["_source"]["name"], "intro": item["_source"]["intro"]} for item in response["hits"]["hits"]]
        st.dataframe(pd.DataFrame(employees))
    ```

6. 这是多路召回 RAG 应用的核心逻辑，它组合了两种不同的检索方式：
    - **向量检索 (Milvus)**: 与之前一样，将用户问题转换为 embedding，然后在 Milvus 中进行语义相似度搜索，获得第一批候选 ID (`milvus_ids`)。
    - **全文检索 (Elasticsearch)**: 将用户的原始问题作为关键词，在 Elasticsearch 的 `intro` 字段上执行 `match` 查询，进行全文搜索，获得第二批候选 ID (`es_ids`) 和相关性得分。
    - **结果合并**: 将从 Milvus 和 Elasticsearch 两路召回的 ID 列表进行合并，并使用 `set` 去重，得到一个更全面的候选文档 ID 集合。
    - **数据获取与生成**: 使用合并后的 `ids` 从 MySQL 中获取完整的员工信息，然后构建提示并调用 LLM 生成答案。这个过程与之前的 RAG 应用相同，但现在它基于一个更丰富的上下文信息。

    ```python
    if st.button("Send") and query:
        with st.spinner("Processing..."):
            # Retrieve relevant chunks in Milvus
            query_embedding = get_embedding(query)
            res = milvus_client.search(
                collection_name=milvus_collection_name,
                anns_field="embedding",
                data=[query_embedding],
                limit=1,
                search_params={"metric_type": "COSINE"},
                output_fields=["id", "embedding"]
            )

            retrieved_milvus = [{"id": item.entity.id, "embedding": item.entity.embedding} for item in res[0]]
            milvus_ids = [item.entity.id for item in res[0]]

            # Retrieve relevant chunks in Elasticsearch
            es_query = {
                "query": {
                    "match": {
                        "intro": query
                    }
                },
                "size": 1
            }
            res = es.search(index=ES_INDEX_NAME, body=es_query)
            retrieved_es = [{"id": item["_id"], "score": item["_score"]} for item in res["hits"]["hits"]]
            es_ids = [int(item["_id"]) for item in res["hits"]["hits"]]

            # Merge the ids from Milvus and Elasticsearch
            ids = list(set(milvus_ids + es_ids))

            # Retrieve relevant chunks in RDB
            retrieved_mysql = []
            with Session(engine) as session:
                employees = session.exec(select(Employee).where(Employee.id.in_(ids))).all()
                retrieved_mysql = [{'id': employee.id, 'name': employee.name, 'intro': employee.intro} for employee in employees]

            if retrieved_milvus and retrieved_mysql:
                if mode == "Retrieval Only":
                    st.write("### Retrieved Embedding and ID in Milvus:")
                    st.dataframe(pd.DataFrame(retrieved_milvus))

                    st.write("### Retrieved ID and Score in Elasticsearch:")
                    st.dataframe(pd.DataFrame(retrieved_es))

                    st.write("### Retrieved Data in MySQL:")
                    st.dataframe(pd.DataFrame(retrieved_mysql))
                else:
                    intro = [employee['intro'] for employee in retrieved_mysql]

                    # Build RAG prompt
                    context = "\n".join(intro)
                    prompt = RAG_PROMPT_TEMPLATE.format(context=context, question=query)

                    # Call LLM to generate answer
                    response = completion(
                        model=llm_model,
                        messages=[{"content": prompt, "role": "user"}],
                    )

                    st.markdown(f"### 🤖 {llm_model}")
                    st.markdown(LLM_RESPONSE_STYLE, unsafe_allow_html=True)

                    # show the response
                    st.markdown(
                        f'<div class="llm-response">{response.choices[0].message.content}</div>',
                        unsafe_allow_html=True,
                    )

                    with st.expander("📚 Retrieved Knowledge", expanded=False):
                        st.write("### Retrieved Embedding and ID in Milvus:")
                        st.dataframe(pd.DataFrame(retrieved_milvus))

                        st.write("### Retrieved ID and Score in Elasticsearch:")
                        st.dataframe(pd.DataFrame(retrieved_es))

                        st.write("### Retrieved Data in MySQL:")
                        st.dataframe(pd.DataFrame(retrieved_mysql))

            else:
                st.info("No relevant information found")
    ```

### 步骤 4. 多路召回 RAG AI 应用程序的总结

我们改进了 RAG 应用的能力，让准确率进一步提升了！但是缺点也很明显，我们现在有三个数据存储，代码也变得极其复杂，每次数据的更新都需要同时编写三处，数据同步和 Error Handling 也变得复杂无比。是的，我们让 AI 应用程序变得更优秀了，but at what cost? 我们牺牲了应用程序的可维护性，换来了服务的能力提升。

那我们有没有一个办法，在不牺牲应用程序的可维护性的前提下，提升服务能力呢？有的！下一步，我们将使用 TiDB 作为存储，简化我们的服务。