跳到文章正文
Grok 实用教程

Grok流式输出怎么接入?Python逐段显示、完成标记与中断保存

通过xAI兼容SDK读取Grok文本流,正确处理空片段和结束原因,把中断结果保存为草稿,区分界面…

让回答边生成边显示,可以减少用户盯着空白界面的时间,但“已经出现文字”不代表请求已成功完成。Grok文本流式输出使用SSE,本教程用Python兼容SDK接收正文、检查结束原因并保存中断草稿,避免网络断开后把半段结果当成最终交付。

第一步:确认模型能力与运行环境

官方说明所有支持文本输出的模型可使用流式,包括聊天和图像理解;图片生成输出不支持同样方式。本例使用grok-4.7文本,不把图片生成接口改一个stream参数就当成可流式。需要自己的xAI API权限、额度和环境变量XAI_API_KEY,第三方聊天会员不能替代API访问。

安装Python、openai兼容SDK和httpx。这里使用Chat Completions的choices与delta结构,不与Responses API事件名混用。官方建议用SDK解析事件流,初次接入无需手工把每行网络数据当成JSON拆解;示例未实际执行付费调用。

第二步:接收正文并保留完成状态

import os
from pathlib import Path
import httpx
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["XAI_API_KEY"],
    base_url="https://api.x.ai/v1",
    timeout=httpx.Timeout(3600.0, connect=20.0),
    max_retries=0,
)
stream = client.chat.completions.create(
    model="grok-4.7",
    messages=[{"role": "user", "content": "请用三条要点解释如何核对CSV列名,不调用工具。"}],
    stream=True,
)
parts = []
finished = False
reason = None
try:
    for chunk in stream:
        if not chunk.choices:
            continue
        choice = chunk.choices[0]
        text = choice.delta.content
        if text:
            parts.append(text)
            print(text, end="", flush=True)
        if choice.finish_reason is not None:
            reason = choice.finish_reason
            finished = reason == "stop"
finally:
    stream.close()
    draft = Path("grok-output.partial.txt")
    draft.write_text("".join(parts), encoding="utf-8")
    if finished and parts:
        draft.replace("grok-output.txt")
print("\n结束原因:", reason, ";完整正文:", finished and bool(parts))

示例只追加delta.content,不把其他类型片段混入正文,且先判断choices是否为空。print使用flush使终端及时显示;若客户端界面仍整段出现,还需检查自己的界面更新或代理缓冲,不能据此认定模型没有发送流。

第三步:区分完成、停止和异常

示例以本次纯文本任务的stop结束原因为完整标记,其他结束原因保留partial文件供复查。如果网络在循环中抛错,finally仍尝试保存已收到片段,但异常会继续上抛,不会在正常结尾打印成功。实际产品还应向用户显示中断状态。

不要把用户手动取消、超时、内容截断和正常完成都映射成绿色完成图标。任务涉及工具时还需单独处理工具调用流程,不能沿用本例的纯文本完成判断。空正文也不应被当成有内容的最终答案,应检查状态和错误。

第四步:调整等待而非盲目重复

官方提醒推理模型可能需要更长超时,例子给出较长读取等待,连接超时则单独设置。应按应用需求选择合理等待与取消机制,不能保证每个请求都在固定秒数返回。用户取消时关闭连接,并保留草稿状态,不悄悄重发整项请求。

验收用例:正常流结束后只有最终文件;中途断网时已有正文写入partial而不宣称完成;空片段不报错;非stop结束保留草稿;界面展示的正文与最终保存内容一致。用自己的测试环境记录实际行为,不将未执行检查写为通过。

若准备重试,先判断上次是否仍在服务端处理及是否已有可用结果,避免多次付费和重复显示。断点续写是新的上下文请求,不等于原请求自动从断点恢复;没有相应设计时应明确告诉用户重新生成的范围。

第五步:保存时保留状态与版本

正式应用应把任务ID、开始时间、结束状态和输出版本分开记录,敏感提示内容按自己的数据规则处理。流式请求的每个片段并非独立回答,不能简单按片段数计算使用次数或费用。成本应按实际响应信息和控制台记录核对。

完成标准是终端或界面能逐段显示、正文拼接正确、正常完成与异常中断有区别,保存文件也与状态一致。流式改善的是等待体验,它不提高事实准确性的保证,最终内容仍要按任务自己的验收标准检查。

官方资料

资料核对日期:2026年10月27日。

本文地址:https://www.163264.com/18228