使用 Google Chat API 流式传输消息

本指南介绍了如何在 Google Chat 中使用 Chat API 中的 StreamMessage 资源以增量方式流式传输消息。借助流式传输，Google Chat 应用可以在最终确定消息之前显示实时进度指示器、中间推理或工具执行步骤，以及增量文本回答。

流式传输专为由生成式 AI 或长时间运行的后端工作流提供支持的 Chat 应用而设计，在这些应用中，生成完整回答需要几秒钟或几分钟的时间。

消息流式传输的运作方式

流式消息在成为聊天室中的标准 Message 资源之前，会经历活跃流式状态：

  1. 启动串流 (spaces.streamMessages.create)：Chat 应用在话题式聊天室中创建 StreamMessage，并引用其回复的用户互动消息。该 API 会在会话中显示实时进度指示器，并返回 StreamMessage 资源名称和初始 nextChunkToken。
  2. 更新串流（appendStreamMessageContent 或 patch）：当 Chat 应用生成输出时，它会使用最新的 nextChunkToken 发送依序更新：
    • 调用 spaces.streamMessages.appendStreamMessageContent 以串联新的文本块、附加中间步骤、添加内嵌引用或更新 progressText。
    • 调用 spaces.streamMessages.patch 以替换流式传输的内容块或就地更新 progressText。
  3. 完成或舍弃数据流（commit 或 delete）：
    • 调用 spaces.streamMessages.commit 以最终确定数据流并将其转换为永久 Message 资源。您还可以在提交时附加仅限提交的元素，例如卡片、辅助微件和消息底部参考来源。
    • 调用 spaces.streamMessages.delete 可舍弃活跃流，而无需在聊天室中保存消息。

在流式传输处于活跃状态时，只有正在流式传输的引用消息 (referencedMessage) 的发送者可以点击停止来取消生成。Chat 会向您的 Chat 应用发送取消系统命令事件，以便您停止后端工作，并提交部分输出或删除数据流。

前提条件

如需使用 Chat API 流式传输消息，请完成以下前提条件：

  • 可访问 Google Chat 的 Google Workspace 商务版或企业版账号。
  • 设置环境：
  • 将您的 Chat 应用添加到串联式 Chat 聊天室（或已包含至少一条消息的 1 对 1 私信），并获取要回复的可访问消息的资源名称 (spaces/SPACE_ID/messages/MESSAGE_ID)。

  • 对于 Python 示例，请安装 Python 版 Google API 客户端库和 google-auth，并初始化 Chat API 服务客户端：

    pip install google-api-python-client google-auth
    from google.oauth2 import service_account
from googleapiclient.discovery import build

SCOPES = ["https://www.googleapis.com/auth/chat.bot"]
creds = service_account.Credentials.from_service_account_file(
    "credentials.json", scopes=SCOPES
)
chat = build("chat", "v1", credentials=creds)

启动数据流

如需开始流式传输回答，请调用 spaces.streamMessages.create 并在请求正文中传递 StreamMessage：

  • referencedMessage（必需）：触发响应的消息的资源名称，格式为 spaces/{space}/messages/{message}。引用的消息必须属于同一聊天室，并且可供您的 Chat 应用访问。Chat 会在与 referencedMessage 相同的会话中发布流式回复。
  • progressText（必需）：显示在进度指示器旁边的非空纯文本状态标签（最多 250 个字符），例如 "Thinking..."。
  • markupSyntax（可选）：消息的文本格式语法，例如 MARKUP_SYNTAX_MARKDOWN 或 MARKUP_SYNTAX_CHAT（如果为 MARKUP_SYNTAX_UNSPECIFIED，则默认为 Chat 应用配置的语法）。如果您打算添加内嵌引用，请设置 MARKUP_SYNTAX_MARKDOWN。
  • streamContentChunks（可选）：要立即显示的初始 text、step 或 citation 块。

保存响应中的 name (spaces/{space}/streamMessages/{stream_message}) 和 nextChunkToken，以便在后续调用中使用。

HTTP

POST https://chat.googleapis.com/v1/spaces/SPACE_ID/streamMessages HTTP/1.1
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/json

{
  "referencedMessage": "spaces/SPACE_ID/messages/MESSAGE_ID",
  "progressText": "Thinking...",
  "markupSyntax": "MARKUP_SYNTAX_MARKDOWN"
}

Python

stream = (
    chat.spaces()
    .streamMessages()
    .create(
        parent="spaces/SPACE_ID",
        body={
            "referencedMessage": "spaces/SPACE_ID/messages/MESSAGE_ID",
            "progressText": "Thinking...",
            "markupSyntax": "MARKUP_SYNTAX_MARKDOWN",
        },
    )
    .execute()
)

stream_name = stream["name"]
chunk_token = stream["nextChunkToken"]

该方法会返回一个 StreamMessage 资源：

{
  "name": "spaces/SPACE_ID/streamMessages/STREAM_MESSAGE_ID",
  "progressText": "Thinking...",
  "markupSyntax": "MARKUP_SYNTAX_MARKDOWN",
  "nextChunkToken": "INITIAL_CHUNK_TOKEN"
}

附加内容块

当 Chat 应用生成中间步骤或文本词元时，调用 spaces.streamMessages.appendStreamMessageContent：

  • 在 chunkToken 中传递最新的 nextChunkToken。每次成功调用都会返回一个新的 nextChunkToken，您必须在下一次更新或提交请求中传递该值。
  • text 块会直接串接到之前流式传输的文本。根据需要在分块之间添加空格或换行符。
  • step 个分块会附加到消息中显示的中间步骤列表中。

  • progressText 为可选项。如果您省略 progressText 或将其设置为 ""，Chat 会继续显示之前设置的 progressText。

HTTP

POST https://chat.googleapis.com/v1/spaces/SPACE_ID/streamMessages/STREAM_MESSAGE_ID:appendStreamMessageContent HTTP/1.1
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/json

{
  "chunkToken": "INITIAL_CHUNK_TOKEN",
  "progressText": "Searching documentation...",
  "streamContentChunks": [
    {
      "step": {
        "title": "Search knowledge base",
        "description": "Found 3 matching articles."
      }
    },
    {
      "text": "See <chat-citation data-id=\"c1\">guide</chat-citation>:\n"
    },
    {
      "citation": {
        "id": "c1",
        "citedSources": [
          {
            "title": "System Architecture Guide",
            "uri": "https://example.com/docs/arch"
          }
        ]
      }
    }
  ]
}

Python

stream = (
    chat.spaces()
    .streamMessages()
    .appendStreamMessageContent(
        name=stream_name,
        body={
            "chunkToken": chunk_token,
            "progressText": "Searching documentation...",
            "streamContentChunks": [
                {
                    "step": {
                        "title": "Search knowledge base",
                        "description": "Found 3 matching articles.",
                    }
                },
                {
                    "text": (
                        'See <chat-citation data-id="c1">'
                        "guide</chat-citation>:\n"
                    )
                },
                {
                    "citation": {
                        "id": "c1",
                        "citedSources": [
                            {
                                "title": "System Architecture Guide",
                                "uri": "https://example.com/docs/arch",
                            }
                        ],
                    }
                },
            ],
        },
    )
    .execute()
)

chunk_token = stream["nextChunkToken"]

替换流式传输的内容或进度文本

如果您需要重写之前流式传输的内容（例如，在运行后处理或安全过滤器之后），或者仅更新 progressText 而不附加分块，请调用 spaces.streamMessages.patch：

  • 将 updateMask 设置为要更新的字段的英文逗号分隔列表：progressText、streamContentChunks 或 progressText,streamContentChunks。
  • 在 StreamMessage 请求正文中添加最新的 nextChunkToken，并记录响应中返回的新 nextChunkToken。
  • 当 streamContentChunks 处于 updateMask 状态时，提供的分块会替换之前流式传输的内容，而不是附加到其后。

  • 当 progressText 位于 updateMask 中时，progressText 必须是不超过 250 个字符的非空纯文本。

HTTP

PATCH https://chat.googleapis.com/v1/spaces/SPACE_ID/streamMessages/STREAM_MESSAGE_ID?updateMask=progressText,streamContentChunks HTTP/1.1
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/json

{
  "nextChunkToken": "LATEST_CHUNK_TOKEN",
  "progressText": "Formatting response...",
  "streamContentChunks": [
    {
      "text": "Updated response text."
    }
  ]
}

Python

stream = (
    chat.spaces()
    .streamMessages()
    .patch(
        name=stream_name,
        updateMask="progressText,streamContentChunks",
        body={
            "nextChunkToken": chunk_token,
            "progressText": "Formatting response...",
            "streamContentChunks": [
                {"text": "Updated response text."}
            ],
        },
    )
    .execute()
)

chunk_token = stream["nextChunkToken"]

完成数据流

生成完成后，使用最新的 chunkToken 调用 spaces.streamMessages.commit，以将数据流最终确定为永久 Message 资源：

  • 您可以选择在提交请求中包含最终 streamContentChunks，包括仅提交分块类型（card、accessoryWidget 和 citedSource）。

  • 如果您已在之前的调用中流式传输了非空内容，则可以在提交时省略 streamContentChunks。最终提交的消息不能为空。

HTTP

POST https://chat.googleapis.com/v1/spaces/SPACE_ID/streamMessages/STREAM_MESSAGE_ID:commit HTTP/1.1
Authorization: Bearer ACCESS_TOKEN
Content-Type: application/json

{
  "chunkToken": "LATEST_CHUNK_TOKEN",
  "streamContentChunks": [
    {
      "text": "All checks passed."
    },
    {
      "citedSource": {
        "title": "Deployment Runbook",
        "uri": "https://example.com/docs/runbook"
      }
    },
    {
      "card": {
        "cardId": "details-card",
        "card": {
          "header": {
            "title": "Build Status"
          },
          "sections": [
            {
              "widgets": [
                {
                  "textParagraph": {
                    "text": "All 42 checks succeeded."
                  }
                }
              ]
            }
          ]
        }
      }
    }
  ]
}

Python

status_card = {
    "cardId": "details-card",
    "card": {
        "header": {"title": "Build Status"},
        "sections": [
            {
                "widgets": [
                    {"textParagraph": {"text": "All 42 checks succeeded."}}
                ]
            }
        ],
    },
}

commit_response = (
    chat.spaces()
    .streamMessages()
    .commit(
        name=stream_name,
        body={
            "chunkToken": chunk_token,
            "streamContentChunks": [
                {"text": "All checks passed."},
                {
                    "citedSource": {
                        "title": "Deployment Runbook",
                        "uri": "https://example.com/docs/runbook",
                    }
                },
                {"card": status_card},
            ],
        },
    )
    .execute()
)

final_message = commit_response["message"]
print(f"Committed message: {final_message['name']}")

舍弃活跃直播

如果发生不可恢复的错误，或者用户取消生成，并且您不想在会话串中留下部分回复，请调用 spaces.streamMessages.delete：

HTTP

DELETE https://chat.googleapis.com/v1/spaces/SPACE_ID/streamMessages/STREAM_MESSAGE_ID HTTP/1.1
Authorization: Bearer ACCESS_TOKEN

Python

chat.spaces().streamMessages().delete(name=stream_name).execute()

将数据流提交到 Message 后，如果您需要删除最终确定的消息，请改为调用 spaces.messages.delete。

支持的内容分块类型

streamContentChunks 中的每个条目都指定了 StreamContentChunk 联合中的一个字段。某些分块类型可以在整个数据流中发送，而其他分块类型仅在 spaces.streamMessages.commit 中接受：

块字段 支持的方法 最终 Message 字段 备注
text create、appendStreamMessageContent、patch、commit text 增量文本。已于 appendStreamMessageContent串联；已于 patch替换。 最终提交的文本不能超过 32,000 字节。
step create、appendStreamMessageContent、patch、commit elements.steps.steps[] 包含纯文本 title（必需）和可选 description 的中间处理步骤。
citation create、appendStreamMessageContent、patch、commit elements.citations[] 内嵌 Citation，其中包含字母数字 id（少于 63 个字符，允许使用下划线）和 citedSources。需要 text 中有 markupSyntax: "MARKUP_SYNTAX_MARKDOWN" 和匹配的 <chat-citation data-id="ID">...</chat-citation> 标记。发送带有现有 id 的 citation 会覆盖之前的引用。
card 仅 commit cardsV2[] 互动式或信息性 CardWithId 卡片。
accessoryWidget 仅 commit accessoryWidgets[] 消息下方显示的 AccessoryWidget 按钮，例如反馈按钮。
citedSource 仅 commit elements.citedSources[] CitedSource 页脚链接显示在消息下方，不带内嵌引用标签。

处理用户发起的流取消

在流式传输处于活跃状态时，只有正在流式传输的引用消息 (referencedMessage) 的发送者才能点击流式传输消息上的停止。当用户停止直播时，Chat 会向您的 Chat 应用端点发送一个系统命令互动事件，其中包含 cancelStream.streamMessage（已停止直播的 spaces/{space}/streamMessages/{stream_message} 资源名称）。

传入 JSON 载荷的结构取决于您是将 Chat 应用构建为 Google Workspace 插件，还是使用 Chat API 互动事件：

Google Workspace 加购项

{
  "commonEventObject": {
    "hostApp": "CHAT"
  },
  "chat": {
    "user": {
      "name": "users/USER_ID"
    },
    "space": {
      "name": "spaces/SPACE_ID"
    },
    "systemCommandPayload": {
      "systemCommandMetadata": {
        "cancelStream": {
          "streamMessage": "spaces/SPACE_ID/streamMessages/STREAM_ID"
        }
      }
    }
  }
}

Google Chat API

{
  "type": "SYSTEM_COMMAND",
  "space": {
    "name": "spaces/SPACE_ID"
  },
  "user": {
    "name": "users/USER_ID"
  },
  "systemCommandMetadata": {
    "cancelStream": {
      "streamMessage": "spaces/SPACE_ID/streamMessages/STREAM_ID"
    }
  }
}

当您的 Chat 应用收到 cancelStream 事件时：

  1. 向后台工作器发出信号，让其停止为 streamMessage 生成更多词元或运行下游工具。
  2. 使用以下方法之一解决活跃数据流问题：
    • 保留部分输出：使用最新的 chunkToken 调用 spaces.streamMessages.commit（可选择附加备注，例如 "\n\n_Generation stopped._"）。
    • 舍弃回复：调用 spaces.streamMessages.delete，这样就不会向会话串保存任何内容。
  3. 从互动事件处理脚本返回空的 HTTP 200 OK JSON 响应 ({})。

限制

消息流式传输存在以下限制：

  • 平台可用性：消息流式传输仅在网页版 Chat 中可见。即将支持 Android 和 iOS。
  • 流超时：活跃的 StreamMessage 会在创建 10 分钟后超时。您的 Chat 应用必须在开始直播后的 10 分钟内调用 spaces.streamMessages.commit 或 spaces.streamMessages.delete。