使用Gradio实现OpenAI API异步流式聊天机器人-Python教程-PHP中文网

使用Gradio实现OpenAI API异步流式聊天机器人

碧海醫心

发布： 2025-10-25 10:11:18

原创

829人浏览过

使用gradio实现openai api异步流式聊天机器人

本文详细介绍了如何使用Gradio的`ChatInterface`与OpenAI API实现异步流式聊天机器人。核心在于解决`async generator`直接`yield`导致`ValueError`的问题，通过在异步生成器中累积部分消息并实时`yield`当前完整消息，从而实现响应内容的逐字或逐句显示，提供流畅的用户体验。

构建异步流式聊天机器人：Gradio与OpenAI API实践

在构建现代交互式应用时，实时响应能力至关重要。对于聊天机器人而言，这意味着用户输入后，不应等待整个回复生成完毕才显示，而是应该逐字或逐句地流式传输内容。Gradio的ChatInterface结合OpenAI API的流式传输功能，是实现这一目标的强大组合。然而，在实际开发中，开发者可能会遇到异步生成器与Gradio接口集成时的挑战，特别是关于如何正确处理流式输出。

理解挑战：ValueError与异步生成器

在使用OpenAI API进行流式传输时，我们通常会定义一个异步生成器函数，例如：

async def chat_with_gpt_problematic(prompt):
    stream = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:
        # 问题所在：直接yield delta content
        yield chunk.choices[0].delta.content

登录后复制

当尝试将这样的函数直接与Gradio的ChatInterface或其他期望特定生成器行为的组件结合时，可能会遇到ValueError: a coroutine was expected, got <async_generator object chat_with_gpt at 0x...>。这个错误表明，Gradio或其内部机制在调用我们的函数时，可能期望一个可以直接await的协程（返回一个最终结果），而不是一个异步生成器对象本身。虽然async for chunk in stream内部的print(chunk.choices[0].delta.content)能够正常打印出流式内容，但yield的方式并未能被Gradio正确地解析为持续更新的流。

Gradio的ChatInterface在处理流式输出时，通常期望其fn参数返回一个生成器，该生成器每次yield的都是当前累积的完整消息，而不是消息的片段（delta）。

解决方案：累积并实时yield完整消息

解决上述问题的关键在于，在异步生成器中累积OpenAI API返回的增量内容（delta），并在每次接收到新内容时，yield出当前已经累积的完整消息。这样，Gradio就能接收到不断增长的字符串，并实时更新UI。

Giiso写作机器人

Giiso写作机器人，让写作更简单

查看详情

以下是修正后的异步生成器函数示例：

import gradio as gr
from openai import AsyncOpenAI
import os

# 确保在环境变量中设置了 OPENAI_API_KEY
# client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# 假设 client 已经正确初始化

async def stream_chat_response(input_text, history):
    # 构造消息列表，包括历史记录
    # history 是一个列表，每个元素是 [user_message, bot_message]
    messages = []
    for human, assistant in history:
        messages.append({"role": "user", "content": human})
        messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": input_text})

    stream = await client.chat.completions.create(
        model="gpt-4", # 或 "gpt-3.5-turbo"
        messages=messages,
        stream=True,
    )

    partial_message = ""
    async for chunk in stream:
        # 检查 delta.content 是否存在，因为有时 chunk 可能只包含 role 信息
        if chunk.choices[0].delta.content is not None:
            partial_message += chunk.choices[0].delta.content
            # 每次收到新内容时，yield 累积的完整消息
            yield partial_message

登录后复制

代码解析：

messages构建： 在实际的聊天机器人中，需要将用户的当前输入和之前的聊天历史（history参数）一并发送给OpenAI API，以维持对话上下文。
partial_message = ""： 初始化一个空字符串，用于累积来自API的流式响应。
async for chunk in stream:： 异步迭代OpenAI API返回的流式块。
if chunk.choices[0].delta.content is not None:： 确保当前块包含实际的内容增量。OpenAI API有时会发送只包含角色信息（如{"role": "assistant"}）而无content的块。
partial_message += chunk.choices[0].delta.content： 将当前块的内容追加到partial_message中。
yield partial_message： 这是关键一步。每次partial_message更新后，我们都将其yield出去。Gradio的ChatInterface会捕获这些yield的值，并将其显示为聊天机器人响应的最新状态，从而实现逐字或逐句的实时更新效果。

整合到Gradio ChatInterface

现在，我们将这个修正后的异步生成器函数集成到Gradio的ChatInterface中：

import gradio as gr
from openai import AsyncOpenAI
import os

# 确保 OPENAI_API_KEY 环境变量已设置
# 示例：export OPENAI_API_KEY="your_openai_api_key_here"
# 或者直接在这里赋值 client = AsyncOpenAI(api_key="your_openai_api_key_here")
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# 修正后的异步流式响应函数
async def stream_chat_response(input_text, history):
    messages = []
    for human, assistant in history:
        messages.append({"role": "user", "content": human})
        messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": input_text})

    stream = await client.chat.completions.create(
        model="gpt-4", # 可以根据需求选择模型
        messages=messages,
        stream=True,
    )

    partial_message = ""
    async for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            partial_message += chunk.choices[0].delta.content
            yield partial_message

# Gradio ChatInterface 启动
if __name__ == "__main__":
    gr.ChatInterface(
        stream_chat_response,
        chatbot=gr.Chatbot(height=400),
        textbox=gr.Textbox(placeholder="向我提问...", container=False, scale=7),
        title="OpenAI 异步流式聊天机器人",
        description="使用Gradio和OpenAI API构建的实时流式聊天机器人。",
        examples=[
            ["什么是异步编程？"],
            ["解释大型语言模型的工作原理。"],
            ["给我讲个关于人工智能的笑话。"]
        ],
        retry_btn="重试",
        undo_btn="撤销",
        clear_btn="清空",
    ).queue().launch() # 使用 .queue() 可以在高并发下更好地管理请求

登录后复制

注意事项与最佳实践

API Key 安全： 永远不要将API Key直接硬编码到代码中。使用环境变量（如os.environ.get("OPENAI_API_KEY")）是更安全的选择。
错误处理： 在实际应用中，应添加try-except块来捕获API请求可能出现的错误（如网络问题、认证失败、API限速等），并向用户提供友好的错误提示。
历史记录管理： Gradio的ChatInterface会自动管理history参数。在stream_chat_response函数中，正确地将history转换为OpenAI API所需的messages格式至关重要，以确保对话的连贯性。
模型选择： 根据应用需求和成本考虑，选择合适的OpenAI模型（如gpt-3.5-turbo或gpt-4）。
Gradio queue()： 在launch()之前调用.queue()可以为您的应用添加请求队列，这对于处理并发用户请求和提高稳定性非常有益。
初始空块处理： OpenAI API有时可能会发送delta.content为None的块（例如，只包含role信息）。代码中的if chunk.choices[0].delta.content is not None:已经考虑了这种情况。

总结

通过在异步生成器中巧妙地累积部分消息并实时yield当前完整的消息，我们成功解决了Gradio ChatInterface与OpenAI API异步流式传输的集成问题。这种方法不仅避免了ValueError，还为用户提供了流畅、实时的聊天体验，是构建高性能、用户友好型AI聊天机器人的关键技术。理解Gradio如何处理生成器输出，以及OpenAI API流式传输的特性，是实现此类应用的核心。

以上就是使用Gradio实现OpenAI API异步流式聊天机器人的详细内容，更多请关注php中文网其它相关文章！