Python 调用流式接口，拼接 delta.content 并完整输出

下面提供两种常用方式的完整代码：

方式一：使用 requests 调用（通用 SSE 流式接口）

import requests
import json
def call_streaming_api(url, headers, payload):
    """
    调用流式接口，拼接 delta.content 并完整输出
    """
    response = requests.post(
        url,
        headers=headers,
        json=payload,
        stream=True              # 开启流式响应
    )
    response.raise_for_status()
    full_content = ""           # 用于拼接完整内容
    # 逐行读取 SSE 流
    for line in response.iter_lines(decode_unicode=True):
        # 跳过空行和注释行
        if not line or line.startswith(":"):
            continue
        # 去掉 "data: " 前缀
        if line.startswith("data: "):
            line = line[6:]
        # 检查是否为结束标志 [DONE]
        if line.strip() == "[DONE]":
            break
        try:
            chunk = json.loads(line)
        except json.JSONDecodeError:
            continue
        # ========== 核心提取逻辑 ==========
        # 适配 OpenAI 兼容格式: {"choices":[{"delta":{"content":"xxx"}}]}
        choices = chunk.get("choices", [])
        if choices:
            delta = choices[0].get("delta", {})
            content = delta.get("content", "")
            if content:
                full_content += content
                print(content, end="", flush=True)   # 实时打印
    print()  # 换行
    return full_content
# ===================== 使用示例 =====================
if __name__ == "__main__":
    # ---- 请替换为你自己的接口地址和密钥 ----
    API_URL  = "https://api.openai.com/v1/chat/completions"
    API_KEY  = "sk-xxxxxxxxxxxxxxxxxxxxxxxx"
    HEADERS = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }
    PAYLOAD = {
        "model": "gpt-3.5-turbo",       # 替换为你的模型名
        "messages": [
            {"role": "user", "content": "请用一句话介绍Python"}
        ],
        "stream": True                   # 开启流式
    }
    print("===== 开始请求 =====")
    result = call_streaming_api(API_URL, HEADERS, PAYLOAD)
    print("===== 完整结果 =====")
    print(result)

方式二：使用 openai 官方 SDK（推荐）

from openai import OpenAI
def call_openai_stream(api_key, base_url, model, messages):
    """
    使用 OpenAI SDK 调用流式接口，拼接 delta.content 并完整输出
    """
    client = OpenAI(
        api_key=api_key,
        base_url=base_url              # 兼容第三方接口时修改此项
    )
    full_content = ""
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True
    )
    for chunk in stream:
        # ========== 核心提取逻辑 ==========
        if chunk.choices and chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            full_content += content
            print(content, end="", flush=True)   # 实时打印
    print()
    return full_content
# ===================== 使用示例 =====================
if __name__ == "__main__":
    result = call_openai_stream(
        api_key  = "sk-xxxxxxxxxxxxxxxxxxxxxxxx",
        base_url = "https://api.openai.com/v1",    # 第三方接口改这里
        model    = "gpt-3.5-turbo",
        messages = [
            {"role": "user", "content": "请用一句话介绍Python"}
        ]
    )
    print("===== 完整结果 =====")
    print(result)

运行效果示意

===== 开始请求 =====
Python是一种简洁优雅的通用编程语言。
===== 完整结果 =====
Python是一种简洁优雅的通用编程语言。

关键说明