Async 流式编程

在现下大语言模型即LLM应用呈爆发态势的情形里, 用户体验是至关重要的一切。设想一下, 当用户问出了一个复杂的问题, 然而你的应用却突兀地卡顿了10秒钟时间, 最终才一次性地“蹦”出了一大段文字来。这样的一种体验真的是极为糟糕的。

用户期望体会到“即时反馈”, 期望目睹AI如同打字机那般, 逐个字呈现出想法。这便是 （流式传输） 的神奇之处。

为了使这种体验更为极致, 我们所需的并发利器是，它能够保证你的程序于等待AI生成下一个字之际, 仍可抽出援手去响应其他用户的操作, 而非呆傻地阻塞于彼处。

今儿这份指南, 我们会联合现代化的工具链, 一步一步带着你掌握＋ 的关键开发诀窍, 塑造一个如丝般平顺的AI对话应用。

1. 核心概念速览

在动代码之前，我们先扫清两个关键概念：

1.1 的 (异步 I/O)

传统代码具备同步特性, 如同于在窗口排队办事, 只要有一人未完成办理, 后续之人皆需等待。

是用于处理并发的标准库, 它能够让你的程序, 在碰到耗时操作, 像是等待网络请求返回这种情况时, 暂且“挂起”当下任务, 进而去执行别的任务, 待数据返回之后再继续进行处理。

在 开发中，我们只需记住三个关键词：

1.2 原理

普通 API 请求呈现的是“全量返回形态”: 即你下达点菜指令, 厨师完成全部菜品制作后, 将所有菜品一次性呈上。该模式是基于 SSE (-Sent )这种方式: 也就是你点菜后, 厨师每完成一道菜品便即刻呈上一道, 连绵不绝, 持续提供直至菜品全部上完。

结合点: 我们要借助 的异步客户端来发起一个流式请求, 接着运用一个异步循环 , 让它去“衔取并且接纳”服务器抛洒过来的一个个数据包。

️ 2. 现代化环境初始化：使用 uv

基于uv的虚拟开发环境

告别那繁琐的venv, 告别那慢吞吞的pip, 我们要使用, 界的新宠, uv, 来极速管理我们的项目环境。

2.1 安装 uv (如果还没安装)

# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

2.2 初始化项目

在终端执行以下命令，创建一个新项目并进入目录：

uv init async-streaming-demo
cd async-streaming-demo

2.3 添加依赖

我们需要 官方库，以及用于方便加载环境变量的 -。

# uv 会自动创建虚拟环境并飞速安装依赖
uv add openai python-dotenv

注： 和 os 是 标准库，无需安装。

️ 3. 灵活的配置：环境变量与客户端初始化

兼容的模型服务

历经实际开发的过程环境, 鉴于安全以及灵活性方面的考量, 举例来说像切换至Azure、或者是本地的之类具有兼容性的服务这般情况, 我们就绝对不可以将API Key以及地址以硬编码的方式放置于代码当中。

我们会依照最佳实践, 借助环境变量去配置客户端, 我们所需的是以下三个变量。

3.1 创建配置文件 .env

在项目的根目录那儿, 创立一个名为 .env的文件, 把你实际的配置填进去。

示例 1：使用官方 （绝大多数国内模型服务都兼容）

OPENAPI_ENDPOINT="https://api.openai.com/v1"
OPENAPI_API_KEY="sk-你的OpenAI密钥"
OPENAPI_MODEL="gpt-3.5-turbo"

示例 2：使用本地 (举例)

OPENAPI_ENDPOINT="http://localhost:11434/v1"
OPENAPI_API_KEY="ollama" # 本地服务通常随便填一个非空值即可
OPENAPI_MODEL="llama3"

3.2 代码中的初始化

于代码里面, 我们将会运用加载那些变量, 并且去初始化客户端, 留意需采用而非同步的那种方式。

# (代码片段，完整代码在最后)
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI
# 加载 .env 文件
load_dotenv()
# 初始化异步客户端，传入自定义的 endpoint 和 key
client = AsyncOpenAI(
    api_key=os.getenv("OPENAPI_API_KEY"),
    base_url=os.getenv("OPENAPI_ENDPOINT")
)
# 获取模型名称
target_model = os.getenv("OPENAPI_MODEL")

4. 解构

如果你将其设定为 =True , 那么 API 所返回的并非是一个规模极大的 JSON , 而是一连串体积微小的 Chunk（数据块）。

流式 vs 非流式

我们需要重点关注结构上的差异：

一个典型的 Chunk 结构示例：

{
  "id": "chatcmpl-xyz",
  "object": "chat.completion.chunk",
  "created": 123456789,
  "model": "gpt-3.5-turbo",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "好"  // <-- 我们要的就是这个！
      },
      "finish_reason": null
    }
  ]
}

处理关键点：

借助 async for chunk in 来对数据流展开遍历, 从中抽取出 chunk..delta. 可要通过妥当的方式去判断其是否为空，毕竟在流起头的那部分像角色信息以及流末尾的结束标志所涵盖的数据包之中, 常常呈现为 None形态。在进行打印操作时启用 flush=True, 从而担保内容能够即刻得以显示, 不会被终端进行缓存处理, 不然就无法达成类似理想中“打字机”那样的效果。5.这里有着完备的实战 Demo。

好了, 所有事情都已经准备齐全。接着把如下的代码复制到你名为 hello.py, 或者 main.py 的文件当中。还要保证你已经配置好了带点的 env 的文件。

然后在终端运行：uv run hello.py

import asyncio
import os
import sys
from dotenv import load_dotenv
from openai import AsyncOpenAI
# --- 1. 配置加载 ---
# 加载项目根目录下的 .env 文件中的环境变量
load_dotenv()
# 读取必要的配置项
api_endpoint = os.getenv("OPENAPI_ENDPOINT")
api_key = os.getenv("OPENAPI_API_KEY")
target_model = os.getenv("OPENAPI_MODEL")
# 简单的健壮性检查：确保环境变量已设置
if not all([api_endpoint, api_key, target_model]):
    print(" 错误: 缺少必要的环境变量。")
    print("请确保在 .env 文件中设置了 OPENAPI_ENDPOINT, OPENAPI_API_KEY, 和 OPENAPI_MODEL。")
    sys.exit(1)
print(f" 配置已加载:")
print(f"   - Endpoint: {api_endpoint}")
print(f"   - Model: {target_model}")
print("-" * 40)
# --- 2. 初始化异步客户端 ---
# 关键点：
# 1. 使用 AsyncOpenAI 而不是同步的 OpenAI
# 2. 显式传入 base_url，使其能连接到自定义的服务端点 (官方或第三方)
client = AsyncOpenAI(
    api_key=api_key,
    base_url=api_endpoint
)
async def stream_chat(prompt: str):
    """
    异步发送 prompt 并以流式打印回复的核心协程。
    """
    print(f" User: {prompt}\n")
    print(f" AI ({target_model}): ", end="", flush=True) # 准备开始输出，提示当前模型
    try:
        # --- 3. 发起流式请求 ---
        # 使用 await 等待连接建立
        stream = await client.chat.completions.create(
            model=target_model,  # 使用环境变量中指定的模型
            messages=[
                {"role": "system", "content": "你是一个思维敏捷、回答简洁的AI助手。"},
                {"role": "user", "content": prompt},
            ],
            stream=True, # <--- 【核心】开启流式模式
        )
        # --- 4. 异步迭代处理响应流 ---
        # stream 是一个异步迭代器，必须用 async for 来遍历
        # 程序会在等待下一个 chunk 时释放控制权，不会阻塞
        async for chunk in stream:
            # 提取增量内容。注意：流式模式下内容在 'delta' 中，而不是 'message' 中
            content = chunk.choices[0].delta.content
            
            # 必须检查 content 是否存在，因为第一个和最后一个 chunk 的 content 可能是 None
            if content:
                # 实时打印出来
                # end="" 防止print自动换行
                # flush=True 强制刷新缓冲区，确保立刻看到字符跳出，实现打字机效果
                print(content, end="", flush=True)
        
        print("\n\n 回复结束")
    except Exception as e:
        # 捕获网络错误、认证错误等
        print(f"\n\n 发生错误: {e}")
        print("请检查你的网络连接、API Key 以及 Endpoint 设置是否正确。")
async def main():
    """
    主入口协程
    """
    print(" 程序启动 (Async Mode)...")
    
    # 这里可以放入你想测试的问题
    test_prompt = "请用 Python 写一个斐波那契数列生成器，并简单解释原理。"
    
    # 等待流式对话任务完成
    await stream_chat(test_prompt)
    
    print(" 程序退出")
if __name__ == "__main__":
    # --- 5. 启动异步事件循环 ---
    # asyncio.run 是运行最高层级入口点的标准方法
    asyncio.run(main())

总结

运行那上面的代码, 你就将看到, AI的回复, 会如同变魔术那般, 逐字去显现。

在借助这篇教程之后, 你已然把控住了去搭建现代具备高性能性质的 LLM 应用的的基础要点:

对环境运用uv进行管理, 借助.env对服务地址以及模型予以灵活配置, 通过和async for达成非阻塞这种流式的输出。6核心参考文档(Links)。

于深入代码范围之先, 提议把这些官方表述记录标识留存备用, 当遭遇问题之际能够随时翻阅查看:

SDK () 官方文档