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

1. 大規模言語モデルノードでの大規模言語モデルの呼び出し

DataWorks の Data Studio (新バージョン) では、大規模言語モデルノードを使用してデータを処理できます。大規模言語モデルノードでは、大規模言語モデルサービスを構成し、モデルを呼び出すことができます。

2. Shell ノードでの大規模言語モデルの呼び出し

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

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

   説明

   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. [実行] をクリックして、デプロイされたモデルサービスを呼び出し、コマンドを実行します。

   説明

   • ノードのスケジューリング設定：モデルサービスを呼び出すノードを定期的に実行する場合は、右側の [スケジューリング] セクションに移動します。モデルサービスとのネットワーク接続設定が完了しているリソースグループを選択します。次に、[スケジューリングポリシー] セクションでスケジューリングプロパティを設定します。

   • ノードの公開：タスクを本番環境で実行する必要がある場合は、 アイコンをクリックしてタスクを公開します。ノードは、本番環境に公開された後にのみ定期的に実行されます。

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

この例では、Python ノードで大規模言語モデルを呼び出して、命令に従って詩を作成する方法を示します。

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

   カスタムイメージパラメーター	設定の説明
   イメージ名/ID	DataWorks イメージリストから Python ノードに適したイメージを選択できます。
   サポートされるタスクタイプ	Python を選択します。
   パッケージ	Python 3： 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"[DEBUG] Received line: {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"[WARNING] JSON parsing failed: {e}, original text: {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 セクションで、ネットワーク接続設定が完了しているリソースグループと、ステップ 1 で作成した requests ライブラリがインストールされているカスタム [イメージ] を選択します。

4. [実行] をクリックして、デプロイされたモデルサービスを呼び出し、コマンドを実行します。

   説明

   • ノードのスケジューリング設定：モデルサービスを呼び出すノードを定期的に実行するには、右側の [スケジューリング設定] セクションに移動します。ネットワーク接続設定が完了しているリソースグループと、ステップ 1 で作成した requests ライブラリがインストールされているカスタム [イメージ] を選択します。次に、[スケジューリングポリシー] セクションでスケジューリングプロパティを設定します。

   • ノードの公開：タスクを本番環境で実行する必要がある場合は、 アイコンをクリックしてタスクを公開します。ノードは、本番環境に公開された後にのみ定期的に実行されます。

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

DataWorks の Data Studio (新バージョン) では、大規模言語モデルノードを使用してデータを処理できます。大規模言語モデルノードでは、大規模言語モデルサービスを構成し、ベクトルモデルを呼び出すことができます。

2. Shell ノードでのベクトルモデルの呼び出し

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

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

   説明

   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": "This is a piece of text that needs to be converted into a vector",
       "model": "bge-m3"
         }'

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

3. [実行] をクリックしてコマンドを実行し、デプロイされたモデルサービスを呼び出します。

   説明

   • ノードのスケジューリング設定：モデルサービスを呼び出すノードを定期的に実行するには、右側の [スケジューリング] セクションに移動します。モデルサービスのネットワーク接続設定が完了しているリソースグループを選択します。次に、[スケジューリングポリシー] セクションでスケジューリングプロパティを設定します。

   • ノードの公開：タスクを本番環境で実行する必要がある場合は、 アイコンをクリックしてタスクを公開します。ノードは、本番環境に公開された後にのみ定期的に実行されます。

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

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

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

   カスタムイメージパラメーター	設定の説明
   イメージ名/ID	DataWorks イメージリストから Python ノードに適したイメージを選択できます。
   サポートされるタスクタイプ	Python を選択します。
   パッケージ	Python 3： 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": "Test text",
       "model": "bge-m3"
   }
   
   try:
       response = requests.post(api_url, headers=headers, data=json.dumps(payload))
       print("Response status code:", response.status_code)
       print("Response content:", response.text)  # 詳細なエラーメッセージを表示します。
   except Exception as e:
       print("Request exception:", e)

3. ノードの内容を編集した後、ノードエディターの右側にある Run Configuration セクションで、ネットワーク接続設定が完了しているリソースグループと、ステップ 1 で作成した requests ライブラリがインストールされているカスタム [イメージ] を選択します。

4. [実行] をクリックしてコマンドを実行し、デプロイされたモデルサービスを呼び出します。

   説明

   • ノードのスケジューリング設定：モデルサービスを呼び出すノードを定期的に実行するには、右側の [スケジューリング設定] セクションに移動します。モデルサービスへのネットワーク接続設定が完了しているリソースグループを選択します。次に、[スケジューリングポリシー] セクションでスケジューリングプロパティを設定します。

   • ノードの公開：タスクを本番環境で実行する必要がある場合は、 アイコンをクリックしてタスクを公開します。ノードは、本番環境に公開された後にのみ定期的に実行されます。