解决AWS上Spacy长文本应用的API Gateway超时问题

我在AWS上部署了基于Spacy的NLP模型，以API形式对外提供服务，调用链路为：AWS API Gateway → Lambda → SageMaker Endpoint。处理长文本时，因API Gateway的29秒超时限制，请求经常失败。

我已经尝试对长文档分块，并用以下代码并行处理：

for doc in clf.pipe(cleaned_text_chunks, n_process = 2, batch_size = 64):
    docs.append(doc)
ner_output = Doc.from_docs(docs)

但调整n_process参数、升级实例到ml.m4.2xlarge都没带来性能提升，现在需要可行的性能优化方案。

相关配置 #

Spacy配置 #

[paths]
train = null
dev = null
vectors = null
init_tok2vec = null

[system]
gpu_allocator = null
seed = 0

[nlp]
lang = "xx"
pipeline = ["tok2vec","ner"]
batch_size = 1000
disabled = []
before_creation = null
after_creation = null
after_pipeline_creation = null
tokenizer = {"@tokenizers":"spacy.Tokenizer.v1"}

[components]

[components.ner]
factory = "ner"
incorrect_spans_key = "incorrect_spans"
moves = null
scorer = {"@scorers":"spacy.ner_scorer.v1"}
update_with_oracle_cut_size = 100

[components.ner.model]
@architectures = "spacy.TransitionBasedParser.v2"
state_type = "ner"
extra_state_tokens = false
hidden_width = 64
maxout_pieces = 2
use_upper = true
nO = null

[components.ner.model.tok2vec]
@architectures = "spacy.Tok2VecListener.v1"
width = ${components.tok2vec.model.encode.width}
upstream = "*"

[components.tok2vec]
factory = "tok2vec"

[components.tok2vec.model]
@architectures = "spacy.Tok2Vec.v2"

[components.tok2vec.model.embed]
@architectures = "spacy.MultiHashEmbed.v2"
width = ${components.tok2vec.model.encode.width}
attrs = ["NORM","PREFIX","SUFFIX","SHAPE"]
rows = [5000,2500,2500,2500]
include_static_vectors = true

[components.tok2vec.model.encode]
@architectures = "spacy.MaxoutWindowEncoder.v2"
width = 256
depth = 8
window_size = 1
maxout_pieces = 3

[corpora]

[corpora.dev]
@readers = "spacy.Corpus.v1"
path = ${paths.dev}
max_length = 0
gold_preproc = false
limit = 0
augmenter = null

[corpora.train]
@readers = "spacy.Corpus.v1"
path = ${paths.train}
max_length = 0
gold_preproc = false
limit = 0
augmenter = null

[training]
dev_corpus = "corpora.dev"
train_corpus = "corpora.train"
seed = ${system.seed}
gpu_allocator = ${system.gpu_allocator}
dropout = 0.1
accumulate_gradient = 1
patience = 16000
max_epochs = 0
max_steps = 20000
eval_frequency = 200
frozen_components = []
annotating_components = []
before_to_disk = null

[training.batcher]
@batchers = "spacy.batch_by_words.v1"
discard_oversize = false
tolerance = 0.2
get_length = null

[training.batcher.size]
@schedules = "compounding.v1"
start = 100
stop = 1000
compound = 1.001
t = 0.0

[training.logger]
@loggers = "spacy.ConsoleLogger.v1"
progress_bar = false

[training.optimizer]
@optimizers = "Adam.v1"
beta1 = 0.9
beta2 = 0.999
L2_is_weight_decay = true
L2 = 0.01
grad_clip = 1.0
use_averages = false
eps = 0.00000001
learn_rate = 0.001

[training.score_weights]
ents_f = 1.0
ents_p = 0.0
ents_r = 0.0
ents_per_type = null

[pretraining]

[initialize]
vectors = ${paths.vectors}
init_tok2vec = ${paths.init_tok2vec}
vocab_data = null
lookups = null
before_init = null
after_init = null

[initialize.components]

[initialize.components.ner]

[initialize.components.ner.labels]
@readers = "spacy.read_labels.v1"
path = "/labels/ner.json"

[initialize.tokenizer]

Flask启动脚本 #

#!/usr/bin/env python

# This file implements the scoring service shell. You don't necessarily need to modify it for various
# algorithms. It starts nginx and gunicorn with the correct configurations and then simply waits until
# gunicorn exits.
#
# The flask server is specified to be the app object in wsgi.py
#
# We set the following parameters:
#
# Parameter                Environment Variable              Default Value
# ---------                --------------------              -------------
# number of workers        MODEL_SERVER_WORKERS              the number of CPU cores
# timeout                  MODEL_SERVER_TIMEOUT              60 seconds

from __future__ import print_function
import multiprocessing
import os
import signal
import subprocess
import sys

cpu_count = multiprocessing.cpu_count()

# model_server_timeout = os.environ.get('MODEL_SERVER_TIMEOUT', 60)
model_server_timeout = 3600
model_server_workers = int(os.environ.get('MODEL_SERVER_WORKERS', cpu_count))

def sigterm_handler(nginx_pid, gunicorn_pid):
    try:
        os.kill(nginx_pid, signal.SIGQUIT)
    except OSError:
        pass
    try:
        os.kill(gunicorn_pid, signal.SIGTERM)
    except OSError:
        pass

    sys.exit(0)

def start_server():
    print('Starting the inference server with {} workers.'.format(model_server_workers))


    # link the log streams to stdout/err so they will be logged to the container logs
    subprocess.check_call(['ln', '-sf', '/dev/stdout', '/var/log/nginx/access.log'])
    subprocess.check_call(['ln', '-sf', '/dev/stderr', '/var/log/nginx/error.log'])

    nginx = subprocess.Popen(['nginx', '-c', '/opt/program/nginx.conf'])
    gunicorn = subprocess.Popen(['gunicorn',
                                 '--timeout', str(model_server_timeout),
                                 '-k', 'gevent',
                                 '-b', 'unix:/tmp/gunicorn.sock',
                                 '-w', str(model_server_workers),
                                 'wsgi:app'])

    signal.signal(signal.SIGTERM, lambda a, b: sigterm_handler(nginx.pid, gunicorn.pid))

    # If either subprocess exits, so do we.
    pids = set([nginx.pid, gunicorn.pid])
    while True:
        pid, _ = os.wait()
        if pid in pids:
            break

    sigterm_handler(nginx.pid, gunicorn.pid)
    print('Inference server exiting')

# The main routine just invokes the start function.

if __name__ == '__main__':
    start_server()

nginx.conf脚本 #

worker_processes 1;
  daemon off; # Prevent forking


  pid /tmp/nginx.pid;
  error_log /var/log/nginx/error.log;

  events {
    # defaults
  }

  http {
    include /etc/nginx/mime.types;
    default_type application/octet-stream;
    access_log /var/log/nginx/access.log combined;
    proxy_connect_timeout  3600s;
    proxy_send_timeout  3600s;
    proxy_read_timeout  3600s;
    upstream gunicorn {
      server unix:/tmp/gunicorn.sock;
    }

    server {
      listen 8080 deferred;
      client_max_body_size 5m;

      keepalive_timeout 3600s;
      proxy_connect_timeout  3600s;
      proxy_send_timeout  3600s;
      proxy_read_timeout  3600s;

      location ~ ^/(ping|invocations) {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $http_host;
        proxy_set_header Connection "";
        proxy_redirect off;
        proxy_pass http://gunicorn;
        proxy_http_version 1.1;
        proxy_connect_timeout  3600s;
        proxy_send_timeout  3600s;
        proxy_read_timeout  3600s;
      }

      location / {
        return 404 "{}";
      }
    }
  }

端点配置 #

一、突破API Gateway超时限制：异步调用改造 #

API Gateway的29秒硬限制无法直接突破，必须切换异步调用模式：

• 方案1：API Gateway + SQS + Lambda + SageMaker
  1. API Gateway接收请求后，立即将任务写入SQS队列，返回任务ID给客户端
  2. 后台Lambda监听SQS消息，触发SageMaker处理长文本
  3. 处理完成后将结果存入DynamoDB，客户端通过轮询或SNS推送获取结果
• 方案2：API Gateway异步集成
  开启API Gateway异步支持，配置Lambda作为后端，请求进入后API Gateway直接返回202 Accepted，Lambda处理完成后通过回调URL通知客户端

二、优化Spacy推理性能 #

1. 启用GPU加速 #

当前Spacy配置未启用GPU，换成ml.g4dn系列GPU实例（如ml.g4dn.xlarge），修改配置：

[system]
gpu_allocator = "pytorch"

加载模型时指定GPU设备：

nlp = spacy.load("model", enable=["tok2vec", "ner"], device=0)

GPU对NLP模型推理速度提升显著，即使是传统Tok2Vec+NER架构也能获得明显收益。

2. 优化并行处理逻辑 #

• 调整分块大小：每个文本块控制在500-1000个token，过小的块会增加进程切换开销
• 设置n_process等于实例CPU核心数（ml.m4.2xlarge有8核，设为8）
• 对齐batch_size与Spacy配置中的nlp.batch_size（当前为1000），避免内部重复分块
• 改用multiprocessing.Pool手动控制进程，减少Spacy内部并行的额外开销：

from multiprocessing import Pool

def process_chunk(chunk):
    return clf(chunk)

with Pool(processes=8) as pool:
    docs = pool.map(process_chunk, cleaned_text_chunks)
ner_output = Doc.from_docs(docs)

3. 模型轻量化改造 #

• 缩减Tok2Vec规模：将tok2vec.model.encode的depth从8降到4，width从256降到128，以小幅精度损失换取速度提升
• 确保只加载必要组件：保持pipeline = ["tok2vec","ner"]，禁用所有未使用的组件
• 替换为Transformer架构：如果业务允许，用预训练Transformer（如bert-base-multilingual-cased）替换Tok2Vec，结合GPU可大幅提升推理效率

三、SageMaker端点优化 #

1. 实例类型选择 #

• CPU实例选ml.c5系列：比m4系列更适合计算密集型任务，如ml.c5.2xlarge，CPU性能更高
• 配置多实例端点：开启自动扩缩容，避免单实例过载，提升并发处理能力

2. 容器配置调整 #

• 替换gunicorn worker类型：将gevent改为sync，Spacy是CPU密集型任务，协程无法提升性能，反而增加开销：

gunicorn = subprocess.Popen(['gunicorn',
                             '--timeout', str(model_server_timeout),
                             '-k', 'sync',
                             '-b', 'unix:/tmp/gunicorn.sock',
                             '-w', str(model_server_workers),
                             'wsgi:app'])

• 调整worker数量：设为cpu_count * 2，但不超过实例核心数的2倍，避免上下文切换过载

3. 批量转换替代实时端点 #

如果长文本处理是批量任务，使用SageMaker批量转换API，效率远高于实时端点，且不受API Gateway超时限制

四、Lambda层优化 #

• 将Spacy模型打包成Lambda层，减小部署包体积，加快冷启动速度
• 调整Lambda超时到最大值（15分钟），确保能等待SageMaker长文本处理完成（异步方案无需此操作）

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