LLM部署，并发控制，流式响应（Python，Qwen2+FastAPI）

佑岸以沫

前言随着生成式人工智能的快速发展，部分场景希望能过自主部署大型语言模型（LLM）服务器用于推理服务，而相关教程博文尽管很多，但存在孤立零散现象，各功能没有打通实现，使得在工程实践中令人困惑。本文从工程实践的角度，着重从“并发控制”与“流式响应”两方面展开。FastAPI是一个简单高效的Web框架，可以快速实现大型语言模型（LLM）服务器独立部署，目前很多博客也写了相关教程。但在工程实践中，我们希望LLM服务器可以同时处理多个请求，并实现“当请求达到一定数量后，直接拒绝后续的推理请求”功能，防止服务器过载以及排队时间过长影响用户体验（不如直接告知服务器繁忙），本文在“并发控制”上做了进一步的相关工作。此外，对于较为复杂的问题（推理复杂、文本过长），希望能够“边推理便输出内容”，即“流式响应”，可以极大提高用户体验（与之相对的我称之为“一次响应”，即生成完后一次性全部返回），并在前述的并发控制框架之下实现。因此，本文实现了工程实践下并发控制下LLM服务器部署并提供流式响应。为便于理解，首先给出所需要的库（LLM服务器环境）。#WebfromfastapiimportFastAPI,Request,HTTPExceptionfromfastapi.responsesimportStreamingResponsefromsse_starlette.sseimportEventSourceResponseimportuvicorn#异步库importasyncio#线程池执行器fromconcurrent.futuresimportThreadPoolExecutorfromthreadingimportThread#LLMfromtransformersimportAutoTokenizer,AutoModelForCausalLM,GenerationConfigfromtransformersimportTextStreamer,TextIteratorStreamer#后者是可迭代的importtorchimportjsonimportdatetime客户端以Python支撑的Web为例，仅使用requests库即可。（本文创作于2024年7月）一、并发控制模型的加载使用FastAPI对LLM进行封装提供API服务。（为便于演示，使用Qwen2-0.5B，实际应用可以替换为其他模型）#创建FastAPI应用app=FastAPI()#主函数入口if__name__=='__main__':#加载预训练的分词器和模型now_model_place="E:\LLM\Qwen2_0.5B\Qwen2-0.5B-Instruct"model_name_or_path=now_model_place#加载分词器tokenizer=AutoTokenizer.from_pretrained(model_name_or_path,use_fast=False)#加载预训练模型，数据类型和自动选择设备model=AutoModelForCausalLM.from_pretrained(model_name_or_path,device_map="auto",torch_dtype=torch.bfloat16).eval()#启动FastAPI应用#用6006端口可以将autodl的端口映射到本地，从而在本地使用apiuvicorn.run(app,host='0.0.0.0',port=6006,workers=1)#在指定端口和主机上启动应用其中需要注意的是，.eval() 是将模型设置为评估模式，在预测或评估时使用，实现禁用特定于训练的操作如Dropout防止误修改模型权重。排队法对于基于Python的FastAPI，有多种并发与限流方法，比如asyncio.Semaphore（信号量）、令牌桶/漏桶算法等。在这里，本文采用排队法（基于asyncio.Queue）来跟踪正在处理与等待的请求，最终实现同时处理多个请求，并实现“当请求达到一定数量后，直接拒绝后续的推理请求”的功能。采用async-await的异步形式。首先创建队列，队列大小为10，即允许最大并发量为10。然后监听post请求，解析给定的prompt输入内容。如果当前队里已满，直接返回服务器繁忙提示；队列未满则加入队列，执行推理，结束后释放相应的队列占用，伪代码如下：#创建队列来跟踪请求数量（并发队列，数量为10）asyncio_queue=asyncio.Queue(maxsize=10)#处理POST请求的端点@app.post("/")asyncdefcreate_item(request:Request):#获取prompt（LLM的输入内容）json_post_raw=awaitrequest.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的“繁忙”提示ifasyncio_queue.full():now=datetime.datetime.now()#获取当前时间time=now.strftime("%Y-%m-%d%H:%M:%S")#格式化时间为字符串#连接成功，但是LLM服务器繁忙return{"response":"503HOT！LLM服务器推理繁忙，请稍后再试","status":200,"time":time}#将请求加入队列awaitasyncio_queue.put(None)try:#进行LLM推理returnanswer#返回响应finally:#任务完成，从队列中移除请求awaitasyncio_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()withThreadPoolExecutor()asexecutor:#使用执行器执行同步的函数#LLM_carry为LLM推理函数，prompt为输入内容answer=awaitloop.run_in_executor(executor,LLM_carry,prompt)#answer=LLM_carry(prompt)returnanswer#返回响应finally:#任务完成，从队列中移除请求awaitasyncio_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异步）defLLM_carry(prompt):globalmodel,tokenizer#声明全局变量以便在函数内部使用模型和分词器messages=[{"role":"system","content":"你是一个热情、积极向上而客观严谨的接待员，为客人提供问答服务。"},#{"role":"system","content":"Youareahelpfulassistant."},{"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')#使用model.generate()方法直接生成文本（一次响应）#generated_ids是由数字构成的tensorgenerated_ids=model.generate(model_inputs.input_ids,#模型输入的input_idsmax_new_tokens=512#最大新生成的token数量)#从生成的ID中提取新生成的ID部分generated_ids=[output_ids[len(input_ids):]forinput_ids,output_idsinzip(model_inputs.input_ids,generated_ids)]#使用分词器的batch_decode方法将生成的ID解码回文本，并跳过特殊tokenresponse=tokenizer.batch_decode(generated_ids,skip_special_tokens=True)[0]now=datetime.datetime.now()#获取当前时间time=now.strftime("%Y-%m-%d%H:%M:%S")#格式化时间为字符串#构建响应JSONanswer={"response":response,"status":200,"time":time}#构建日志信息log="["+time+"]"+'",prompt:"'+prompt+'",response:"'+repr(response)+'"'print(log)#打印日志torch_gc()#执行GPU内存清理returnanswer#返回响应可以看出，演示代码主要是使用了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异步）defLLM_carry(prompt):globalmodel,tokenizer#声明全局变量以便在函数内部使用模型和分词器messages=[{"role":"system","content":"你是一个热情、积极向上而客观严谨的接待员，为客人提供问答服务。"},#{"role":"system","content":"Youareahelpfulassistant."},{"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')#使用流式传输模式（更加流畅和动态的交互体验）#自定义、可迭代的流式输出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流式响应asyncdefstream_response():fortextinstreamer:iftext:print(text)此时，在服务器的终端上就可以流式（逐token）输出生成的文本了。FastAPI传输生成的文本不仅需要流式输出，还需要通过FastAPI来进行流式传输到客户端，即服务器实时响应，经收集整理可分为以下几种方案【4】【5】【6】【7】： HTTP：传统的“请求—响应”模型，客户端主动发请求，服务端被动地返回响应；可采用长轮询形式（需要维护大量长连接）。WebSocket：全双工通信，实现实时双向通信；太“重”了。流式传输（基于HTTP/1.1的分块传输）：FastAPI中的StreamingResponse，通常用于流式传输原始数据，如文件流或视频流，一般要求要传输的数据是确定的，适用于需要流式传输大量数据或长时间数据的场景，更有通用性。SSE（Server-sentEvents）：允许服务器主动向客户端发送数据的技术，使用标准的HTTP协议，仅提供服务端到客户端的单向通信。FastAPI中的EventSourceResponse（需要安装 sse_starlette进行拓展），提供了更适合LLM推理场景的特性，如事件驱动的设计和更好的实时性，同时更方便传输复杂的数据结构。（重点）显然，选择3、4方案可以有效解决流式传输问题，并且在实际使用中都可以实现需要的效果。考虑当前主流方向，采用SSE的主动推送方案（代码注释部分中也包括实现了的方案3）#FastAPI流式响应（StreamingResponse与EventSourceResponse）asyncdefstream_response():fortextinstreamer:iftext:#print(repr(text))#用yield迭代推送对话内容#StreamingResponse下的返回信息#yieldtext#EventSourceResponse下的返回信息yield{"event":"message","retry":15000,"data":repr(text)#防止\n换行符等传输过程中丢失}#returnStreamingResponse(stream_response(),media_type="application/octet-stream")returnEventSourceResponse(stream_response())需要说明的是，在采用EventSourceResponse直接返回text时，会出现换行符丢失的情况，产生的根本原因不明，这在以md语法输出的LLM推理中是致命问题。SSE响应内容格式定义：event:{event}\r\ndata:{data}\r\nretry:{retry}\r\n\r\n使用repr()显示字符形式来进行检查，发现LLM推理流式输出的token，在传输过程中被拆分了（原因不明），导致客户端解析错误。因此，通过返回字符形式的repr(text)使得能够在客户端正确解析。客户端获取来自服务器的流式响应需要使用合适的方式进行接收，本文以基于Python的Web为例，使用requests库的post请求，分别实现接收来自“一次响应”、StreamingResponse的流式传输、EventSourceResponse的SSE的数据。#直接返回defget_completion(prompt):headers={'Content-Type':'application/json'}data={"prompt":prompt}response=requests.post(url='http://127.0.0.1:6006',headers=headers,data=json.dumps(data))returnresponse.json()['response']#流式响应defget_completion_stream(prompt):headers={'Content-Type':'application/octet-stream'}data={"prompt":prompt}#LLM服务器采用StreamingResponsetry:response=requests.post(url='http://127.0.0.1:6006',headers=headers,data=json.dumps(data),stream=True)ifresponse.status_code==200:#列表，用于拼接流式返回的生成文本all_chunk_response=[]#chunk_size:默认为1，正常情况下要设置一个比较大的值，否则获取到一个字节数据就会走到下面的处理逻辑##decode_unicode:iter_content()函数遍历的数据是bytes类型的，这个参数可以控制是否将bytes转为strforchunkinresponse.iter_content(chunk_size=512,decode_unicode=True):#print(chunk)#StreamingResponse下的返回信息处理#all_chunk_response.append(chunk)#EventSourceResponse下的返回信息处理#解析SSE响应内容格式，分割出所需数据chunk_data=chunk.split("\r\ndata:")[1].split("\r\nretry:")[0]yieldchunk_dataall_chunk_response.append(chunk_data)all_chunk_response_text=''.join(all_chunk_response)#打印日志now=datetime.datetime.now()#获取当前时间time=now.strftime("%Y-%m-%d%H:%M:%S.%f")#格式化时间为字符串（微秒）log="["+time+"]"+all_chunk_response_text#print(log)else:print(response)exceptrequests.RequestExceptionase:print(f"Requestfailed:{e}")if__name__=='__main__':#print(get_completion('你好'))fordatainget_completion_stream('什么是大模型技术？有什么影响？'):print(data,end="")#流式显示数据在实际使用过程中，将客户端示例代码的主函数部分使用yield包装成迭代器调用即可。三、代码实现该部分给出本文实现的完整代码。LLM服务器（服务端）待进一步整理后完整开源。Web服务器（客户端）同二、客户端获取部分代码。四、小结本文实现了工程实践下并发控制下LLM服务器部署并提供流式响应。使得LLM服务器可以同时处理多个请求，并实现“当请求达到一定数量后，直接拒绝后续的推理请求”功能，防止服务器过载以及排队时间过长影响用户体验（直接告知服务器繁忙）。此外，对于较为复杂的问题（推理复杂、文本过长）生成回答，在并发架构下实现了“边推理便输出内容”的“流式响应”，可有效提高用户体验，优化人机交互实现。参考感谢智谱清言提供交互思考支持，以及其他同志的无私分享。【1】通义千问本地部署教程Qwen-7B-ChatQwen1.5-1.8BWindows-详细认真版_qwen1.5本地部署-CSDN博客 【2】transformers模块中的模型推理流式输出【3】【Qwen2部署实战】Qwen2初体验：用Transformers打造智能聊天机器人_fromtransformers.models可以加载qwen吗-CSDN博客【4】解密SSE，用Python像ChatGPT一样返回流式响应【5】使用FastAPI与aiohttp进行SSE响应开发【6】Python使用fastAPI实现一个流式传输接口_fastapi流式-CSDN博客【7】结合实例理解流式输出的几种实现方法