DataWorks:大規模言語モデルの使用

1. 大規模言語モデルノードの使用

DataWorks の Data Development は、大規模言語モデルノードを使用したデータ処理をサポートしています。このノードでは、大規模言語モデルサービスを設定して呼び出すことができます。

2. シェルノードによる大規模言語モデルの呼び出し

この例では、シェルノードから大規模言語モデルを呼び出して、特定の質問に回答する方法を示します。

1. シェルノードを作成し、次のサンプルコマンドを追加します。

   説明

   http で始まるモデルサービスエンドポイントと DW で始まるトークンを、「事前準備」セクションで取得した実際の値に置き換えてください。

   curl -X POST http://ms-xxx.cn-beijing.dataworks-model.aliyuncs.com/v1/completions -H "Authorization: DW-ms-xxx" -H "Content-Type: application/json" -d '{"prompt":"AI、機械学習、ディープラーニングの違いと関連性", "stream":"false", "max_tokens": 1024}' -v    

2. ノードを編集した後、ノードエディターの右側にある Run Configuration パネルで、モデルサービスとのネットワーク接続設定が完了しているリソースグループを選択します。

3. Run をクリックして、デプロイ済みのモデルサービスを呼び出します。

   説明

   • ノードスケジューリング設定：ノードをスケジュール実行する必要がある場合は、右側の Scheduling Settings パネルでネットワーク接続設定が完了しているリソースグループを選択します。その後、Scheduling Policy でスケジューリングプロパティを設定します。

   • ノード公開：本番環境でタスクを実行するには、画面上の アイコンをクリックして公開ワークフローを開始します。本番環境に公開された後にのみ、ノードはスケジュールに従って実行されます。

3. Python ノードによる大規模言語モデルの呼び出し

この例では、Python ノードから大規模言語モデルを呼び出して詩を生成する方法を示します。

1. この例では、Python の requests ライブラリが必要です。Alibaba Cloud イメージに基づいてカスタムイメージを作成し、この依存関係をインストールします。次のパラメーターを使用します。

   カスタムイメージパラメーター	説明
   イメージ名/ID	DataWorks イメージリスト から Python 互換のイメージを選択します。
   サポートされるタスクタイプ	Python を選択します。
   インストールパッケージ	Python3： requests。 スクリプト： /home/tops/bin/pip3 install 'urllib3<2.0'。

2. Python ノードを作成し、次のサンプルコードを追加します。

   説明

   http で始まるモデルサービスエンドポイントと DW で始まるトークンを、「事前準備」セクションで取得した実際の値に置き換えてください。

   import requests
   import json
   import time
   import sys
   def stream_print_response():
       httpUrl = "http://ms-xxxx.cn-beijing.dataworks-model.aliyuncs.com"
       apikey = "DW-ms-xxxx"
       url = httpUrl + "/v1/completions"
       headers = {
           "Authorization": apikey,
           "Content-Type": "application/json"
       }
       data = {
           "prompt": "春についての詩を書いてください",
           "stream": True,
           "max_tokens": 512
       }
       try:
           response = requests.post(url, headers=headers, json=data, stream=True)
           response.raise_for_status()
           full_text = ""  # 損失を防ぐために完全なレスポンスを蓄積
           buffer = ""     # オプション: 不完全な JSON 行を処理
           for line in response.iter_lines():
               if not line:
                   continue  # 空行をスキップ
               line_str = line.decode('utf-8').strip()
               # print(f"[デバッグ] 受信した行: {line_str}")  # デバッグ用
               if line_str.startswith("data:"):
                   data_str = line_str[5:].strip()  # "data: " を削除
                   if data_str == "[DONE]":
                       print("\n[ストリーミング応答が終了しました]")
                       break
                   # JSON の解析を試行
                   try:
                       parsed = json.loads(data_str)
                       choices = parsed.get("choices", [])
                       if choices:
                           delta_text = choices[0].get("text", "")
                           if delta_text:
                               # 完全なテキストに蓄積
                               full_text += delta_text
                               # 文字を 1 つずつ出力
                               for char in delta_text:
                                   print(char, end='', flush=True)
                                   sys.stdout.flush()
                                   time.sleep(0.03)  # タイプライター効果
                   except json.JSONDecodeError as e:
                       # print(f"[警告] JSON の解析に失敗しました: {e}, raw: {data_str}")
                       continue
           print(f"\n\n[完全なレスポンスの長さ: {len(full_text)} 文字]")
           print(f"[ 完全なコンテンツ]:\n{full_text}")
       except requests.exceptions.RequestException as e:
           print(f" リクエストに失敗しました: {e}")
       except Exception as e:
           print(f" その他のエラー: {e}")
   if __name__ == "__main__":
       stream_print_response()
   

3. ノードを編集した後、右側の Run Configuration パネルで、ネットワーク接続設定が完了しているリソースグループと、requests ライブラリを含む、ステップ1 で作成したカスタム ミラー を選択します。

4. Run をクリックして、デプロイ済みのモデルサービスを呼び出します。

   説明

   • ノードスケジューリング設定：ノードをスケジュール実行する必要がある場合は、Scheduling Settings パネルで、ネットワーク接続設定が完了しているリソースグループと、requests ライブラリを含む、ステップ1 で作成したカスタム ミラー を選択します。その後、Scheduling Policy でスケジューリングプロパティを設定します。

   • ノード公開：本番環境でタスクを実行するには、画面上の アイコンをクリックして公開ワークフローを開始します。本番環境に公開された後にのみ、ノードはスケジュールに従って実行されます。

1. 大規模言語モデルノードによるベクトルモデルサービスの呼び出し

DataWorks の Data Development は、大規模言語モデルノードを使用したデータ処理をサポートしています。このノードでは、ベクトルモデルサービスを設定して呼び出すことができます。

2. シェルノードによるベクトルモデルの呼び出し

この例では、シェルノードからベクトルモデルを呼び出してテキストをベクトルに変換する方法を示します。

1. シェルノードを作成し、次のサンプルコマンドを追加します。

   説明

   http で始まるモデルサービスエンドポイントと DW で始まるトークンを、「事前準備」セクションで取得した実際の値に置き換えてください。

   curl -X POST "http://ms-xxx.cn-beijing.dataworks-model.aliyuncs.com/v1/embeddings" \
     -H "Authorization: DW-ms-xxx" \
     -H "Content-Type: application/json" \
     -d '{
       "input": "これはベクトルに変換する必要があるテキストです",
       "model": "bge-m3"
         }'

2. ノードを編集した後、ノードエディターの右側にある Run Configuration パネルで、モデルサービスとのネットワーク接続設定が完了しているリソースグループを選択します。

3. Run をクリックして、デプロイ済みのモデルサービスを呼び出します。

   説明

   • ノードスケジューリング設定：ノードをスケジュール実行する必要がある場合は、Scheduling Settings パネルでネットワーク接続設定が完了しているリソースグループを選択します。その後、Scheduling Policy でスケジューリングプロパティを設定します。

   • ノード公開：本番環境でタスクを実行するには、画面上の アイコンをクリックして公開ワークフローを開始します。本番環境に公開された後にのみ、ノードはスケジュールに従って実行されます。

3. Python ノードによるベクトルモデルの呼び出し

この例では、Python ノードからベクトルモデルを呼び出してテキストをベクトルに変換する方法を示します。

1. この例では、Python の requests ライブラリが必要です。Alibaba Cloud イメージに基づいてカスタムイメージを作成し、この依存関係をインストールします。次のパラメーターを使用します。

   カスタムイメージパラメーター	説明
   イメージ名/ID	DataWorks イメージリスト から Python 互換のイメージを選択します。
   サポートされるタスクタイプ	Python を選択します。
   インストールパッケージ	Python3： requests。 スクリプト： /home/tops/bin/pip3 install 'urllib3<2.0'。

2. Python ノードを作成し、次のサンプルコードを追加します。

   説明

   http で始まるモデルサービスエンドポイントと DW で始まるトークンを、「事前準備」セクションで取得した実際の値に置き換えてください。

   import requests
   import json
   import sys
   # API エンドポイントとトークンに置き換えてください
   api_url = "http://ms-xxx.cn-beijing.dataworks-model.aliyuncs.com" + "/v1/embeddings"
   token = "DW-ms-xxx"
   print("api_url:"+api_url)
   print("token:"+token)
   headers = {
       "Authorization": f"{token}",
       "Content-Type": "application/json"
   }
   payload = {
       "input": "テストテキスト",
       "model": "bge-m3"
   }
   try:
       response = requests.post(api_url, headers=headers, data=json.dumps(payload))
       print("レスポンスステータスコード:", response.status_code)
       print("レスポンスコンテンツ:", response.text)  # 詳細なエラーメッセージの表示
   except Exception as e:
       print("リクエストエラー:", e)

3. ノードを編集した後、右側の Run Configuration パネルで、ネットワーク接続設定が完了しているリソースグループと、requests ライブラリを含む、ステップ1 で作成したカスタム ミラー を選択します。

4. Run をクリックして、デプロイ済みのモデルサービスを呼び出します。

   説明

   • ノードスケジューリング設定：ノードをスケジュール実行する必要がある場合は、Scheduling Settings パネルでネットワーク接続設定が完了しているリソースグループを選択します。その後、Scheduling Policy でスケジューリングプロパティを設定します。

   • ノード公開：本番環境でタスクを実行するには、画面上の アイコンをクリックして公開ワークフローを開始します。本番環境に公開された後にのみ、ノードはスケジュールに従って実行されます。