ナレッジベース API

# このサンプルコードは参照用です。本番環境で直接使用しないでください。
import os

from alibabacloud_bailian20231229 import models as bailian_20231229_models
from alibabacloud_bailian20231229.client import Client as bailian20231229Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

def check_environment_variables():
    """必要な環境変数を確認し、不足しているものを報告します。"""
    required_vars = {
        'ALIBABA_CLOUD_ACCESS_KEY_ID': 'ご利用の Alibaba Cloud AccessKey ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': 'ご利用の Alibaba Cloud AccessKey secret',
        'WORKSPACE_ID': 'ご利用の Model Studio ワークスペース ID'
    }
    missing_vars = []
    for var, description in required_vars.items():
        if not os.environ.get(var):
            missing_vars.append(var)
            print(f"エラー: {var} 環境変数を設定してください ({description})。")

    return len(missing_vars) == 0

def create_client() -> bailian20231229Client:
    """
    クライアントを作成して設定します。

    Returns:
        bailian20231229Client: 設定済みのクライアント。
    """
    config = open_api_models.Config(
        access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
        access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
    )
        # これはパブリッククラウドのエンドポイントです。必要に応じて変更できます。
    config.endpoint = 'bailian.ap-southeast-1.aliyuncs.com'
    return bailian20231229Client(config)

def retrieve_index(client, workspace_id, index_id, query):
    """
    指定されたナレッジベースから情報を取得します。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        index_id (str): ナレッジベース ID。
        query (str): 検索クエリ。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    retrieve_request = bailian_20231229_models.RetrieveRequest(
        index_id=index_id,
        query=query
    )
    runtime = util_models.RuntimeOptions()
    return client.retrieve_with_options(workspace_id, retrieve_request, headers, runtime)

def main():
    """
    Model Studio サービスを使用してナレッジベースから取得します。

    Returns:
        str or None: 呼び出しが成功した場合、取得されたテキストセグメント。それ以外の場合は None が返されます。
    """
    if not check_environment_variables():
        print("環境変数の検証に失敗しました。")
        return
    try:
        print("ステップ 1: クライアントを作成します。")
        client = create_client()
        print("ステップ 2: ナレッジベースから取得します。")
        index_id = input("ナレッジベース ID を入力してください: ")  # これは CreateIndex 操作によって返される Data.Id 値です。Model Studio コンソールのナレッジベースページからも取得できます。
        query = input("検索クエリを入力してください: ")
        workspace_id = os.environ.get('WORKSPACE_ID')
        resp = retrieve_index(client, workspace_id, index_id, query)
        result = UtilClient.to_jsonstring(resp.body)
        print(result)
    except Exception as e:
        print(f"エラーが発生しました: {e}")
        return None

if __name__ == '__main__':
    main()

# このサンプルコードは参照用です。本番環境では使用しないでください。
import hashlib
import os
import time

import requests
from alibabacloud_bailian20231229 import models as bailian_20231229_models
from alibabacloud_bailian20231229.client import Client as bailian20231229Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models

def check_environment_variables():
    """必要な環境変数を確認し、不足しているものがあればプロンプトを表示します。"""
    required_vars = {
        'ALIBABA_CLOUD_ACCESS_KEY_ID': 'Alibaba Cloud AccessKey ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': 'Alibaba Cloud AccessKey secret',
        'WORKSPACE_ID': 'Model Studio workspace ID'
    }
    missing_vars = []
    for var, description in required_vars.items():
        if not os.environ.get(var):
            missing_vars.append(var)
            print(f"エラー: {var} 環境変数を設定してください ({description})。")

    return len(missing_vars) == 0

# クライアントを作成します。
def create_client() -> bailian20231229Client:
    """
    クライアントを作成して設定します。

    Returns:
        bailian20231229Client: 設定済みのクライアント。
    """
    config = open_api_models.Config(
        access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
        access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
    )
    # これはパブリックエンドポイントです。必要に応じて変更できます。
    config.endpoint = 'bailian.ap-southeast-1.aliyuncs.com'
    return bailian20231229Client(config)

def calculate_md5(file_path: str) -> str:
    """
    ファイルの MD5 ハッシュを計算します。

    Args:
        file_path (str): ファイルのローカルパス。

    Returns:
        str: ファイルの MD5 ハッシュ。
    """
    md5_hash = hashlib.md5()

    # ファイルをバイナリモードで読み込みます。
    with open(file_path, "rb") as f:
        # 大きなファイルの場合、メモリ使用量を抑えるためにチャンクで読み込みます。
        for chunk in iter(lambda: f.read(4096), b""):
            md5_hash.update(chunk)

    return md5_hash.hexdigest()

def get_file_size(file_path: str) -> int:
    """
    ファイルのサイズをバイト単位で取得します。
    Args:
        file_path (str): ファイルのローカルパス。
    Returns:
        int: ファイルサイズ (バイト単位)。
    """
    return os.path.getsize(file_path)

# ファイルアップロードリースを申請します。
def apply_lease(client, category_id, file_name, file_md5, file_size, workspace_id):
    """
    Model Studio サービスからファイルアップロードリースを申請します。

    Args:
        client (bailian20231229Client): クライアント。
        category_id (str): カテゴリ ID。
        file_name (str): ファイルの名前。
        file_md5 (str): ファイルの MD5 ハッシュ。
        file_size (int): ファイルサイズ (バイト単位)。
        workspace_id (str): ワークスペース ID。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    request = bailian_20231229_models.ApplyFileUploadLeaseRequest(
        file_name=file_name,
        md_5=file_md5,
        size_in_bytes=file_size,
    )
    runtime = util_models.RuntimeOptions()
    return client.apply_file_upload_lease_with_options(category_id, workspace_id, request, headers, runtime)

# ファイルを一時ストレージにアップロードします。
def upload_file(pre_signed_url, headers, file_path):
    """
    Model Studio サービスにファイルをアップロードします。
    Args:
        pre_signed_url (str): アップロードリースからの URL。
        headers (dict): アップロード用のリクエストヘッダー。
        file_path (str): ファイルのローカルパス。
    """
    with open(file_path, 'rb') as f:
        file_content = f.read()
    upload_headers = {
        "X-bailian-extra": headers["X-bailian-extra"],
        "Content-Type": headers["Content-Type"]
    }
    response = requests.put(pre_signed_url, data=file_content, headers=upload_headers)
    response.raise_for_status()

# ファイルをカテゴリに追加します。
def add_file(client: bailian20231229Client, lease_id: str, parser: str, category_id: str, workspace_id: str):
    """
    Model Studio サービスの指定されたカテゴリにファイルを追加します。

    Args:
        client (bailian20231229Client): クライアント。
        lease_id (str): リース ID。
        parser (str): ファイルに使用するパーサー。
        category_id (str): カテゴリ ID。
        workspace_id (str): ワークスペース ID。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    request = bailian_20231229_models.AddFileRequest(
        lease_id=lease_id,
        parser=parser,
        category_id=category_id,
    )
    runtime = util_models.RuntimeOptions()
    return client.add_file_with_options(workspace_id, request, headers, runtime)

# ファイルの解析ステータスをクエリします。
def describe_file(client, workspace_id, file_id):
    """
    ファイルの解析ステータスをクエリします。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        file_id (str): ファイル ID。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    runtime = util_models.RuntimeOptions()
    return client.describe_file_with_options(workspace_id, file_id, headers, runtime)

# ドキュメントを追加するジョブを送信します。
def submit_index_add_documents_job(client, workspace_id, index_id, file_id, source_type):
    """
    解析済みのファイルをナレッジベースに追加します。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        index_id (str): ナレッジベース ID。
        file_id (str): ファイル ID。
        source_type(str): データの型。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    submit_index_add_documents_job_request = bailian_20231229_models.SubmitIndexAddDocumentsJobRequest(
        index_id=index_id,
        document_ids=[file_id],
        source_type=source_type
    )
    runtime = util_models.RuntimeOptions()
    return client.submit_index_add_documents_job_with_options(workspace_id, submit_index_add_documents_job_request,
                                                              headers, runtime)

# ドキュメント追加ジョブが完了するのを待ちます。
def get_index_job_status(client, workspace_id, job_id, index_id):
    """
    インデックスジョブのステータスをクエリします。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        index_id (str): ナレッジベース ID。
        job_id (str): ジョブ ID。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    get_index_job_status_request = bailian_20231229_models.GetIndexJobStatusRequest(
        index_id=index_id,
        job_id=job_id
    )
    runtime = util_models.RuntimeOptions()
    return client.get_index_job_status_with_options(workspace_id, get_index_job_status_request, headers, runtime)

# 古いファイルを削除します。
def delete_index_document(client, workspace_id, index_id, file_id):
    """
    指定されたナレッジベースから 1 つ以上のファイルを永久に削除します。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        index_id (str): ナレッジベース ID。
        file_id (str): ファイル ID。

    Returns:
        Model Studio サービスからのレスポンス。
    """
    headers = {}
    delete_index_document_request = bailian_20231229_models.DeleteIndexDocumentRequest(
        index_id=index_id,
        document_ids=[file_id]
    )
    runtime = util_models.RuntimeOptions()
    return client.delete_index_document_with_options(workspace_id, delete_index_document_request, headers, runtime)

def update_knowledge_base(
        file_path: str,
        workspace_id: str,
        index_id: str,
        old_file_id: str
):
    """
    Model Studio のナレッジベースを更新します。
    Args:
        file_path (str): 更新されたファイルのローカルパス。
        workspace_id (str): ワークスペース ID。
        index_id (str): 更新するナレッジベースの ID。
        old_file_id (str): 置き換えられるファイルの FileId。
    Returns:
        str or None: 成功した場合はナレッジベース ID、それ以外の場合は None。
    """
    # デフォルト値を設定します。
    category_id = 'default'
    parser = 'DASHSCOPE_DOCMIND'
    source_type = 'DATA_CENTER_FILE'
    try:
        # ステップ 1: クライアントを作成します。
        print("ステップ 1: クライアントを作成します。")
        client = create_client()
        # ステップ 2: ファイル情報を準備します。
        print("ステップ 2: ファイル情報を準備します。")
        file_name = os.path.basename(file_path)
        file_md5 = calculate_md5(file_path)
        file_size = get_file_size(file_path)
        # ステップ 3: Model Studio からアップロードリースを申請します。
        print("ステップ 3: Model Studio からアップロードリースを申請します。")
        lease_response = apply_lease(client, category_id, file_name, file_md5, file_size, workspace_id)
        lease_id = lease_response.body.data.file_upload_lease_id
        upload_url = lease_response.body.data.param.url
        upload_headers = lease_response.body.data.param.headers
        # ステップ 4: ファイルを一時ストレージにアップロードします。
        print("ステップ 4: ファイルを一時ストレージにアップロードします。")
        upload_file(upload_url, upload_headers, file_path)
        # ステップ 5: ファイルをカテゴリに追加します。
        print("ステップ 5: ファイルをカテゴリに追加します。")
        add_response = add_file(client, lease_id, parser, category_id, workspace_id)
        file_id = add_response.body.data.file_id
        # ステップ 6: Model Studio でファイルステータスを確認します。
        print("ステップ 6: Model Studio でファイルステータスを確認します。")
        while True:
            describe_response = describe_file(client, workspace_id, file_id)
            status = describe_response.body.data.status
            print(f"現在のファイルステータス: {status}")
            if status == 'INIT':
                print("ファイルは解析待ちです。お待ちください...")
            elif status == 'PARSING':
                print("ファイルは解析中です。お待ちください...")
            elif status == 'PARSE_SUCCESS':
                print("ファイルの解析が正常に完了しました。")
                break
            else:
                print(f"不明なファイルステータス: {status}。テクニカルサポートにお問い合わせください。")
                return None
            time.sleep(5)
        # ステップ 7: ドキュメントを追加するジョブを送信します。
        print("ステップ 7: ドキュメントを追加するジョブを送信します。")
        index_add_response = submit_index_add_documents_job(client, workspace_id, index_id, file_id, source_type)
        job_id = index_add_response.body.data.id
        # ステップ 8: インデックス作成ジョブが完了するのを待ちます。
        print("ステップ 8: インデックス作成ジョブが完了するのを待ちます。")
        while True:
            get_index_job_status_response = get_index_job_status(client, workspace_id, job_id, index_id)
            status = get_index_job_status_response.body.data.status
            print(f"現在のインデックスジョブのステータス: {status}")
            if status == 'COMPLETED':
                break
            time.sleep(5)
        # ステップ 9: 古いファイルを削除します。
        print("ステップ 9: 古いファイルを削除します。")
        delete_index_document(client, workspace_id, index_id, old_file_id)
        print("Model Studio ナレッジベースが正常に更新されました。")
        return index_id
    except Exception as e:
        print(f"エラーが発生しました: {e}")
        return None

def main():
    if not check_environment_variables():
        print("環境変数チェックに失敗しました。")
        return
    file_path = input("更新されたファイルのローカルパスを入力してください (例: Linux システムの場合 /path/to/Model_Studio_Phone_Series_Introduction.docx):")
    index_id = input("更新するナレッジベースの ID を入力してください: ")  # これは CreateIndex 操作によって返される Data.Id です。Model Studio コンソールのナレッジベースページからも取得できます。
    old_file_id = input("置き換えられるファイルの FileId を入力してください: ")  # これは AddFile 操作によって返される FileId です。Model Studio コンソールのアプリケーションデータページでファイル名の横にある ID アイコンをクリックして取得することもできます。
    workspace_id = os.environ.get('WORKSPACE_ID')
    update_knowledge_base(file_path, workspace_id, index_id, old_file_id)

if __name__ == '__main__':
    main()

# このサンプルコードは参照用です。本番環境では使用しないでください。
import os

from alibabacloud_bailian20231229 import models as bailian_20231229_models
from alibabacloud_bailian20231229.client import Client as bailian20231229Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

def check_environment_variables():
    """必要な環境変数が設定されていることを確認します。"""
    required_vars = {
        'ALIBABA_CLOUD_ACCESS_KEY_ID': 'Alibaba Cloud AccessKey ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': 'Alibaba Cloud AccessKey secret',
        'WORKSPACE_ID': 'Alibaba Cloud Model Studio workspace ID'
    }
    missing_vars = []
    for var, description in required_vars.items():
        if not os.environ.get(var):
            missing_vars.append(var)
            print(f"エラー: {var} 環境変数を設定してください ({description})")

    return len(missing_vars) == 0

# クライアントを作成します。
def create_client() -> bailian20231229Client:
    """
    クライアントを作成して設定します。

    Returns:
        bailian20231229Client: 設定済みのクライアント。
    """
    config = open_api_models.Config(
        access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
        access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
    )
    # 以下のエンドポイントはシンガポールリージョンのパブリッククラウド用です。必要に応じて変更してください。
    config.endpoint = 'bailian.ap-southeast-1.aliyuncs.com'
    return bailian20231229Client(config)

# ナレッジベースを一覧表示します。
def list_indices(client, workspace_id):
    """
    指定されたワークスペース内のナレッジベースを一覧表示します。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。

    Returns:
        Alibaba Cloud Model Studio サービスからのレスポンス。
    """
    headers = {}
    list_indices_request = bailian_20231229_models.ListIndicesRequest()
    runtime = util_models.RuntimeOptions()
    return client.list_indices_with_options(workspace_id, list_indices_request, headers, runtime)

# ナレッジベースを削除します。
def delete_index(client, workspace_id, index_id):
    """
    指定されたナレッジベースを永久に削除します。

    Args:
        client (bailian20231229Client): クライアント。
        workspace_id (str): ワークスペース ID。
        index_id (str): ナレッジベース ID。

    Returns:
        Alibaba Cloud Model Studio サービスからのレスポンス。
    """
    headers = {}
    delete_index_request = bailian_20231229_models.DeleteIndexRequest(
        index_id=index_id
    )
    runtime = util_models.RuntimeOptions()
    return client.delete_index_with_options(workspace_id, delete_index_request, headers, runtime)

def main():
    if not check_environment_variables():
        print("環境変数チェックに失敗しました。")
        return
    try:
        start_option = input(
            "実行する操作を選択してください:\n1. ナレッジベースを一覧表示\n2. ナレッジベースを削除\nオプションを入力 (1 または 2):")
        if start_option == '1':
            # ナレッジベースを一覧表示します。
            print("\nナレッジベースを一覧表示しています...")
            workspace_id = os.environ.get('WORKSPACE_ID')
            client = create_client()
            list_indices_response = list_indices(client, workspace_id)
            print(UtilClient.to_jsonstring(list_indices_response.body.data))
        elif start_option == '2':
            print("\nナレッジベースを削除しています...")
            workspace_id = os.environ.get('WORKSPACE_ID')
            index_id = input("ナレッジベース ID を入力してください: ")  # これは `CreateIndex` レスポンスの `Data.Id` 値です。Model Studio コンソールのナレッジベースページでも確認できます。
            # 削除前に確認します。
            while True:
                confirm = input(f"ナレッジベース {index_id} を永久に削除してもよろしいですか？ (y/n): ").strip().lower()
                if confirm == 'y':
                    break
                elif confirm == 'n':
                    print("削除はキャンセルされました。")
                    return
                else:
                    print("無効な入力です。y または n を入力してください。")
            client = create_client()
            resp = delete_index(client, workspace_id, index_id)
            if resp.body.status == 200:
                print(f"ナレッジベース {index_id} は正常に削除されました。")
            else:
                err_info = UtilClient.to_jsonstring(resp.body)
                print(f"エラーが発生しました: {err_info}")
        else:
            print("無効なオプションです。プログラムを終了します。")
            return
    except Exception as e:
        print(f"エラーが発生しました: {e}")
        return

if __name__ == '__main__':
    main()