OpenRath流式输出实战：实时接收LLM增量响应并统计token消耗

OpenRath是一个 PyTorch 风格的多智能体、多会话运行时框架，它的流式输出能力让你可以在 LLM 生成过程中实时接收增量响应（streaming delta），并在一次运行结束后直接统计 token 消耗（token usage）。本文面向新手，用一个官方示例讲清楚：如何开启流式输出、增量事件长什么样、token 账本在哪里读，以及如何用 token 预算护栏防止费用失控。

为什么流式输出对 LLM 应用很重要

传统的一次性complete()调用要等模型完整生成后才能拿到内容，长回答意味着长时间"干等"，用户体感差。流式输出则让模型边生成、边推送小块文本，UI 可以逐字渲染，延迟被大幅压缩。

OpenRath 的设计是Session 优先：会话是显式的 Python 对象，携带有序的 ChunkTable、沙箱位置和血缘元数据。流式内容到达时逐块转发给你，而最终的累积结果（包括cumulative_usagetoken 账本）会被原子地写回会话——你既能看到过程，也能拿到汇总。

核心概念：RathLLMStreamDelta 增量事件

所有流式片段都会归一化为统一的 RathLLMStreamDelta 数据对象，跨 OpenAI、Anthropic 等不同供应商保持同一形状。它包含几个互相独立的字段，任一子集都可能有值：

• content_delta：助手文本片段，逐块推送的主体内容；
• finish_reason：结束原因（stop/length/tool_calls等），只在最后一个片段上出现；
• usage：本次调用的 token 用量（prompt_tokens、completion_tokens、total_tokens），只在流末尾事件填充；
• tool_call_*系列：工具调用的名称与参数增量，多个并行工具调用用tool_call_index区分。

各供应商的原始事件会被翻译层映射成这种统一增量，比如 Anthropic 侧的映射逻辑见 anthropic/stream_deltas.py，OpenAI 兼容客户端见 openai/client.py。

最简上手：给 Agent 传一个 on_event 回调

开启流式输出的方式极其简单——在构建 flow.Agent 时传入on_event回调即可。下面是官方示例 example/07_streaming.py 的核心逻辑：

from rath import flow from rath.llm import RathLLMStreamDelta from rath.session import Session def on_event(delta: RathLLMStreamDelta) -> None: if delta.content_delta: # 有文本增量就实时打印 sys.stdout.write(delta.content_delta) sys.stdout.flush() if delta.finish_reason is not None: # 结束时打印 finish_reason sys.stdout.write(f"\n[finish_reason={delta.finish_reason}]\n") agent = flow.Agent("You are a concise assistant.", provider_from_env(), on_event=on_event) user = Session.from_user_message("Count from 1 to 5.").to("local") out = agent(user)

运行后即可看到数字逐字出现在终端，而不是整段一次性输出。如果你只想快速看到效果，项目还提供了便捷回调 stream_to_stdout()，直接on_event=stream_to_stdout()就能把流式内容打到标准输出。

💡 小贴士：on_event为None（默认）时循环以非流式方式运行；一旦提供回调，运行时会自动切换到满足StreamingChatClient协议的路径（见 session/loop.py），无需你手动管理 SSE 连接。

统计 token 消耗：一次读取 cumulative_usage

很多新手最关心的问题是：流式输出结束后，token 花在哪了？OpenRath 把每轮补全的 usage 累加到会话上，返回的Session直接暴露 cumulative_usage 属性：

if out.cumulative_usage is not None: print(f"total={out.cumulative_usage.total_tokens} " f"(prompt={out.cumulative_usage.prompt_tokens}, " f"completion={out.cumulative_usage.completion_tokens})")

这个账本跨多轮工具调用自动累加：每次 LLM 补全的 usage 通过 add_usage() 求和，工具轮次再多也不用你自己维护计数器。会话 merge 时 usage 也会合并（session.py），所以多分支工作流下的总消耗同样可追溯。

进阶：用预算护栏防止费用失控

生产环境里，一次失控的会话循环可能烧掉大量 token。Provider 提供两个护栏字段：

• budget_total_tokens：设置 token 预算上限，当累计用量首次超过该值时触发；
• on_budget_exceeded：超限回调。回调内抛出 BudgetExceededError 即可中止整个会话循环。

from rath.llm import Provider from rath.llm.budget import BudgetExceededError def guard(): raise BudgetExceededError("token budget exceeded, aborting loop") provider = Provider( api_key="sk-...", model="your-model", budget_total_tokens=4000, # 单次循环最多 4000 token on_budget_exceeded=guard, )

护栏是"每会话闩锁"：同一循环内只在首次越限时触发一次，不会反复回调。不设置回调时框架只打一条 warning，不会自动中断——把"超预算即停止"的决策权留给了你。

实战要点清单 📝

1. 一个回调搞定流式：on_event收到的是统一形状的RathLLMStreamDelta，供应商差异由 OpenRath 内部抹平；
2. finish_reason判断收尾：非None表示该 choice 已结束，可安全刷新 UI；
3. token 账本看会话：跑完读out.cumulative_usage，跨轮自动累加，无需手动记账；
4. 费用护栏要设：长会话或工具循环密集的场景，务必配置budget_total_tokens；
5. 多智能体场景同样适用：每个 Agent 可独立配置on_event，在 Workflow 中组合时流式能力随会话一起传递。

参考资料

• 流式示例：example/07_streaming.py
• 增量与用量类型：src/rath/llm/chat_response.py
• 会话循环与流式切换逻辑：src/rath/session/loop.py
• 预算护栏异常：src/rath/llm/budget.py
• 示例 README（含全部示例索引）：example/README.md

掌握on_event回调 +cumulative_usage读取 +budget_total_tokens护栏这三件套，你的 OpenRath 应用就同时拥有了"实时感"和"成本可控"——这正是流式输出在生产 LLM 应用中的核心价值。