问题

我已在Sagemaker中部署了dolly2模型，尝试为嵌入创建一些向量。代码在生成文本方面工作正常，但在将inference.py更改为处理嵌入后，出现了以下错误：

otocore.errorfactory.ModelError: 调用InvokeEndpoint操作时发生错误 (ModelError)：从主节点接收到客户端错误 (400)，消息为“{
"code": 400,
"type": "InternalServerException",
"message": "(\"You need to define one of the following ['audio-classification', 'automatic-speech-recognition', 'feature-extraction', 'text-classification', 'token-classification', 'question-answering', 'table-question-answering', 'visual-question-answering', 'document-question-answering', 'fill-mask', 'summarization', 'translation', 'text2text-generation', 'text-generation', 'zero-shot-classification', 'zero-shot-image-classification', 'conversational', 'image-classification', 'image-segmentation', 'image-to-text', 'object-detection', 'zero-shot-object-detection', 'depth-estimation', 'video-classification'] as env 'HF_TASK'.", 403)"
}

下面您还可以看到我用于嵌入的代码：

```python
import json
import os
import boto3
from transformers import pipeline

def invoke_sagemaker_endpoint():
    # 创建SageMaker客户端
    sagemaker_client = boto3.client("sagemaker-runtime")

    # 定义端点名称和有效负载
    endpoint_name = 'XXX'  # 替换为您的SageMaker端点名称
    payload = {"inputs": "这是一个大文档。"}  # 根据模型期望的有效负载格式更新

    # 将请求发送到SageMaker端点
    response = sagemaker_client.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Body=json.dumps(payload),
    )

    # 解析响应并提取嵌入向量
    response_body = response["Body"].read().decode("utf-8")
    response_json = json.loads(response_body)

    if "embeddings" in response_json:
        embeddings = response_json["embeddings"]
        embeddings_vector = embeddings[0]  # 嵌入以列表形式返回
        return embeddings_vector
    else:
        return None

if __name__ == "__main__":
    # 将HF_TASK环境变量设置为'feature-extraction'以获取嵌入
    os.environ["HF_TASK"] = "feature-extraction"
    # 调用SageMaker端点
    embeddings_vector = invoke_sagemaker_endpoint()

    if embeddings_vector:
        print(embeddings_vector)
    else:
        print("响应中未找到嵌入。")

以及inference.py：

import torch
from transformers import pipeline

def model_fn(model_dir):
    model = pipeline(
        "text-generation",
        model=model_dir,
        torch_dtype=torch.bfloat16,
        trust_remote_code=True,
        device_map="auto",
        model_kwargs={"load_in_8bit": True},
    )
    tokenizer = model.tokenizer
    embeddings_model = model.model

    def generate_embeddings(inputs):
        inputs = tokenizer(inputs, truncation=True, padding="longest", return_tensors="pt")
        with torch.no_grad():
            outputs = embeddings_model(**inputs)
        embeddings = outputs.last_hidden_state.mean(dim=1).squeeze(0).tolist()
        return embeddings

    def retrieve_qa(question, context):
        inputs = {"question": question, "context": context}
        qa_outputs = model(question, context)
        return qa_outputs

    return model, generate_embeddings, retrieve_qa

更改了推理（inference）和HF（Hugging Face），重新部署到Sagemaker，并从API Gateway调用，而不是从Sagemaker端点调用。

<details>
<summary>英文:</summary>

I have deployed the dolly2 model in sagemaker and I am trying to create some vectors for embeddings, the code works just fine for text generation but after changing the inference.py to handle embeddings, I am getting the error below

otocore.errorfactory.ModelError: An error occurred (ModelError) when calling the InvokeEndpoint operation: Received client error (400) from primary with message "{
"code": 400,
"type": "InternalServerException",
"message": "("You need to define one of the following [\u0027audio-classification\u0027, \u0027automatic-speech-recognition\u0027, \u0027feature-extraction\u0027, \u0027text-classification\u0027, \u0027token-classification\u0027, \u0027question-answering\u0027, \u0027table-question-answering\u0027, \u0027visual-question-answering\u0027, \u0027document-question-answering\u0027, \u0027fill-mask\u0027, \u0027summarization\u0027, \u0027translation\u0027, \u0027text2text-generation\u0027, \u0027text-generation\u0027, \u0027zero-shot-classification\u0027, \u0027zero-shot-image-classification\u0027, \u0027conversational\u0027, \u0027image-classification\u0027, \u0027image-segmentation\u0027, \u0027image-to-text\u0027, \u0027object-detection\u0027, \u0027zero-shot-object-detection\u0027, \u0027depth-estimation\u0027, \u0027video-classification\u0027] as env \u0027HF_TASK\u0027.", 403)"
}

Below you can also see the code that I am using for the embedding 

import json
import os
import boto3
from transformers import pipeline

def invoke_sagemaker_endpoint():
# Create a SageMaker client
sagemaker_client = boto3.client("sagemaker-runtime")

# Define the endpoint name and payload
endpoint_name = 'XXX'  # Replace with your SageMaker endpoint name
payload = {"inputs": "This is a large document."}  # Update the payload format as expected by the model

# Send the request to the SageMaker endpoint
response = sagemaker_client.invoke_endpoint(
    EndpointName=endpoint_name,
    ContentType="application/json",
    Body=json.dumps(payload),
)

# Parse the response and extract the embeddings vector
response_body = response["Body"].read().decode("utf-8")
response_json = json.loads(response_body)

if "embeddings" in response_json:
    embeddings = response_json["embeddings"]
    embeddings_vector = embeddings[0]  # embeddings are returned as a list
    return embeddings_vector
else:
    return None

if name == "main":
# Set the HF_TASK environment variable to 'feature-extraction' for the embeddings
os.environ["HF_TASK"] = "feature-extraction"
# Invoke the SageMaker endpoint
embeddings_vector = invoke_sagemaker_endpoint()

if embeddings_vector:
    print(embeddings_vector)
else:
    print("No embeddings found in the response.")

and the inference.py 

import torch
from transformers import pipeline

def model_fn(model_dir):
model = pipeline(
"text-generation",
model=model_dir,
torch_dtype=torch.bfloat16,
trust_remote_code=True,
device_map="auto",
model_kwargs={"load_in_8bit": True},
)
tokenizer = model.tokenizer
embeddings_model = model.model

def generate_embeddings(inputs):
    inputs = tokenizer(inputs, truncation=True, padding="longest", return_tensors="pt")
    with torch.no_grad():
        outputs = embeddings_model(**inputs)
    embeddings = outputs.last_hidden_state.mean(dim=1).squeeze(0).tolist()
    return embeddings

def retrieve_qa(question, context):
    inputs = {"question": question, "context": context}
    qa_outputs = model(question, context)
    return qa_outputs

return model, generate_embeddings, retrieve_qa

Changed the inference, the HF, redeployed to sagemaker, called from api gateway instead of sagemaker endpoint

</details>


# 答案1
**得分**: 1

嗨 [Arpel](https://stackoverflow.com/users/13038760/arpel)，看起来你正在混合两种不同的方法来部署 Hugging Face 模型作为 SageMaker 端点。

供日后参考，我看到你尝试设置 `HF_TASK` 环境变量，但是你正在调用 boto3 的实例上执行这个操作 — 这与将托管你的模型并执行推断的实例是_分开的_。请参考[此指南]()以获取关于在 SageMaker 和 HuggingFace 中进行**非自定义**推断的具体信息。

因为你想让模型执行两个任务 — 嵌入和问答 — 你正确地认识到你将需要一个自定义的 `inference.py` 文件。要采用这种方法，你需要执行以下步骤：

* 使用 git 克隆 Hugging Face 模型
* 创建一个 `code/` 目录（在模型目录内），并添加一个 `inference.py` 文件
* 在推断文件中包含两个函数，这些函数**必须**分别命名为 `model_fn()` 和 `predict_fn()`。前者仅在初始化端点时使用，必须返回模型和标记器，后者用于每个推断请求。你可以使用 `predict_fn()` 来包含自定义逻辑。
* 使用所有模型工件（包括自定义推断代码）创建一个 tarball（`model.tar.gz`）。其格式如下。

```plaintext
model.tar.gz/
|- pytorch_model.bin
|- ....
|- code/
  |- inference.py
  |- requirements.txt 

• 最后，将 tarball 上传到 S3，并在创建模型/端点时将 S3 URI 传递给 SageMaker。

Hugging Face 提供了一个很棒的笔记本，涵盖了整个过程。这是我迄今为止找到的最好的指南。如果你逐字复制它，只修改 inference.py 脚本，你应该能够成功。

以下是我之前使用过的 inference.py 的示例，如你所见，Hugging Face Pipelines 也可以使用！

from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification
from DirectQuoteUtils import reformat
import torch
import os

def model_fn(model_dir):
    tokenizer = AutoTokenizer.from_pretrained(model_dir)
    model = AutoModelForTokenClassification.from_pretrained(model_dir)
    pipe = pipeline("ner", model=model, tokenizer=tokenizer)
    return pipe

def predict_fn(data, pipeline):
    pipe = pipeline
    outputs = []
    
    # 模型输入的格式：
    # {               # 字符串列表
    #     "inputs": ["Donald Trump is the president of the US", "Joe Biden is the United States president"]
    # }
    
    modelData = pipe(data['inputs'])
    
    for prediction in modelData:
        cleanPred = reformat(prediction)
        outputs.append(cleanPred)
        
    return {
        # "device": device, # 用于检查是否使用了 CUDA 的有用信息
        "outputs": outputs
    }

model.tar.gz/
|- pytorch_model.bin
|- ....
|- code/
  |- inference.py
  |- requirements.txt 

from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification
from DirectQuoteUtils import reformat
import torch
import os

def model_fn(model_dir):
    tokenizer = AutoTokenizer.from_pretrained(model_dir)
    model = AutoModelForTokenClassification.from_pretrained(model_dir)
    pipe = pipeline("ner", model=model, tokenizer=tokenizer)
    return pipe

def predict_fn(data, pipeline):
    pipe = pipeline
    outputs = []
    
    # FORMAT FOR MODEL INPUT:
    # {               # list of strings
    #     "inputs": ["Donald Trump is the president of the US", "Joe Biden is the United States president"]
    # }
    
    modelData = pipe(data['inputs'])
    
    for prediction in modelData:
        cleanPred = reformat(prediction)
        outputs.append(cleanPred)
        
    return {
        # "device": device, # handy to check if CUDA is being used
        "outputs": outputs
    }