人工智能AI 技能AI 评测【免费下载链接】skillsPublic repository for Agent Skills项目地址https://gitcode.com/GitHub_Trending/skills3/skills点击查看免费下载流式Streaming是调用 Claude API 处理长输出、长思考或高max_tokens请求时最关键的工程手段它既能避免 SDK 的 HTTP 空闲连接超时又能让用户在首 token 到达后立刻看到输出。本文以 Claude API Python 流式文档 为骨架结合仓库内 Python README、Tool Use 文档、cURL SSE 参考 与 模型迁移指南 等源码级资料系统讲解同步/异步流式、低层事件、thinking 块处理、工具调用流式、最终消息获取、进度统计、错误处理与六大最佳实践让你能够写出可直接上生产环境的流式代码。为什么默认就应该用流式在 claude-api 技能中流式不是一个可选项而是 SKILL.md 的默认策略任何可能涉及长输入、长输出或高max_tokens的请求都应默认流式因为非流式请求在输出量较大时会命中 SDK 的 HTTP 超时。底层原因在 模型迁移指南 中有明确说明max_tokens大于约 16000 时任何模型都必须流式否则超出 SDK 的 HTTP 超时风险流式场景下除 Haiku 4.5上限 64K外当前所有模型都能达到 128K 输出上限SDK 默认请求超时为 10 分钟空闲连接会被丢弃长输出非流式请求极易超时。而流式配合get_final_message()既可以逐 token 展示也能在结束后拿到与create()完全一致的完整Message对象一举两得。快速开始client.messages.stream()上下文管理器官方推荐的流式入口是client.messages.stream(...)上下文管理器。它会在背后帮你累积事件状态并暴露text_stream迭代器与get_final_message()两个便捷接口。同步示例with client.messages.stream( modelclaude-opus-5, max_tokens64000, messages[{role: user, content: Write a story}] ) as stream: for text in stream.text_stream: print(text, end, flushTrue)要点max_tokens64000是流式请求的推荐默认值——流式不担心超时应给模型留足输出空间非流式则建议~16000见 README 与模型迁移指南 附近的建议print(text, end, flushTrue)中的flushTrue保证每个 token 立即刷到终端是流式输出的标准写法默认模型为claude-opus-5见 SKILL.md 默认策略你可以在 SKILL.md 模型表 中查到全部可用模型 ID。异步版本async with async_client.messages.stream( modelclaude-opus-5, max_tokens64000, messages[{role: user, content: Write a story}] ) as stream: async for text in stream.text_stream: print(text, end, flushTrue)异步版本只需将client换成async_client anthropic.AsyncAnthropic()初始化方式见 Python README将with换成async withfor换成async for。与messages.create()的对应关系client.messages.stream(...)与client.messages.create(...)接受几乎相同的请求参数model、max_tokens、messages、system、tools、thinking等区别仅在于返回的是可迭代的流对象而非一次性Message。流结束后通过stream.get_final_message()拿到的对象与create()的返回值同构可直接读取content、usage、stop_reason等字段。低层方式messages.create(streamTrue)如果你只需要原始事件迭代器、且希望内存占用更低可以跳过messages.stream()直接向messages.create()传streamTruefor event in client.messages.create( modelclaude-opus-5, max_tokens64000, messages[{role: user, content: Write a story}], streamTrue, ): print(event.type)注意这种形式不会为你做任何最终消息累积。每次迭代产出一个原始事件对象如content_block_delta不会自动拼接文本也没有get_final_message()可用。messages.stream()之所以是推荐方式正是因为它在低层事件之上额外提供了状态累积与文本/消息便捷访问。从 SDK 升级视角看anthropic1.x 中isinstance(x, anthropic.Stream)已不再匹配messages.stream()对象——需改用anthropic.lib.streaming.MessageStream/AsyncMessageStream类型检查见 sdk-upgrade.md。Stream仅指create(streamTrue)返回的原始流。从底层认识流SSE 事件类型无论 SDK 如何封装底层传输都是 Server-Sent EventsSSE。原始事件顺序与含义如下表继承自 流式文档事件类型说明触发时机message_start包含消息元数据仅流开始时一次content_block_start新的内容块开始text / tool_use 块开始时content_block_delta增量内容更新每个 token/分片content_block_stop内容块完成块结束时message_delta消息级更新携带stop_reason、usagemessage_stop消息完成仅流结束时一次cURL 参考 给出了这些事件的原始报文形态可与 SDK 事件一一对应event: message_start data: {type:message_start,message:{id:msg_...,type:message,...}} event: content_block_start data: {type:content_block_start,index:0,content_block:{type:text,text:}} event: content_block_delta data: {type:content_block_delta,index:0,delta:{type:text_delta,text:Hello}} event: content_block_stop data: {type:content_block_stop,index:0} event: message_delta data: {type:message_delta,delta:{stop_reason:end_turn},usage:{output_tokens:12}} event: message_stop data: {type:message_stop}理解这张事件表对排查流式问题很有帮助例如 token 用量出现在message_delta中、stop_reason也在message_delta中、而每个文本 token 都是content_block_delta里的text_delta。处理不同类型的内容块thinking、text 与 tool_useClaude 可能返回文本text、思考thinking或工具调用tool_use内容块。流式场景下需要在事件层面区分它们with client.messages.stream( modelclaude-opus-5, max_tokens64000, thinking{type: adaptive, display: summarized}, # display 为选填默认省略空 thinking 文本 messages[{role: user, content: Analyze this problem}] ) as stream: for event in stream: if event.type content_block_start: if event.content_block.type thinking: print(\n[Thinking...]) elif event.content_block.type text: print(\n[Response:]) elif event.type content_block_delta: if event.delta.type thinking_delta: print(event.delta.thinking, end, flushTrue) elif event.delta.type text_delta: print(event.delta.text, end, flushTrue)几个关键事实依据 README 思考与精力配置 与 模型迁移指南Fable 5 / Claude Opus 5 / Opus 4.8 / 4.7 / 4.6 等新模型请用thinking: {type: adaptive}让模型动态决定思考时机与深度旧模型才使用{type: enabled, budget_tokens: N}budget_tokens在新模型上已移除直接返回 400Claude Opus 5 上省略thinking参数即等价于 adaptive思考默认开启与 Opus 4.8/4.7省略即不思考行为不同display: summarized返回可读的思考摘要默认omitted会让thinking块以空文本流出——在流式 UI 上表现为输出前一段“长停顿”。如果要把推理过程展示给用户应显式设置display: summarizedthinking_delta/text_delta分别承载思考与正文的增量文本事件结构如代码所示。与工具调用结合流式 工具执行当请求携带tools时Claude 可能在流式输出末尾以tool_use停止。两种官方路径方式一messages.stream() 手动继续需要逐 token 流式 工具with client.messages.stream( modelclaude-opus-5, max_tokens64000, toolstools, messagesmessages ) as stream: for text in stream.text_stream: print(text, end, flushTrue) response stream.get_final_message() # 如果 response.stop_reason tool_use继续执行工具并回传结果执行完工具后把tool_result以 user 消息追加回messages再发起下一轮流式请求即可完成 agent 循环。完整的工具循环写法见 Python Tool Use 文档的手动循环示例其中还包含stop_reason pause_turn时的重发处理。方式二Tool Runner 原生流式推荐Python 工具运行器Tool Runner本身支持流式Tool Use 文档 展示了用beta_tool定义类型化工具后交给client.beta.messages.tool_runner(...)驱动完整 agent 循环的方式。流式文档指出给tool_runner(...)传streamTrue每次迭代产出一个可逐事件消费的流并用get_final_message()取到每轮累积的完整消息。两种方式的取舍在 Tool Use 概念文档 中有明确对比Tool Runner 由 SDK 提供循环harness你只需写工具函数仅当你需要 Runner 无法暴露的控制如自定义传输、非 beta 依赖、SDK Runner 不支持的单 token 流式时才用手动循环手动循环必须循环到stop_reason end_turn始终追加完整response.content保留tool_use块并保证每个tool_result携带匹配的tool_use_id。注意Tool Use 文档 提示Python Runner 目前不会自动续跑pause_turn遇到该场景需镜像历史后用新 Runner 重放暂停回合——若你的服务端工具回合可能长时间运行请阅读该节给出的重启模式。获取完整消息get_final_message()即使逐 token 打印了文本流结束后仍能拿到完整的Message对象用于读取用量、stop_reason、内容块等with client.messages.stream( modelclaude-opus-5, max_tokens64000, messages[{role: user, content: Hello}] ) as stream: for text in stream.text_stream: print(text, end, flushTrue) # 流结束后获取完整消息 final_message stream.get_final_message() print(f\n\nTokens used: {final_message.usage.output_tokens})final_message与client.messages.create(...)的返回值同构因此可以继续使用 README 中的响应辅助能力如final_message._request_id、final_message.to_json()。流式进度更新统计 token 与内容结合content_block_delta与message_delta可以边流式输出边统计 token 用量def stream_with_progress(client, **kwargs): 流式输出并附带进度统计。 total_tokens 0 content_parts [] with client.messages.stream(**kwargs) as stream: for event in stream: if event.type content_block_delta: if event.delta.type text_delta: text event.delta.text content_parts.append(text) print(text, end, flushTrue) elif event.type message_delta: if event.usage and event.usage.output_tokens is not None: total_tokens event.usage.output_tokens final_message stream.get_final_message() print(f\n\n[Tokens used: {total_tokens}]) return .join(content_parts)设计要点文本增量来自content_block_delta的text_delta逐段累积到content_parts最终拼出完整文本token 用量只出现在message_delta事件的usage.output_tokens字段且最后一次message_delta的值为最终值与 事件类型表 完全对应final_message同样可以在流结束后再次核对usage作为统计校验。该模式非常适合聊天 UI、长任务进度条或 Fable 5.1 / Opus 5 这类单次请求可能运行数分钟的场景见 模型迁移指南关于长回合的建议。流式错误处理流式请求的错误处理与普通请求一致使用 SDK 的类型化异常try: with client.messages.stream( modelclaude-opus-5, max_tokens64000, messages[{role: user, content: Write a story}] ) as stream: for text in stream.text_stream: print(text, end, flushTrue) except anthropic.APIConnectionError: print(\nConnection lost. Please retry.) except anthropic.RateLimitError: print(\nRate limited. Please wait and retry.) except anthropic.APIStatusError as e: print(f\nAPI error: {e.status_code})结合 README 的错误处理章节完整的异常链建议按“最具体优先”排列anthropic.BadRequestError400 类请求错误anthropic.AuthenticationErrorAPI Key 无效anthropic.PermissionDeniedError权限不足anthropic.NotFoundError模型或端点无效anthropic.RateLimitError429可读retry-after头SDK 默认自动重试anthropic.APIStatusError其余状态码500应重试4xx 一般不重试anthropic.APIConnectionError网络层失败。注意 SKILL.md 的警示不要只捕获一个宽泛的APIStatusError否则会丢失“可重试429/5xx/网络与不可重试400/404”的区分同时SDK 已内置 429 与 5xx 的指数退避重试默认 2 次可用max_retries配置自定义重试逻辑只在需要超出 SDK 行为时才编写。最佳实践清单综合 流式文档的 Best Practices、README 与 SKILL.md 常见陷阱始终 flush 输出flushTrue让 token 立即显示避免整段憋到流结束处理部分响应流被中断时可能只拿到不完整内容业务上要容忍截断跟踪 token 用量message_delta事件携带usage用于计费与进度展示设置超时流式请求同样需要合理timeout默认 10 分钟可用with_options(timeout...)按请求覆盖见 README 客户端配置默认使用流式即便不需要逐 token 展示也用.get_final_message()取完整响应——这能获得超时保护而无需手工处理单个事件大max_tokens非流式会被拒绝SDK 会估算请求超过约 10 分钟则抛ValueError此时必须传streamTrue/ 用messages.stream()或显式覆盖timeout以解除保护max_tokens取值的分水岭非流式默认~16000保持响应在 HTTP 超时内流式默认~64000超时不再是问题给模型留空间128K 输出上限在当前模型上只有流式可用见 模型迁移指南结构化输出 流式如需流式返回 Pydantic 模型请用client.messages.stream(..., output_formatModel)并通过stream.get_final_message().parsed_output取结果messages.parse(streamTrue)参数在 SDK 1.x 已移除见 sdk-upgrade.md不要重新实现 SDK 能力文本拼接、最终消息累积、重试均由 SDK 提供避免自造轮子见 SKILL.md 陷阱清单。小结本指南完整覆盖了 Claude API Python 流式的全部核心路径messages.stream()的同步/异步用法、create(streamTrue)的低层事件迭代、thinking/text/tool_use 三种内容块的逐事件处理、工具调用与 Tool Runner 的流式结合、get_final_message()的完整消息获取、进度统计、类型化错误处理以及九条生产级最佳实践。配合 流式文档 本身和 Python README 使用即可在聊天 UI、长文档生成、Agent 工具循环等场景中稳定、低延迟地消费 Claude 的输出。赞分享人工智能AI 技能AI 评测【免费下载链接】skillsPublic repository for Agent Skills项目地址https://gitcode.com/GitHub_Trending/skills3/skills点击查看免费下载相关推荐RikkaHub 实战Claude API 流式输出Streaming完整指南RikkaHub 实战Claude API 流式输出Streaming完整指南 导读 本文以开源项目 RikkaHub一个支持多 LLM 提供商的 An人工智能大模型AI 应用移动开发交互助手openai-agents-python 流式Streaming编程全指南从 Runner.run_streamed() 到事件流处理实战openai agents python 流式Streaming编程全指南从 Runner.run_streamed 到事件流处理实战 流式Stream人工智能AI AgentAgent 框架多智能体工具调用MCP ClientsOpen-Assistant 项目全景解析从数据众包、模型训练到推理部署的开源助手技术栈Open Assistant 项目全景解析从数据众包、模型训练到推理部署的开源助手技术栈 Open Assistant下文简称 OA是一个以聊天对话为核心人工智能大模型强化学习微调数据标注后端前端上一篇Wand-Enhancer 免费解锁 Wand 专业版实测30 天亲测报告与完整使用教程下一篇forgecode 工具服务化迁移实战从直接基础设施依赖到纯业务逻辑 Service 架构创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考