对于基于 Python 的 FastAPI，有多种并发与限流方法，比如 asyncio.Semaphore（信号量）、令牌桶/漏桶算法等。在这里，本文采用排队法（基于 asyncio.Queue）来跟踪正在处理与等待的请求，最终实现同时处理多个请求，并实现“当请求达到一定数量后，直接拒绝后续的推理请求”的功能。

采用 async-await 的异步形式。

首先创建队列，队列大小为10，即允许最大并发量为10。然后监听 post 请求，解析给定的 prompt 输入内容。如果当前队里已满，直接返回服务器繁忙提示；队列未满则加入队列，执行推理，结束后释放相应的队列占用，伪代码如下：

# 创建队列来跟踪请求数量（并发队列，数量为10）

asyncio_queue = asyncio.Queue(maxsize = 10)

# 处理POST请求的端点

@app.post("/")

async def create_item(request: Request):

# 获取 prompt （LLM的输入内容）

json_post_raw = await request.json() # 获取POST请求的JSON数据

json_post = json.dumps(json_post_raw) # 将JSON数据转换为字符串

json_post_list = json.loads(json_post) # 将字符串转换为Python对象

prompt = json_post_list.get('prompt') # 获取请求中的提示

# 获取了令牌许可：加入了请求队列（并发队列）

# 队列已满，则返回 503HOT 的“繁忙”提示

if asyncio_queue.full():

now = datetime.datetime.now() # 获取当前时间

time = now.strftime("%Y-%m-%d %H:%M:%S") # 格式化时间为字符串

# 连接成功，但是LLM服务器繁忙

return {"response": "503HOT！LLM服务器推理繁忙，请稍后再试",

"status": 200,

"time": time}

# 将请求加入队列

await asyncio_queue.put(None)

try:

# 进行LLM推理

return answer # 返回响应

finally:

# 任务完成，从队列中移除请求

await asyncio_queue.get()

asyncio_queue.task_done()

需要注意的是，本文所列方法均为单个工作进程，即在 Uvicorn 中 workers=1。若依托多核采用多个工作进程，需要使用外部的队列等缓存机制来实现分布式下的状态共享。

异步与同步

（重点）

若在上述“ LLM 推理”直接调用传统的推理形式，会发现在实际使用中，请求推理会逐个执行，同时后续请求会一直等待，并没有返回“LLM服务器繁忙”的提示。

这是由于，“ LLM 推理”属于同步任务，其基于深度学习，涉及到了不支持异步的库（如 PyTorch），造成阻塞。Python 的异步特性是基于事件循环的，而在单个线程中，当执行阻塞操作时，事件循环会被阻塞，无法执行其他任务。因此，即使使用了异步编程技术，如果底层的模型推理库不支持异步操作，造成在异步函数中调用了同步函数，那么在执行模型推理时仍然可能会出现阻塞。

但实际上，LLM 推理涉及过多，无法重新写成完全异步的形式。查阅 AI，给出了如下解决方案：

使用支持异步操作的库进行 LLM 推理使用批处理，把多个推理请求合并成一个，一次性执行使用分布式系统，多个服务器分散负载使用 asyncio.run_in_executor 将同步任务放入线程池或进程池

显然，在当前情况下，且资源受限，采用方案4为最佳选择，同时可以提高单机的并发能力，即解决方案：将同步的模型推理代码放入线程池中执行，从而避免阻塞事件循环，集成到异步框架下。

try:

# 创建一个线程执行器

loop = asyncio.get_event_loop()

with ThreadPoolExecutor() as executor:

# 使用执行器执行同步的函数

# LLM_carry为LLM推理函数，prompt为输入内容

answer = await loop.run_in_executor(executor, LLM_carry, prompt)

#answer = LLM_carry(prompt)

return answer # 返回响应

finally:

# 任务完成，从队列中移除请求

await asyncio_queue.get()

asyncio_queue.task_done()

在这里，使用了 concurrent.futures.ThreadPoolExecutor 来提供线程池执行器，线程池执行器允许你将可调用的对象（通常是函数）提交到线程池中执行，可以重用线程，这减少了线程创建和销毁的开销，并且在处理大量短生命周期的任务时能够提高性能。

在实际使用中，发现当使用 LLM（Qwen2-7B）运行时，GPU 占用首先直接达到 16GB 左右保持稳定（无推理任务）。尔后随并发下推理任务 GPU 占用逐渐小幅度增加，推理结束后存在回落（即：GPU 内存加载时大幅度占用，并发推理时小幅度增加与回落）。并且可以明显发现，LLM 推理确实是并行执行（现象：A\B\C 顺序请求，一段时间后 B 先返回响应，紧接着 A\B 返回响应）。推测原因：

（1）大模型在初始加载时占用大量显存。因为模型参数和必要的数据结构需要被加载到 GPU 内存中。这些数据包括模型的权重、梯度、优化器状态等。一旦模型被加载，这些显存就会一直被占用，直到模型被卸载或者 Python 进程结束。

（2）后续的并发推理中，显存占用的小幅度增加。因为在模型推理过程中，可能会生成一些临时变量，这些变量用于存储中间计算结果等。

上述技术方法与实验现象，为单机并发推理提供了理论与实践的支撑。

二、流式响应

当前，最容易实现的，是 LLM 服务器进行推理后，将生成的结果一次性返回到客户端（“一次响应”）。但当上下文过长、问题复杂时，推理时间会很长，导致用户使用体验感不佳（较长时间盯着加载图标转圈）。因此，第二部分着重解决“边推理便输出内容”（“流式响应”）。

LLM的流式输出

首先给出“一次响应”下的代码（改编自 Qwen2 示例代码）

# 大模型推理（这应该是一个同步的阻塞任务，导致该线程只进行此任务，无法处理其他的async异步）

def LLM_carry(prompt):

global model, tokenizer # 声明全局变量以便在函数内部使用模型和分词器

messages = [

{"role": "system", "content": "你是一个热情、积极向上而客观严谨的接待员，为客人提供问答服务。"},

#{"role": "system", "content": "You are a helpful assistant."},

{"role": "user", "content": prompt}

]

torch_gc() # 执行GPU内存清理

# 使用分词器的apply_chat_template方法来格式化消息

input_ids = tokenizer.apply_chat_template(

messages, # 要格式化的消息

tokenize=False, # 不进行分词

add_generation_prompt=True # 添加生成提示

)

# 将格式化后的文本转换为模型输入，并转换为PyTorch张量，然后移动到指定的设备（cuda）

model_inputs = tokenizer([input_ids], return_tensors="pt").to('cuda')code>

# 使用model.generate()方法直接生成文本（一次响应）

# generated_ids 是由数字构成的 tensor

generated_ids = model.generate(

model_inputs.input_ids, # 模型输入的input_ids

max_new_tokens=512 # 最大新生成的token数量

)

# 从生成的ID中提取新生成的ID部分

generated_ids = [

output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)

]

# 使用分词器的batch_decode方法将生成的ID解码回文本，并跳过特殊token

response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]

now = datetime.datetime.now() # 获取当前时间

time = now.strftime("%Y-%m-%d %H:%M:%S") # 格式化时间为字符串

# 构建响应JSON

answer = {

"response": response,

"status": 200,

"time": time

}

# 构建日志信息

log = "[" + time + "] " + '", prompt:"' + prompt + '", response:"' + repr(response) + '"'

print(log) # 打印日志

torch_gc() # 执行GPU内存清理

return answer # 返回响应

可以看出，演示代码主要是使用了 model.generate() 方法来直接进行文本生成。

对于 Qwen2 的流式输出，存在 swift.llm 的解决方案【1】（同为阿里在 LLM 领域布局），但截止2024年7月初，本人参考 Qwen-7B-Chat 教程对 Qwen2 进行流式输出尚未成功（确实本人水平有限）。因此只能采用其他方法。

目前（2024年7月），LLM 领域虽然百花齐放，但大多数投入开发的仍属于 Transformer 结构之下，因此，也离不开 transformers 这一开源库。经资料搜集，transformers 确实提供了模型推理的流式输出【2】，并且确实可以用于Qwen2【3】，因此采用 transformers 来实现Qwen2的“流式响应”。

（重点）

transformers 库有两个流式输出函数，TextStreamer 与 TextIterateStreamer，前者属于终端进行输出，后者是返回一个可迭代对象，更加具有自定义性，因此我们使用 TextIterateStreamer。

# 大模型推理（这应该是一个同步的阻塞任务，导致该线程只进行此任务，无法处理其他的async异步）

def LLM_carry(prompt):

global model, tokenizer # 声明全局变量以便在函数内部使用模型和分词器

messages = [

{"role": "system", "content": "你是一个热情、积极向上而客观严谨的接待员，为客人提供问答服务。"},

#{"role": "system", "content": "You are a helpful assistant."},

{"role": "user", "content": prompt}

]

torch_gc() # 执行GPU内存清理

# 使用分词器的apply_chat_template方法来格式化消息

input_ids = tokenizer.apply_chat_template(

messages, # 要格式化的消息

tokenize=False, # 不进行分词

add_generation_prompt=True # 添加生成提示

)

# 将格式化后的文本转换为模型输入，并转换为PyTorch张量，然后移动到指定的设备（cuda）

model_inputs = tokenizer([input_ids], return_tensors="pt").to('cuda')code>

# 使用流式传输模式（更加流畅和动态的交互体验）

# 自定义、可迭代的流式输出

streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)

#另一种执行流式输出的写法

#generation_kwargs = dict(model_inputs, streamer=streamer, max_new_tokens=512)

#thread = Thread(target=model.generate, kwargs=generation_kwargs)

#thread.start()

# 执行流式输出

model.generate(model_inputs.input_ids, max_new_tokens=512, streamer=streamer)

# FastAPI流式响应

async def stream_response():

for text in streamer:

if text:

print(text)

此时，在服务器的终端上就可以流式（逐 token）输出生成的文本了。

FastAPI传输