AWS Lambda容器化函数pickle解码遇UnicodeDecodeError问题排查

错误的根源在于AWS Lambda的API Gateway集成（包括SAM本地模拟环境）会自动将HTTP请求体按UTF-8字符串解析，而pickle序列化生成的是二进制数据，包含大量非UTF-8兼容的字节（比如报错中的0x80）。这个解码操作发生在你的Lambda代码执行之前——也就是说，当你的lambda_handler拿到event['body']时，API Gateway已经尝试对二进制请求体做UTF-8解码，直接触发了UnicodeDecodeError，你的pickle.loads代码根本没机会执行。

有两种可行的修复思路，根据你的场景选择：

方案一：用Base64编码传输二进制数据（推荐，无需修改API配置） #

将pickle的二进制数据转成Base64字符串（纯UTF-8兼容文本）传输，避免API Gateway的自动解码错误。

客户端代码修改 #

import requests
import pickle
import base64

texts = ['My first sentence', 'My second sentence']
# 序列化后转Base64字符串
pickled_data = pickle.dumps(texts)
base64_data = base64.b64encode(pickled_data).decode('utf-8')

my_url = "http://localhost:3000/wherewasit-embeddings/"
# 发送Base64字符串
response = requests.post(my_url, data=base64_data)
# 更规范的方式是用JSON格式传输
# response = requests.post(my_url, json={"payload": base64_data})

print(response)
# 如果需要解析返回结果：
# result_base64 = response.text
# result = pickle.loads(base64.b64decode(result_base64))

服务端Lambda代码修改 #

import pickle
import base64
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

def get_embeddings_raw(texts):
    try:
        return model.encode(texts)
    except:
        print("It failed!")
        return None

def lambda_handler(event, context):
    # 解析Base64字符串
    base64_data = event['body']
    pickled_data = base64.b64decode(base64_data)
    texts = pickle.loads(pickled_data)
    
    # 如果用JSON传输，替换为：
    # import json
    # body_json = json.loads(event['body'])
    # pickled_data = base64.b64decode(body_json['payload'])
    # texts = pickle.loads(pickled_data)
    
    result = get_embeddings_raw(texts)
    # 结果同样用Base64编码返回
    result_base64 = base64.b64encode(pickle.dumps(result)).decode('utf-8')
    return {
        "statusCode": 200,
        "body": result_base64
    }

方案二：配置API Gateway接受二进制媒体类型 #

如果必须直接传输二进制数据，需要在SAM模板中配置API Gateway的BinaryMediaTypes，允许特定类型的二进制请求体。

SAM模板（template.yaml）修改 #

Resources:
  WherewasitEmbeddingsFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./your-code-path
      Handler: app.lambda_handler
      Runtime: python3.9
      Events:
        EmbeddingsApi:
          Type: Api
          Properties:
            Path: /wherewasit-embeddings/
            Method: post
            RestApiId:
              Ref: EmbeddingsApi
  EmbeddingsApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      # 配置接受所有二进制类型，或指定application/octet-stream
      BinaryMediaTypes:
        - "*/*"

客户端代码修改 #

发送请求时设置Content-Type为二进制类型：

import requests
import pickle

texts = ['My first sentence', 'My second sentence']
pickled_data = pickle.dumps(texts)

my_url = "http://localhost:3000/wherewasit-embeddings/"
response = requests.post(
    my_url, 
    data=pickled_data, 
    headers={"Content-Type": "application/octet-stream"}
)

print(response)
# 解析返回结果：
# result = pickle.loads(base64.b64decode(response.content))

服务端Lambda代码修改 #

API Gateway会将二进制请求体转成Base64字符串存入event['body']，需要先解码：

import pickle
import base64
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

def get_embeddings_raw(texts):
    try:
        return model.encode(texts)
    except:
        print("It failed!")
        return None

def lambda_handler(event, context):
    # 解码Base64格式的请求体
    pickled_data = base64.b64decode(event['body'])
    texts = pickle.loads(pickled_data)
    
    result = get_embeddings_raw(texts)
    # 返回二进制结果时，需标记为Base64编码并设置正确的Content-Type
    return {
        "statusCode": 200,
        "body": base64.b64encode(pickle.dumps(result)).decode('utf-8'),
        "headers": {
            "Content-Type": "application/octet-stream"
        },
        "isBase64Encoded": True
    }

内容的提问来源于stack exchange，提问作者AlwaysLearning