流式输出与成本统计

本文来自《AI 应用开发课程》月份 1 课程文档，已整理为网站文章版本。

学习目标

学完本节后，你应当能够：

理解流式输出的价值与限制。
实现最小流式打印。
设计月份 1 的基础成本统计方式。
明白为什么日志、token 和成本记录是调试抓手。

前置知识

已能完成普通文本请求

1. 什么是流式输出

普通输出：

等模型全部生成完成后一次性返回

流式输出：

模型边生成边返回片段

对用户来说，流式的直接价值是：

首字时间更短
体验更接近实时对话

2. 最小流式心智模型

flowchart LR
    A["发送请求"] --> B["模型开始生成"]
    B --> C["分块返回 chunk"]
    C --> D["程序逐块消费"]
    D --> E["终端或 API 持续输出"]

3. 最小流式示例思路

不同 SDK 或兼容接口的流式细节会有差异，但课程中的抽象目标固定为：

client 层暴露 stream()
上层逐块消费
每个块先打印，再拼接为最终完整文本

示例接口形状：

from collections.abc import AsyncIterator


class LLMClient:
    async def stream(self, messages: list[dict]) -> AsyncIterator[str]:
        ...

4. 为什么流式逻辑要单独抽象

因为它和普通 generate() 的关注点不同：

普通调用关心最终结果
流式调用关心片段消费、连接关闭和异常中断

不要把两种逻辑强行写成一个难懂的大函数。

5. 成本统计为什么必须在月份 1 开始

因为后续你会更频繁调用模型。如果现在不养成记录习惯，后面很难回答这些问题：

哪个功能最耗 token？
哪次实验为什么花费变高？
哪个 prompt 更长、成本更高？

6. 月份 1 的最低成本记录方式

建议至少记录：

调用时间
模型名
输入消息条数
输出字符数
是否成功
错误信息

如果接口响应里能取到 usage，就一并记录输入 token 和输出 token。

最小记录示例：

logger.info(
    "llm_call_finished",
    extra={
        "model": model,
        "message_count": len(messages),
        "output_length": len(answer),
        "success": True,
    },
)

7. 实操任务

设计一个 stream() 方法签名
在终端中逐块打印模型输出
最后将片段拼接为完整文本
每次请求输出一条调用日志

8. 自测题

为什么流式输出不能简单等价于普通输出？
为什么课程要求从月份 1 开始记录成本或调用信息？
如果接口没有直接返回 token 用量，月份 1 至少应该记录什么？

9. 作业与验收

作业：

在 prompt_lab 中增加 stream_demo.py
至少记录一次成功请求和一次失败请求日志

验收标准：

流式输出可以逐块看到
完整文本最终可拼接
日志里能定位请求是否成功

10. 常见错误

流式数据不打印也不积累，最后什么都没拿到
没有 finally 或关闭逻辑，导致资源未释放
只在成功时记日志，失败时没有排查信息

11. 本章与前文关系

前面你已经学会：

普通文本输出
结构化输出

但一个真实 AI 应用还需要另外两件东西：

输出过程本身要对用户友好
调用过程要对开发者可观察

流式输出对应第一点，成本与调用记录对应第二点。

12. 本章在研发助手项目中的位置

研发助手项目如果只有“等几秒后一次性返回结果”，在体验上会比较粗糙。而如果完全不记录：

调了哪个模型
调了几次
哪次失败

那你后面调试和复盘会非常痛苦。

因此，本章是在补两条非常实际的能力线：

用户体验线
运维与调试线

13. 基于官方文档确认的流式返回事实

DeepSeek 官方对话补全文档中，明确展示了流式返回的 chunk 形式，返回对象类型为 chat.completion.chunk，并以 data: ... 方式逐条推送。

这意味着：

你不能把流式调用当成普通 JSON 一次性读取
你需要逐行消费或逐 chunk 消费
上层通常要一边接收一边拼接

来源：

DeepSeek API Docs: 对话补全

14. 错误示例 vs 正确示例

错误示例：把流式接口当普通接口解析

response = await client.post(...)
data = response.json()

这适用于非流式调用，不适用于真正的流式响应。

正确示例：逐块消费

async for line in response.aiter_lines():
    ...

你要逐步处理 chunk，而不是等待一个最终完整对象。

15. 完整文件级示例：`stream_demo.py`

"""流式输出与最小调用记录示例。"""

from __future__ import annotations

import logging
from collections.abc import AsyncIterator


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger(__name__)


async def consume_stream(stream: AsyncIterator[str]) -> str:
    """逐块消费流式结果，并拼接为完整文本。"""

    chunks: list[str] = []

    async for chunk in stream:
        print(chunk, end="", flush=True)
        chunks.append(chunk)

    print()
    return "".join(chunks)


def record_call_summary(
    model: str,
    message_count: int,
    output_text: str,
    success: bool,
) -> None:
    """记录月份 1 最小调用摘要。"""

    logger.info(
        "llm_call_summary | model=%s | message_count=%s | output_length=%s | success=%s",
        model,
        message_count,
        len(output_text),
        success,
    )

16. 逐段解释这份完整示例

为什么流式消费要同时“打印 + 拼接”

因为它服务于两个目标：

打印：让用户及时看到内容
拼接：让程序最终仍然拿到完整结果，方便后续继续处理

为什么最小记录先只保留 `output_length`

因为月份 1 当前阶段不要求你做专业 token 计费板。只要你先开始记录：

模型
调用次数
输出长度
成功/失败

就已经有了后续排查和比较的基本抓手。

17. 为什么“只记录成功调用”是危险的

因为开发中真正最有价值的信息往往在失败调用里。

如果你只记录成功情况，后面很难回答：

为什么某次流式输出中断了
为什么某次结构化输出返回为空
为什么某次工具调用前模型就失败了

月份 1 的最小原则是：失败调用也要留下最少上下文。

18. 一个更接近工程的增强版记录思路

你可以把成功和失败统一收束到一个记录函数里：

def record_call_result(model: str, stage: str, success: bool, detail: str) -> None:
    logger.info(
        "stage=%s | model=%s | success=%s | detail=%s",
        stage,
        model,
        success,
        detail,
    )

这在后面的：

普通调用
结构化输出
Tool Calling
FastAPI 服务

里都可以复用。

19. 调试与排错：本章最常见问题

问题一：终端没有逐块输出

可能原因：

没有真正迭代流
chunk 根本没被打印
缓冲区没有及时 flush

问题二：最终拿不到完整文本

通常是你只打印没积累，或者 chunk 清洗逻辑写错了。

问题三：日志里没有失败信息

说明你的记录逻辑只放在成功路径里，异常路径没有覆盖。

20. 本章完成后你应该具备的能力

完成本章后，你应当做到：

能解释流式输出和普通输出的程序消费差异。
能写出最小流式消费函数。
能为调用记录设计最小字段集。
能理解为什么日志和成本记录从月份 1 就必须出现。

21. 如果你卡在这里，先回看哪几章

异步概念不稳：回看 05-asyncio基础.md
模型调用链路不稳：回看 01-LLM基本概念与DeepSeek接入.md

22. 从本章过渡到下一章的桥接说明

接下来进入 05-Prompt实验手册.md。

你现在已经具备了请求、消息、结构化输出和流式消费的基础能力。下一步不再是补底层机制，而是学会用“实验”而不是“感觉”优化 Prompt。这会让你从“能调 API”进一步进入“能比较方案并得出结论”。