vllm加速（以Qwen2.5-7B-instruction为例）与流式响应
1. vllm介绍
-
什么是vllm?
vLLM 是一个高性能的大型语言模型推理引擎，采用创新的内存管理和执行架构，显著提升了大模型推理的速度和效率。它支持高度并发的请求处理，能够同时服务数千名用户，并且兼容多种深度学习框架，方便集成到现有的机器学习流程中。通过一个名为PagedAttention的新型注意力算法来解决传统LLM在生产环境中部署时所遇到的高内存消耗和计算成本的挑战。PagedAttention算法能有效管理注意力机制中的键和值，将它们分割成更小、更易于管理的块，从而减少了vLLM的内存占用，并使其吞吐量超过传统LLM服务方法。
-
为什么需要vllm:
1) 推理加速
2) 模型多并发推理
LLM是同步的，无法进行多并发推理的。正常情况下，当有多个请求依次短时间内进入LLM时，LLM一次只能处理一个问题，其它请求就会进入堵塞状态，直到该问题处理后才会处理下一个。如果强行使用异步或者线程池让模型处理多个请求，只能导致模型内部报错（这个作者尝试了下，确实会报错）。
具体实现原理：待续…
2. vllm中大模型创建的两种方法：LLM（同步）和AsyncLLMEngine（异步）
在 VLLM 框架中，AsyncLLMEngine 和 LLM 的主要区别在于它们的设计目的和使用方式：
- LLM
同步执行：LLM 是一个同步的语言模型接口，通常用于简单的推理任务。在调用时，用户会等待模型处理完请求，才会返回结果。
易于使用：对于初学者或简单应用，LLM 提供了一个直接且简单的接口，可以快速获取推理结果。
执行llm.generate()函数返回的是文本结果，无法做api的流式输出 - AsyncLLMEngine
异步执行：AsyncLLMEngine 设计用于处理高并发请求，使用异步编程模型，允许同时处理多个请求而不阻塞主线程。
性能优化：适合需要高吞吐量和低延迟的应用场景，如实时聊天机器人或在线服务，能够更有效地利用系统资源。
复杂性：使用 AsyncLLMEngine 可能需要开发者对异步编程有一定的理解，但它能显著提升处理能力。
执行llm.generate()函数返回的是一个生成器，用于流式输出 - 总结
如果你的应用需要处理大量并发请求或对响应时间有严格要求，建议使用 AsyncLLMEngine。
对于简单的任务或低并发场景，LLM 可能更为合适和易于实现。
3. 流式输出、非流失输出和vllm的同步、异步关系
对于vllm同步：无论是流式还是非流式输出，vllm的LLM函数创建的模型对象通常以同步的方式工作，处理多并发情况时只能以队列形式一个个输出。对于非流式输出，它会阻塞直到生成完成并返回结果；对于流式输出，它也可以逐步返回数据给前端，但这是假流式，因为后端以及把所有的文本都输出了，然后我们又把文本一个个传给前端。
对vllm异步：异步引擎同样可以支持流式和非流式输出，但它允许你以非阻塞的方式处理这些输出。你可以启动一个生成任务而不等待它完成，然后根据需要逐步获取流式输出，或者在任务完成后一次性获取非流式输出（也是并发状态）。这为高并发环境下的应用提供了更好的性能和灵活性。
总结：流式输出和非流式输出关注的是输出的传输方式，而AsyncLLMEngine和LLM则更多地涉及到执行模式（同步 vs 异步）。两者可以组合使用，例如，你可以使用AsyncLLMEngine来异步地处理流式输出，从而在高并发环境中获得最佳性能和用户体验。
4. vllm的LLM（同步）
from transformers import AutoTokenizer
from vllm import LLM, SamplingParamstokenizer = AutoTokenizer.from_pretrained("模型地址")
sampling_params = SamplingParams(temperature=0.7, top_p=0.8, repetition_penalty=1, max_tokens=1024)
llm = LLM(model="模型地址")prompt = "请帮我写一篇800字的关于“春天”的作文"
messages = [{"role": "system", "content": "You are Qwen, created by Alibaba Cloud. You are a helpful assistant."},{"role": "user", "content": prompt}
]
text = tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True
)# text为文本和dict都可，但是对格式有特殊要求。
outputs = llm.generate([text], sampling_params=sampling_params)
# # Print the outputs.
for output in outputs:prompt = output.promptgenerated_text = output.outputs[0].textprint(f"Prompt: {prompt!r}, Generated text: {generated_text!r}")
上述代码中对于llm.generate的输入，研究代码底层函数，可以发现llm.generate的输入prompts可以是数组，即多个问题批量处理（批量处理还是一个线程，并不是说llm的批量处理是使用多个线程处理的），也可以是单个元素：dict或者文本。
为dict时，dict必须包含prompt_token_ids的key，value为最终文本的编码。如果不包含prompt_token_ids的属性时，就必须存在prompt这个key（value为未编码的最终文本，即|im_read|这种最终文本格式），用于底层编码为embedding，并交给llm生成答案。所以说如果对llm.generate传入的是向量编码，就必须是最终文本的向量编码，并且放在prompt_token_ids这个key中。
为str时，则必须是这个大模型的训练时最后交给llm推理答案的格式（注意：上述代码中的text已经apply_chat_template过了，变成|im_read|那种格式了），即|im_read|这种最终文本格式。而代码底层会如果判断为str，就会直接将其视为prompt最终文本，并编码为向量保存在prompt_token_ids中交给llm生成答案。
根据上述延伸一下：当使用vllm加速自己微调过的模型时，微调模型时的训练集最后交给llm进行推理的格式，必须要和vllm得apply_chat_templat之后得格式一样（如果vllm输入为文本而并非编码向量的话），即训练和推理时，模型的输入格式要相同。
还需要注意llm.generate是没有经过包装的chat函数，它接收的输入如果是对话指令则进行对话，如果是文本，则直接进行文本续写。需要要看传入的text是一段文本还是{‘role’: ‘user’, ‘content’：’‘}格式的对话，决定生产的文本类型。如果是chat函数函数，则只能进行对话，且只需要输入问题，而不需要apply_chat_template函数包装。
5. vllm异步的AsyncLLMEngine（异步）
# 后端代码
import os
import json
import uuid
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import StreamingResponse
from transformers import AutoTokenizer
from vllm import AsyncEngineArgs, AsyncLLMEngine, SamplingParams# 响应对象
class ChatCompletionStreamResponse(BaseModel):success: boolcontent: str = ''# 将python对象转化为json对象
def jsonify(data: "BaseModel") -> str:try: # pydantic v2return json.dumps(data.model_dump(exclude_unset=True), ensure_ascii=False)except AttributeError: # pydantic v1return data.json(exclude_unset=True, ensure_ascii=False)def event_serializer(obj):return {'success': obj.success, 'content': obj.content}app = FastAPI()
tokenizer = AutoTokenizer.from_pretrained("/media/houfengzhen/张璐璐/chentao/Qwen2.5-7B-Instruct")
engine_args = {"model": "/media/houfengzhen/张璐璐/chentao/Qwen2.5-7B-Instruct","trust_remote_code": True,"dtype": 'bfloat16',"enforce_eager": True,"tensor_parallel_size": 1,"gpu_memory_utilization": 0.9
}
prompt = "你是谁"
messages = [{"role": "system", "content": "You are Qwen, created by Alibaba Cloud. You are a helpful assistant."},{"role": "user", "content": prompt}
]
text = tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True
)llm = AsyncLLMEngine.from_engine_args(AsyncEngineArgs(**engine_args))
@app.post("/api")
async def chat_complete():# 注意此处为流式输出，非流式输出直接将生成的内容全接接收完后，直接return即可# 流式输出的返回需要StreamingResponse包装！！！！return StreamingResponse(generate_text(), media_type="text/event-stream")async def generate_text():sampling_params = SamplingParams(repetition_penalty=1.0or 1.0, # repetition_penalty must > 0temperature=0.7,stop_token_ids=[tokenizer.eos_token_id] + tokenizer.additional_special_tokens_ids,max_tokens=512,skip_special_tokens=True)request_id = "chatcmpl-{}".format(uuid.uuid4().hex)# 注意此处的输入，text和embdding都不能为[],只能是单个元素：str或者{}。# embdding需要为dict，且{}中存在key: prompt_token_ids、prompt# text 则是最终文本即可result_generator = llm.generate(text,sampling_params=sampling_params,request_id=request_id,lora_request=None,)# result_generator = llm.generate(# {'prompt_token_ids': [151644, 8948, 198, 56568, 101909, 99471, 47764, 104799, 102064, 101057, 151645, 198, 151644, 872, 198, 36667, 52183, 27369, 28311, 6, 29991, 6, 30440, 99250, 105395, 17714, 5122, 675, 198, 6, 18, 24, 8908, 108, 223, 73670, 111138, 102992, 102204, 99661, 11319, 6, 30440, 99250, 105395, 17714, 5122, 18, 24, 13, 10479, 1231, 8683, 438, 264, 25202, 18630, 18239, 320, 43, 934, 86427, 27705, 13, 16, 13, 220, 56007, 36987, 104595, 99556, 28946, 18987, 91282, 69249, 100630, 100165, 11319, 6, 30440, 99250, 105395, 17714, 5122, 35, 13, 16, 13, 1207, 25, 10479, 374, 5230, 304, 279, 7271, 315, 1036, 90799, 48481, 854, 5267, 27705, 13, 20, 13, 220, 56007, 5122, 100165, 114532, 110124, 2073, 99204, 113484, 9370, 101939, 18987, 11319, 6, 30440, 99250, 105395, 17714, 5122, 35, 13, 20, 10003, 25, 10479, 374, 6509, 264, 1036, 37555, 1682, 854, 1939, 100345, 104120, 36667, 52183, 27369, 37945, 44063, 6, 105043, 100165, 6, 105395, 17714, 105205, 3837, 101097, 66017, 105395, 104813, 109949, 3837, 110263, 42855, 30868, 101090, 17177, 1773, 151645, 198, 151644, 77091, 198],# 'prompt':None},# sampling_params=sampling_params,# request_id=request_id,# lora_request=None,# )generated_text = ""async for result in result_generator:generated_text = result.outputs[0].textprint(generated_text)# 流式输出特殊处理'''为什么要加 '\n'？此处需要使用json.dumps函数将每一次的Python对象转化为json字符串当通过HTTP或WebSocket等协议发送流式数据时，客户端需要一种方式来区分接收到的不同数据块。换行符 \n 可以作为一个简单的 分隔符，使得每一行代表一个独立的数据包。这有助于前端解析器更容易地区分和处理每个单独的消息，而不会将多个消息混淆为一个.例如，如果你不使用任何分隔符，那么所有发送的JSON对象可能会被合并成一个连续的字符串，导致前端难以正确解析这些数据。假设你发送了两个JSON对象：{"success": true, "content": "Hello"}{"success": true, "content": " World"}没有分隔符的情况下，它们会被视为一个无效的JSON字符串，因为JSON格式不允许两个对象直接相连。但是，如果我们在每个JSON对象后面加上换行符：{"success": true, "content": "Hello"}\n{"success": true, "content": " World"}\n这样，前端可以轻松地按行读取（r.iter_lines()）并解析每个JSON对象，确保每个消息都被正确处理''''''# 1. 方式一# chunk = ChatCompletionStreamResponse(success=True, content=generated_text)# yield jsonify(chunk) + '\n'# 2.方式二# 这个 event_serializer 函数的作用是处理那些默认情况下无法被 JSON 序列化的对象，比如自定义类实例、日期时间对象、集合等。在此处是将其转化为dict对象，再交给json.dumps处理# chunk = ChatCompletionStreamResponse(success=True, content=generated_text)# yield json.dumps(chunk, default=event_serializer) + '\n'# 3.方式三yield json.dumps({'success': True, 'content': generated_text}) + '\n'if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("API_PORT", 7777)), workers=1)
import json
import requestsurl_api = "http://localhost:7777/api"# 这种请求更安全，另一种请求请见 ‘fastAPI接口的请求与响应——基础：https://editor.csdn.net/md/?articleId=144560081’一文
with requests.post(url_api, stream=True) as r:r.raise_for_status() # 检查请求是否成功for line in r.iter_lines():if line: # 过滤掉保持连接的空行print(json.loads(line.decode('utf-8')))
上述代码模型可以处理多个请求，且流式响应。如果不想流式响应，直接将流式输出的内容拼接，一次性返回即可。
6. 注意：
在发送请求后，接收流式响应前，如果执行力print(response.content)，可能会导致流式输出变成假流式输出：即等待后端全部输出完成后，前端才开始输出。
这是因为 response.content 属性会强制 requests 库读取整个HTTP响应体，并将其作为一个字节字符串返回。当你访问这个属性时，实际上是在告诉 requests 立刻获取所有的响应数据，这与流式处理的概念相悖。流式处理的目的是允许你逐步处理接收到的数据，而不是一次性读取所有内容。因此，当你使用 stream=True 参数进行请求时，你不应该使用 response.content 或 response.text，因为它们都会触发对整个响应体的读取。相反，你应该使用 response.iter_content() 或 response.iter_lines() 来逐块或逐行处理响应。