解析 ‘Streaming JSON Parser’：如何在模型输出的同时，实时解析并展示部分生成的结构化内容？

各位同学，各位同仁，大家好！

今天，我们齐聚一堂，共同探讨一个在现代数据处理和用户体验领域日益关键的话题——“Streaming JSON Parser”，特别是在如何实时解析并展示部分生成的结构化内容这一特定场景下的应用。在大型语言模型（LLM）和实时API交互日益普及的今天，我们经常面临这样的挑战：一个庞大的JSON结构正在生成中，但我们希望在它完全生成之前，就能看到并操作其中已经完成的部分。这不仅仅是为了节约时间，更是为了提供流畅、响应迅速的用户体验。

传统的JSON解析方式，无论是DOM（Document Object Model）风格的一次性加载整个文档到内存，还是SAX（Simple API for XML，其JSON对应物通常是事件驱动解析）风格的事件流处理，都各有其局限性。DOM解析器在处理大型JSON或流式数据时会消耗大量内存，并引入显著的延迟，因为它必须等待整个文档接收完毕才能开始构建内存中的对象图。SAX风格的解析器虽然内存效率更高，通过回调函数处理遇到的每个“事件”（如开始对象、结束键、值等），但它通常只报告“完整”的事件。例如，它会在遇到一个完整的字符串值后才报告这个值。这对于我们“实时展示部分生成内容”的需求来说，仍然不够精细。

设想一下，当一个LLM正在生成一个包含复杂JSON结构的回应时，比如一个包含多个字段和嵌套对象的任务列表或报告。用户可能希望在第一个任务的标题刚刚生成出来时就能看到，而不是等到所有任务、所有字段都填充完毕，甚至整个JSON结构都闭合之后。这就要求我们的解析器能够理解“未完成”的JSON片段，并将其解释为有意义的中间状态。这正是我们今天讲座的核心：构建一个能够实时、渐进式地解析JSON流，并能够处理和展示其部分生成内容的解析器。

JSON基础：流的本质与挑战

在深入解析器设计之前，我们首先要回顾一下JSON（JavaScript Object Notation）的基础。JSON是一种轻量级的数据交换格式，易于人阅读和编写，也易于机器解析和生成。它的基本结构包括：

• 对象（Object）：无序的键值对集合，用花括号 {} 包裹。键是字符串，值可以是任意JSON类型。

  {"name": "Alice", "age": 30}

• 数组（Array）：有序的值集合，用方括号 [] 包裹。值可以是任意JSON类型。

  ["apple", "banana", "orange"]

• 值（Value）：可以是字符串（String）、数字（Number）、布尔值（true 或 false）、null、对象（Object）或数组（Array）。

JSON数据在网络传输时，是以字符流的形式逐个字符发送的。这意味着我们的解析器将接收到的是一系列字符，而不是一个完整的、预先确定的文档。

表 1: JSON基本数据类型及其表示

JSON类型	描述	示例
Object	键值对的无序集合	{"key": "value", "num": 123}
Array	值的有序集合	["item1", 2, true]
String	Unicode字符序列，双引号包裹	"Hello, World!", "escapedn"
Number	整数或浮点数	123, -4.5, 1.2e+3
Boolean	逻辑值	true, false
Null	空值	null

流式解析的挑战在于：

1. 不完整性（Incompleteness）：我们接收到的JSON数据可能在任何位置中断。例如，我们可能收到 {"name": "Al，这显然不是一个完整的JSON。
2. 上下文依赖（Context Dependency）：一个字符的含义取决于它在JSON结构中的位置。例如，: 在键之后表示键值分隔，但在其他地方可能只是一个普通字符。
3. 实时更新（Real-time Updates）：解析器不仅要识别完整结构，还要能够识别并报告那些正在形成中的结构，例如一个正在被逐字接收的字符串。

核心概念：状态机与事件驱动

要解决上述挑战，我们需要结合两个核心概念：状态机（State Machine）和事件驱动（Event-Driven）解析。

状态机

一个解析器可以被建模为一个状态机。它在任何时刻都处于一个特定的状态，这个状态告诉解析器它当前期望接收什么类型的JSON元素。例如，在接收到 { 之后，解析器会进入 EXPECT_KEY 状态；在接收到键之后，会进入 EXPECT_COLON 状态。当接收到一个新字符时，解析器会根据当前状态和接收到的字符进行转换，并可能执行相应的操作。

状态转换的例子：

• 当前状态：EXPECT_VALUE
• 接收字符："
• 新状态：IN_STRING (开始解析字符串)
• 操作：触发 on_start_string 事件

这种状态机模型对于处理不完整输入至关重要。即使输入流中断，解析器也知道它当前期望什么，从而可以在后续输入到达时无缝恢复。

事件驱动

事件驱动模型意味着解析器在识别出JSON结构的特定部分时，会触发一个事件，并通过回调函数通知外部组件。例如：

• on_start_object()：遇到 {
• on_end_object()：遇到 }
• on_key(key_name)：识别出完整的键名
• on_value(value)：识别出完整的字面值（字符串、数字、布尔、null）
• on_partial_string(current_string)：关键！当一个字符串正在形成时，周期性地报告其当前内容。
• on_partial_number(current_number_str)：类似地，报告正在形成的数字。

这些带有“partial”的事件是实现实时展示的核心。它们允许外部UI在接收到完整值之前，就能够显示一个“占位符”或正在更新的文本。

缓冲区与前瞻（Lookahead）

在流式解析中，我们通常需要一个内部缓冲区来累积接收到的字符块。解析器会从缓冲区中逐个字符地读取，并可能需要少量的前瞻能力来判断当前字符的上下文。例如，要区分 f (可能是 false 的开始) 和 fu (普通字符串的一部分)，可能需要查看后续字符。然而，对于JSON这样相对简单和上下文无关的文法，很多情况下单字符前瞻或基于状态的期望就足够了。

实时展示的架构考量

为了实现“实时解析并展示部分生成内容”，我们需要一个协同工作的系统：

表 2: 实时展示系统架构组件

组件	职责	示例技术
生产者 (Producer)	生成JSON数据流，通常通过网络API逐块发送。	LLM API (如OpenAI Streaming API), WebSockets
消费者 (Consumer) / Streaming JSON Parser	接收数据块，内部维护状态机，解析JSON，并触发事件。处理不完整数据和错误。	Python自定义解析器, asyncio
数据结构层 (Data Structure Layer)	根据解析器事件动态构建和更新一个表示当前JSON状态的数据结构。它可能包含占位符和部分值。	树形结构 (字典/列表的嵌套), MutableMapping
展示层 (Display Layer)	监听数据结构层的更新，并将其可视化给用户。	Web UI (React/Vue/Angular), Console UI, Desktop App

关键在于解析器能够以足够细的粒度触发事件，并且数据结构层能够高效地响应这些事件，更新其内部表示，进而通知展示层刷新。

设计一个流式JSON解析器：逐步实现

我们将用Python来演示如何构建这样一个解析器。

1. 核心结构：解析器类与状态管理

我们的解析器将是一个类，它维护当前的解析状态、一个用于跟踪对象/数组嵌套的栈，以及一个用于累积字符的缓冲区。

import collections
import json

# 定义解析器可能处于的状态
class ParserState:
    EXPECT_VALUE = 1        # 期待一个值 (在对象开始、数组开始、逗号之后、冒号之后)
    EXPECT_KEY = 2          # 期待一个键 (在对象开始、逗号之后)
    EXPECT_COLON = 3        # 期待冒号 (在键之后)
    EXPECT_COMMA_OR_END_OBJ = 4 # 期待逗号或对象结束符 (在值之后，对象内部)
    EXPECT_COMMA_OR_END_ARRAY = 5 # 期待逗号或数组结束符 (在值之后，数组内部)

    IN_STRING = 6           # 正在解析字符串
    IN_NUMBER = 7           # 正在解析数字
    IN_TRUE = 8             # 正在解析 'true'
    IN_FALSE = 9            # 正在解析 'false'
    IN_NULL = 10            # 正在解析 'null'

    # 特殊状态，表示已经完成一个顶级JSON元素，等待下一个 (如果支持多个顶层JSON)
    # 对于单个JSON文档，一旦完成，通常进入FINISHED
    FINISHED = 11           # 解析完成

# 用于跟踪对象或数组的上下文
class ParsingContext:
    OBJECT = 1
    ARRAY = 2

    def __init__(self, type_):
        self.type = type_
        self.current_key = None # 仅用于OBJECT类型，存储当前正在解析的键

class StreamingJsonParser:
    def __init__(self):
        self._buffer = collections.deque() # 字符缓冲区
        self._state = ParserState.EXPECT_VALUE # 初始状态
        self._context_stack = [] # 存储ParsingContext，跟踪嵌套层级

        # 临时存储正在构建的字符串或数字
        self._current_string_buffer = [] 
        self._current_number_buffer = []
        self._escaped_char = False # 标志是否正在处理转义字符

        # 回调函数集合，外部可以注册监听器
        self.on_start_object = lambda path: None
        self.on_end_object = lambda path: None
        self.on_start_array = lambda path: None
        self.on_end_array = lambda path: None
        self.on_key = lambda path, key: None
        self.on_value = lambda path, value: None
        self.on_partial_string = lambda path, current_string: None
        self.on_partial_number = lambda path, current_number_str: None
        self.on_error = lambda error_msg: None

        # 内部状态，用于构建路径
        self._current_path = [] # 记录当前解析到的JSON路径，例如 ['data', 0, 'item']

    def _get_current_path(self):
        # 动态生成当前路径，用于回调
        path = []
        for ctx in self._context_stack:
            if ctx.type == ParsingContext.OBJECT:
                if ctx.current_key:
                    path.append(ctx.current_key)
            elif ctx.type == ParsingContext.ARRAY:
                # 对于数组，我们通常需要知道索引。这里简化为Placeholder，
                # 实际应用中需要更复杂的逻辑来跟踪已完成的元素数量
                path.append(f"[{len(path)}]") # 临时表示
        return tuple(path) # 返回元组作为不可变路径

    def feed_chunk(self, chunk: str):
        """
        接收新的数据块并加入缓冲区，然后尝试解析。
        """
        for char in chunk:
            self._buffer.append(char)

        while self._buffer:
            if not self._process_char(self._buffer[0]):
                # 如果当前字符无法处理 (例如，需要更多输入才能完成一个 token)，
                # 则停止循环，等待更多数据
                break
            else:
                self._buffer.popleft() # 成功处理，移除字符

    def _error(self, message):
        """报告解析错误"""
        self.on_error(f"Parsing error at path {self._get_current_path()}: {message}")
        self._state = ParserState.FINISHED # 错误后停止解析

    def _push_context(self, type_):
        self._context_stack.append(ParsingContext(type_))

    def _pop_context(self):
        if self._context_stack:
            return self._context_stack.pop()
        return None

    def _current_context(self):
        if self._context_stack:
            return self._context_stack[-1]
        return None

2. 字符处理与状态转换核心逻辑

_process_char 方法是解析器的核心，它根据当前状态和输入字符进行状态转换和操作。

    def _process_char(self, char):
        current_path = self._get_current_path()

        # 处理字符串转义
        if self._state == ParserState.IN_STRING:
            if self._escaped_char:
                # 处理转义字符，例如 " \ / b f n r t uXXXX
                # 简化处理：直接添加，实际需要解析 uXXXX
                self._current_string_buffer.append(char)
                self._escaped_char = False
                self.on_partial_string(current_path, "".join(self._current_string_buffer))
                return True
            elif char == '\':
                self._escaped_char = True
                return True
            elif char == '"':
                # 字符串结束
                value = "".join(self._current_string_buffer)
                self._current_string_buffer = []
                self._state = self._get_next_state_after_value()
                self._emit_value(value)
                return True
            else:
                self._current_string_buffer.append(char)
                self.on_partial_string(current_path, "".join(self._current_string_buffer))
                return True

        # 处理数字
        if self._state == ParserState.IN_NUMBER:
            if char.isdigit() or char in '.-+eE': # 允许数字、小数点、符号、指数
                self._current_number_buffer.append(char)
                self.on_partial_number(current_path, "".join(self._current_number_buffer))
                return True
            else:
                # 数字结束，回退一个字符，因为当前字符不属于数字
                # 这里我们不popleft，让外层循环再次处理这个非数字字符
                # 先尝试转换数字，如果无效则报错
                num_str = "".join(self._current_number_buffer)
                self._current_number_buffer = []
                try:
                    value = json.loads(num_str) # 使用json库的loads来解析数字，更健壮
                    self._state = self._get_next_state_after_value()
                    self._emit_value(value)
                    return False # 表示当前字符未被处理，让外层循环再次处理
                except json.JSONDecodeError:
                    self._error(f"Invalid number format: {num_str}")
                    return True # 错误已处理，可以继续或停止

        # 处理布尔和null
        if self._state == ParserState.IN_TRUE:
            return self._handle_keyword(char, 'true', True)
        elif self._state == ParserState.IN_FALSE:
            return self._handle_keyword(char, 'false', False)
        elif self._state == ParserState.IN_NULL:
            return self._handle_keyword(char, 'null', None)

        # 跳过空白字符
        if char.isspace():
            return True

        # 根据当前状态和字符进行转换
        if self._state == ParserState.EXPECT_VALUE:
            if char == '{':
                self.on_start_object(current_path)
                self._push_context(ParsingContext.OBJECT)
                self._state = ParserState.EXPECT_KEY
            elif char == '[':
                self.on_start_array(current_path)
                self._push_context(ParsingContext.ARRAY)
                self._state = ParserState.EXPECT_VALUE
            elif char == '"':
                self._current_string_buffer = []
                self._state = ParserState.IN_STRING
            elif char.isdigit() or char == '-': # 数字的开始
                self._current_number_buffer = [char]
                self._state = ParserState.IN_NUMBER
                self.on_partial_number(current_path, char)
            elif char == 't':
                self._current_string_buffer = [char]
                self._state = ParserState.IN_TRUE
            elif char == 'f':
                self._current_string_buffer = [char]
                self._state = ParserState.IN_FALSE
            elif char == 'n':
                self._current_string_buffer = [char]
                self._state = ParserState.IN_NULL
            else:
                self._error(f"Unexpected character '{char}' while expecting a value.")
            return True

        elif self._state == ParserState.EXPECT_KEY:
            if char == '"':
                self._current_string_buffer = []
                self._state = ParserState.IN_STRING # 键也是字符串
            elif char == '}':
                # 对象结束
                self.on_end_object(current_path)
                self._pop_context()
                self._state = self._get_next_state_after_value() # 对象本身也是一个值
            else:
                self._error(f"Unexpected character '{char}' while expecting a key or '}}'.")
            return True

        elif self._state == ParserState.EXPECT_COLON:
            if char == ':':
                self._state = ParserState.EXPECT_VALUE
            else:
                self._error(f"Unexpected character '{char}' while expecting ':'.")
            return True

        elif self._state == ParserState.EXPECT_COMMA_OR_END_OBJ:
            if char == ',':
                self._state = ParserState.EXPECT_KEY
            elif char == '}':
                self.on_end_object(current_path)
                self._pop_context()
                self._state = self._get_next_state_after_value()
            else:
                self._error(f"Unexpected character '{char}' while expecting ',' or '}}'.")
            return True

        elif self._state == ParserState.EXPECT_COMMA_OR_END_ARRAY:
            if char == ',':
                self._state = ParserState.EXPECT_VALUE
            elif char == ']':
                self.on_end_array(current_path)
                self._pop_context()
                self._state = self._get_next_state_after_value()
            else:
                self._error(f"Unexpected character '{char}' while expecting ',' or ']'.")
            return True

        elif self._state == ParserState.FINISHED:
            # 如果已经解析完成，忽略后续字符，或者根据需要报告错误
            return True 

        return False # 未知状态或需要更多字符来完成当前token

辅助方法：

    def _handle_keyword(self, char, keyword, value):
        """处理 'true', 'false', 'null' 关键字的解析"""
        current_path = self._get_current_path()
        self._current_string_buffer.append(char)
        partial_keyword = "".join(self._current_string_buffer)

        if len(partial_keyword) > len(keyword):
            self._error(f"Invalid keyword: {partial_keyword}")
            return True

        # 检查是否匹配到目前为止的关键字
        if partial_keyword != keyword[:len(partial_keyword)]:
            self._error(f"Unexpected character '{char}' while parsing '{keyword}'.")
            return True

        # 如果关键字完整
        if partial_keyword == keyword:
            self._current_string_buffer = []
            self._state = self._get_next_state_after_value()
            self._emit_value(value)
            return True

        # 关键字未完成，但目前匹配
        # 对于关键字，我们不需要on_partial_string事件，因为它们是固定值
        # 如果需要，可以在这里添加一个 on_partial_keyword 事件
        return True

    def _get_next_state_after_value(self):
        """根据当前上下文，确定解析完一个值之后的下一个状态"""
        if not self._context_stack:
            return ParserState.FINISHED # 顶级值解析完成

        current_ctx = self._current_context()
        if current_ctx.type == ParsingContext.OBJECT:
            return ParserState.EXPECT_COMMA_OR_END_OBJ
        elif current_ctx.type == ParsingContext.ARRAY:
            return ParserState.EXPECT_COMMA_OR_END_ARRAY
        return ParserState.FINISHED # 不应该发生

    def _emit_value(self, value):
        """根据上下文发出on_key或on_value事件"""
        current_path = self._get_current_path()
        if self._context_stack and self._current_context().type == ParsingContext.OBJECT:
            if self._current_context().current_key is None:
                # 这是一个键，而不是值
                self._current_context().current_key = value
                self.on_key(current_path, value)
                self._state = ParserState.EXPECT_COLON
            else:
                # 这是一个值，与之前的键对应
                key = self._current_context().current_key
                # 实际路径应该是父路径 + key
                path_for_value = tuple(list(current_path[:-1]) + [key]) if current_path else (key,)
                self.on_value(path_for_value, value)
                self._current_context().current_key = None # 重置，等待下一个键
        else:
            # 数组中的值或顶层值
            self.on_value(current_path, value)

3. 实时显示层：构建和更新中间数据结构

为了实时展示，我们需要一个数据结构来表示我们已经解析出的JSON内容。这个数据结构需要能够处理“部分”值。一个简单的实现是使用字典和列表的嵌套，并在遇到部分值时存储一个特殊的“占位符”对象。

import collections

class PartialValue:
    """表示一个正在被填充的、不完整的值"""
    def __init__(self, initial_content=""):
        self.content = initial_content

    def update(self, new_content):
        self.content = new_content

    def __repr__(self):
        return f"Partial('{self.content}')"

    def __str__(self):
        return self.content

class RealtimeJsonViewer:
    def __init__(self):
        self.parsed_data = None # 存储构建中的JSON结构
        self.path_to_node_map = {} # 映射路径到实际节点，方便快速更新

    def _get_or_create_node(self, path, is_container=False):
        """
        根据路径获取或创建节点。
        path: 元组，例如 ('root', 'items', 0, 'name')
        is_container: 如果为True，表示要创建的是一个对象或数组
        """
        if not path: # 顶级节点
            if self.parsed_data is None:
                self.parsed_data = {} if is_container else PartialValue()
            return self.parsed_data

        current_node = self.parsed_data
        if current_node is None:
            # 如果顶级节点还未初始化，先初始化它
            self.parsed_data = {} # 假设顶级是对象或数组
            current_node = self.parsed_data

        for i, segment in enumerate(path):
            if isinstance(current_node, PartialValue):
                # 如果父节点是PartialValue，说明之前的值是部分值，现在要更新为容器
                # 这是一种简化，实际中可能需要更精细的错误处理或类型推断
                parent_path = path[:i]
                if not parent_path: # 顶级节点是PartialValue
                    self.parsed_data = {} if is_container else []
                    current_node = self.parsed_data
                else:
                    # 需要找到父节点，并替换其子节点
                    # 这是一个复杂的情况，可能需要更强大的path_to_node_map
                    # 暂时简化：假设PartialValue不会直接作为容器的父节点
                    raise TypeError(f"Cannot add child to PartialValue at path {parent_path}")

            if isinstance(segment, int) or (isinstance(segment, str) and segment.startswith('[') and segment.endswith(']')):
                # 数组索引
                try:
                    idx = int(segment.strip('[]')) if isinstance(segment, str) else segment
                except ValueError:
                    idx = -1 # 无法解析的索引，按列表末尾处理

                while len(current_node) <= idx:
                    current_node.append(None) # 填充None直到索引有效

                if i == len(path) - 1: # 最后一个段，直接返回或创建
                    if current_node[idx] is None:
                        current_node[idx] = {} if is_container else [] if is_container else PartialValue()
                    return current_node[idx]
                else:
                    if current_node[idx] is None:
                        current_node[idx] = {} if isinstance(path[i+1], str) and not path[i+1].startswith('[') else []
                    current_node = current_node[idx]
            else:
                # 对象键
                if i == len(path) - 1: # 最后一个段
                    if segment not in current_node:
                        current_node[segment] = {} if is_container else [] if is_container else PartialValue()
                    return current_node[segment]
                else:
                    if segment not in current_node or current_node[segment] is None:
                        # 确保路径存在，如果下一个是键就创建对象，否则创建数组
                        current_node[segment] = {} if isinstance(path[i+1], str) and not path[i+1].startswith('[') else []
                    current_node = current_node[segment]
        return current_node # 理论上不会走到这里

    def _set_node_value(self, path, value):
        """根据路径设置节点的值"""
        if not path: # 顶级节点
            self.parsed_data = value
            return

        parent_path = path[:-1]
        last_segment = path[-1]

        parent_node = self._get_or_create_node(parent_path, is_container=True) # 确保父节点是容器

        if isinstance(last_segment, int) or (isinstance(last_segment, str) and last_segment.startswith('[')):
            try:
                idx = int(last_segment.strip('[]')) if isinstance(last_segment, str) else last_segment
                while len(parent_node) <= idx:
                    parent_node.append(None)
                parent_node[idx] = value
            except ValueError:
                print(f"Warning: Could not set array index {last_segment} at path {parent_path}")
        else:
            parent_node[last_segment] = value

        # 更新path_to_node_map，但对于简单值，直接存储值即可，不需要映射
        # 对于PartialValue，我们才需要映射以便更新

    def attach_parser(self, parser: StreamingJsonParser):
        """将Viewer的回调函数注册到解析器"""
        parser.on_start_object = self._on_start_object
        parser.on_end_object = self._on_end_object
        parser.on_start_array = self._on_start_array
        parser.on_end_array = self._on_end_array
        parser.on_key = self._on_key
        parser.on_value = self._on_value
        parser.on_partial_string = self._on_partial_string
        parser.on_partial_number = self._on_partial_number
        parser.on_error = self._on_error

    def _on_start_object(self, path):
        # 确保路径上的节点是对象
        self._set_node_value(path, {})
        # self._get_or_create_node(path, is_container=True) # 确保父节点是对象

    def _on_end_object(self, path):
        # 对象结束，可能需要做一些清理或最终化处理
        pass

    def _on_start_array(self, path):
        # 确保路径上的节点是数组
        self._set_node_value(path, [])
        # self._get_or_create_node(path, is_container=True)

    def _on_end_array(self, path):
        # 数组结束
        pass

    def _on_key(self, path, key):
        # 键已经解析完成，此时path是父对象的path，key是当前键名
        # 此时不需要直接更新parsed_data，因为键只是一个标识符，它对应的值才是内容
        # 实际的UI显示中，键名可以立即显示
        pass

    def _on_value(self, path, value):
        # 完整的值已解析
        self._set_node_value(path, value)

    def _on_partial_string(self, path, current_string):
        # 字符串部分更新
        node = self._get_or_create_node(path)
        if isinstance(node, PartialValue):
            node.update(current_string)
        else:
            # 如果之前是完整值，现在又来部分值，说明逻辑有误或需要替换
            self._set_node_value(path, PartialValue(current_string))

    def _on_partial_number(self, path, current_number_str):
        # 数字部分更新
        node = self._get_or_create_node(path)
        if isinstance(node, PartialValue):
            node.update(current_number_str)
        else:
            self._set_node_value(path, PartialValue(current_number_str))

    def _on_error(self, error_msg):
        print(f"Viewer received error: {error_msg}")

    def render(self):
        """
        将当前解析到的结构渲染为可读的字符串。
        实际UI中会更复杂，这里仅为控制台演示。
        """
        if self.parsed_data is None:
            return "No data yet..."

        # 递归地将PartialValue转换为其内容字符串
        def _to_display_friendly(node):
            if isinstance(node, PartialValue):
                return f"'{node.content}...' (partial)"
            elif isinstance(node, dict):
                return {k: _to_display_friendly(v) for k, v in node.items()}
            elif isinstance(node, list):
                return [_to_display_friendly(item) for item in node]
            return node

        return json.dumps(_to_display_friendly(self.parsed_data), indent=2, ensure_ascii=False)

4. 驱动示例：模拟LLM输出

现在，我们来模拟一个LLM逐字输出JSON的场景，并观察我们的解析器和查看器如何实时工作。

import time
import sys

# 模拟一个LLM逐字符生成JSON输出
def simulate_llm_stream(json_string_data, chunk_size=1, delay=0.01):
    for i in range(0, len(json_string_data), chunk_size):
        yield json_string_data[i : i + chunk_size]
        time.sleep(delay)

if __name__ == "__main__":
    # 复杂的LLM输出示例
    llm_output_json = """
    {
      "task_id": "gen-001",
      "status": "in_progress",
      "summary": "Generating a detailed report on streaming JSON parsers for real-time display.",
      "steps": [
        {
          "step_number": 1,
          "description": "Initialize parser and viewer components.",
          "completed": true,
          "output": "Parser and viewer set up."
        },
        {
          "step_number": 2,
          "description": "Simulate data stream from LLM.",
          "completed": false,
          "progress": 50,
          "details": {
            "current_chunk": "{"key": "va",
            "expected_next": "lue"}"
          }
        },
        {
          "step_number": 3,
          "description": "Process incoming JSON chunks and update display.",
          "completed": false,
          "progress": 0,
          "estimated_time_remaining_seconds": 300
        }
      ],
      "final_report_link": null
    }
    """

    parser = StreamingJsonParser()
    viewer = RealtimeJsonViewer()
    viewer.attach_parser(parser)

    print("--- Starting Streaming JSON Parsing and Real-time Display Simulation ---")
    print("-----------------------------------------------------------------------n")

    # 用于清空控制台行的辅助函数 (仅适用于简单控制台)
    def clear_previous_lines(num_lines):
        for _ in range(num_lines):
            sys.stdout.write("33[F") # Move cursor up one line
            sys.stdout.write("33[K") # Clear line from cursor to end

    previous_output_lines = 0

    for chunk in simulate_llm_stream(llm_output_json, chunk_size=2, delay=0.01):
        parser.feed_chunk(chunk)
        current_display = viewer.render()

        # 在控制台实时更新显示
        current_output_lines = current_display.count('n') + 1
        clear_previous_lines(previous_output_lines)
        print(current_display)
        previous_output_lines = current_output_lines

        time.sleep(0.05) # 稍作延迟，以便观察更新

    print("n--- Simulation Complete ---")
    print(f"Final Parsed Data:n{json.dumps(viewer.parsed_data, indent=2, ensure_ascii=False)}")

通过运行这个示例，您将看到控制台上的JSON输出是如何逐行、逐字段地实时构建和更新的。Partial('...') 标记了那些还在填充中的字符串或数字。当它们完成后，这些标记就会消失，显示出完整的值。

高级议题与考量

1. 性能优化

对于极高吞吐量的场景，Python的纯文本处理可能不够高效。可以考虑：

• C扩展：使用Cython或直接编写C扩展来处理核心的字符读取和状态转换逻辑。
• 更快的缓冲区：collections.deque 已经很高效，但更底层的字节操作可能更快。
• 预编译正则：在某些识别模式（如数字）上，使用预编译正则表达式可以加速。

2. 错误恢复与容错

当前的解析器在遇到不符合JSON规范的字符时会直接报错并停止。在生产环境中，可能需要更健壮的错误恢复策略：

• 跳过错误区域：尝试跳过直到找到下一个可能的同步点（如 {, [, } ]）。
• 报告警告而非停止：记录错误，但尝试继续解析剩余部分。
• 部分有效性：即使JSON结构不完整或有错误，也能识别并报告其中有效的子结构。

3. 异步集成

在现代Web服务或GUI应用中，解析器通常需要与异步I/O（如网络请求）相结合。Python的 asyncio 框架是理想选择。解析器的 feed_chunk 方法可以设计为协程，或者在一个单独的线程中运行，并通过消息队列与主事件循环通信。

# 异步集成示例 (概念性代码)
import asyncio

class AsyncStreamingJsonParser(StreamingJsonParser):
    def __init__(self):
        super().__init__()
        self._input_queue = asyncio.Queue()

    async def _process_chunks_loop(self):
        while True:
            chunk = await self._input_queue.get()
            if chunk is None: # 信号：停止
                break
            super().feed_chunk(chunk) # 调用同步的feed_chunk
            # 可以在这里加入 asyncio.sleep(0) 来让出控制权，避免阻塞事件循环

    async def feed_chunk_async(self, chunk: str):
        await self._input_queue.put(chunk)

    async def start(self):
        self._processing_task = asyncio.create_task(self._process_chunks_loop())

    async def stop(self):
        await self._input_queue.put(None) # 发送停止信号
        await self._processing_task

# 使用示例
async def main():
    async_parser = AsyncStreamingJsonParser()
    viewer = RealtimeJsonViewer()
    viewer.attach_parser(async_parser)
    await async_parser.start()

    # 模拟异步数据源
    async def data_source():
        llm_output_json = "{ "data": "hello world" }"
        for i in range(len(llm_output_json)):
            await async_parser.feed_chunk_async(llm_output_json[i])
            await asyncio.sleep(0.01)
        await async_parser.stop()

    # 模拟UI更新循环
    async def ui_update():
        while True:
            # print(viewer.render()) # 实际UI会在这里更新
            await asyncio.sleep(0.1)
            if async_parser._state == ParserState.FINISHED:
                break

    await asyncio.gather(data_source(), ui_update())

# asyncio.run(main())

4. 与现有库的比较

市面上有一些优秀的流式JSON解析库，如Python的 ijson 和 json_stream。它们专注于高效地处理大型JSON文件，避免一次性加载到内存，并提供SAX风格的事件或迭代器接口。然而，它们通常只在识别出“完整”的JSON元素时才报告事件。对于我们“实时展示部分内容”的需求，这些库可能需要额外的封装和逻辑来实现 on_partial_string 或 on_partial_number 这样的事件，因为它们的API设计初衷不是为了实时字符级更新。我们今天构建的解析器，其核心差异就在于对这种“不完整但可展示”状态的显式支持。

5. 安全性考量

在处理来自不可信源的JSON流时，需要考虑潜在的安全问题：

• 深度嵌套：恶意JSON可能包含极其深的嵌套结构，导致栈溢出。解析器应有深度限制。
• 超长字符串/数字：过长的字符串或数字可能耗尽内存。缓冲区应有大小限制。
• 缓慢或恶意数据流：攻击者可能发送极慢或格式错误的数据，以消耗服务器资源。需要有超时和错误处理机制。

实际应用场景

这种实时、渐进式JSON解析能力在多个领域具有重要价值：

1. 大型语言模型（LLM）的实时响应：这是我们讲座的最初动机。当LLM生成复杂的结构化输出（如代码、JSON数据、Markdown表格）时，用户可以立即看到和交互已生成的部分，极大地提升用户体验。
2. 实时仪表盘与监控：从实时数据流（如传感器数据、日志、API事件）中解析JSON，并立即更新UI，显示正在发生的变化。
3. 云服务API的渐进式加载：一些云服务API在处理耗时请求时，会以流式方式返回部分结果。客户端可以使用这种解析器来渐进式地展示结果。
4. 开发工具：在调试或开发过程中，实时显示正在接收和解析的JSON数据流，有助于问题排查。

结语

今天，我们深入探讨了“Streaming JSON Parser”在实时展示部分生成结构化内容这一特定场景下的设计与实现。我们了解了JSON流的挑战，通过状态机和事件驱动模型，构建了一个能够识别并报告部分值的解析器。结合一个动态更新的数据结构层，我们成功地模拟了LLM输出的实时展示。

这项技术不仅仅是理论上的创新，它在提升用户体验、优化资源利用以及应对现代异步数据流挑战方面，都具有显著的实践意义。希望今天的讲座能为您在构建下一代实时应用时提供有益的思路和工具。

谢谢大家！