08. Streaming 与可观测性

学习目标

理解 stream 和 invoke 的体验差异。
学会用流式输出改善 CLI 或 UI 的等待体验。
知道调试 Agent 时应该记录哪些关键事件。
了解 LangSmith 的作用，以及为什么它是可选但有价值的工具。

Streaming

普通 invoke 会等待模型生成完整回复后一次性返回。Streaming 会在模型生成片段时不断返回 chunk，让用户更早看到输出。

运行示例：

bash

uv run python examples/08_streaming_cli.py

这个脚本使用 model.stream(question)，每拿到一个 chunk 就立即打印其中的 content。对 CLI、聊天界面和长答案生成来说，Streaming 能减少“程序卡住了”的感觉，也方便用户提前判断回答方向是否正确。

Streaming 改变的是输出方式，不代表模型更聪明，也不会自动解决工具调用、检索质量或上下文不足的问题。它主要改善交互体验。

调试 Agent 要看什么

Agent 出错时，单看最终答案通常不够。更可靠的调试信息包括：

输入：用户原始问题、系统提示、消息历史和传入模型的上下文。
工具调用：模型决定调用哪个 tool，参数是什么，调用顺序是什么。
工具输出：外部 API、数据库、检索器或本地函数返回了什么。
检索结果：RAG 命中了哪些文档片段，来源和相似度是否合理。
中间消息：模型是否理解了工具结果，是否跳过了必要步骤。
最终答案：答案是否基于上下文，是否暴露了不该暴露的信息。
错误和耗时：失败发生在哪一步，超时、网络错误、模型格式错误分别是什么。

这些日志应该帮助你复现问题，而不是只堆满控制台。生产环境里还要注意脱敏，不要把密钥、隐私数据或客户敏感文本直接写进日志。

LangSmith

LangSmith 是 LangChain 生态里的可观测性和评估平台。它可以记录一次链路运行中的输入、模型调用、tool call、tool output、retrieval、最终答案、耗时和错误，适合排查复杂 Agent 为什么做出某个决定。

本教程的本地脚本不强制使用 LangSmith。你可以先通过 print 和结构化日志理解数据流；当 Agent 变复杂、需要团队协作调试、需要保存 traces 或做评估集回归时，再接入 LangSmith 会更有价值。

代码阅读重点

阅读 examples/08_streaming_cli.py 时，重点看 model.stream(question)。

invoke() 会等完整回答生成后一次性返回。stream() 会逐步返回 chunk，CLI 可以边生成边打印。对用户体验来说，streaming 的价值不是让模型更快，而是让用户更早看到反馈。

这个示例没有引入 Agent 和 Tool，是为了把 streaming 机制单独讲清楚。后续如果要给 Agent 加 streaming，需要同时考虑模型输出、工具调用事件和中间状态展示。

应该记录哪些日志

最小可用日志应该包含：

request_id：一次请求的唯一标识。
user_input：用户原始输入。
rendered_messages：最终发给模型的消息。
tool_calls：模型请求的工具名称和参数。
tool_outputs：工具返回值和异常。
retrieved_docs：RAG 检索到的来源和 chunk。
final_answer：最终给用户的回答。
latency：模型调用、工具调用和总耗时。

日志不是为了堆数据，而是为了回答“为什么这次 Agent 这样做”。如果日志不能帮助定位问题，它就只是噪声。

实验练习

把 streaming 示例改成普通 invoke()，比较等待体验。
在 streaming 循环里统计 chunk 数量。
故意关闭 Ollama，确认错误提示不会输出 traceback。

自测问题

streaming 会提升模型推理质量吗？
为什么只记录最终回答不足以调试 Agent？
LangSmith 主要解决开发体验、评估和监控中的哪类问题？

深入理解：可观测性不是打印更多文本

很多初学者会把可观测性理解成“多加几个 print”。这在学习阶段有用，但在真实项目里不够。好的可观测性应该能回答三个问题：

这次请求经过了哪些步骤？
每一步输入、输出、耗时和错误是什么？
这次行为和过去的成功/失败样例有什么不同？

因此，日志最好是结构化的，而不是散落的自然语言。比如 tool call 日志应该包含工具名、参数、耗时、是否成功、错误类型，而不是只打印“调用工具失败”。

关键指标

Agent 项目至少应该关注这些指标：

成功率：请求是否得到可接受回答。
工具调用成功率：模型是否选择了正确工具，工具是否执行成功。
检索命中率：RAG 是否找到了期望来源。
平均延迟和 P95 延迟：用户等待多久。
重试次数：是否频繁因为格式或工具错误重试。
人工接管率：多少请求需要人类介入。
拒答率：模型是否在不该回答时拒答，在该回答时误拒答。

这些指标比“感觉模型变好了”更可靠。尤其是换模型、改 prompt、改 chunk 策略时，要看指标是否整体改善。

Trace 应该如何阅读

一次 Agent trace 可以按时间线阅读：

用户输入是什么？
system prompt 和历史消息是什么？
模型第一次决定了什么？
如果调用工具，参数是什么？
工具返回了什么？
模型如何使用工具结果？
最终回答是否忠于上下文？

当你能沿着这条链路解释一次失败，才算真正具备调试 Agent 的能力。

完整示例

python

from __future__ import annotations

import sys
from pathlib import Path

sys.path.append(str(Path(__file__).resolve().parents[1]))

from langchain_ollama import ChatOllama

from examples.common.config import load_config, ollama_client_kwargs
from examples.common.ollama_check import assert_model_available, assert_ollama_running


DEFAULT_QUESTION = "用三句话解释 LangChain Agent。"


def _friendly_error(exc: RuntimeError, model: str) -> str:
    message = str(exc)
    if message.startswith("Cannot reach Ollama."):
        return "无法连接 Ollama。请先打开 Ollama 应用，或运行 `ollama serve`，并检查 OLLAMA_BASE_URL。"
    if message.startswith("Cannot read Ollama model list"):
        return "无法读取 Ollama 模型列表。请检查 Ollama 是否正常运行，以及 OLLAMA_BASE_URL 是否正确。"
    if message.startswith("Ollama model list"):
        return "Ollama 返回的模型列表格式异常。请检查 OLLAMA_BASE_URL 是否指向 Ollama。"
    if message.startswith("Ollama model `"):
        return f"当前配置的 Ollama 模型未安装。请先运行 `ollama pull {model}`。"
    return message


def main() -> None:
    config = load_config()
    try:
        assert_ollama_running(config.ollama_base_url)
        assert_model_available(config.ollama_base_url, config.ollama_model)
    except RuntimeError as exc:
        print(f"错误：{_friendly_error(exc, config.ollama_model)}")
        return

    model = ChatOllama(
        model=config.ollama_model,
        base_url=config.ollama_base_url,
        client_kwargs=ollama_client_kwargs(),
        temperature=0,
    )

    question = input(f"问题（直接回车使用默认问题：{DEFAULT_QUESTION}）：").strip()
    if not question:
        question = DEFAULT_QUESTION

    print("\nAI：", end="", flush=True)
    try:
        for chunk in model.stream(question):
            content = getattr(chunk, "content", "")
            if content:
                print(content, end="", flush=True)
    except Exception as exc:
        print(f"\n错误：流式输出失败：{exc}")
        return

    print()


if __name__ == "__main__":
    main()

08. Streaming 与可观测性 ​

学习目标 ​

Streaming ​

调试 Agent 要看什么 ​

LangSmith ​

代码阅读重点 ​

应该记录哪些日志 ​

实验练习 ​

自测问题 ​

深入理解：可观测性不是打印更多文本 ​

关键指标 ​

Trace 应该如何阅读 ​

完整示例 ​