代码收藏家技术教程 2024-08-28

【AsyncOpenAI与OpenAI API对比】异步调用流式输出大比拼

在异步函数中使用AsyncOpenAI与直接从openai导入OpenAI的区别

在现代的Python编程中，异步编程已成为提高应用程序效率和响应性的关键技术。尤其是在处理I/O操作密集的任务时，使用异步编程可以显著提高程序的响应速度和性能。在这篇文章中，我们将讨论如何在异步函数async def api_predict()中使用AsyncOpenAI来完成任务，并对其与直接从openai库中导入OpenAI的区别进行分析。

1. 背景介绍

1.1 同步与异步编程

同步编程是指程序在执行一个任务时，必须等待该任务完成才能继续执行下一个任务。相反，异步编程允许程序在等待某个任务完成的同时，继续执行其他任务，从而提高程序的效率和响应速度。

1.2 OpenAI API

OpenAI提供了一套强大的API，可以用于生成文本、回答问题等任务。在Python中，我们通常使用openai库来与OpenAI API进行交互。随着异步编程的普及，OpenAI也推出了支持异步操作的AsyncOpenAI。

2. 使用`AsyncOpenAI`的示例

下面是一个使用AsyncOpenAI进行异步API调用的示例代码：

from openai import AsyncOpenAI

openai_api_base = "http://192.168.10.61:7862/v1"
openai_api_key = "EMPTY"
client = AsyncOpenAI(api_key=openai_api_key, base_url=openai_api_base)

async def api_predict(conversation):
    response = await client.chat.completions.create(
        model="Qwen",
        messages=conversation,
        stream=True
    )
    async for chunk in response:
        content = chunk.choices[0].delta.content
        if content:
            yield content

在这个示例中，我们首先从openai库中导入了AsyncOpenAI，然后配置了API的基本信息。接下来，我们定义了一个异步函数api_predict()，该函数使用await关键字等待API响应，并通过异步迭代器async for来处理响应中的数据块。

3. 与直接导入`OpenAI`的区别

3.1 同步操作与异步操作

直接从openai库中导入OpenAI适用于同步操作，例如：

from openai import OpenAI

openai_api_base = "http://192.168.10.61:7862/v1"
openai_api_key = "EMPTY"
client = OpenAI(api_key=openai_api_key, base_url=openai_api_base)

def api_predict(conversation):
    response = client.chat.completions.create(
        model="Qwen",
        messages=conversation
    )
    for chunk in response:
        content = chunk.choices[0].delta.content
        if content:
            yield content

在这个示例中，api_predict()函数是同步的，意味着在API调用完成之前，函数会阻塞当前线程。最终返回全部内容，输出非流式。
等同于

from openai import OpenAI

client = OpenAI()

def api_predict():
    response = client.chat.completions.create(
        model="Qwen",
        messages=conversation
    )
    return response.choices[0].message.content

3.2 性能与效率

使用AsyncOpenAI进行异步调用可以提高性能和效率，特别是在处理大量I/O操作时。异步编程允许程序在等待API响应的同时，执行其他任务，从而避免了线程阻塞。

3.3 代码结构

异步代码与同步代码在结构上有所不同。异步代码使用async和await关键字，并且通常与异步框架（如asyncio）一起使用。这要求开发者对异步编程有一定的了解和掌握。

3.4 并发性能

在处理多个请求时,AsyncOpenAI可以显著提高性能。

3.5 事件循环兼容性:

AsyncOpenAI可以无缝集成到使用asyncio的应用中。

4. 结论

对于需要高并发或实时数据流的应用，AsyncOpenAI是更好的选择。它提供了更大的灵活性和更好的性能，特别是在处理多个API调用或流式响应时。然而，对于简单的单次调用或不需要异步功能的脚本，传统的OpenAI客户端可能更为简单直接。

5.补充

在全局使用AsyncOpenAI的状态下，
批处理输出(Batch output：是指将所有数据处理完毕后一次性输出全部结果,而不是像流式输出那样逐步、实时地输出数据)也称非流式输出
是无法正常获取到输出的，获取到的是<coroutine object AsyncCompletions.create at 0x7f8ee3f93990>
所以在AsyncOpenAI使用流式输出时还想同时使用非流式输出，参考下面的代码：

async def extract_summary(text):
    tokenizer = AutoTokenizer.from_pretrained('/home/hyh/model/Qwen2-7B-Instruct')
    openai_api_base = "http://192.168.10.61:7862/v1"
    openai_api_key = "EMPTY"
    client = OpenAI(api_key=openai_api_key,base_url=openai_api_base,)
    if isinstance(text,str):
        if len(text) < 30:
            return text
        tokens = tokenizer.tokenize(text)
        max_length = 30000 # 32768
        truncated_tokens = tokens[:max_length]
        long_text = tokenizer.convert_tokens_to_string(truncated_tokens)
        chat_response = await client.chat.completions.create(
            model="Qwen",
            messages=[
                {"role": "system", "content": "总结下面的文本形成摘要,如果文本中有前言、摘要、目录、序章等，则仅参考这一部分进行总结。\n"},
                {"role": "user", "content": long_text},
            ],
            temperature=0
        )
        summary = chat_response.choices[0].message.content
        return summary
    else:
        raise ValueError("extract_summary函数获取到了非字符串")

其实就是多了一个await等待模型回答完，另外如果别的异步函数调用了这个函数，还需要在extract_summary再加await

作者：内卷焦虑人士