## モジュール 4: マルチチャネル検索によるRAG AIアプリケーションの改善 モジュール3の冒頭で、RAGが検索チャネルとして埋め込みベクトルだけでなく、他の方法も使用できると述べました。次のような方法を使用できます： - BoW - TF-IDF - Word2Vec これらを使ってテキストベクトルを生成し、意味的類似性を計算します。 同時に、グラフデータベースを導入し、テキストをグラフに抽出してグラフを検索することもできます。つまり、GraphRAGです。 もちろん、以下で説明する方法、つまり全文検索を第二の検索チャネルとして使用することもできます。これにより、RAGに別の検索角度が与えられ、検索されるコンテンツが豊かになります。 全文検索とは、大量のテキストの中からキーワードを使ってテキストチャンクを迅速に検索できる技術の一種です。検索エンジンや広告などのビジネスでよく使用されます。全文検索の一般的な実装方法の一つに、転置インデックスの使用があります。その中核となるプロセスは次のとおりです： 1. テキストを保存する際、トークナイザーを使用して個々の単語に分割します。 2. 単語をキー、テキストIDを値としてインデックスを作成します。 3. キーワードを検索する際、このインデックスを利用してテキストIDのリストを迅速に取得します。 幸いなことに、このプロセスを自分で実装する必要はありません。従来のアプローチでは、検索エンジンを導入することができます。ElasticSearchは一般的に使用される検索エンジンであり、マルチチャネル検索の基盤ストレージとして使用できます。 ### 手順 1. マルチチャネルRAG AIアプリケーションのStreamlitサービスの実行 もう一度新しいターミナルを作成します： ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-rag-streamlit-start-0606.jpg) 次に、新しいマルチチャネル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アプリケーションのデモWebページの使用 - [RAG AIアプリケーション](http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8503/)を開きます。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-ui-0606.jpg) - `All Employees in the MySQL`、`All Employees ID and Embedding Mapping in the Milvus`、`All Employees in the Elasticsearch`という名前のフォルダを開き、MySQL、Milvus、Elasticsearchの現在のデータを確認します。モジュール3でMySQLとMilvusにデータを保存したため、現在Elasticsearchにはデータがなく、MySQLとMilvusにはデータが存在するはずです。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-data-start-0613.jpg) - `Reset`をクリックしてデータを削除します。 - これで、`All Employees in the MySQL`と`All Employees ID and Embedding Mapping in the Milvus`のフォルダが空であることが確認できます。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-data-reset-0606.jpg) - `Save Data`をクリックしてデータを保存します。これにより、データがMySQL、Milvus、Elasticsearchに保存されます。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-data-saved-0613.jpg) - コンテキスト検索のために`Retrieval Only`をクリックします。 - 質問を入力します： ```text Who is Cheese Wang at PingCAP? ``` - `Send`ボタンをクリックして検索を開始します。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-retrieve-0606.jpg) - この検索では、2つの異なる検索チャネルを使用したため、Milvusの埋め込みベクトルから取得したIDとElasticsearchのスコアの両方があります。ElasticsearchはデフォルトでTF-IDFに基づくBM25アルゴリズムを使用して類似度スコアを計算します。スコアが高いほどテキストが類似していると考えることができます。 - 次に、`Select Mode`オプションを`RAG Q&A`に変更し、`Send`をクリックします。 ![90%](https://lab-static.pingcap.com/quick-demo/demo-422/422-improved-rag-streamlit-qna-0606.jpg) > [!NOTE] > このデモでは、マルチチャネル検索からの情報はシングルチャネル検索からの情報と同じですが、これはこの例の複雑さが不足しているためです。スペースの都合上、ここでは証明は行いませんが、以下を参照してください： > > - 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"|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`とマッピングが定義され、保存されるドキュメントの構造（`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. データベース操作ロジックも拡張されました。 - **リセットボタン**: MySQLとMilvusからデータをクリアするのに加えて、Elasticsearchのインデックスも削除して再作成し、環境の完全なリセットを保証します。 - **データ保存ボタン**: MySQLとMilvusにデータを保存した後、`employees`リストを反復処理し、各従業員の情報をElasticsearchにドキュメントとしてインデックス付けします。これにより、従業員の紹介文コンテンツが全文検索可能になります。 ```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. マルチチャネル検索の効果を示すために、Elasticsearchに保存されているすべての従業員データを表示するための新しい展開可能なセクションがUIに追加されます。 ```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. これは、2つの異なる検索方法を組み合わせたマルチチャネルRAGアプリケーションのコアロジックです： - **ベクトル検索（Milvus）**: 以前と同様に、ユーザーの質問を埋め込みに変換し、Milvusで意味的類似性検索を実行して、最初の候補IDバッチ（`milvus_ids`）を取得します。 - **全文検索（Elasticsearch）**: ユーザーの元の質問をキーワードとして使用し、Elasticsearchの`intro`フィールドで`match`クエリを実行して全文検索を行い、2番目の候補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'
{response.choices[0].message.content}
', unsafe_allow_html=True, ) with st.expander("📚 Retrieved Knowledge", expanded=False): 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アプリケーションの能力を向上させ、精度をさらに高めました！しかし、欠点も明らかです。現在、3つのデータストアがあり、コードは非常に複雑になっています。データの更新のたびに3つの場所でコードを記述する必要があります。データの同期とエラー処理は信じられないほど複雑になっています。はい、AIアプリケーションを改善しましたが、その代償は何だったのでしょうか？サービスの能力向上と引き換えに、アプリケーションの保守性を犠牲にしました。 では、アプリケーションの保守性を犠牲にすることなくサービスの能力を向上させる方法はあるのでしょうか？はい！次のステップでは、TiDBをストレージとして使用し、サービスを簡素化します。