使用模型_大数据开发治理平台 DataWorks(DataWorks)-阿里云帮助中心

在网络连通的前提下，部署后的模型可在DataWorks同一地域的任意工作空间中使用。以下将为您介绍如何在DataWorks中调用和使用已部署的模型。

前提条件

已为DataWorks工作空间绑定Serverless资源组。
已在Serverless资源组部署大模型服务，详情请参见模型部署。

准备工作

在调用模型前，您需要确保调用模型的应用可正常访问您的大模型服务。在实际调用模型时，您需要先获取模型服务调用地址及API Key信息用于模型身份认证，以便后续调用模型。

1. 配置网络连通

请确保用于模型调用的DataWorks资源组绑定的VPC，已经在模型可连通的VPC列表中。

查看资源组绑定的VPC。
1. 前往DataWorks资源组列表页，在顶部切换至目标资源组所在地域，然后在资源组列表中找到目标资源组。
2. 单击目标资源组右侧操作列的网络设置，进入资源组专有网络绑定页。
3. 在数据调度 & 数据集成下查看对应的交换机网段。
查看模型可连通的VPC列表或为模型添加网络绑定，可参见管理模型网络。

2. 获取调用信息

模型部署完成后，会自动生成内网域名，请前往需要调用的模型服务详情页获取该模型服务的内网域名。域名格式为：http://<model-service-id>.<region>.dataworks-model.aliyuncs.com。获取方式可参见查看模型服务。

3. 获取API Key

权限验证信息，可通过管理API Key获取API Key信息。

调用大语言模型服务

可参考以下示例调用部署在DataWorks资源组的大型语言模型服务。更多命令细节可参见OpenAI-Compatible Server。

示例一：在Shell节点调用大型语言模型

本示例展示如何在Shell节点调用大型语言模型回答特定问题。

创建Shell节点，在Shell节点添加以下示例命令：

说明

请将代码中以http开头的大模型服务调用地址和以DW开头的Token信息替换为您在准备工作中获取的实际值。

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

编辑节点内容后，请在节点编辑页面右侧的调试配置中，选择已与服务模型完成网络连通配置的资源组。
单击运行节点，即可调用已部署的服务模型执行相关命令。
说明
- 节点调度：若调用服务模型的节点需要周期性调度执行，您需要在节点右侧的调度配置中选择已与服务模型完成网络连通配置的资源组，并在调度策略配置相关的调度属性。
- 节点发布：若任务需要发布至生产环境执行，请单击界面图标唤起发布流程，通过该流程将任务发布至生产环境。节点只有在发布至生产环境后，才会进行周期性调度。

示例二：在Python节点调用大型语言模型

本示例展示如何在Python节点通过调用大型语言模型完成写诗指令。

当前示例依赖Python的requests库，请参考以下主要参数，基于DataWorks官方镜像创建自定义镜像安装该依赖环境。

自定义镜像参数	配置说明
镜像名称/ID	您可基于DataWorks镜像列表选择适用于Python节点的镜像。
支持任务类型	选择Python。
安装包	Python3：`requests`。 Script：`/home/tops/bin/pip3 install 'urllib3<2.0'`。

创建Python节点，在Python节点添加如下示例代码：

说明

请将代码中以http开头的大模型服务调用地址和以DW开头的Token信息替换为您在准备工作中获取的实际值。

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] 收到行: {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

                            # 逐字打印新增的字符
                            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}, 原文: {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()

编辑节点内容后，在节点编辑页面右侧的调试配置中，选择已完成网络连通配置的资源组和步骤1中安装了requests库的自定义镜像。
单击运行节点，即可调用已部署的服务模型执行相关命令。
说明
- 节点调度：若调用服务模型的节点需要周期性调度执行，您需要在节点右侧的调度配置中选择已完成网络连通配置的资源组和步骤1中安装了requests库的自定义镜像，并在调度策略配置相关的调度属性。
- 节点发布：若任务需要发布至生产环境执行，请单击界面图标唤起发布流程，通过该流程将任务发布至生产环境。节点只有在发布至生产环境后，才会进行周期性调度。

调用向量模型服务

可参考以下示例调用部署在DataWorks资源组的向量（BGE-M3）模型服务。

示例一：在Shell节点调用向量模型

本示例展示如何在Shell节点通过调用向量模型将文字转化为向量。

创建Shell节点，在Shell节点添加以下示例命令：

说明

请将代码中以http开头的大模型服务调用地址和以DW开头的Token信息替换为您在准备工作中获取的实际值。

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"
        }'

编辑节点内容后，请在节点编辑页面右侧的调试配置中，选择已与服务模型完成网络连通配置的资源组。
单击运行节点，即可调用已部署的服务模型执行相关命令。
说明
- 节点调度：若调用服务模型的节点需要周期性调度执行，您需要在节点右侧的调度配置中选择已与服务模型完成网络连通配置的资源组，并在调度策略配置相关的调度属性。
- 节点发布：若任务需要发布至生产环境执行，请单击界面图标唤起发布流程，通过该流程将任务发布至生产环境。节点只有在发布至生产环境后，才会进行周期性调度。

示例二：在Python节点调用向量模型

本示例展示如何在Python节点通过调用向量模型将文字转化为向量。

当前示例依赖Python的requests库，请参考以下主要参数，基于DataWorks官方镜像创建自定义镜像安装该依赖环境。

自定义镜像参数	配置说明
镜像名称/ID	您可基于DataWorks镜像列表选择适用于Python节点的镜像。
支持任务类型	选择Python。
安装包	Python3：`requests`。 Script：`/home/tops/bin/pip3 install 'urllib3<2.0'`。

创建Python节点，在Python节点添加如下示例代码：

说明

请将代码中以http开头的大模型服务调用地址和以DW开头的Token信息替换为您在准备工作中获取的实际值。

import requests
import json
import sys

# 替换为你的API地址和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": "测试文本",
    "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)

编辑节点内容后，在节点编辑页面右侧的调试配置中，选择已完成网络连通配置的资源组和步骤1中安装了requests库的自定义镜像。
单击运行节点，即可调用已部署的服务模型执行相关命令。
说明
- 节点调度：若调用服务模型的节点需要周期性调度执行，您需要在节点右侧的调度配置中选择已与服务模型完成网络连通配置的资源组，并在调度策略配置相关的调度属性。
- 节点发布：若任务需要发布至生产环境执行，请单击界面图标唤起发布流程，通过该流程将任务发布至生产环境。节点只有在发布至生产环境后，才会进行周期性调度。