DataWorks:Use large language models

1. Use a large language model node

DataWorks Data Development supports using a large language model node to process data. In this node, you can configure and invoke a large language model service.

2. Shell node invoking large language model

This example shows how to invoke a large language model from a Shell node to answer a specific question.

1. Create a Shell node and add the following sample command:

   Note

   Replace the model service endpoint starting with http and the token starting with DW with the actual values you obtained in the Preparations section.

   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":"Differences and connections among AI, machine learning, and deep learning", "stream":"false", "max_tokens": 1024}' -v    

2. After editing the node, in the Run Configuration panel on the right side of the node editor, select the resource group that has completed network connectivity configuration with the model service.

3. Click Run to invoke the deployed model service.

   Note

   • Node scheduling configuration: If the node needs to run on a schedule, select the resource group with completed network connectivity configuration in the Scheduling Settings panel on the right. Then configure scheduling properties in Scheduling Policy.

   • Node publishing: To run the task in a production environment, click the icon in the interface to start the publishing workflow. Only after publishing to the production environment will the node run on a schedule.

3. Invoke a large language model with a Python node

This example shows how to invoke a large language model from a Python node to generate a poem.

1. This example requires the Python requests library. Create a custom image based on an Alibaba Cloud Image to install this dependency. Use the following parameters:

   Custom image parameter	Description
   Image name/ID	Select a Python-compatible image from the DataWorks image list.
   Supported task type	Select Python.
   Installation packages	Python3: requests. Script: /home/tops/bin/pip3 install 'urllib3<2.0'.

2. Create a Python node and add the following sample code:

   Note

   Replace the model service endpoint starting with http and the token starting with DW with the actual values you obtained in the Preparations section.

   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": "Write a poem about spring",
           "stream": True,
           "max_tokens": 512
       }
       try:
           response = requests.post(url, headers=headers, json=data, stream=True)
           response.raise_for_status()
           full_text = ""  # Accumulate full response to prevent loss
           buffer = ""     # Optional: handle incomplete JSON lines
           for line in response.iter_lines():
               if not line:
                   continue  # Skip empty lines
               line_str = line.decode('utf-8').strip()
               # print(f"[DEBUG] Received line: {line_str}")  # For debugging
               if line_str.startswith("data:"):
                   data_str = line_str[5:].strip()  # Remove "data: "
                   if data_str == "[DONE]":
                       print("\n[Streaming response ended]")
                       break
                   # Attempt to parse JSON
                   try:
                       parsed = json.loads(data_str)
                       choices = parsed.get("choices", [])
                       if choices:
                           delta_text = choices[0].get("text", "")
                           if delta_text:
                               # Accumulate into full text
                               full_text += delta_text
                               # Print characters one by one
                               for char in delta_text:
                                   print(char, end='', flush=True)
                                   sys.stdout.flush()
                                   time.sleep(0.03)  # Typewriter effect
                   except json.JSONDecodeError as e:
                       # print(f"[Warning] JSON parsing failed: {e}, raw: {data_str}")
                       continue
           print(f"\n\n[Full response length: {len(full_text)} characters]")
           print(f"[ Full content]:\n{full_text}")
       except requests.exceptions.RequestException as e:
           print(f" Request failed: {e}")
       except Exception as e:
           print(f" Other error: {e}")
   if __name__ == "__main__":
       stream_print_response()
   

3. After editing the node, in the Run Configuration panel on the right, select the resource group with completed network connectivity configuration and the custom Image from step 1 that includes the requests library.

4. Click Run to invoke the deployed model service.

   Note

   • Node scheduling configuration: If the node needs to run on a schedule, select the resource group with completed network connectivity configuration and the custom Image from step 1 that includes the requests library in the Scheduling Settings panel. Then configure scheduling properties in Scheduling Policy.

   • Node publishing: To run the task in a production environment, click the icon in the interface to start the publishing workflow. Only after publishing to the production environment will the node run on a schedule.

1. LLM node invokes vector model service

DataWorks Data Development supports using a large model node to process data. In this node, you can configure and invoke a vector model service.

2. Shell node invokes vector model

This example shows how to convert text into vectors by invoking a vector model from a Shell node.

1. Create a Shell node and add the following sample command:

   Note

   Replace the model service endpoint starting with http and the token starting with DW with the actual values you obtained in the Preparations section.

   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. After editing the node, in the Run Configuration panel on the right, select the resource group that has completed network connectivity configuration with the model service.

3. Click Run to invoke the deployed model service.

   Note

   • Node scheduling configuration: If the node needs to run on a schedule, select the resource group with completed network connectivity configuration in the Scheduling Settings panel. Then configure scheduling properties in Scheduling Policy.

   • Node publishing: To run the task in a production environment, click the icon in the interface to start the publishing workflow. Only after publishing to the production environment will the node run on a schedule.

3. Invoke a vector model from a Python node

This example shows how to convert text into vectors by invoking a vector model from a Python node.

1. This example requires the Python requests library. Create a custom image based on an Alibaba Cloud Image to install this dependency. Use the following parameters:

   Custom image parameter	Description
   Image name/ID	Select a Python-compatible image from the DataWorks image list.
   Supported task type	Select Python.
   Installation packages	Python3: requests. Script: /home/tops/bin/pip3 install 'urllib3<2.0'.

2. Create a Python node and add the following sample code:

   Note

   Replace the model service endpoint starting with http and the token starting with DW with the actual values you obtained in the Preparations section.

   import requests
   import json
   import sys
   # Replace with your API endpoint and token
   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)  # View detailed error messages
   except Exception as e:
       print("Request error:", e)

3. After editing the node, in the Run Configuration panel on the right, select the resource group with completed network connectivity configuration and the custom Image from step 1 that includes the requests library.

4. Click Run to invoke the deployed model service.

   Note

   • Node scheduling configuration: If the node needs to run on a schedule, select the resource group with completed network connectivity configuration in the Scheduling Settings panel. Then configure scheduling properties in Scheduling Policy.

   • Node publishing: To run the task in a production environment, click the icon in the interface to start the publishing workflow. Only after publishing to the production environment will the node run on a schedule.