AI Search Open Platform 上にエンドツーエンドのマルチモーダルデータ前処理パイプラインを構築し、ドキュメントと画像を解析し、コンテンツをチャンキングして、テキスト埋め込みを生成します。
ユースケース
マルチモーダルデータ前処理パイプラインは、ドキュメント解析、画像解析、ドキュメントチャンキング、テキスト埋め込み、スパーステキスト埋め込みの 5 つのサービスを通じて、非構造化ドキュメントと画像を処理します。すべてのサービスは、AI Search Open Platform API を介して呼び出します。料金は実際の使用量に基づきます。
前提条件
-
AI Search Open Platform が有効化されていること。サービスの有効化。
-
API エンドポイントと API キーが用意されていること。サービスエンドポイントの照会、API キーの管理。
サービスは、インターネット経由または VPC 経由で呼び出すことができます。クロスリージョンの VPC 呼び出しは、China (Shanghai)、China (Hangzhou)、China (Shenzhen)、China (Beijing)、China (Zhangjiakou)、China (Qingdao) リージョンでサポートしています。
マルチモーダルデータ前処理パイプラインの構築
使いやすさを考慮し、AI Search Open Platform では以下の 4 種類の開発フレームワークを提供しています。
-
Java SDK
-
Python SDK
-
既に LangChain 開発フレームワークを使用している場合は、LangChain を選択します。
-
既に LlamaIndex 開発フレームワークを使用している場合は、LlamaIndex を選択します。
ステップ 1：サービスの選択とコードのダウンロード
この手順では、Python SDK を使用してマルチモーダルデータ前処理パイプラインを構築します。
-
AI Search Open Platform コンソールにログインします。
-
China (Shanghai) リージョンを選択し、AI 検索オープンプラットフォーム に切り替えてから、ターゲットのワークスペースに切り替えます。
説明現在、AI Search Open Platform は、中国 (上海) およびドイツ (フランクフルト) リージョンでのみ利用可能です。
中国 (杭州)、中国 (深圳)、中国 (北京)、中国 (張家口)、および中国 (青島) リージョンのユーザーは、VPC エンドポイントを使用してリージョン間で AI Search Open Platform サービスを呼び出すことができます。
-
左側のナビゲーションペインで、シナリオセンター を選択します。[マルチモーダルデータの前処理 - データの解析とベクトル化] を見つけて、進む をクリックします。
-
ドロップダウンリストから、必要なサービスを選択します。各サービスの詳細情報は 詳細 ページで確認できます。
説明-
各サービスにはサービス ID (
service_id) が必要です。たとえば、ドキュメント解析サービスの ID はops-document-analyze-001です。 -
生成されたコード内の
service_idは、サービスを切り替えると自動的に更新されます。コードをダウンロードした後に手動でservice_idを変更することもできます。
段階
説明
ドキュメント内容解析
ドキュメント内容解析サービス (
ops-document-analyze-001) ：非構造化ドキュメント (テキスト、表、画像) からタイトルや段落などの論理構造を抽出し、構造化されたコンテンツを返します。画像内容解析
-
画像内容理解サービス (
ops-image-analyze-vlm-001) ：マルチモーダル LLM を使用して画像コンテンツを解析し、テキストを認識して、画像の検索や質疑応答に利用します。 -
画像テキスト認識サービス (
ops-image-analyze-ocr-001) ：OCR を使用して画像内のテキストを認識し、検索や質疑応答に利用します。
ドキュメントチャンキング
ドキュメントチャンキングサービス (
ops-document-split-001) ：HTML、Markdown、TXT 形式の構造化データを段落、セマンティクス、またはカスタムルールに従って分割します。また、リッチテキストからコード、画像、表を抽出します。テキスト埋め込み
-
OpenSearch テキスト埋め込みサービス-001 (
ops-text-embedding-001) ：40 以上の言語をサポート。最大入力：300 トークン。出力：1,536 次元ベクトル。 -
OpenSearch 一般テキスト埋め込みサービス-002 (
ops-text-embedding-002) ：100 以上の言語をサポート。最大入力：8,192 トークン。出力：1,024 次元ベクトル。 -
OpenSearch テキスト埋め込みサービス-中国語-001 (
ops-text-embedding-zh-001) ：中国語専用。最大入力：1,024 トークン。出力：768 次元ベクトル。 -
OpenSearch テキスト埋め込みサービス-英語-001 (
ops-text-embedding-en-001) ：英語専用。最大入力：512 トークン。出力：768 次元ベクトル。
スパーステキスト埋め込み
テキストをスパースベクトル表現に変換します。スパースベクトルはストレージ要件が少なく、キーワードと用語の頻度を捉えます。密ベクトルと組み合わせてハイブリッド検索を行うことで、検索精度を向上させます。
OpenSearch スパーステキスト埋め込みサービス (
ops-text-sparse-embedding-001) ：100 以上の言語をサポート。最大入力：8,192 トークン。 -
サービスを選択した後、設定完了、コード照会へ をクリックしてコードを表示し、ダウンロードします。次の表は、データ前処理パイプラインのランタイム フローを説明しています。
|
関数 |
説明 |
|
ドキュメントと画像の解析、ドキュメントのチャンキング、テキスト埋め込みなど、ドキュメントの処理を行います。 |
|
コード照会 で [ドキュメントの解析とベクトル化] を選択し、コードのコピー または ファイルのダウンロード をクリックして、コードをローカルに保存します。
ステップ 2：環境の設定とパイプラインのテスト
コードをダウンロードした後、次のパラメーターを設定します。
|
カテゴリ |
パラメーター |
説明 |
|
AI Search Open Platform |
api_key |
認証用の API キー。API キーの管理。 |
|
aisearch_endpoint |
API エンドポイント。サービスエンドポイントの照会。 説明
API は、インターネット経由または VPC 経由で呼び出すことができます。 |
|
|
workspace_name |
AI Search Open Platform のワークスペース名。 |
|
|
service_id |
サービス ID。 |
Python 3.8.1 以降でコードを実行して、結果を確認します。
ドキュメント 「AI Search Open Platform の概要」 を前処理した際のサンプル出力：
C:\Users\issuser\PycharmProjects\pythonProject630Demo\.venv\Scripts\python.exe C:\Users\issuser\PycharmProjects\pythonProject3\.venv\Lib\site-packages\alibabacloud_searchplat2...
document_analyze task_id:f6dd92de-01e4-46a2-96c4-b80e7b120239
document_analyze done
document-split done, chunks count: 6 rich text count:9
image analyze :https://img.alicdn.com/imgextra/i2/O1CN01bYc1m81RrcSAyOjMu_!!6000000002165-54-tps-60-60.apng
https://img.alicdn.com/imgextra/i2/O1CN01bYc1m81RrcSAyOjMu_!!6000000002165-54-tps-60-60.apng is unanalysable.
image analyze :https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/3873436171/p802381.png
image analyze :https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/0517992271/p821030.png
image analyze ://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
https://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is unanalysable.
image analyze ://img.alicdn.com/tfs/TB1..50QpXXXXX7XpXXXXXXXXXX-40-40.png
image analyze :https://img.alicdn.com/tfs/TB1UdKEM6TpK1RjSZKPXXa3UpXa-256-256.png
image analyze :https://img.alicdn.com/tfs/TB1A0dINW6qK1RjSZFmXXX0PFXa-258-258.jpg
image analyze ://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
https://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is unanalysable.
image analyze ://img.alicdn.com/tfs/TB1..50QpXXXXX7XpXXXXXXXXXX-40-40.png
text-embedding done.
Process finished with exit code 0完全なパイプラインコード：
# マルチモーダルデータ処理パイプライン
# 要件：
# Python バージョン： 3.8.1 以降
# パッケージ要件：
# pip install alibabacloud_searchplat20240529
# AI Search Open Platform の設定
aisearch_endpoint = "xxx.platform-cn-shanghai.opensearch.aliyuncs.com"
api_key = "OS-xxx"
workspace_name = "default"
service_id_config = {"document_analyze": "ops-document-analyze-001",
"split": "ops-document-split-001",
"text_embedding": "ops-text-embedding-001",
"text_sparse_embedding": "ops-text-sparse-embedding-001",
"image_analyze": "ops-image-analyze-ocr-001"}
# 入力ドキュメントの URL。 サンプルドキュメントは AI Search Open Platform の概要です。
document_url = "https://www.alibabacloud.com/help/zh/open-search/search-platform/product-overview/introduction-to-search-platform?spm=a2c4g.11186623.0.0.7ab93526WDzQ8z"
import asyncio
from operator import attrgetter
from typing import List
from Tea.exceptions import TeaException, RetryError
from alibabacloud_tea_openapi.models import Config
from alibabacloud_searchplat20240529.client import Client
from alibabacloud_searchplat20240529.models import GetDocumentSplitRequest, CreateDocumentAnalyzeTaskRequest, \
CreateDocumentAnalyzeTaskRequestDocument, GetDocumentAnalyzeTaskStatusRequest, \
GetDocumentSplitRequestDocument, GetTextEmbeddingRequest, GetTextEmbeddingResponseBodyResultEmbeddings, \
GetTextSparseEmbeddingRequest, GetTextSparseEmbeddingResponseBodyResultSparseEmbeddings, \
GetImageAnalyzeTaskStatusResponse, CreateImageAnalyzeTaskRequest, GetImageAnalyzeTaskStatusRequest, \
CreateImageAnalyzeTaskRequestDocument, CreateImageAnalyzeTaskResponse
async def poll_doc_analyze_task_result(ops_client, task_id, service_id, interval=5):
while True:
request = GetDocumentAnalyzeTaskStatusRequest(task_id=task_id)
response = await ops_client.get_document_analyze_task_status_async(workspace_name, service_id, request)
status = response.body.result.status
if status == "PENDING":
await asyncio.sleep(interval)
elif status == "SUCCESS":
return response
else:
print("error: " + response.body.result.error)
raise Exception("document analyze task failed")
def is_analyzable_url(url:str):
if not url:
return False
image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff'}
return url.lower().endswith(tuple(image_extensions))
async def image_analyze(ops_client, url):
try:
print("image analyze :" + url)
if url.startswith("//"):
url = "https:" + url
if not is_analyzable_url(url):
print(url + " is unanalysable.")
return url
image_analyze_service_id = service_id_config["image_analyze"]
document = CreateImageAnalyzeTaskRequestDocument(
url=url,
)
request = CreateImageAnalyzeTaskRequest(document=document)
response: CreateImageAnalyzeTaskResponse = ops_client.create_image_analyze_task(workspace_name, image_analyze_service_id, request)
task_id = response.body.result.task_id
while True:
request = GetImageAnalyzeTaskStatusRequest(task_id=task_id)
response: GetImageAnalyzeTaskStatusResponse = ops_client.get_image_analyze_task_status(workspace_name, image_analyze_service_id, request)
status = response.body.result.status
if status == "PENDING":
await asyncio.sleep(5)
elif status == "SUCCESS":
return url + response.body.result.data.content
else:
print("image analyze error: " + response.body.result.error)
return url
except Exception as e:
print(f"image analyze Exception : {e}")
def chunk_list(lst, chunk_size):
for i in range(0, len(lst), chunk_size):
yield lst[i:i + chunk_size]
async def document_pipeline_execute(document_url: str = None, document_base64: str = None, file_name: str = None):
# AI Search Open Platform クライアントを作成します。
config = Config(bearer_token=api_key,endpoint=aisearch_endpoint,protocol="http")
ops_client = Client(config=config)
# ステップ 1： ドキュメント/画像の解析
document_analyze_request = CreateDocumentAnalyzeTaskRequest(document=CreateDocumentAnalyzeTaskRequestDocument(url=document_url, content=document_base64,file_name=file_name, file_type='html'))
document_analyze_response = await ops_client.create_document_analyze_task_async(workspace_name=workspace_name,service_id=service_id_config["document_analyze"],request=document_analyze_request)
print("document_analyze task_id:" + document_analyze_response.body.result.task_id)
extraction_result = await poll_doc_analyze_task_result(ops_client, document_analyze_response.body.result.task_id, service_id_config["document_analyze"])
print("document_analyze done")
document_content = extraction_result.body.result.data.content
content_type = extraction_result.body.result.data.content_type
# ステップ 2： ドキュメントのチャンキング
document_split_request = GetDocumentSplitRequest(
GetDocumentSplitRequestDocument(content=document_content, content_type=content_type))
document_split_result = await ops_client.get_document_split_async(workspace_name, service_id_config["split"],
document_split_request)
print("document-split done, chunks count: " + str(len(document_split_result.body.result.chunks))
+ " rich text count:" + str(len(document_split_result.body.result.rich_texts)))
# ステップ 3： テキストの埋め込み
# チャンキング結果を抽出します。 画像チャンクの場合、画像解析サービスがテキストコンテンツを抽出します。
doc_list = ([{"id": chunk.meta.get("id"), "content": chunk.content} for chunk in document_split_result.body.result.chunks]
+ [{"id": chunk.meta.get("id"), "content": chunk.content} for chunk in document_split_result.body.result.rich_texts if chunk.meta.get("type") != "image"]
+ [{"id": chunk.meta.get("id"), "content": await image_analyze(ops_client,chunk.content)} for chunk in document_split_result.body.result.rich_texts if chunk.meta.get("type") == "image"]
)
# 1 回のリクエストで最大 32 個の埋め込みを計算できます。
chunk_size = 32
all_text_embeddings: List[GetTextEmbeddingResponseBodyResultEmbeddings] = []
for chunk in chunk_list([text["content"] for text in doc_list], chunk_size):
response = await ops_client.get_text_embedding_async(workspace_name,service_id_config["text_embedding"],GetTextEmbeddingRequest(chunk))
all_text_embeddings.extend(response.body.result.embeddings)
all_text_sparse_embeddings: List[GetTextSparseEmbeddingResponseBodyResultSparseEmbeddings] = []
for chunk in chunk_list([text["content"] for text in doc_list], chunk_size):
response = await ops_client.get_text_sparse_embedding_async(workspace_name,service_id_config["text_sparse_embedding"],GetTextSparseEmbeddingRequest(chunk,input_type="document",return_token=True))
all_text_sparse_embeddings.extend(response.body.result.sparse_embeddings)
for i in range(len(doc_list)):
doc_list[i]["embedding"] = all_text_embeddings[i].embedding
doc_list[i]["sparse_embedding"] = all_text_sparse_embeddings[i].embedding
print("text-embedding done.")
if __name__ == "__main__":
# 非同期タスクを実行します。
# import nest_asyncio # Jupyter Notebook で実行する場合は、次の 2 行のコメントを解除してください。
# nest_asyncio.apply() # Jupyter Notebook で実行する場合は、次の 2 行のコメントを解除してください。
asyncio.run(document_pipeline_execute(document_url))
# asyncio.run(document_pipeline_execute(document_base64="eHh4eHh4eHg...", file_name="attention.pdf")) # 別の呼び出し方法