【AsyncOpenAI vs OpenAI】在异步函数中调用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)asyncdefapi_predict(conversation):
    response =await client.chat.completions.create(
        model="Qwen",
        messages=conversation,
        stream=True)asyncfor 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)defapi_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