Jacin Blog
AI 大模型开发

openai流式解析

JACIN6 分钟阅读
openai流式
AI 摘要

这篇笔记聚焦 Python 中 OpenAI Chat Completions 的异步流式调用，适用于希望把模型响应按片段实时输出到终端或业务层的开发场景。内容先从环境变量读取入手，强调用 os.getenv 获取 OPENAI_API_KEY、OPENAI_BASE_URL 和 OPENAI_MODEL，并说明 python-dotenv 的安装包名与导入模块名不同：通过 pip install python-dotenv 安装，但代码中从 dotenv 导入 load_dotenv。核心示例封装了一个 AsyncOpenAIOut 类，在初始化阶段创建 AsyncOpenAI 客户端，在 gpt_stream 方法中组合 history、system_prompt 和用户消息，并以 stream=True 发起请求。代码通过 async for 遍历 OpenAI 返回的流式 response，检查 chunk.choices[0].delta.content 后用 yield 逐块产出内容，因此调用方也需要用 async for 消费这个异步生成器。文章还补充了 print(chunk) 在事件循环中的行为：它是同步执行，但在简单逐步输出场景通常可接受；若希望更严格地避免占用事件循环，可以用 asyncio.to_thread(print, chunk) 放到后台线程执行。整体价值在于把 .env 配置、AsyncOpenAI 初始化、消息组装、流式响应解析和 async for/yield 的语义串成一个可运行样例，适合正在接入 OpenAI 兼容接口或学习 Python 异步流处理的开发者。

OpenAI 流式的代码：

首选一般请使用os.getenv 去读环境变量的内容

注意使用pip install python-dotenv 的安装方法

load_dotenv 是这个库提供的一个函数，用于读取 .env 文件并将其中定义的键值对设置为系统的环境变量。

默认情况下，load_dotenv() 会自动查找当前目录下的 .env 文件。如果文件不在默认路径，可以通过参数指定，例如 load_dotenv('/path/to/.env')。

为什么代码里用 dotenv 而不是 python_dotenv？

在 Python 中，导入模块时使用的是库的模块名，而不是 PyPI 上的包名。python-dotenv 这个包安装后，提供了一个名为 dotenv 的模块供导入。这是由库开发者决定的命名约定。例如：

  • PyPI 包名：python-dotenv
  • 导入时的模块名：dotenv

流式代码 类 （一键运行）

python 
from openai import  AsyncOpenAI
from dotenv import load_dotenv
import os

load_dotenv()

class AsyncOpenAIOut:
    def __init__(self):
        self.api_key = os.getenv("OPENAI_API_KEY")
        self.base_url = os.getenv("OPENAI_BASE_URL")
        self.oai_client = AsyncOpenAI(api_key=self.api_key, base_url=self.base_url)
        self.model = os.getenv("OPENAI_MODEL")
    async def gpt_stream(self, user_message: str,model: str = os.getenv("OPENAI_MODEL"),history: list[dict] = [],system_prompt: str = "") :
        messages = []
        if history:
            messages.extend(history)
        
        if system_prompt:
            messages.extend([{"role": "system", "content": system_prompt}])
        
        messages.append({"role": "user", "content": user_message})
        response = await self.oai_client.chat.completions.create(
            model=model,
            messages=messages,
            stream=True
        )
        
        async for chunk in response:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

async_openai_out = AsyncOpenAIOut()

if __name__ == "__main__":
    async def test_gpt_stream():
        async for chunk in async_openai_out.gpt_stream(user_message="写300字作文",system_prompt="You are a helpful assistant."):
            print(chunk)

    import asyncio
    
    asyncio.run(test_gpt_stream())

这里面有几点需要注意：

简短回答：print(chunk) 是 同步操作，会在当前事件循环中执行完毕后才继续，但它不是 I/O 密集型操作，所以不会造成实际的“阻塞”问题，特别是在异步函数中逐步输出内容的场景下，它是可接受的。

想确保异步非阻塞输出:

python 
        async for chunk in async_openai_out.gpt_stream(user_message="写300字作文",system_prompt="You are a helpful assistant."):
            # print(chunk)
            await asyncio.to_thread(print, chunk)  # 在后台线程执行 print

异步生成器（Async Generator） 的用法，结合了 Python 的异步编程（async for）和生成器（yield）机制。

在异步迭代 response 中的每个 chunk，如果它有内容，就通过 yield 一块一块地“流式返回”。

python 
async for chunk in response:
    if chunk.choices[0].delta.content:
        yield chunk.choices[0].delta.content

对比for ： 普通for是同步迭代

async for是异步迭代,比如网络流、WebSocket、OpenAI 的 Stream 响应等。

yield 是生成器的关键，它不是“返回”值，而是“产出”值（可以被迭代一次）。

评论

还没有评论，来发第一个吧