# AIアプリケーション向け統合ストレージとしてのTiDBの活用
## 概要
この演習の最初の4つのモジュールでは、GenAIアプリケーションの様々な要件に対応するために、複数の異種データストア（MySQL、Milvus、ElasticSearchなど）を扱います。これには複数のPythonパッケージが必要となり、開発と保守の課題を実際に体験することができます。しかし、モジュール5では、GenAIの統一データストレージとして、たった1つのPythonパッケージ - pytidb - と1つのデータストア - TiDB - のみを使用します。ここでは、TiDBを使用してGenAIアプリケーションを構築することがいかにシンプルで効率的であるかを直接体験することができます。
> [!IMPORTANT]
> この演習では、必要なモデルを呼び出すためにAmazon Bedrockへの一時的なアクセスが許可されます。演習資料に記載されている使用ガイドラインを厳守し、他のコンテキストでこのアクセス権を利用しないでください。金銭的損失につながるAmazon Bedrockの権限の不正使用に対しては、ユーザーに法的責任を問う権利を留保します。
>
> 現在、TiDB Full-Text Search with SQL機能は、TiDB Serverlessフランクフルト（eu-central-1）とシンガポール（ap-southeast-1）リージョンでのみ利用可能です。この演習でご自身の無料TiDB Cloudデータベースを使用する場合は、TiDBクラスタがフランクフルトまたはシンガポールにデプロイされていることを確認してください。
生成AIは現在、多くの開発者が探求を熱望する最先端分野です。しかし、いくつかの障壁がしばしばこの領域への参入を妨げています。例えば：
- AIの基礎知識の欠如により、どこから始めればよいかわからない。
- 生成AIアプリケーションを構築するための包括的で、ステップバイステップ、かつインタラクティブなガイドの不在。
- データベースや大規模モデルAPIなど、学習に必要なインフラへのアクセスの制限。
これらの課題に対処するため、私たちはこの演習を含む一連の演習を開発しました。これらを通じて、以下のことを学びます：
- 従来のAIアプリケーションの作成方法。以下のストレージを基盤とし、フル機能のアプリケーションを構築します：
- リレーショナルデータベース： 例 `MySQL`
- ベクトルデータベース：例 `Milvus`
- 全文検索エンジン：例 `ElasticSearch`
- `pytidb` を使用して **TiDB Cloud Serverless** をAIアプリケーション向けの統一データストレージ層として活用し、アプリケーションを簡素化する方法。
> [!NOTE]
> この演習では、Streamlitを高速アプリケーション開発フレームワークとして使用し、データアプリケーションの構築を効率化・加速させます。
> 演習を開始するためにご自身のTiDB Cloud Serverlessクラスタを選択するか、サンドボックスのTiDB Cloud Serverlessオプションを選択するかにかかわらず、このハンズオン演習専用のデータベースが作成されます。
### タスク
1. 初めてのシンプルなAIアプリケーションの構築：LLMを呼び出し、質問への回答を提供。
2. RAGを使用してAIアプリケーションを改善し、アプリケーションに記憶機能の付与：
- テキストチャンクの保存ストレージとしてMySQLを使用。
- ベクトルストレージおよび検索エンジンとしてMilvusを使用。
3. 追加の検索チャネルを導入してAIアプリケーションのパフォーマンスを向上。全文検索は、AIアプリケーションに異なる検索視点を提供するための良い選択肢：
- 全文検索エンジンとしてElasticSearchを使用。
4. **TiDBを使用してアプリケーションを書き直し、全体のアーキテクチャを簡素化。**
### 所要時間
- この演習の所要時間は約90分です。
> [!TIP]
> **開始** をクリックして演習を起動します。演習環境の準備が整ったら、下部バーの **次のモジュール** をクリックして **モジュール1** に進んでください。
## モジュール 1: Pythonコード実行の準備
### 手順 1. 演習環境の確認
- 左側の `演習リソース` ボタンをクリックして、演習用にセットアップされた演習環境を確認します。表示される内容は次のとおりです：
- 開発サーバーとアプリケーションサーバーの両方として機能するVS Codeサーバーを実行している `monitor1` という名前のサーバー。
- TiDB Cloud Serverlessクラスタへの接続詳細—選択に応じて、ご自身のTiDB Cloudアカウントまたは演習のサンドボックス環境からのものとなります。
- さらに、このハンズオン演習のデータベース名は `${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}` であり、選択したTiDB Cloud Serverlessクラスタ内に自動で作成されます。
- `演習リソース` ポップアップウィンドウを閉じます。
### 手順 2. IDEでターミナルを開く
- ブラウザで [VS Code](${LAB:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage"|urlencode}) を開きます。
- ポップアップウィンドウで `Yes, I trust the authors` をクリックします。
- 左上の `≡` アイコンをクリックし、`Terminal` を選択し、次に `New Terminal` をクリックして新しいターミナルウィンドウを開きます。
### 手順 3. Python依存関係のインストール
VS Codeターミナルで以下のコマンドを実行し、必要な依存関係をインストールします。
```bash
poetry install
```
```output
Creating virtualenv tidb-lab-unify-storage-wZek_onr-py3.11 in /root/.cache/pypoetry/virtualenvs
Installing dependencies from lock file
Package operations: 93 installs, 0 updates, 0 removals
- Installing attrs (25.3.0)
- Installing rpds-py (0.25.1)
- Installing typing-extensions (4.13.2)
- Installing certifi (2025.4.26)
...
```
## モジュール 2: 初めてのシンプルなAIアプリケーションの構築
最もシンプルなAIアプリケーションから始めましょう。このステップでは、大規模言語モデル（LLM）であるAWSのNovaモデルを呼び出します。このモデルはトレーニング中に多くの知識を学習済みです。生の知識がLLMに変換される方法については、以下の資料を参照してください。これらの資料は後で学習のために保存しておくことができます。それでは、ステップ1から始めましょう。
- Transformerの基礎論文：[Attention Is All You Need](https://arxiv.org/abs/1706.03762)
- YouTuber 3Blue1BrownによるTransformerの視覚的解説：[Transformers (how LLMs work) explained visually](https://www.youtube.com/watch?v=wjZofJX0v4M)
スペースの都合上、ここでは実装の詳細には深入りせず、LLMの使用に焦点を当てます。
### 手順 1. シンプルなAIアプリケーションのStreamlitサービスの実行
- VS Codeターミナルで、`streamlit` コマンドラインツールを使用して以下のスクリプトを実行し、Streamlitサービスを開始します。`streamlit` の前に `poetry run` を追加していることにお気づきかもしれません。これにより、デフォルトのpipではなく、poetryを使用して依存関係を管理できます。
```bash
poetry run streamlit run vanilla_ai_app.py --server.port 8501
```
```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:8501
Network URL: http://172.18.0.7:8501
External URL: http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8501
```
### 手順 2. シンプルなAIアプリケーションのデモWebページの使用
- [RAGデモWebページ](http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8501/)にアクセスしてください。
- これで、`Enter your question:` の下の入力ボックスに質問を入力できます。例えば：
```txt
What is TiDB?
```
- 次に、`Send` ボタンをクリックします。少し待つと、回答が表示されます！
- おめでとうございます🎉！初めてのAIアプリケーションが完成しました。AIに興味のある他の質問を続けてみてください。初期のAIアプリケーションの能力の限界を見るために、もっと難しい質問を試すこともできます。
- 次に、コードの内部で何が起こっているかを見てみましょう。
### 手順 3. シンプルなAIアプリケーションのコード読解
[VS Code](${LAB:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage"|urlencode})に戻り、`vanilla_ai_app.py` というコードファイルをクリックしましょう。
1. スクリプトはまず必要なライブラリをインポートします：LLMと対話するための `litellm`、Webインターフェースを構築するための `streamlit` です。さらに、`common` モジュールから `LLM_RESPONSE_STYLE` をインポートします。これはLLMの応答出力を美しくするための定義済みCSSスタイルです。
```python
from litellm import completion
import streamlit as st
from common import LLM_RESPONSE_STYLE
```
2. 使用するLLMが定義されます。モデルを `bedrock/us.amazon.nova-lite-v1:0` に設定します。`litellm` が使用するモデル名は2部構成です：`/` の前の部分はプロバイダーで、この場合はAmazon Bedrockです。後ろの部分はAmazon内部で定義されたモデル名です。
```python
llm_model = "bedrock/us.amazon.nova-lite-v1:0"
```
3. Streamlitページのタイトルを "🔍 Vanilla AI Q&A Demo" に設定します。
```python
st.title("🔍 Vanilla AI Q&A Demo")
```
4. ページに説明テキストを表示して、アプリケーションの機能をユーザーに伝えます：ユーザーが質問を入力すると、システムが組み込みの知識に基づいて回答します。
```python
st.write(
"Enter your question, and the system will answer it based on the built-in knowledge"
)
```
5. `st.text_input` を使用して、ユーザーが質問を入力できるテキスト入力ボックスを作成します。入力された質問は `query` 変数に保存されます。
```python
query = st.text_input("Enter your question:", "")
```
6. このコードブロックは、ユーザーが "Send" ボタンをクリックし、`query`（ユーザーが入力した質問）が空でない場合のロジックを処理します：
- `with st.spinner("Processing...")` を使用して、システムがリクエストを処理中であることをユーザーに知らせるローディングアニメーションを表示します。
- "Answer the question: " という単純なプロンプトにユーザーの質問 `query` を加えて構築します。
```python
if st.button("Send") and query:
with st.spinner("Processing..."):
prompt = f"Answer the question: {query}"
```
7. `litellm.completion` 関数を呼び出して、LLMと対話し、回答を生成します。
- `model` パラメータは使用するLLMを指定します（つまり `llm_model`）。
- `messages` パラメータは、メッセージの内容（`prompt`）と役割（`user`）を定義する辞書を含むリストです。
```python
# Call LLM to generate answer
response = completion(
model=llm_model,
messages=[{"content": prompt, "role": "user"}],
)
```
8. LLMのモデル名とロボットのアイコンをページに表示します。
```python
st.markdown(f"### 🤖 {llm_model}")
```
9. 定義済みのCSSスタイル `LLM_RESPONSE_STYLE` を適用して、LLMの応答表示を美しくします。`unsafe_allow_html=True` パラメータにより、MarkdownにHTMLとCSSを埋め込むことができます。
```python
st.markdown(LLM_RESPONSE_STYLE, unsafe_allow_html=True)
```
10. LLMによって生成された回答を表示します。回答は `response.choices[0].message.content` から抽出され、`llm-response` クラスを持つHTML `div` タグでラップされてスタイルが適用されます。
```python
# show the response
st.markdown(
f'{response.choices[0].message.content}
',
unsafe_allow_html=True,
)
```
### 手順 4. シンプルなAIアプリケーションのまとめ
シンプルなAIアプリケーションが非常に単純であることがわかります。要するに、Amazon Bedrock APIを呼び出して回答を得るだけです。すべての知識は、LLMがトレーニング中に供給されたものに基づいています。例えば、上記のサンプル質問：
```text
What is TiDB?
```
LLMがこの質問に答えられるのは、Amazonがこのモデルをトレーニングする際に関連するドキュメントや紹介を含んでいたためです。ここで疑問が浮かびます：もしLLMがトレーニング中にこのデータを持っていなかったらどうなるでしょうか？例えば：
```text
Who is Cheese Wang at PingCAP?
```
ご覧の通り、LLMはいくつかの情報を「捏造」しました。これは、LLMがトレーニングされた際に類似の情報が入力されなかったためです。したがって、この従業員とPingCAPとの関係を正しく知ることができません。CEOの名前は会社とより密接に関連しており、質問される可能性が高いため、質問された人物がCEOであると「推測」することしかできません。
そして、この回答は間違っています。なぜなら、この演習の作者は ***Cheese Wang*** であり、彼は PingCAP の CEO ではないからです。
LLMのこのような振る舞いは**ハルシネーション（幻覚）**と呼ばれ、LLMを使用する際には特に注意が必要な現象です。次のモジュールでは、AIアプリケーションを改善し、ハルシネーションの頻度を減らし、モデル自体を変更することなくトレーニングデータ以外の質問に答えられるように追加の知識を提供します。
## モジュール 3: RAGによるAIアプリケーションの改善
モジュール1では、予備的なAIアプリケーションを完成させましたが、LLMがトレーニングデータ外の質問に答える際に深刻なハルシネーション問題に遭遇しました。では、これをどう解決すればよいでしょうか？答えは簡単です。
LLMを呼び出す際に入力に情報を事前に取得して配置し、LLMに内容を要約させることで、トレーニングセットの知識に頼るのではなく、LLMを活用することができます。これにより、以下の利点が得られます：
1. ハルシネーションの削減：TransformerベースのLLMは、コンテキストウィンドウ内でシーケンスベースの予測を行うためにマルチヘッドアテンションモデルを使用します。関連知識をコンテキストウィンドウに含めると、その知識により近い回答が生成されます。表面的には、ハルシネーションの頻度や深刻度が減少します。
2. 外部知識の提供：繰り返しになりますが、TransformerベースのLLMにはコンテキストウィンドウ（APIではトークンサイズの制限として反映される）があるため、すべての情報をプロンプトに含めることはできません。したがって、関連するコンテンツを事前に選択する必要があります。
3. コスト削減：LLMのコンテキストウィンドウが大きくなっているにもかかわらず、コストを考慮し、各質問にあまりにも多くのトークンを浪費すべきではありません。なぜなら、ほとんどのLLMはトークン単位で課金されるためです（トークンはテキストの長さに正比例する変数と考えることができます。テキストが長ければ長いほど、トークンも多くなります）。
したがって、今我々が直面している唯一の問題は：
**LLMが関連情報を検索するのをどのように助けるか？**
これを達成する方法は一つではありません。モジュール2では、そのうちの一つであるテキスト埋め込みベクトル方式を使用します。
> [!NOTE]
> **補足資料**
>
> 埋め込みベクトルとは、埋め込みモデルがテキストを変換して生成するベクトルです。埋め込みモデルとは、意味を区別するために特別に訓練されたモデルの一種です。簡単に言うと、対照学習の訓練中に、多数のサンプルセットを取得し、各セットは以下のように分けられます：
>
> - アンカーサンプル：テキストの一部
> - ポジティブサンプル：アンカーサンプルと意味的に類似したテキストの一部
> - ネガティブサンプルのセット：アンカーサンプルと意味的に類似しない多数のテキストの一部
>
> 埋め込みモデルのタスクは：
>
> - アンカーサンプルとポジティブサンプルの出力ベクトル間の距離を最小化すること
> - アンカーサンプルとネガティブサンプルの出力ベクトル間の距離を最大化すること
>
> より正確に言えば、ここではInfoNCE損失関数を使用してモデルを訓練し、確率的勾配降下法を実行してサンプルに適合させます。
任意読解部分が理解できなくても全く問題ありません。それは原理の説明に過ぎません。ただ知っておくべきことは：2つのテキストの埋め込みベクトル間のコサイン距離が大きいほど、それらの意味は類似しておらず、コサイン距離が小さいほど、それらの意味は類似しているということです。
したがって、この特性を利用して今回の改善を行うことができます。モジュール2では、Milvusをベクトル検索エンジンとして、MySQLをリレーショナルデータベースとして元のテキストを保存します。
> [!NOTE]
> **なぜベクトルデータベースのメタデータを使って元のテキストを保存しないのか？**
>
> これは技術的には可能ですが、トランザクションなど、ベクトルデータベースの能力が不足しているため、完全なデータ一貫性を保証できません。実際のエンジニアリングでは、リレーショナルデータベースとベクトルデータベースの分離アーキテクチャを依然として使用する場合、リレーショナルデータベースをデータの信頼できる情報源として使用することをお勧めします。同時に、元のテキストIDと埋め込みベクトルのデータペアを非同期でベクトルデータベースに更新する必要があります。
### 手順 1. RAG AIアプリケーションのStreamlitサービスの実行
新しいターミナルを作成します：
次に、新しいRAG AIアプリケーションを実行します：
```bash
poetry run streamlit run rag.py --server.port 8502
```
```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:8502
Network URL: http://172.18.0.7:8502
External URL: http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8502
```
### 手順 2. RAG AIアプリケーションのデモWebページの使用
- [RAG AIアプリケーション](http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8502/)を開きます。
- `All Employees in the MySQL` と `All Employees ID and Embedding Mapping in the Milvus` というフォルダを開いて、現在のMySQLとMilvusのデータを確認します。まだデータを保存していないため、これらは空のはずです。
- `Save Data` をクリックして、このデモ用に準備した模擬従業員情報データを保存します。次のステップのコードを読むと、元のデータが確認できます。これは従業員オブジェクトのリストです。
- データを保存すると、`All Employees in the MySQL` フォルダでMySQLのデータを確認できます。`id`、`name`、`intro` の3つの列があります。また、`All Employees ID and Embedding Mapping in the Milvus` では、`id` と `embedding` の関係が保存されていることがわかります。この時点で、Milvusの `id` とMySQLの `id` は1対1でマッピングされています。
- コンテキストの検索のみを行いたい場合は、`Retrieval Only` オプションをクリックしてください。その後、質問を入力します。
- RAG AIアプリケーションがEmbeddingを使用して関連情報を選択できるか試してみましょう。以下の質問を入力します：
```text
Who is Cheese Wang at PingCAP?
```
- `Send` ボタンをクリックして検索を開始します。
- ご覧の通り、`Cheese Wang` に関する情報が得られました。
- 次に、the moment of truth（真実の瞬間）です。RAG AIアプリケーションがシンプルなAIアプリケーションでは回答できない質問に答えられるかどうかをテストしましょう。`Select Mode` オプションを `RAG Q&A` に変更し、`Send` をクリックします。
- 今回は、RAG AIアプリケーションは情報リストではなく、回答を提供します。
- ご覧のように、LLMにトレーニング段階で取得できなかった情報を提供したため、LLMは私をPingCAPのCEOだと考えず、私の実際の情報に基づいて回答しました。
### 手順 3. RAG AIアプリケーションのコード読解
[VS Code](${LAB:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage"|urlencode})に戻り、`rag.py`コードファイルをクリックしましょう。
1. スクリプトはまず、RAGアプリケーションを構築するために必要な多くのライブラリをインポートします：
- `os`, `dotenv`: データベースのパスワードやホストアドレスなどの環境変数をロードして読み込むために使用されます。
- `pandas`: Streamlitでデータを処理し、テーブルを表示するために使用されます。
- `sqlalchemy`, `sqlmodel`: リレーショナルデータベース（この場合はMySQL）に接続し、データモデルを定義するために使用されます。`SQLModel`はSQLAlchemyとPydanticを組み合わせる便利さを提供します。
- `pymilvus`: MilvusのPythonクライアントで、Milvusベクトルデータベースに接続し、操作するために使用されます。
- `litellm`: さまざまな大規模言語モデル（LLM）や埋め込みモデルと標準化された対話を行うためのライブラリです。
- `streamlit`: Webユーザーインターフェースを作成し、実行するために使用されます。
- `common`: 共通モジュールから事前定義されたRAGプロンプトテンプレートとCSSスタイルをインポートします。
```python
import os
import pandas as pd
from sqlalchemy import func
from sqlmodel import Field, SQLModel, Session, create_engine, select, delete
from pymilvus import MilvusClient, DataType
from litellm import completion, embedding
import streamlit as st
from typing import List, Optional, Any
from pydantic_core import MultiHostUrl
from common import LLM_RESPONSE_STYLE, RAG_PROMPT_TEMPLATE
from dotenv import load_dotenv
load_dotenv()
```
2. 初期化と設定：
- `create_engine`: 環境変数から読み取った設定（ホスト、ポート、パスワードなど）を使用してMySQLデータベースへの接続エンジンを作成します。
- `embedding_model`, `llm_model`: 使用する埋め込みモデルとLLMの名前を定義します。
- `MilvusClient`: 指定されたURIで実行されているMilvusサービスに接続するためにMilvusクライアントを初期化します。
- `milvus_collection_name`: Milvusで作成されるコレクションに名前を付けます。
```python
engine = create_engine(
str(
MultiHostUrl.build(
scheme="mysql+pymysql",
username="root",
password=os.getenv("MYSQL_ROOT_PASSWORD"),
host=os.getenv("MYSQL_HOST"),
port=int(os.getenv("MYSQL_PORT", "4000")),
path=os.getenv("MYSQL_DATABASE")
)
)
)
embedding_model = "bedrock/amazon.titan-embed-text-v2:0"
embedding_dimensions = 1024
llm_model = "bedrock/us.amazon.nova-lite-v1:0"
milvus_client = MilvusClient(
uri=f"http://{os.getenv('MILVUS_HOST')}:{os.getenv('MILVUS_GRPC_PORT')}",
token="root:Milvus"
)
milvus_collection_name = "employee_id_mapping"
```
3. テキスト文字列を受け取り、`litellm.embedding` APIを呼び出してテキストを1024次元のベクトルに変換し、そのベクトルを返すヘルパー関数 `get_embedding` を定義します。
```python
def get_embedding(text: str) -> List[float]:
response = embedding(
model=embedding_model,
input=[text],
dimensions=embedding_dimensions
)
return response.data[0]['embedding']
```
4. `SQLModel`を使用して `Employee` テーブルの構造を定義します。このテーブルはMySQLに作成され、元の従業員情報（ID、名前、紹介文）を保存するために使用されます。`SQLModel.metadata.create_all(engine)` は、このテーブルがデータベースに作成されることを保証します。
```python
class Employee(SQLModel, table=True):
__table_args__ = {"extend_existing": True}
id: int | None = Field(default=None, primary_key=True)
name: str
intro: str
SQLModel.metadata.create_all(engine)
```
5. デモ用のサンプルデータとして、3人の従業員の情報を含むリスト `employees` を作成します。
```python
employees = [
Employee(id=1, name="Cheese Wang", intro="Cheese is a Senior Software Engineer of AI Application at PingCAP, the company behind TiDB"),
Employee(id=2, name="Hao Huo", intro="Hao Huo is an engineer of PingCAP, the company behind TiDB | Build pingcap/autoflow / OSSInsight.io"),
Employee(id=3, name="Humble BGL", intro="Humble BGL is the Head of Engagement Innovation at PingCAP, the company behind TiDB")
]
```
6. Milvusのデータスキーマとインデックスを定義します。
- `MilvusClient.create_schema`: コレクションの構造を定義します。主キー `id`（MySQLの `Employee` レコードと関連付けるため）と `embedding` ベクトルフィールドが含まれます。
- `milvus_client.prepare_index_params`: `embedding` フィールドのインデックスパラメータを準備し、`AUTOINDEX` を使用してMilvusに最適なインデックスタイプを自動選択させます。
- `milvus_client.create_collection`: 定義されたスキーマとインデックスを使用してMilvusに新しいコレクションを作成します。これにより、効率的なベクトル類似性検索を実行できます。
```python
schema = MilvusClient.create_schema(auto_id=False, enable_dynamic_field=True)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=embedding_dimensions)
index_params = milvus_client.prepare_index_params()
index_params.add_index(field_name="embedding", index_type="AUTOINDEX")
milvus_client.create_collection(
collection_name=milvus_collection_name,
schema=schema,
index_params=index_params
)
```
7. Streamlitインターフェースを設定し、データベース操作ボタンを提供します。
- **リセットボタン**: クリックすると、MySQLの `Employee` テーブルとMilvusコレクションから関連するすべてのデータがクリアされ、デモ環境をリセットするために使用されます。
- **データ保存ボタン**: クリックすると、MySQLにデータが既に存在するかどうかを確認します。存在しない場合、次の処理を行います：
a. `employees` リストを反復処理し、各従業員の `intro` の埋め込みベクトルを生成します。
b. 完全な従業員情報（ID、名前、紹介文）をMySQLに保存します。
c. 従業員のIDとそれに対応する埋め込みベクトルをMilvusに保存します。
このプロセスは、構造化メタデータとベクトルデータをそれぞれRDBとベクトルデータベースに書き込む方法を示しています。
```python
st.title("🔍 RAG Demo")
st.subheader("Database Operations")
left, right = st.columns(2)
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]
)
st.rerun()
left.markdown("This option will delete all data", unsafe_allow_html=True)
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
)
st.rerun()
else:
st.error("Data already exists")
```
8. 2つの展開可能なセクション（`st.expander`）を作成し、それぞれMySQLとMilvusに現在保存されているデータを表示します。これにより、ユーザーはデータ操作の結果を直感的に確認できます。
```python
with st.expander("📁 All Employees in the MySQL", expanded=False):
with Session(engine) as session:
employees = session.exec(select(Employee)).all()
if employees:
data = [{'id': employee.id, 'name': employee.name, 'intro': employee.intro} for employee in employees]
df = pd.DataFrame(data)
st.dataframe(df)
else:
st.info("No data found in MySQL.")
with st.expander("📁 All Employees ID and Embedding Mapping in the Milvus", expanded=False):
employees = milvus_client.get(
collection_name=milvus_collection_name,
ids=[employee.id for employee in employees],
output_fields=["id", "embedding"]
)
if employees:
data = [{'id': employee['id'], 'embedding': employee['embedding']} for employee in employees]
df = pd.DataFrame(data)
st.dataframe(df)
else:
st.info("No data found in Milvus.")
```
9. これはRAGアプリケーションのコアな対話ロジックです。ユーザーが質問を入力して「送信」をクリックすると：
- **クエリのベクトル化**: まず、`get_embedding` 関数を使用してユーザーの質問テキストをクエリベクトルに変換します。
- **ベクトル検索**: `milvus_client.search` を使用してMilvusコレクションでベクトル類似性検索（コサイン距離を使用）を行い、最も類似したレコードを見つけてその `id` を取得します。
- **データ取得**: Milvusから取得した `id` を使用してMySQLの `Employee` テーブルをクエリし、対応する完全な従業員情報（名前、紹介文など）を取得します。
- **生成**:
- モードが `Retrieval Only` の場合、MilvusとMySQLから取得した生のデータを直接表示します。
- モードが `RAG Q&A` の場合、MySQLから取得した従業員の紹介文（`intro`）をコンテキストとして使用し、ユーザーの元の質問とともに `RAG_PROMPT_TEMPLATE` に入力します。
- 最後に、`litellm.completion` を呼び出して構築されたプロンプトをLLMに送信し、最終的な回答を生成して、美化されたスタイルでインターフェースに表示します。同時に、回答の生成に使用された「知識源」を表示するための展開可能なセクションを提供します。
- ベクトル検索で関連情報が見つからない場合は、ユーザーに通知します。
```python
st.write(
"Enter your question, and the system will retrieve relevant knowledge and generate an answer"
)
mode = st.radio("Select Mode:", ["Retrieval Only", "RAG Q&A"])
query = st.text_input("Enter your question:", "")
if st.button("Send") and query:
with st.spinner("Processing..."):
# Retrieve relevant chunks
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]]
ids = [item.entity.id for item in res[0]]
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 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.write("### Retrieved Embedding and ID in Milvus:")
st.dataframe(pd.DataFrame(retrieved_milvus))
st.write("### Retrieved Data in MySQL:")
st.dataframe(pd.DataFrame(retrieved_mysql))
else:
st.info("No relevant information found")
```
### 手順 4. RAG AIアプリケーションのまとめ
シンプルなAIアプリケーションとRAG AIアプリケーションの長所と短所を明確に比較できます。シンプルなAIアプリケーションは単純ですが、その背景知識はすべてトレーニングセットに組み込まれています。質問がトレーニングセットの範囲外である場合、幻覚を生じる可能性が高いです。RAG AIアプリケーションはこの問題をうまく解決し、モジュール2の冒頭で提案した一連の利点をもたらしますが、最も単純なデモアプリケーションでさえ、かなりの複雑さを伴います。
次に、RAGアプリケーションを最適化して、その精度をさらに向上させる必要があります。
## モジュール 4: マルチチャネル検索によるRAG AIアプリケーションの改善
モジュール2の冒頭で、RAGが検索チャネルとして埋め込みベクトルだけでなく、他の方法も使用できると述べました。次のような方法を使用できます：
- BoW
- TF-IDF
- Word2Vec
これらを使ってテキストベクトルを生成し、意味的類似性を計算します。
同時に、グラフデータベースを導入し、テキストをグラフに抽出してグラフを検索することもできます。つまり、GraphRAGです。
もちろん、以下で説明する方法、つまり全文検索を第二の検索チャネルとして使用することもできます。これにより、RAGに別の検索角度が与えられ、検索されるコンテンツが豊かになります。
全文検索とは、大量のテキストの中からキーワードを使ってテキストチャンクを迅速に検索できる技術の一種です。検索エンジンや広告などのビジネスでよく使用されます。全文検索の一般的な実装方法の一つに、転置インデックスの使用があります。その中核となるプロセスは次のとおりです：
1. テキストを保存する際、トークナイザーを使用して個々の単語に分割します。
2. 単語をキー、テキストIDを値としてインデックスを作成します。
3. キーワードを検索する際、このインデックスを利用してテキストIDのリストを迅速に取得します。
幸いなことに、このプロセスを自分で実装する必要はありません。従来のアプローチでは、検索エンジンを導入することができます。ElasticSearchは一般的に使用される検索エンジンであり、マルチチャネル検索の基盤ストレージとして使用できます。
### 手順 1. マルチチャネルRAG AIアプリケーションのStreamlitサービスの実行
もう一度新しいターミナルを作成します：
次に、新しいマルチチャネル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/)を開きます。
- `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にはデータが存在するはずです。
- `Reset`をクリックしてデータを削除します。
- これで、`All Employees in the MySQL`と`All Employees ID and Embedding Mapping in the Milvus`のフォルダが空であることが確認できます。
- `Save Data`をクリックしてデータを保存します。これにより、データがMySQL、Milvus、Elasticsearchに保存されます。
- コンテキスト検索のために`Retrieval Only`をクリックします。
- 質問を入力します：
```text
Who is Cheese Wang at PingCAP?
```
- `Send`ボタンをクリックして検索を開始します。
- この検索では、2つの異なる検索チャネルを使用したため、Milvusの埋め込みベクトルから取得したIDとElasticsearchのスコアの両方があります。ElasticsearchはデフォルトでTF-IDFに基づくBM25アルゴリズムを使用して類似度スコアを計算します。スコアが高いほどテキストが類似していると考えることができます。
- 次に、`Select Mode`オプションを`RAG Q&A`に変更し、`Send`をクリックします。
> [!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:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage"|urlencode})に戻り、`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をストレージとして使用し、サービスを簡素化します。
## モジュール 5: TiDBによるAIアプリケーションの簡素化
これまでに、マルチチャネルRAGアプリケーションを完成させました。その精度はかなり満足のいくものです。しかし、その複雑さと保守性は理想とはほど遠いものです。これは、従来のソリューションでは、基盤となるストレージとして多くのデータベースを導入する必要があるためです。使用したデータベースを思い出してみましょう：
- MySQL
- Milvus
- ElasticSearch
しかし、今ではより良いソリューションがあります：TiDB Serverless！TiDB ServerlessはMySQL構文と互換性があり、ベクトルと全文検索をサポートしています。
このモジュールでは、`pytidb`ライブラリを使用してマルチチャネルRAGアプリケーションを書き直します。サービスをどの程度簡素化できるか見てみましょう。
### 手順 1. TiDB AIアプリケーションのStreamlitサービスの実行
もう一度新しいターミナルを作成します：
次に、新しいTiDB AIアプリケーションを実行します：
```bash
poetry run streamlit run tidb_rag.py --server.port 8504
```
```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:8504
Network URL: http://172.18.0.7:8504
External URL: http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8504
```
### 手順 2. TiDB AIアプリケーションのデモWebページの使用
- [TiDB AIアプリケーション](http://${LAB:HOST_MONITOR1_PUBLIC_IP}:8504/)を開きます。TiDB Serverlessはこのサーバー上にはなく、初期化が必要なため、全体の初期化には約5分かかります。しばらくお待ちください。初期化後のページは次のようになります：
- `All Employees in the Database`という名前のフォルダを開き、TiDBの現在のデータを確認します。まだデータを保存していないので、空のはずです。
- `Save Data`をクリックしてデータを保存します。
- データを保存した後、`All Employees in the Database`フォルダでTiDBのデータを見ることができます。`id`、`name`、`intro`、`intro_vec`の4つの列があります。TiDBはMySQLよりも1つ多い列`intro_vec`を持っています。これは、TiDBがベクトル型をサポートしているためです。ベクトルを直接TiDBに保存できます。
- `Retrieval Only`をクリックし、質問を入力します：
```text
Who is Cheese Wang at PingCAP?
```
- `Send`ボタンをクリックして検索を開始します。
- ご覧の通り、取得された情報には情報自体に加えて、`_distance`、`_score`、`_match_score`などのメトリクスが含まれています。`pytidb`はこれらのメトリクスを使用してテキストの類似性を計算します。
- `Select Mode`オプションを`RAG Q&A`に変更し、`Send`をクリックして回答を提供します。
- マルチチャネルRAGを完璧に再現しました。では、コードの複雑さを見てみましょう。
### 手順 3. TiDB AIアプリケーションのコード読解
[VS Code](${LAB:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-unify-storage"|urlencode})に戻り、`tidb_rag.py`コードファイルをクリックしましょう。
1. 最も重要な変更点は、インポートされるライブラリです。`sqlalchemy`、`pymilvus`、`elasticsearch`のような複数のデータベース用のクライアントはもう必要ありません。今や、TiDBとの対話のための統一インターフェースとして機能する`pytidb`ライブラリだけが必要です。
```python
import os
import pandas as pd
from litellm import completion
import streamlit as st
from typing import Optional, Any
from pytidb import TiDBClient
from pytidb.schema import TableModel, Field
from pytidb.embeddings import EmbeddingFunction
from common import LLM_RESPONSE_STYLE, RAG_PROMPT_TEMPLATE
```
2. 初期化部分も大幅に簡素化されています。TiDB Serverlessに接続するために`TiDBClient`インスタンスを作成するだけです。`EmbeddingFunction`は`pytidb`のヘルパークラスで、埋め込みモデルの呼び出しをデータテーブルスキーマとシームレスに統合するのに役立ちます。
```python
db = TiDBClient.connect(
host=os.getenv("SERVERLESS_CLUSTER_HOST"),
port=int(os.getenv("SERVERLESS_CLUSTER_PORT")),
username=os.getenv("SERVERLESS_CLUSTER_USERNAME"),
password=os.getenv("SERVERLESS_CLUSTER_PASSWORD"),
database=os.getenv("SERVERLESS_CLUSTER_DATABASE_NAME"),
enable_ssl=True,
)
embedding_model = "bedrock/amazon.titan-embed-text-v2:0"
llm_model = "bedrock/us.amazon.nova-lite-v1:0"
text_embedding_function = EmbeddingFunction(
embedding_model,
timeout=60
)
```
3. これがアプリケーションの簡素化の中核です。`pytidb.schema.TableModel`を使用して`Employee`という名前のテーブルを定義します。
- `id`、`name`、`intro`は標準的なテキストおよび数値フィールドです。
- `intro_vec`: これが最も重要な部分です。これを`text_embedding_function.VectorField`として定義し、`source_field="intro"`を指定します。これにより、`pytidb`はデータの挿入または更新時に`intro`フィールドの内容を自動的に取得し、事前定義された埋め込みモデルを呼び出してベクトルに変換し、その結果を`intro_vec`列に保存するように指示されます。この単一のフィールドの定義が、手動でベクトルを生成してMilvusに保存するすべてのロジックを置き換えます。
```python
table_name = "employees"
class Employee(TableModel, table=True):
__tablename__ = table_name
__table_args__ = {"extend_existing": True}
id: int | None = Field(default=None, primary_key=True)
name: str
intro: str
intro_vec: Optional[Any] = text_embedding_function.VectorField(
source_field="intro",
)
```
4. データとインデックスの準備も非常に簡単になりました。
- `db.open_table`と`db.create_table`はテーブルの存在を保証します。
- `table.create_fts_index("intro")`: **この一行のコードがElasticsearchのセットアップ全体を置き換えます**。これにより`intro`列に全文検索インデックスが作成され、キーワード検索が可能になります。
```python
table = db.open_table(table_name)
if table is None:
table = db.create_table(schema=Employee)
if not table.has_fts_index("intro"):
table.create_fts_index("intro")
```
5. データ操作は単一のコマンドに簡素化されました。
- **リセットボタン**: `table.truncate()` は1つのコマンドで全てのデータをクリアします。複数のデータベースにまたがる削除操作を調整する必要はもうありません。
- **データ保存ボタン**: `table.bulk_insert(employees)` は1つのコマンドで全てのデータを保存します。`pytidb` はバックグラウンドで自動的に処理を行います：各 `employee` レコードの `intro` のベクトルを生成し、そのベクトルを含む完全なデータを TiDB の同じテーブルに書き込みます。全文検索インデックスも自動的に更新されます。
```python
if left.button("Reset", type="primary"):
left.write("Resetting...")
table.truncate()
st.rerun()
if right.button("Save Data"):
right.write("Saving...")
if table.rows() == 0:
table.bulk_insert(employees)
st.rerun()
```
6. これが最も驚くべき簡素化です。マルチチャネル検索の複雑なロジック全体が、今や一行のコードに凝縮されています。
- `table.search(query, search_type="hybrid").text_column("intro").limit(1)`: この一行のコードは、以前は複数のステップが必要だったことを達成します：
1. ユーザーのクエリ`query`を**ベクトル類似性検索**（`intro_vec`列）と**全文検索**（`intro`列）の両方に使用します。
2. `search_type="hybrid"`パラメータは、`pytidb`に両方の検索を自動的に実行し、組み込みの融合アルゴリズムを使用して両チャネルからの結果をランク付けし、マージするように指示します。
3. 統一された、最も関連性の高い結果のリストを返します。
- 埋め込みモデルを手動で呼び出し、MilvusとElasticsearchを別々にクエリし、IDをマージし、最後にMySQLをクエリする必要はもうありません。これらすべてが`pytidb`によって単一のステップでエレガントに処理されます。
```python
if st.button("Send") and query:
with st.spinner("Processing..."):
# Retrieve relevant employees
results = table.search(query, search_type="hybrid").text_column("intro").limit(1)
if results:
if mode == "Retrieval Only":
st.write("### Hybrid Search Results:")
st.dataframe(results.to_pandas())
else:
intro = [employee.intro for employee in results.to_rows()]
# 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("📚 Hybrid Search Retrieved Knowledge", expanded=False):
st.dataframe(results.to_pandas())
else:
st.info("No relevant information found")
```
### 手順 4. TiDB AIアプリケーションのまとめ
同じ要件を満たすためにTiDBを使用することで、コードの実装が大幅に簡素化されたことが明らかです。
## 結論
AIアプリケーションの開発において、リレーショナルデータベース、ベクトル計算、全文検索は頻繁に使用される機能となっています。市場に統一されたソリューションがないため、リレーショナルデータベース、ベクトルデータベース、検索エンジンを導入する必要があります。これにより、私たちのソリューションは非常に複雑になります。この複雑さは、実際にはデータベースの能力不足によって引き起こされています。
TiDBは、コードをよりシンプルで保守しやすくすることでこの問題を解決します。開発者はもはや複数のデータストアを使用してデータを移動させる必要はありません。代わりに、データを統一された方法で一元的に保存し、アクセスできます。これにより、AI分野でのアプリケーション開発の障壁が大幅に低くなります。
```hbsptform
{
"ui": {
"titleLevel": 3
},
"portalId": "4466002",
"formId": "62bfb32a-a60a-4f9d-b2dc-1107a93db568",
"region": "na1",
"sfdcCampaignId": "701RC00000NdFlAYAV"
}
```