LLamafactory API部署与使用异步方式 API 调用优化大模型推理效率

背景介绍

第三方大模型API

目前，市面上有许多第三方大模型 API 服务提供商，通过 API 接口向用户提供多样化的服务。这些平台不仅能提供更多类别和类型的模型选择，还因其用户规模较大，能够以更低的成本从原厂获得服务，再将其转售给用户。此外，这些服务商还支持一些海外 API 服务，例如 ChatGPT 等，为用户提供了更加广泛的选择。

• https://www.gptapi.us/register?aff=9xEy

比如上述网站以 API 接口的形式对外提供的服务，比官方的 API 要便宜。

装包：

pip install langchain langchain_openai

运行下述代码，完成上述网站的注册后，并填上述网站的 api_key 便可通过 python API 调用，就会收到 gpt-4o-mini 大模型的响应。

from langchain_openai import ChatOpenAIllm = ChatOpenAI(model="gpt-4o-mini", base_url="https://www.gptapi.us/v1/",api_key="sk-xxx", # 在这里填入你的密钥)
res = llm.invoke("你是谁？请你简要做一下，自我介绍？")
print(res)

介绍

在部署垂直领域模型时，我们通常会对开源大模型进行微调，并获得相应的 LoRA 权重。在接下来的部分，我将介绍如何使用 LLamafactory 将微调后的 LoRA 模型部署为 API 服务。

在 Python 中调用 API 服务时，如果采用同步方式进行请求，往往会导致请求速度较慢。因为同步方式需要在接收到上一条请求的响应后，才能发起下一条请求。

为了解决这一问题，我将为大家介绍如何通过异步请求的方式，在短时间内发送大量请求，从而提升 API 调用效率。

LLamafactory 部署API

关于 LLamafactory 的下载与微调模型，点击查看我的这篇博客：Qwen2.5-7B-Instruct 模型微调与vllm部署详细流程实战.https://blog.csdn.net/sjxgghg/article/details/144016723

vllm_api.yaml 的文件内容如下：

model_name_or_path: qwen/Qwen2.5-7B-Instruct
adapter_name_or_path: ../saves/qwen2.5-7B/ner_epoch5/
template: qwen
finetuning_type: lora
infer_backend: vllm
vllm_enforce_eager: true# llamafactory-cli chat lora_vllm.yaml
# llamafactory-cli webchat lora_vllm.yaml
# API_PORT=8000 llamafactory-cli api lora_vllm.yaml

使用下述命令便可把大模型以 API 部署的方式，部署到8000端口：

API_PORT=8000 llamafactory-cli api vllm_api.yaml

LangChain 的 invoke 方法是常用的调用方式，但该方法并不支持异步操作。如果读者想了解同步与异步在速度上的差距，可以自行尝试使用一个 for 循环调用 invoke 方法，并对比其性能表现。

import os
from langchain_openai import ChatOpenAI
client = ChatOpenAI(model="gpt-3.5-turbo", api_key="{}".format(os.environ.get("API_KEY", "0")),base_url="http://localhost:{}/v1".format(os.environ.get("API_PORT", 8000)),
)res = llm.invoke("你是谁？请你简要做一下，自我介绍？")
print(res)

在项目文件夹下，新建一个 .env 文件， 其中 API_KEY 的值便是API接口调用的 API_KEY。

API_KEY=sk-12345678

LLamafactory 通过API部署的大模型地址是: http://localhost:8000/v1
API_KEY 是.env 文件中 API_KEY：sk-12345678

大模型 API 调用工具类

使用异步协程加快 API 的调用速度，可以参考我们前面的这篇文章：大模型 API 异步调用优化：高效并发与令牌池设计实践.https://blog.csdn.net/sjxgghg/article/details/143858730

我们在前面一篇文章的基础上，对异步类再封装了一下。

装包：

pip install langchain tqdm aiolimiter python-dotenv

import os
import random
import asyncio
import pandas as pd
from tqdm import tqdm
from typing import List
from dataclasses import dataclass, field
from aiolimiter import AsyncLimiter
from langchain_openai import ChatOpenAI
from dotenv import load_dotenvload_dotenv()def generate_arithmetic_expression(num: int):"""生成数学计算的公式和结果"""# 定义操作符和数字范围，除法operators = ["+", "-", "*"]expression = (f"{random.randint(1, 100)} {random.choice(operators)} {random.randint(1, 100)}")num -= 1for _ in range(num):expression = f"{expression} {random.choice(operators)} {random.randint(1, 100)}"result = eval(expression)expression = expression.replace("*", "x")return expression, result@dataclass
class AsyncLLMAPI:"""大模型API的调用类"""base_url: strapi_key: str  # 每个API的key不一样uid: int = 0cnt: int = 0  # 统计每个API被调用了多少次model: str = "gpt-3.5-turbo"llm: ChatOpenAI = field(init=False)  # 自动创建的对象，不需要用户传入num_per_second: int = 6  # 限速每秒调用6次def __post_init__(self):# 初始化 llm 对象self.llm = self.create_llm()# 创建限速器，每秒最多发出 5 个请求self.limiter = AsyncLimiter(self.num_per_second, 1)def create_llm(self):# 创建 llm 对象return ChatOpenAI(model=self.model,base_url=self.base_url,api_key=self.api_key,)async def __call__(self, text):# 异步协程 限速self.cnt += 1async with self.limiter:return await self.llm.agenerate([text])@staticmethodasync def _run_task_with_progress(task, pbar):"""包装任务以更新进度条"""result = await taskpbar.update(1)return result@staticmethoddef async_run(llms: List["AsyncLLMAPI"],data: List[str],keyword: str = "",  # 文件导出名output_dir: str = "output",chunk_size=500,):async def _func(llms, data):"""异步请求处理一小块数据"""results = [llms[i % len(llms)](text) for i, text in enumerate(data)]with tqdm(total=len(results)) as pbar:results = await asyncio.gather(*[AsyncLLMAPI._run_task_with_progress(task, pbar)for task in results])return resultsidx = 0all_df = []while idx < len(data):file = f"{idx}_{keyword}.csv"file_dir = os.path.join(output_dir, file)if os.path.exists(file_dir):print(f"{file_dir} already exist! Just skip.")tmp_df = pd.read_csv(file_dir)else:tmp_data = data[idx : idx + chunk_size]loop = asyncio.get_event_loop()tmp_result = loop.run_until_complete(_func(llms=llms, data=tmp_data))tmp_result = [item.generations[0][0].text for item in tmp_result]tmp_df = pd.DataFrame({"infer": tmp_result})# 如果文件夹不存在，则创建if not os.path.exists(tmp_folder := os.path.dirname(file_dir)):os.makedirs(tmp_folder)tmp_df.to_csv(file_dir, index=False)all_df.append(tmp_df)idx += chunk_sizeall_df = pd.concat(all_df)all_df.to_csv(os.path.join(output_dir, f"all_{keyword}.csv"), index=False)return all_dfif __name__ == "__main__":# 生成 数学计算数据集texts = []labels = []for _ in range(1000):text, label = generate_arithmetic_expression(2)texts.append(text)labels.append(label)llm = AsyncLLMAPI(base_url="http://localhost:{}/v1".format(os.environ.get("API_PORT", 8000)),api_key="{}".format(os.environ.get("API_KEY", "0")),)AsyncLLMAPI.async_run([llm], texts, keyword="数学计算", output_dir="output", chunk_size=500)

使用异步类，在短时间内向对方服务器，发送大量的请求可能会导致服务器拒绝响应。
由于使用了异步的请求，则必须在所有的请求都完成后才能拿到结果。为了避免程序中途崩溃导致前面的请求的数据丢失，故 使用 chunk_size 对请求的数据进行切分，每完成一块数据的请求则把该块数据保存到csv文件中。

本文使用 generate_arithmetic_expression 生成1000条数学计算式，调用大模型 API 完成计算。

运行效果如下：

原始的 1000 条数据，设置chunk_size为500，故拆分为2块500条，分批进行处理。

为了避免程序崩垮，分批进行异步推理，若程序崩溃了，可重新运行，程序会从上一次崩溃的点重新运行。（要保证数据集输入的一致！）

最终的输出文件是 all_数学计算.csv ，它是所有分快csv文件的汇总。

项目开源

https://github.com/JieShenAI/csdn/tree/main/24/11/async_llm_api

• vllm_api.yaml 是 llamafactory 的API部署的配置；
• core.py 是主要代码；