转载&存档丨AI 标准协议及调用（1）

AI 标准协议及调用

前言

AI近期来发展迅速，这段时间刷视频和朋友圈总是看到大家又在说什么前端已死、后端已死，还有什么你们搞大模型的都是码奸之类的话，这些也都是AI迅速发展的表现。在这种环境下，学习开发的人很难不焦虑，有些时候甚至会想还有必要学基础的编程吗，直接全部vibe coding不就好了吗，但其实不是这样的，编程能力从来不是跳跃式获得的，所有的学习都是一条平滑上升的曲线，要学好计算机，先从最基本的coding学起，学习前端、后端，再到全栈、agent,逐渐再转向研发大模型，这样才算是健全的学习道路，而非是从一开始就跑去学习大模型。本质上，AI 不是起点，而是建立在扎实工程能力之上的。接下来就会讲解在学习AI的过程中最基础的AI标准协议及调用。

现在的 API 协议已经相当标准化和通用化

参考 HTTP 协议

在讲 AI API 协议之前，我们先复习一下 HTTP 协议，这在之前也有讲过。HTTP（超文本传输协议）是互联网通信的基础，也是现在AI应用中最常用的传输层协议,它定义了客户端和服务器之间如何交换数据的规则：

• 统一的请求格式：包含 GET、POST、PUT、DELETE 等方法
• 标准化的状态码：200 成功、404 未找到(这个大家应该都见过)、500 服务器错误等
• 通用的头部字段：Content-Type、Authorization 等等，这些就不赘述了

AI API 协议

同样的道理，AI 模型的调用也需要标准化的协议，：

1. 统一接口：开发者可以用相似的方式调用不同的模型，只需要改变某些参数就可以，之后会接触到的mcp其实也是一种统一接口
2. 降低学习成本：掌握一种协议后，可以快速迁移到其他兼容服务，多数协议其实差别不大，只需要改一些字段
3. 生态系统建设：标准化促进了工具库、框架的发展，这对开发者来说十分友好
4. 互操作性：应用可以轻松切换不同的 AI 服务提供商，包括国内的和国外的

目前主流的 AI API 协议和工具主要有以下几种：

主流协议

• OpenAI API 协议：由 OpenAI 制定，已成为事实上的行业标准，被广泛采用
• Anthropic (Claude) API 协议：由 Anthropic 为 Claude 系列模型设计，强调安全性

其他协议和工具

• Google Gemini API：Google 的多模态 AI 协议，支持文本、图像、视频等多种输入
• LiteLLM：统一的 API 接口，支持 100+ 种 LLM 模型，一套代码调用所有模型
• OpenRouter：AI 模型聚合平台，提供统一的 OpenAI 兼容接口访问多家模型
• Ollama：本地运行开源模型的工具，提供 OpenAI 兼容的 API 接口

本文将详细讲解 OpenAI 和 Claude 两种主流协议，其他工具会简要介绍使用方法。

OpenAI API 协议详解

协议概述

OpenAI API 协议是目前最广泛使用的 AI API 标准，许多国内外厂商都提供了兼容接口，包括：

• 阿里云通义千问（Qwen）
• DeepSeek
• 智谱 AI（GLM）
• 月之暗面（Kimi）
• 百度文心一言(这个和似了其实区别不大)

核心端点（Endpoint）

主要的 API 端点是 /v1/chat/completions，用于对话式交互。

请求参数详解

首先，让我们看一个完整的 API 请求示例，了解整体结构：

{
  "model": "gpt-3.5-turbo",
  "messages": [
    {
      "role": "system",
      "content": "你是一个有帮助的 AI 助手，擅长回答技术问题。"
    },
    {
      "role": "user",
      "content": "什么是机器学习？"
    }
  ],
  "temperature": 0.7,
  "max_tokens": 2000,
  "stream": false,
  "stop": null
}

接下来，我们逐个解释这些参数的含义和用法。

必需参数

model (string)

• 含义：指定要使用的模型名称
• 示例："gpt-3.5-turbo"、"deepseek-chat"、"qwen-turbo"
• 说明：不同服务商的模型名称不同，需查阅对应文档

messages (array)

• 含义：对话历史记录，包含用户和助手的消息
• 结构：每条消息是一个对象，包含 role 和 content 字段
• 示例：

[
  { "role": "system", "content": "你是一个有帮助的助手" },
  { "role": "user", "content": "什么是机器学习？" },
  { "role": "assistant", "content": "机器学习是..." },
  { "role": "user", "content": "能举个例子吗？" }
]

role 的类型：

• system：系统提示词，定义 AI 的行为和角色
• user：用户的输入
• assistant：AI 的回复（可能包含 tool_calls 字段，表示需要调用工具）
• tool：工具调用的返回结果（用于 Function Calling），必须包含：
• tool_call_id：对应 assistant 消息中的工具调用 ID
• content：工具执行的结果（通常是 JSON 字符串）

常用可选参数

temperature (number, 0-2)

• 含义：控制输出的随机性
• 默认值：有些模型通常为 1.0(但在实际生产中一般不需要自己调整这个)
• 说明：
• 接近 0：输出更确定、保守
• 接近 2：输出更随机、创造性
• 使用建议：代码生成用 0.2-0.5，创意写作用 0.7-1.2

max_tokens (integer)

• 含义：生成的最大 token 数量
• 说明：1 个 token 约等于 0.75 个英文单词，或 0.5 个中文字符(好像各家的说法都不太一样，这个简单了解即可)
• 注意：设置过小可能导致回复被截断

stream (boolean)

• 含义：是否启用流式输出
• 默认值：false
• 说明：
• true：逐字返回，适合实时显示
• false：等待完整响应后返回

stop (string or array)

• 含义：停止序列，遇到时停止生成
• 示例：["###", "END"]
• 用途：控制输出格式，防止生成过多内容

tools (array)

• 含义：定义模型可以调用的工具（Function Calling）
• 用途：让模型能够调用外部函数或 API
• 详见后文 Tool Call 章节

响应格式

成功响应示例：

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1677652288,
  "model": "gpt-3.5-turbo",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "机器学习是人工智能的一个分支..."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 20,
    "completion_tokens": 50,
    "total_tokens": 70
  }
}

响应字段说明：

• id：本次请求的唯一标识符
• object：对象类型，通常为 chat.completion
• created：创建时间戳
• model：实际使用的模型
• choices：生成的回复列表（通常只有一个）
• index：回复的索引
• message：消息对象
  • role：角色（assistant）
  • content：生成的文本内容
• finish_reason：结束原因
  • stop：自然结束
  • length：达到 max_tokens 限制
  • content_filter：内容被过滤
  • tool_calls：需要调用工具
• usage：token 使用情况
• prompt_tokens：输入消耗的 token
• completion_tokens：输出消耗的 token
• total_tokens：总计

Anthropic (Claude) API 协议详解

协议特点

Claude API 协议与 OpenAI 有相似之处，但也有独特设计：

• 更强调安全性和可控性
• 支持更长的上下文窗口
• 提供了更细粒度的控制选项

核心端点

主要端点是 /v1/messages。

请求参数详解

我们先看一个完整的 Claude API 请求示例：

{
  "model": "claude-3-5-sonnet-20241022",
  "max_tokens": 2000,
  "system": "你是一个有帮助的 AI 助手，擅长回答技术问题。",
  "messages": [
    {
      "role": "user",
      "content": "什么是机器学习？"
    }
  ],
  "temperature": 0.7
}

这里要注意一下 Claude API 与 OpenAI 的主要区别：

• system 提示词是独立参数，不在 messages 数组中
• max_tokens 是必需参数
• messages 中不包含 system 角色的消息

接下来详细解释各个参数：

必需参数

model (string)

• 含义：指定要使用的 Claude 模型
• 示例："claude-3-5-sonnet-20241022"、"claude-3-opus-20240229"、"claude-3-haiku-20240307"
• 说明：不同模型在性能、速度和成本上有差异

messages (array)

• 含义：对话消息列表
• 结构与 OpenAI 类似，但有细微差异
• 注意：Claude 的 system 提示词是单独的参数，不在 messages 中
• 示例：

[
  {
    "role": "user",
    "content": "什么是机器学习？"
  },
  {
    "role": "assistant",
    "content": "机器学习是人工智能的一个分支..."
  },
  {
    "role": "user",
    "content": "能举个例子吗？"
  }
]

max_tokens (integer)

• 含义：生成的最大 token 数量
• 必需参数（与 OpenAI 不同，OpenAI 中是可选的）
• 必须明确指定最大生成长度
• 建议：根据实际需求设置，避免设置过大浪费成本

可选参数

system (string)

• 含义：系统提示词，定义 AI 的行为和角色
• 独立参数，不在 messages 数组中
• 示例："你是一个专业的 Python 编程助手，代码要简洁高效。"

temperature (number, 0-1)

• 范围：0 到 1（注意：Claude 的范围是 0-1，而 OpenAI 是 0-2）
• 默认值：1.0

响应格式

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [
    {
      "type": "text",
      "text": "机器学习是..."
    }
  ],
  "model": "claude-3-5-sonnet-20241022",
  "stop_reason": "end_turn",
  "usage": {
    "input_tokens": 20,
    "output_tokens": 50
  }
}

其他 AI 协议和工具简介

Google Gemini API 协议详解

Google Gemini 是一个强大的多模态 AI 模型，支持文本、图像、音频、视频等多种输入。

协议特点

• 原生多模态：无需额外配置即可处理文本、图像、音频、视频
• 独特的消息结构：使用 contents 和 parts 而非 messages
• Gemini 3 新特性：
• 支持 thinking_level 控制推理深度
• 引入 thought_signature 维持多轮对话逻辑连贯
• 函数调用支持唯一 id，实现并行调用
• 函数结果使用专门的 tool 角色

核心端点

主要端点是 /v1beta/models/{model}:generateContent，用于生成内容。

请求参数详解

首先看一个完整的 Gemini API 请求示例（使用 REST API）：

{
  // --- 系统指令（可选但推荐） ---
  "system_instruction": {
    "parts": [
      {
        "text": "你是一个专业的 AI 助手，擅长解释技术概念。请用清晰、易懂的语言回答问题。"
      }
    ]
  },

  // --- 对话内容 ---
  "contents": [
    {
      "role": "user",
      "parts": [
        {
          "text": "什么是机器学习？请详细解释。"
        }
      ]
    }
  ],

  // --- 生成配置 ---
  "generationConfig": {
    "temperature": 0.7, // 1. 控制随机性 (0-2)
    "topP": 0.95, // 2. 核采样阈值
    "topK": 40, // 3. 候选 token 数量
    "maxOutputTokens": 2048, // 4. 最大输出长度
    "stopSequences": [], // 停止序列

    // --- Gemini 3 核心新参数 ---
    "thinking_level": "HIGH" // 5. 【核心】替代旧版 budget，控制推理深度 (MINIMAL/MEDIUM/HIGH)
  },

  // --- 安全设置 ---
  "safetySettings": [
    {
      "category": "HARM_CATEGORY_HARASSMENT",
      "threshold": "BLOCK_MEDIUM_AND_ABOVE"
    }
  ]
}

必需参数

contents (array)

• 含义：对话内容列表，包含用户和模型的消息
• 结构：每个 content 包含 role 和 parts 字段
• 角色类型：
• user：用户输入
• model：模型回复（注意：不是 assistant）
• tool：工具/函数调用结果（Gemini 3 新增）
• 示例：

{
  "contents": [
    {
      "role": "user",
      "parts": [{ "text": "你好，请介绍一下自己" }]
    },
    {
      "role": "model",
      "parts": [
        { "text": "你好！我是 Gemini，一个由 Google 开发的 AI 助手..." }
      ]
    },
    {
      "role": "user",
      "parts": [{ "text": "你能做什么？" }]
    }
  ]
}

parts (array)

• 含义：消息的组成部分，支持多种类型
• 类型：
• text：文本内容
• inline_data：内联数据（如图像的 base64）
• file_data：文件引用
• function_call：函数调用请求（模型生成）
• function_response：函数调用结果（用户返回）

可选参数

generationConfig (object)

生成配置对象，控制输出行为：

• temperature (number, 0-2)：控制随机性

• topP (number, 0-1)：核采样参数，默认 0.95

• topK (integer)：Top-K 采样，默认 40

• maxOutputTokens (integer)：最大输出 token 数，默认 2048

• stopSequences (array)：停止序列列表

• candidateCount (integer)：生成候选数量，默认 1

• thinking_level (string)：推理深度（Gemini 3 新增）

• MINIMAL：快速响应，适合简单问题

• MEDIUM：平衡速度和质量

• HIGH：深度推理，适合复杂问题

safetySettings (array)

安全设置，控制内容过滤：

• 类别：
• HARM_CATEGORY_HARASSMENT：骚扰
• HARM_CATEGORY_HATE_SPEECH：仇恨言论
• HARM_CATEGORY_SEXUALLY_EXPLICIT：色情内容
• HARM_CATEGORY_DANGEROUS_CONTENT：危险内容
• 阈值：
• BLOCK_NONE：不阻止
• BLOCK_ONLY_HIGH：仅阻止高风险
• BLOCK_MEDIUM_AND_ABOVE：阻止中等及以上风险
• BLOCK_LOW_AND_ABOVE：阻止低等及以上风险

systemInstruction (object)

系统指令，类似 OpenAI 的 system message：

{
  "systemInstruction": {
    "parts": [
      {
        "text": "你是一个专业的 Python 编程助手，代码要简洁高效。"
      }
    ]
  },
  "contents": [...]
}

响应格式

2026 最新 Gemini 3 响应示例 (JSON)：

{
  "candidates": [
    {
      "content": {
        "role": "model",
        "parts": [
          {
            // 1. 【新增】推理思考过程（仅在开启 thinking_level 时出现）
            "thought": "首先，我需要定义机器学习的三个核心要素：数据、算法和模型。然后..."
          },
          {
            "text": "机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进，而无需明确编程..."
          }
        ],
        // 2. 【核心新增】思维签名，用于多轮对话维持逻辑连贯
        "thought_signature": "asdf897asdf_logic_chain_v3"
      },
      "finishReason": "STOP",
      "safetyRatings": [
        {
          "category": "HARM_CATEGORY_HARASSMENT",
          "probability": "NEGLIGIBLE",
          "blocked": false // 新增：更直观的布尔值判断
        }
      ],
      "avg_logprobs": -0.15, // 3. 【新增】平均对数概率，用于评估回答的置信度
      "groundingMetadata": {
        // 4. 【增强】联网搜索溯源
        "searchEntryPoint": {
          "renderedContent": "..."
        },
        "groundingChunks": [
          {
            "web": {
              "uri": "[https://wikipedia.org/](https://wikipedia.org/)...",
              "title": "Machine Learning"
            }
          }
        ]
      }
    }
  ],
  "usageMetadata": {
    "promptTokenCount": 10,
    "candidatesTokenCount": 150,
    "totalTokenCount": 160,
    // 5. 【新增】分项计费统计
    "cachedContentTokenCount": 0,
    "reasoningTokenCount": 45, // 思考过程消耗的 Token
    "mediaTokenCount": 0
  }
}

Gemini 3 核心新增字段说明：

• parts[].thought：推理思考过程（需开启 thinking_level）
• thought_signature：思维签名，用于多轮对话维持逻辑连贯性
• 重要：在函数调用的多轮交互中，必须将 thought_signature 包含在历史记录中，否则模型会丢失”为什么要调用这个函数”的逻辑上下文
• avg_logprobs：平均对数概率，评估回答置信度（越接近 0 越自信）
• groundingMetadata：联网搜索溯源信息
• searchEntryPoint：搜索入口
• groundingChunks：引用的网页来源
• usageMetadata 新增字段：
• cachedContentTokenCount：缓存内容 Token 数
• reasoningTokenCount：推理过程消耗的 Token
• mediaTokenCount：多模态内容消耗的 Token

通用响应字段说明：

• finishReason：结束原因
• STOP：自然结束
• MAX_TOKENS：达到最大 token 限制
• SAFETY：触发安全过滤
• RECITATION：检测到重复内容
• safetyRatings：安全评级
• category：安全类别
• probability：风险概率（NEGLIGIBLE/LOW/MEDIUM/HIGH）
• blocked：是否被阻止

函数调用（Function Calling）

Gemini 3 函数调用的关键变化：

1. 强制要求 ID 匹配：为了支持并行函数调用（Parallel Function Calling）和复杂的长程推理，Gemini 3 现在生成的 function_call 对象中包含一个唯一的 id。当你返回 function_response 时，必须带上对应的 id，否则在多轮对话或并行调用时，模型会因为无法对齐”哪个结果对应哪个请求”而报错（400 Error）。

2. 思维签名（thought_signature）：在函数调用的多轮交互中，你必须确保模型返回的 thought_signature 被包含在历史记录中，否则模型会丢失”为什么要调用这个函数”的逻辑上下文。

3. 专门的 tool 角色：虽然在某些早期实现中会将函数结果标记为 user 角色，但 Gemini 3 的标准做法是引入了专门的 tool 角色（或在某些 SDK 中称为 function 角色）。将函数结果发回给模型时，消息的 role 应当设为 tool 或 function，而不是 user。这有助于模型区分”用户说的话”和”工具返回的客观事实”。

与 OpenAI 协议的主要区别

特性	OpenAI	Gemini
消息结构	messages	contents
消息组成	content (string)	parts (array)
助手角色名	assistant	model
工具结果角色	tool	tool (Gemini 3)
系统提示	role: "system"	systemInstruction
生成配置	顶层参数	generationConfig 对象
安全设置	无（依赖内容审核）	safetySettings 数组
多模态支持	需要特殊格式	原生支持，使用 parts 数组
推理深度控制	无	thinking_level
思维签名	无	thought_signature
函数调用 ID	tool_calls[].id	function_call.id
函数结果 ID	tool_call_id	function_response.id
函数参数格式	JSON 字符串	对象
联网搜索溯源	无	groundingMetadata
Token 分项统计	基础统计	详细分项（推理、媒体等）

获取 API Key

1. 访问 Google AI Studio
2. 登录 Google 账号
3. 点击 “Get API Key” 创建新的 API Key
4. 复制 API Key 并保存到环境变量

最佳实践

1. 多模态优势：充分利用原生多模态能力，无需额外配置
2. 安全设置：根据应用场景调整安全阈值
3. 流式输出：对于长回复，使用流式输出提升用户体验
4. 函数调用：利用 Function Calling 实现复杂的工具集成

流式输出（Streaming）详解

流式输出是什么东西？

流式输出是指 AI 模型逐步生成并返回响应内容，而不是等待全部内容生成完毕后一次性返回。

非流式 vs 流式

非流式输出：

用户提问 → AI 思考 → 等待... → 完整回复一次性显示

用户体验：需要等待较长时间，看到的是突然出现的完整文本。

流式输出：

用户提问 → AI 思考 → 逐字显示 → 逐字显示 → 逐字显示 → 完成

用户体验：立即看到响应开始，文字逐渐出现，类似打字效果。

流式输出有什么用？

在大语言模型（LLM）爆发之前，普通的网页请求（比如看新闻、查天气）基本都是“全量返回”的：服务器把所有内容一次性计算好、打包，再传给浏览器，由浏览器统一渲染。但在大模型时代，这种方式就不太合适了。因为模型本质上是一个“概率预测器”，它不是一次性生成整段内容，而是按 Token（字/词）逐步向外预测、逐步生成的。如果仍然等到 LLM 把所有内容都生成完再“全量返回”，用户往往需要经历较长的等待时间，体验会明显变差。在这种情况下，流式输出就显得很有必要：

1. 改善用户体验：用户不需要长时间等待，立即看到响应开始
2. 降低感知延迟：即使总时间相同，流式输出让用户感觉更快
3. 实时反馈：用户可以提前看到部分内容，决定是否继续等待
4. 处理长文本：对于长回复，流式输出避免了长时间的空白等待

来讲讲 SSE（Server-Sent Events）

SSE（Server-Sent Events）是一种服务器向客户端推送数据的技术，AI API 的流式输出就是基于 SSE 实现的。

SSE 的特点

1. 单向通信：服务器 → 客户端（客户端不能通过 SSE 发送数据）
2. 基于 HTTP：使用标准 HTTP 协议，无需特殊协议
3. 自动重连：连接断开后会自动重连
4. 文本格式：传输的是文本数据，通常是 JSON

SSE 的数据格式

SSE 使用特定的文本格式传输数据：

data: {"content": "你"}

data: {"content": "好"}

data: {"content": "，"}

data: {"content": "我"}

data: {"content": "是"}

data: [DONE]

每条消息以 data: 开头，以两个换行符 \n\n 结束。

SSE vs WebSocket

特性	SSE	WebSocket
通信方向	单向（服务器 → 客户端）	双向（客户端 ↔ 服务器）
协议	HTTP	WebSocket 协议（ws://）
连接建立	简单，标准 HTTP 请求	需要握手升级
浏览器支持	原生支持（EventSource API）	原生支持（WebSocket API）
自动重连	内置自动重连	需要手动实现
数据格式	文本（通常 JSON）	文本或二进制
防火墙友好	是（使用标准 HTTP）	可能被阻止
适用场景	服务器推送、实时通知、AI 流式	聊天、游戏、实时协作

为什么 AI API 使用 SSE 而不是 WebSocket？

1. 单向通信足够：AI 生成响应是单向的，不需要双向通信
2. 更简单：SSE 基于 HTTP，无需额外的协议升级
3. 更好的兼容性：HTTP 更容易通过代理、负载均衡器
4. 自动重连：SSE 内置重连机制，更可靠
5. 标准化：OpenAI 等厂商都采用 SSE，已成为事实标准

如何实现流式输出

在实现流式输出之前，我们需要理解其背后的工作原理和处理逻辑。

流式输出的工作原理

1. 服务器端的生成过程

AI 模型生成文本的过程本质上是逐个 token 预测的：

输入: "什么是机器学习？"
↓
模型预测: "机" (第1个token)
↓
模型预测: "器" (第2个token，基于前面的上下文)
↓
模型预测: "学" (第3个token)
↓
... 持续预测直到结束

在非流式模式下，服务器会等待所有 token 生成完毕后，一次性返回完整结果。而在流式模式下，服务器每生成一个或几个 token，就立即通过 SSE 推送给客户端。

2. SSE 数据传输格式

服务器通过 SSE 发送的数据格式如下：

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"机"},"finish_reason":null}]}

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"器"},"finish_reason":null}]}

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"学"},"finish_reason":null}]}

data: [DONE]

每条消息：

• 以 data: 开头
• 包含一个 JSON 对象（称为 chunk）
• 以两个换行符 \n\n 结束
• 最后一条消息是 data: [DONE] 表示流结束

3. 客户端的处理流程

客户端需要按照以下步骤处理流式响应：

1. 建立 HTTP 连接（设置 stream=True）
   ↓
2. 持续监听服务器推送的数据
   ↓
3. 每收到一个 chunk：
   - 解析 JSON 数据
   - 提取 delta.content（本次新增的内容）
   - 立即显示给用户
   - 累积到完整内容中
   ↓
4. 检测到 finish_reason 不为 null 或收到 [DONE]
   ↓
5. 关闭连接，流式输出完成

关于 SSE 的”粘包”处理（重要）

由于网络传输的特性，客户端收到的一个数据块并不一定正好是一个完整的 data: {...}\n\n。

常见现象：

• 有时一次读到两个完整的 chunk
• 有时只读到半个 chunk（JSON 被截断）
• 有时一个 chunk 被拆分到多次读取中

解决方案：

# 客户端需要维护一个缓冲区
buffer = ""

for chunk in response.iter_content():
    # 1. 将新数据追加到缓冲区
    buffer += chunk.decode('utf-8')

    # 2. 寻找完整的消息（以 \n\n 分隔）
    while '\n\n' in buffer:
        # 3. 提取一个完整的消息
        message, buffer = buffer.split('\n\n', 1)

        # 4. 解析并处理
        if message.startswith('data: '):
            data = message[6:]  # 去掉 "data: " 前缀
            if data == '[DONE]':
                break
            json_obj = json.loads(data)
            # 处理 json_obj...

关键点：

• 使用缓冲区累积接收到的数据
• 只有找到 \n\n 分隔符时才解析
• 未完成的部分保留在缓冲区中，等待下次数据到达

4. 关键技术点

• 增量更新（Delta）：每个 chunk 只包含新增的内容片段，不是完整内容
• 实时显示：使用 print(..., end="", flush=True) 或类似机制立即输出，不等待换行
• 内容累积：客户端需要自己拼接所有 chunk 的内容，得到完整文本
• 结束检测：通过 finish_reason 字段判断是否结束（stop、length 等）

性能优化：Flush 机制

在实际部署中，如果后端服务器（如 Nginx）开启了缓存（Buffering），流式效果会失效——文字会一坨一坨地蹦出来，而不是平滑流出。

问题原因：

• Nginx 等反向代理默认会缓冲响应内容
• 只有缓冲区满了或响应结束时才会发送给客户端
• 这导致流式传输的实时性丧失

解决方案：

# 后端代码中必须设置响应头
response.headers['X-Accel-Buffering'] = 'no'  # 针对 Nginx
response.headers['Cache-Control'] = 'no-cache'
response.headers['Connection'] = 'keep-alive'

Nginx 配置（可选）：

location /api/chat {
    proxy_pass http://backend;
    proxy_buffering off;  # 关闭缓冲
    proxy_cache off;      # 关闭缓存
}

确保每一个 chunk 都能实时流出，不被中间层缓存。

5. 流式 vs 非流式的数据对比

非流式响应（一次性返回）：

{
  "choices": [
    {
      "message": {
        "content": "机器学习是人工智能的一个分支..." // 完整内容
      },
      "finish_reason": "stop"
    }
  ]
}

流式响应（多次返回）：

// 第1次
{"choices": [{"delta": {"content": "机"}, "finish_reason": null}]}

// 第2次
{"choices": [{"delta": {"content": "器"}, "finish_reason": null}]}

// 第3次
{"choices": [{"delta": {"content": "学"}, "finish_reason": null}]}

// ...

// 最后一次
{"choices": [{"delta": {}, "finish_reason": "stop"}]}

注意：

• 流式使用 delta 字段（增量），非流式使用 message 字段（完整）
• 流式的 finish_reason 在最后一个 chunk 才不为 null
• 流式需要客户端自己拼接所有 delta.content

6. Delta 模式的例外：工具调用（Function Calling）

当涉及到 Function Calling 时，流式的 delta 结构会发生变化。函数参数是以字符串形式逐段传输的：

// 第1个 chunk：开始传输函数调用
{
  "choices": [{
    "delta": {
      "tool_calls": [{
        "index": 0,
        "id": "call_abc123",
        "type": "function",
        "function": {
          "name": "get_weather",
          "arguments": ""
        }
      }]
    },
    "finish_reason": null
  }]
}

// 第2个 chunk：传输参数的第一部分
{
  "choices": [{
    "delta": {
      "tool_calls": [{
        "index": 0,
        "function": {
          "arguments": "{\"locat"
        }
      }]
    },
    "finish_reason": null
  }]
}

// 第3个 chunk：传输参数的第二部分
{
  "choices": [{
    "delta": {
      "tool_calls": [{
        "index": 0,
        "function": {
          "arguments": "ion\": \"Shang"
        }
      }]
    },
    "finish_reason": null
  }]
}

// 第4个 chunk：传输参数的最后部分
{
  "choices": [{
    "delta": {
      "tool_calls": [{
        "index": 0,
        "function": {
          "arguments": "hai\"}"
        }
      }]
    },
    "finish_reason": null
  }]
}

// 最后一个 chunk
{
  "choices": [{
    "delta": {},
    "finish_reason": "tool_calls"
  }]
}

处理要点：

• 函数参数（arguments）是 JSON 字符串，会被拆分成多个片段
• 客户端需要拼接所有 arguments 片段，最后再解析为 JSON 对象
• 完整的参数示例：{"location": "Shanghai"}
• finish_reason 为 "tool_calls" 表示需要执行工具调用

# 客户端处理示例
tool_calls = {}

for chunk in stream:
    delta = chunk['choices'][0]['delta']

    if 'tool_calls' in delta:
        for tool_call in delta['tool_calls']:
            index = tool_call['index']

            # 初始化工具调用
            if index not in tool_calls:
                tool_calls[index] = {
                    'id': tool_call.get('id', ''),
                    'type': tool_call.get('type', ''),
                    'function': {
                        'name': tool_call.get('function', {}).get('name', ''),
                        'arguments': ''
                    }
                }

            # 累积参数字符串
            if 'function' in tool_call and 'arguments' in tool_call['function']:
                tool_calls[index]['function']['arguments'] += tool_call['function']['arguments']

# 流结束后，解析完整的参数
for tool_call in tool_calls.values():
    args_str = tool_call['function']['arguments']
    tool_call['function']['arguments'] = json.loads(args_str)

JavaScript 实现

使用 OpenAI SDK：

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});

async function streamChat(userMessage) {
  const stream = await client.chat.completions.create({
    model: "gpt-3.5-turbo",
    messages: [{ role: "user", content: userMessage }],
    stream: true,
  });

  process.stdout.write("AI: ");

  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || "";
    process.stdout.write(content);
  }

  console.log("\n");
}

// 使用
streamChat("介绍一下 JavaScript 的闭包");

使用原生 Fetch API：

async function streamChatFetch(userMessage) {
  const response = await fetch("[https://api.openai.com/v1/chat/completions](https://api.openai.com/v1/chat/completions)", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
    },
    body: JSON.stringify({
      model: "gpt-3.5-turbo",
      messages: [{ role: "user", content: userMessage }],
      stream: true,
    }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  process.stdout.write("AI: ");

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    // 解码数据
    const chunk = decoder.decode(value);
    const lines = chunk.split("\n");

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const data = line.slice(6);

        if (data === "[DONE]") {
          console.log("\n");
          return;
        }

        try {
          const parsed = JSON.parse(data);
          const content = parsed.choices[0]?.delta?.content || "";
          process.stdout.write(content);
        } catch (e) {
          // 忽略解析错误
        }
      }
    }
  }
}

// 使用
streamChatFetch("什么是 Promise？");

前端实现（浏览器）

使用 EventSource API：

// 注意：OpenAI API 不支持直接使用 EventSource（需要 POST 请求）
// 这里展示的是通用的 SSE 客户端实现

function streamChatBrowser(userMessage) {
  // 需要后端代理，因为 EventSource 只支持 GET 请求
  const eventSource = new EventSource(
    `/api/chat/stream?message=${encodeURIComponent(userMessage)}`,
  );

  const outputDiv = document.getElementById("output");

  eventSource.onmessage = (event) => {
    if (event.data === "[DONE]") {
      eventSource.close();
      return;
    }

    try {
      const data = JSON.parse(event.data);
      const content = data.choices[0]?.delta?.content || "";
      outputDiv.textContent += content;
    } catch (e) {
      console.error("解析错误:", e);
    }
  };

  eventSource.onerror = (error) => {
    console.error("SSE 错误:", error);
    eventSource.close();
  };
}

使用 Fetch API（推荐）：

async function streamChatBrowser(userMessage) {
  const response = await fetch("/api/chat", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      message: userMessage,
    }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  const outputDiv = document.getElementById("output");

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const chunk = decoder.decode(value);
    const lines = chunk.split("\n");

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const data = line.slice(6);

        if (data === "[DONE]") {
          return;
        }

        try {
          const parsed = JSON.parse(data);
          const content = parsed.choices[0]?.delta?.content || "";
          outputDiv.textContent += content;
        } catch (e) {
          // 忽略解析错误
        }
      }
    }
  }
}

流式输出的最佳实践

1. 错误处理：流式输出中途可能中断，需要妥善处理错误
2. 缓冲处理：可能一次收到多个 chunk，需要正确解析
3. 用户体验：添加打字动画效果，提升视觉体验
4. 取消机制：允许用户中途取消生成
5. 内容累积：保存完整内容，方便后续使用