RAGは通常、「Retrieval-Augmented Generation（検索拡張生成）」の略称です。これは検索と生成を組み合わせた機械学習モデルであり、テキスト生成や質問応答システムなどの自然言語処理タスクによく使用されます。

以下の手順で、JD Cloud 公式ドキュメントに基づいた RAG システムを完成させます。

• データ収集

• 知識ベースを構築する

• ベクトル検索

• プロンプトワードとモデル

データ収集

データ収集は、RAG導入プロセス全体の中で、間違いなく最も労力を要する部分です。データ収集には、収集、クレンジング、フォーマット、セグメンテーションといった作業が含まれます。ここでは、JD Cloudの公式ドキュメントをナレッジベースの基盤として使用します。ドキュメントの形式は、おおよそ以下のとおりです。

 {
"content": "DDoS IP High-Defense と Web アプリケーション ファイアウォール ソリューションの組み合わせの説明\n========================\n\n\nDDoS IP High-Defense + Web アプリケーション ファイアウォールは、ゲーム、金融、電子商取引、インターネット、政府機関/企業など、JD Cloud の内外のさまざまなタイプのユーザーに適用できる、レイヤー 3 からレイヤー 7 までのセキュリティ保護システムを提供します。\n\n\n導入アーキテクチャ\n====\n\n\n[![\\\"導入アーキテクチャ\\\"](\\\"https://jdcloud-portal.oss.cn-north-1.jcloudcs.com/cn/image/Advanced%20Anti-DDoS/Best-Practice02.png\\\")](\\\"https://jdcloud-portal.oss.cn-north-1.jcloudcs.com/cn/image/Advanced%20Anti-DDoS/Best-Practice02.png\\\") DDoS IP保護+Webアプリケーションファイアウォールの最適な導入アーキテクチャは次のとおりです。* JD Cloudのセキュリティディスパッチセンターは、DNSを介してユーザーのドメイン名をDDoS IP保護CNAMEレコードに解決します。* 通常のユーザーアクセストラフィックとDDoS攻撃トラフィックは、DDoS IP保護によってクリーンアップされ、Webアプリケーションファイアウォールにルーティングされます。* 悪意のある攻撃者のリクエストは、Webアプリケーションファイアウォールによってフィルタリングされ、ユーザーのオリジンサーバーに返されます。* Webアプリケーションファイアウォールは、 JD Cloud、その他のクラウドプロバイダー、IDCなどを含む、あらゆるパブリックネットワークサーバー。ソリューションの利点：1. ユーザーのオリジンサーバーはDDoS IP保護とWebアプリケーションファイアウォールの背後に配置されているため、オリジンサーバーのIPアドレスは効果的に隠蔽されます。2. CNAMEアクセスの設定が簡単で、運用・保守担当者の作業負荷を軽減します。
タイトル: DDoS IP高防御とWebアプリケーションファイアウォールソリューションの組み合わせの説明
"product": "DDoS IP高保護",
「URL」: 「https://docs.jdcloud.com/cn/anti-ddos-pro/anti-ddos-pro-and-waf」
}

各データ エントリは、次の 4 つのフィールドを含む JSON ファイルです: "content": ドキュメント コンテンツ、 "title": ドキュメント タイトル、 "product": 関連製品、および "url": ドキュメントのオンライン アドレス。

ベクターデータベースの選択とRetrieverの実装

ベクターデータベースは、RAGシステムのメモリセンターとして機能します。現在、オープンソースのベクターデータベースは数多く存在し、どれが優れているかについては様々な意見があります。このプロジェクトでは、ベクターデータベースとしてClickHouseを選択しました。ClickHouseを選択した主な理由は次のとおりです。

• ck の langchain コミュニティでの統合実装は非常に良好で、データの挿入は比較的スムーズです。

• ベクター クエリは SQL をサポートし、学習曲線が緩やかで、使いやすいです。

• JD Cloud では関連製品を提供しており、専門チームによるサポートも受けられるため、安心してご利用いただけます。

文書のベクトル化とデータベース入力プロセス

ドキュメントのベクトル化と検索のプロセスを簡素化するために、以下のコードに示すように、最初に longchain の Retriever ツールセットを使用してドキュメントをベクトル化しました。

 libs.jd_doc_json_loader から JD_DOC_Loader をインポートします。
langchain_community.document_loaders から DirectoryLoader をインポートします
root_dir = "/root/jd_docs
ローダー = ディレクトリローダー(
'/root/jd_docs'、glob="**/*.json"、loader_cls=JD_DOC_Loader)
ドキュメント = loader.load()

langchain コミュニティでは特定の形式用のローダーが提供されていないため、読み込みプロセスを実装するために JD_DOC_Loader をカスタマイズしました。

 JSONをインポート
インポートログ
pathlibからPathをインポート
入力からインポートIterator、Optional、Union
langchain_core.documentsからDocumentをインポートする
langchain_community.document_loaders.base から BaseLoader をインポートします。
langchain_community.document_loaders.helpersからdetect_file_encodingsをインポートします
ロガー = logging.getLogger(__name__)
クラス JD_DOC_Loader(BaseLoader):
テキストファイルを読み込みます。
引数:
file_path: ロードするファイルへのパス。
encoding: 使用するファイルエンコーディング。`None` の場合、ファイルは読み込まれます
デフォルトのシステム エンコーディングを使用します。
autodetect_encoding: ファイルのエンコーディングを自動検出するかどうか
指定されたエンコードが失敗した場合。
「」
デフ__init__(
自己、
file_path: Union[str, Path],
エンコーディング: オプション[str] = なし、
autodetect_encoding: bool = False、
):
「ファイルパスで初期化します。」
self.file_path = ファイルパス
self.encoding = エンコーディング
self.autodetect_encoding = 自動検出エンコーディング
def lazy_load(self) -> Iterator[ドキュメント]:
「ファイルパスから読み込みます。」
テキスト = ""
from_url = ""
試す：
open(self.file_path, encoding=self.encoding) を f として実行します:
doc_data = json.load(f)
テキスト = doc_data["コンテンツ"]
タイトル = doc_data["タイトル"]
製品 = doc_data["製品"]
from_url = doc_data["url"]
# テキスト = f.read()
UnicodeDecodeError を除き、e:
self.autodetect_encoding の場合:
検出されたエンコーディング = 検出ファイルエンコーディング(self.file_path)
検出されたエンコードの場合:
logger.debug(f"エンコードを試行しています: {encoding.encoding}")
試す：
open(self.file_path, encoding=encoding.encoding) を f として実行します:
テキスト = f.read()
壊す
UnicodeDecodeErrorを除く:
                        続く
それ以外：
eからRuntimeError(f"{self.file_path}の読み込みエラー")が発生します
except Exception を e として:
eからRuntimeError(f"{self.file_path}の読み込みエラー")が発生します
# メタデータ = {"source": str(self.file_path)}
メタデータ = {"source": from_url, "title": title, "product": product}
ドキュメントを生成します(page_content=テキスト、metadata=メタデータ)

上記のコードは主に JSON ファイルを解析し、ドキュメントの `page_content` フィールドと `metadata` フィールドにデータを入力します。

次に、Langchain の ClickHouse ベクター ツールセットを使用して、ドキュメントをデータベースにインポートします。

 langchain_community.vectorstores.clickhouse を clickhouse としてインポートします。
langchain.embeddings から HuggingFaceEmbeddings をインポートします
model_kwargs = {"デバイス": "cuda"}
埋め込み = HuggingFaceEmbeddings(
model_name="/root/models/moka-ai-m3e-large", model_kwargs=model_kwargs)
設定 = clickhouse.ClickhouseSettings(
テーブル="jd_docs_m3e_with_url", ユーザー名="default", パスワード="xxxxxx", ホスト="10.0.1.94")
docsearch = clickhouse.Clickhouse.from_documents(
ドキュメント、埋め込み、config=settings)

入庫完了後、以下の検査を実施いたします。

 langchain_community.vectorstores.clickhouse を clickhouse としてインポートします。
langchain.embeddings から HuggingFaceEmbeddings をインポートします
model_kwargs = {"デバイス": "cuda"}~~~~
埋め込み = HuggingFaceEmbeddings(
model_name="/root/models/moka-ai-m3e-large", model_kwargs=model_kwargs)
設定 = clickhouse.ClickhouseSettings(
テーブル="jd_docs_m3e_with_url_splited", ユーザー名="default", パスワード="xxxx", ホスト="10.0.1.94")
ck_db = clickhouse.Clickhouse(埋め込み、config=設定)
ck_retriever = ck_db.as_retriever(
search_type="類似度スコアしきい値", search_kwargs={'スコアしきい値': 0.9})
ck_retriever.get_relevant_documents("MySQL RDS の作成方法")

ナレッジ ベースが準備できたら、シンプルな RESTful サービスを構築できます。そのためには FastAPI を使用します。

 fastapiからFastAPIをインポートする
pydantic から BaseModel をインポート
singleton_decorator からシングルトンをインポート
langchain_community.embeddings から HuggingFaceEmbeddings をインポートします
langchain_community.vectorstores.clickhouse を clickhouse としてインポートします。
uvicornをインポートする
JSONをインポート
アプリ = FastAPI()
app = FastAPI(docs_url=なし)
app.host = "0.0.0.0"
model_kwargs = {"デバイス": "cuda"}
埋め込み = HuggingFaceEmbeddings(
model_name="/root/models/moka-ai-m3e-large", model_kwargs=model_kwargs)
設定 = clickhouse.ClickhouseSettings(
テーブル="jd_docs_m3e_with_url_splited", ユーザー名="default", パスワード="xxxx", ホスト="10.0.1.94")
ck_db = clickhouse.Clickhouse(埋め込み、config=設定)
ck_retriever = ck_db.as_retriever(
search_type="類似度", search_kwargs={"k": 3})
クラスの質問(BaseModel):
コンテンツ: str
@app.get("/")
非同期定義root():
{"ok"} を返す
@app.post("/retriever")
非同期定義取得(質問: 質問):
グローバル ck_retriever
結果 = ck_retriever.invoke(質問.コンテンツ)
結果を返す
__name__ == '__main__' の場合:
uvicorn.run(app='retriever_api:app', ホスト="0.0.0.0",
ポート=8000、リロード=True)

返される構造はおおよそ次のようになります。

 [
{
"page_content": "Cloud Cache Redis - Redis 移行ソリューション\n###RedisSyncer 操作手順\n####データ検証\n```\nwget https://github.com/TraceNature/rediscompare/releases/download/v1.0.0/rediscompare-1.0.0-linux-amd64.tar.gz\nrediscompare compare single2single --saddr \\\"10.0.1.101:6479\\\" --spassword \\\"redistest0102\\\" --taddr \\\"10.0.1.102:6479\\\" --tpassword \\\"redistest0102\\\" --comparetimes 3\n\n``` \n**Github アドレス:** [https://github.com/TraceNature/redissyncer-server](\\\"https://github.com/TraceNature/redissyncer-server\\\")",
「メタデータ」: {
"product": "クラウドキャッシュRedis",
「ソース」: "https://docs.jdcloud.com/cn/jcs-for-redis/doc-2",
タイトル: Redis 移行ソリューション
},
「タイプ」: 「ドキュメント」
},
{
"page_content": "Cloud Cache Redis - Redis 移行ソリューション\n###RedisSyncer 操作手順\n####データ検証\n```\nwget https://github.com/TraceNature/rediscompare/releases/download/v1.0.0/rediscompare-1.0.0-linux-amd64.tar.gz\nrediscompare compare single2single --saddr \\\"10.0.1.101:6479\\\" --spassword \\\"redistest0102\\\" --taddr \\\"10.0.1.102:6479\\\" --tpassword \\\"redistest0102\\\" --comparetimes 3\n\n``` \n**Github アドレス:** [https://github.com/TraceNature/redissyncer-server](\\\"https://github.com/TraceNature/redissyncer-server\\\")",
「メタデータ」: {
"product": "クラウドキャッシュRedis",
「ソース」: "https://docs.jdcloud.com/cn/jcs-for-redis/doc-2",
タイトル: Redis 移行ソリューション
},
「タイプ」: 「ドキュメント」
},
{
"page_content": "Cloud Cache Redis - Redis 移行ソリューション\n###RedisSyncer 操作手順\n####データ検証\n```\nwget https://github.com/TraceNature/rediscompare/releases/download/v1.0.0/rediscompare-1.0.0-linux-amd64.tar.gz\nrediscompare compare single2single --saddr \\\"10.0.1.101:6479\\\" --spassword \\\"redistest0102\\\" --taddr \\\"10.0.1.102:6479\\\" --tpassword \\\"redistest0102\\\" --comparetimes 3\n\n``` \n**Github アドレス:** [https://github.com/TraceNature/redissyncer-server](\\\"https://github.com/TraceNature/redissyncer-server\\\")",
「メタデータ」: {
"product": "クラウドキャッシュRedis",
「ソース」: "https://docs.jdcloud.com/cn/jcs-for-redis/doc-2",
タイトル: Redis 移行ソリューション
},
「タイプ」: 「ドキュメント」
}
]

距離が最小のベクトルのリストを返します。

モデルとプロンプトを組み合わせて質問に答えます。

計算リソースを節約するため、qwen 1.8Bモデルを選択しました。1枚のV100カードには、1つのqwenモデルと1つのm3e-large埋め込みモデルを収容できます。

• 応答サービス

fastapiからFastAPIをインポートする
pydantic から BaseModel をインポート
langchain_community.llms から VLLM をインポートします
トランスフォーマーからAutoTokenizerをインポート
langchain.promptsからPromptTemplateをインポートする
輸入リクエスト
uvicornをインポートする
JSONをインポート
インポートログ
アプリ = FastAPI()
app = FastAPI(docs_url=なし)
app.host = "0.0.0.0"
ロガー = logging.getLogger()
logger.setLevel(ログ記録.INFO)
to_console = ログ記録.StreamHandler()
logger.addHandler(コンソールへ)
# ロードモデル
# model_name = "/root/models/Llama3-Chinese-8B-Instruct"
model_name = "/root/models/Qwen1.5-1.8B-Chat"
トークナイザー = AutoTokenizer.from_pretrained(モデル名)
llm_llama3 = VLLM(
モデル=モデル名、
トークナイザー=トークナイザー、
タスク="テキスト生成",
温度=0.2、
do_sample=True、
繰り返しペナルティ=1.1、
return_full_text=False、
最大新規トークン数=900、
）
＃ プロンプト
prompt_template = """
あなたはクラウドテクノロジーの専門家です。以下のコンテキストを使用して質問に回答してください。
答えが分からない場合は、分からないとだけ言ってください。
質問に中国語で答えてください。
質問: {質問}
コンテキスト: {context}
答え：
「」
プロンプト = プロンプトテンプレート(
input_variables=["コンテキスト", "質問"],
テンプレート=プロンプトテンプレート、
）
get_context_list(q: str) を定義します。
url = "http://10.0.0.7:8000/retriever"
ペイロード = {"content": q}
res = リクエスト.post(url, json=ペイロード)
res.textを返す
クラスの質問(BaseModel):
コンテンツ: str
@app.get("/")
非同期定義root():
{"ok"} を返す
@app.post("/answer")
非同期定義回答(q: 質問):
logger.info("呼び出し!!!")
グローバルプロンプト
グローバル llm_llama3
context_list_str = get_context_list(q.content)
context_list = json.loads(context_list_str)
コンテキスト = ""
ソースリスト = []
context_list の context_json の場合:
コンテキスト = コンテキスト+context_json["ページコンテンツ"]
source_list.append(context_json["メタデータ"]["ソース"])
p = prompt.format(コンテキスト=コンテキスト、質問=q.content)
答え = llm_llama3(p)
結果 = {
「回答」：回答、
「ソース」: ソースリスト
}
結果を返す
__name__ == '__main__' の場合:
uvicorn.run(app='retriever_api:app', ホスト="0.0.0.0",
ポート=8888、リロード=True)

このコードは、Retriever インターフェースを使用して質問に類似したドキュメントを検索し、これらのドキュメントをコンテキストとして使用してプロンプトをモデルにプッシュし、回答を生成します。
主要なサービスが準備できたら、Gradio を使用してユーザー インターフェイスの作成を開始できます。

• グラディオサービス

JSONをインポート
gradio を gr としてインポート
輸入リクエスト
def greet(名前, 強度):
"Hello, " + name + "!" * int(intensity) を返します
def 回答(質問):
url = "http://127.0.0.1:8888/answer"
ペイロード = {"content": 質問}
res = リクエスト.post(url, json=ペイロード)
res_json = json.loads(res.text)
[res_json["回答"], res_json["ソース"]]を返す
デモ = gr.Interface(
fn=回答、
# inputs=["テキスト", "スライダー"],
入力=[gr.Textbox(label="question", lines=5)],
# 出力=[gr.TextArea(label="answer", lines=5),
# gr.JSON(ラベル="urls", 値=リスト)]
出力=[gr.Markdown(label="回答"),
gr.JSON(ラベル="urls", 値=リスト)]
）
demo.launch(サーバー名="0.0.0.0")