🚀 LLM Stream Parser：让大模型流式输出解析更优雅

前言

在使用大语言模型（LLM）进行应用开发时，流式输出是提升用户体验的关键技术。但你是否遇到过这样的困扰：

• 需要解析模型输出的工具调用，但 XML 标签被分割到多个 chunk 中导致解析失败？
• 想要按步骤展示模型的工作流程，但缺乏一个优雅的解析方案？

今天给大家介绍一个专门解决这些问题的 Python 库 —— LLM Stream Parser。

什么是 LLM Stream Parser？

LLM Stream Parser 是一个用于实时解析大语言模型流式响应的 Python 库，它支持基于 XML 标签提取内容，能够智能处理标签和内容被分割到多个数据块的情况。

核心特性

✅ 智能标签解析 - 自动处理标签被分割的情况
✅ 实时流式输出 - 支持标签内内容的实时流式展示
✅ 异步流式处理 - 完美对接 LLM 的异步流式输出
✅ 灵活的标签配置 - 支持自定义多个标签和步骤名称
✅ 类型安全 - 基于 Pydantic 的数据模型，提供完整的类型提示

快速上手

安装

pip install llm-stream-parser

或使用 uv：

uv add llm-stream-parser

基础用法

假设你的 LLM 输出包含多个步骤标签，比如 <analysis>、<calculation>、<summary>，解析器会自动处理标签被分割的情况：

from llm_stream_parser import StreamParser

# 定义自定义标签
custom_tags = {
    "analysis": "分析",
    "calculation": "计算",
    "summary": "总结"
}

parser = StreamParser(tags=custom_tags)

# 模拟标签内容被切割成多个 chunk 的场景
chunks = [
    "<anal",           # 标签被分割
    "ysis>这是分析内容的第一部分",
    "，这是第二部分</a",
    "nalysis>",        # 标签闭合
    "<calcu",
    "lation>计算过程：1+1=",
    "2</calc",
    "ulation>",
    "<sum",
    "mary>总结内容在",
    "多个chunk中</summar",
    "y>"
]

# 逐块解析
messages = []
for chunk in chunks:
    messages.extend(parser.parse_chunk(chunk))

# 处理流结束后的剩余内容
final = parser.finalize()
if final:
    messages.append(final)

# 输出结果
for msg in messages:
    print(f"{msg.step_name}: {msg.content}")

输出结果：

分析: 这是分析内容的第一部分，这是第二部分
计算: 计算过程：1+1=2
总结: 总结内容在多个chunk中

进阶功能

实时流式输出

启用 enable_tags_streaming 后，标签内的内容会实时输出，而不是等待标签闭合：

from llm_stream_parser import StreamParser

parser = StreamParser(
    tags={"think": "思考中", "tools": "工具调用"},
    enable_tags_streaming=True  # 启用标签内内容流式输出
)

# 模拟 LLM 流式输出
chunks = [
    "让我思考",
    "一下...",
    "正在分析",
    "问题...",
    "需要调用工具：<tools>",
    "<get_weather>",
    "北京",
    "</get_weather>",
    "</tools>",
    "这是最终答案。"
]

for chunk in chunks:
    for msg in parser.parse_chunk(chunk):
        print(f"{msg.step_name}: {msg.content} [标签闭合: {msg.is_complete}]")

输出结果：

思考中: 让我思考 [标签闭合: False]
思考中: 一下... [标签闭合: False]
思考中: 正在分析 [标签闭合: False]
思考中: 让我思考一下...正在分析问题... [标签闭合: True]
回答: 需要调用工具： [标签闭合: True]
工具调用: <get_weather>北京 [标签闭合: False]
工具调用: <get_weather>北京</get_weather> [标签闭合: True]
回答: 这是最终答案。 [标签闭合: False]

异步流式处理

直接对接 LLM 的异步流式输出：

import asyncio
from llm_stream_parser import process_llm_stream

async def main():
    # 模拟 LLM 流式响应
    async def mock_stream():
        yield "让我分析一下..."
        yield "<analysis>这是分析内容</analysis>"
        yield "这是最终答案。"

    # 封装异步流式输出
    async for msg in process_llm_stream(
            mock_stream(),
            tags={"analysis": "分析"},
            enable_tags_streaming=True
    ):
        print(f"{msg.step_name}: {msg.content} [标签闭合: {msg.is_complete}]")

asyncio.run(main())

实际应用场景

1. 展示模型的多步骤执行过程

parser = StreamParser(tags={
    "analysis": "分析",
    "planning": "规划",
    "execution": "执行",
    "summary": "总结"
})

# LLM 输出包含多个标签，可以按步骤实时展示行为

2. 基于 XML 的工具调用解析

parser = StreamParser(tags={
    "tools": "工具调用",
})

# LLM 输出: "我需要查询天气。<tools>get_weather(city='北京')</tools>"
# 解析后可以分别处理工具调用和结果

数据模型

解析器返回的 StreamMessage 对象包含以下字段：

字段	类型	说明
step	int	步骤序号（按 step_name 分组自增）
step_name	str	步骤名称（如"思考"、“工具调用”、“回答”）
content	Any	消息内容
is_complete	bool	标签是否闭合（True 表示闭合，False 表示流式输出中）

总结

LLM Stream Parser 是一个轻量级但功能强大的库，它解决了 LLM 流式输出中标签解析的痛点。无论你是想要：

• 🎯 实时展示模型的思考过程
• 🔧 解析工具调用
• 📊 按步骤展示模型的工作流程

LLM Stream Parser 都能为你提供优雅的解决方案。

项目地址：https://github.com/AriesYB/llm-stream-parser
PyPI 地址：https://pypi.org/project/llm-stream-parser/

如果你在使用 LLM 开发应用时遇到了流式输出解析的问题，不妨试试 LLM Stream Parser！