喜欢的条友记得点赞、关注、转发、收藏，你们的支持就是我最大的动力源泉。

安装 Ubuntu24.04 + LangChain + Ollama + Qwen2 看前期教程：

流式处理对于使基于 LLM 的应用程序能够响应最终用户至关重要。

LangChain（如聊天模型、输出解析器、提示、检索器和代理）实现了LangChain可运行接口。

此接口提供了两种常规的流式处理内容方法：

Sync stream 和 Async astream：流式处理的默认，用于从链流式传输最终输出。
async astream_events 和 async astream_log：它们提供了一种从链中流式传输中间步骤和最终输出的方法。

让我们看一下这两种方法，并尝试了解如何使用它们。

使用 Stream

所有 Runnable 对象都实现了一个名为 stream 的同步方法和一个名为 astream 的异步变体。

这些方法旨在以块的形式流式传输最终输出，并在每个块可用时立即生成每个块。

只有当程序中的所有步骤都知道如何处理 input 流时，才能进行流式处理;即，一次处理一个 input 块，并产生相应的输出块。

这种处理的复杂性可能会有所不同，从简单的任务（如发出 LLM 生成的令牌）到更具挑战性的任务（如在整个 JSON 完成之前流式传输 JSON 结果的一部分）。

开始探索流媒体的最佳位置是使用 LLM 应用程序中最重要的组件 - LLM 本身！

LLM 和聊天模型

大型语言模型及其聊天变体是基于 LLM 的应用程序的主要瓶颈。

大型语言模型可能需要几秒钟才能生成对查询的完整响应。这比应用程序感觉对最终用户的响应速度要慢得多 ~200-300 毫秒的阈值。

使应用程序感觉响应更灵敏的关键策略是显示中间进度;即，逐个令牌流式传输模型令牌的输出。

让我们从同步stream API 开始：

from langchain_community.llms.ollama import Ollama

model = Ollama(
    model="llama3.1"
)

chunks = []
for chunk in model.stream("天为什么是蓝的？"):
    chunks.append(chunk)
    print(chunk, end="", flush=True)

运行结果：

让我们检查其中一个 chunk 的前五个值：

print(chunks[0] + chunks[1] + chunks[2] + chunks[3] + chunks[4])

运行结果：

使用输入流

如果在生成输出时您想要流式传输 JSON 该怎么办？

如果您依靠json.loads来解析部分 JSON，由于该部分 JSON 不是有效的 JSON，解析将会失败。

您可能完全不知道该怎么做，并声称无法流式传输 JSON。

嗯，事实证明有一种方法可以做到这一点——解析器需要对输入流进行操作，并尝试将部分 JSON“自动完成”到有效状态。

让我们看看这样一个解析器的实际操作，以理解这意味着什么。

from langchain_community.llms.ollama import Ollama

model = Ollama(
    model="llama3.1"
)

from langchain_core.output_parsers import JsonOutputParser

chain = (model | JsonOutputParser())

for text in chain.stream(
    "以JSON格式输出法国、西班牙和日本及其人口的列表。"
    '使用外部关键字为“countries”的字典，其中包含国家列表。'
    "每个国家都应该有关键的‘名称’和‘人口’"
):
    print(text, flush=True)

运行结果：

现在，让我们停止流传输。我们将使用前面的示例，并在末尾添加一个提取函数，从最终的 JSON 中提取国家名称。

李成笔记网

专注域名、站长SEO知识分享与实战技巧

【LangChain + Python 开发语言模型 LLM 教程】六、详解流式传输结果

使用 Stream

LLM 和聊天模型

使用输入流