知識庫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': '阿里雲存取金鑰ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': '阿里雲存取金鑰密碼',
        'WORKSPACE_ID': '阿里雲百鍊業務空間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

# 建立用戶端（Client）
def create_client() -> bailian20231229Client:
    """
    建立並配置用戶端（Client）。

    返回:
        bailian20231229Client: 配置好的用戶端（Client）。
    """
    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):
    """
    在指定的知識庫中檢索資訊。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        index_id (str): 知識庫ID。
        query (str): 檢索 query。

    返回:
        阿里雲百鍊服務的響應。
    """
    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():
    """
    使用阿里雲百鍊服務檢索知識庫。

    返回:
        str or None: 如果成功，返回檢索召回的文本切片；否則返回None。
    """
    if not check_environment_variables():
        print("環境變數校正未通過。")
        return
    try:
        print("步驟1：建立Client")
        client = create_client()
        print("步驟2：檢索知識庫")
        index_id = input("請輸入知識庫ID：")  # 即 CreateIndex 介面返回的 Data.Id，您也可以在阿里雲百鍊控制台的知識庫頁面擷取。
        query = input("請輸入檢索query：")
        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': '阿里雲存取金鑰ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': '阿里雲存取金鑰密碼',
        'WORKSPACE_ID': '百鍊工作空間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

# 建立用戶端（Client）
def create_client() -> bailian20231229Client:
    """
    建立並配置用戶端（Client）。

    返回:
        bailian20231229Client: 配置好的用戶端（Client）。
    """
    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值。

    參數:
        file_path (str): 檔案本地路徑。

    返回:
        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:
    """
    擷取檔案大小（以位元組為單位）。
    參數:
        file_path (str): 檔案本地路徑。
    返回:
        int: 檔案大小（以位元組為單位）。
    """
    return os.path.getsize(file_path)

# 申請檔案上傳租約
def apply_lease(client, category_id, file_name, file_md5, file_size, workspace_id):
    """
    從阿里雲百鍊服務申請檔案上傳租約。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        category_id (str): 類目ID。
        file_name (str): 檔案名稱。
        file_md5 (str): 檔案的MD5值。
        file_size (int): 檔案大小（以位元組為單位）。
        workspace_id (str): 業務空間ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    將檔案上傳到阿里雲百鍊服務。
    參數:
        lease_id (str): 租約ID。
        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):
    """
    將檔案添加到阿里雲百鍊服務的指定類目中。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        lease_id (str): 租約ID。
        parser (str): 用於檔案的解析器。
        category_id (str): 類目ID。
        workspace_id (str): 業務空間ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    擷取檔案的基本資料。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        file_id (str): 檔案ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    向一個文檔搜尋類知識庫追加匯入已解析的檔案。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        index_id (str): 知識庫ID。
        file_id (str): 檔案ID。
        source_type(str): 資料類型。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    查詢索引任務狀態。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        index_id (str): 知識庫ID。
        job_id (str): 任務ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    從指定的文檔搜尋類知識庫中永久刪除一個或多個檔案。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        index_id (str): 知識庫ID。
        file_id (str): 檔案ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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
):
    """
    使用阿里雲百鍊服務更新知識庫。
    參數:
        file_path (str): 檔案（更新後的）的實際本地路徑。
        workspace_id (str): 業務空間ID。
        index_id (str): 需要更新的知識庫ID。
        old_file_id (str): 需要更新的檔案的FileID。
    返回:
        str or None: 如果成功，返回知識庫ID；否則返回None。
    """
    # 設定預設值
    category_id = 'default'
    parser = 'DASHSCOPE_DOCMIND'
    source_type = 'DATA_CENTER_FILE'
    try:
        # 步驟1：建立用戶端（Client）
        print("步驟1：建立Client")
        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：申請上傳租約
        print("步驟3：向阿里雲百鍊申請上傳租約")
        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：檢查檔案狀態
        print("步驟6：檢查阿里雲百鍊中的檔案狀態")
        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)
        print("步驟9：刪除舊檔案")
        delete_index_document(client, workspace_id, index_id, old_file_id)
        print("阿里雲百鍊知識庫更新成功！")
        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為例：/xxx/xxx/阿里雲百鍊系列手機產品介紹.docx）：")
    index_id = input("請輸入需要更新的知識庫ID：")  # 即 CreateIndex 介面返回的 Data.Id，您也可以在阿里雲百鍊控制台的知識庫頁面擷取。
    old_file_id = input("請輸入需要更新的檔案的FileID：")  # 即 AddFile 介面返回的 FileId。您也可以在阿里雲百鍊控制台的應用資料頁面，單擊檔案名稱旁的 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': '阿里雲存取金鑰ID',
        'ALIBABA_CLOUD_ACCESS_KEY_SECRET': '阿里雲存取金鑰密碼',
        'WORKSPACE_ID': '阿里雲百鍊業務空間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

# 建立用戶端（Client）
def create_client() -> bailian20231229Client:
    """
    建立並配置用戶端（Client）。

    返回:
        bailian20231229Client: 配置好的用戶端（Client）。
    """
    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):
    """
    擷取指定業務空間下一個或多個知識庫的詳細資料。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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):
    """
    永久性刪除指定的知識庫。

    參數:
        client (bailian20231229Client): 用戶端（Client）。
        workspace_id (str): 業務空間ID。
        index_id (str): 知識庫ID。

    返回:
        阿里雲百鍊服務的響應。
    """
    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，您也可以在阿里雲百鍊控制台的知識庫頁面擷取。
            # 刪除前二次確認
            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()