流式JSON解析：解决LLM应用实时结构化数据输出的关键技术

1. 项目概述：为什么我们需要一个“不完整”的JSON解析器？

如果你最近在折腾大语言模型（LLM）应用，尤其是那些需要模型输出结构化数据的场景，比如让GPT帮你生成一个待办事项列表、一个产品规格表，或者一个API调用参数，那你大概率会跟JSON打交道。标准的做法是在提示词（Prompt）里要求模型“请以JSON格式回复”。理想很丰满，模型开始流式（Streaming）输出了，你满怀期待地准备解析第一个token，结果立马就卡住了——因为一个完整的JSON对象，必须等到最后一个闭合的}或]出现，json.loads才会认它。在这之前，你收到的所有数据都是一堆“无效”的字符串碎片。

这就像等一份快递，快递员告诉你“今天会到”，但你得等到晚上他敲你门的那一刻，才知道包裹里到底是什么。对于需要实时展示、逐词渲染的用户界面来说，这种体验无疑是灾难性的。partial-json-parser这个库，就是为了解决这个“快递拆封前”的等待问题而生的。它的核心能力是：在JSON字符串还处于“残缺”状态时，尽最大努力解析出其中已确定的部分，并允许你指定哪些类型的“残缺”是可以接受的。这样一来，前端界面可以在模型生成的过程中，就逐步渲染出已确定的结构和数据，给用户一种“实时构建”的流畅体验。

我自己在开发AI Agent和复杂提示工程应用时，就深受这个问题的困扰。早期只能粗暴地等待流式响应结束，或者自己写一堆正则和状态机去“猜”结构，既繁琐又容易出错。这个库的出现，相当于提供了一个标准化、高可靠性的“中间件”，把流式JSON解析这个脏活累活给包了。接下来，我会带你深入它的设计思路、实战用法以及那些官方文档里没写的“坑”和技巧。

2. 核心设计思路与方案选型解析

2.1 问题本质：流式传输与结构化数据的矛盾

要理解这个库的价值，得先看清问题的本质。LLM的流式输出是一个字符（或token）一个字符地“吐”出来的。而JSON作为一种严格的结构化数据格式，有其完整的语法规则：字符串必须用双引号闭合，对象必须有配对的{}，数组必须有配对的[]，键值对之间用逗号分隔。

当传输中途中断时，我们得到的可能是一个畸形的字符串：{"name": "Alice", "age": 2。在标准JSON解析器眼里，这就是个语法错误，直接抛异常。但对于应用来说，我们已经能确定一个键为"name"，值为"Alice"的字段，以及一个键为"age"，但值不完整的字段。我们是否有可能安全地提取出已确定的部分（{"name": "Alice"}），并对不完整部分（"age": 2）做出合理推断或标记？

partial-json-parser的设计哲学是：在语法允许的最大安全边界内，进行“最大努力”解析。它不是一个“猜测”引擎，而是一个“语法感知”的补全器。它的解析过程严格遵循JSON语法，只是在遇到未闭合的语法结构时，根据用户设定的规则（Allow标志），决定是将其视为“可接受的不完整”并尝试补全，还是直接忽略该不完整结构及其所属的父节点。

2.2 方案对比：正则、状态机与专用解析器

在遇到这个库之前，常见的解决方案有哪些？

1. 正则表达式（Regex）：尝试用正则去匹配已出现的模式，比如r'"(\w+)":\s*"([^"]*)'去抓取完整的键值对。这种方法极其脆弱，无法处理嵌套结构、转义字符、数字、布尔值等复杂情况，且正则本身难以维护，是典型的“临时解决方案”，不适用于生产环境。
2. 手动状态机：自己编写一个简单的解析器，跟踪当前处于对象、数组、字符串还是键/值状态。这比正则靠谱，但实现一个健壮的、能处理所有JSON边缘情况（如Unicode、各种数字格式、\uXXXX转义）的状态机，工作量巨大，且极易引入bug。
3. 等待完整后再解析：这是最省事但体验最差的方法。它放弃了流式传输的实时性优势。
4. 专用解析器（即本库）：基于成熟的JSON解析器（如Python内置的json模块）进行扩展，在其语法分析的基础上，增加对“未闭合状态”的处理逻辑。这是最稳健、最专业的方案。

partial-json-parser选择了第四条路。它没有重新发明轮子，而是“包装”了标准的JSON解析流程。其核心流程可以概括为：

• 词法/语法分析：首先，它需要理解输入字符串的语法结构，识别出令牌（tokens）如字符串、数字、布尔值、结构符号（{},[],:，,）。
• 残缺检测与补全：当分析到字符串末尾，发现语法结构未闭合时（例如字符串缺引号，对象缺}），根据Allow标志判断是否允许此类型的残缺。
• 安全补全：如果允许，则按照JSON语法规则，插入最小的必要字符（如缺失的"，}，]）来使其语法完整。
• 委托解析：将补全后的（或部分补全的）字符串，交给底层的标准JSON解析器（默认是json.loads）进行最终解析。

这种设计的优势在于，它继承了标准JSON解析器的所有正确性和健壮性，同时通过清晰的Allow标志，将“可以容忍何种不完整”的控制权完全交给了开发者。

2.3 Allow标志位的精妙设计：权限控制粒度

Allow枚举的设计是这个库的精华所在，它体现了对“安全性”和“灵活性”的精细权衡。它不是简单的一个“是否允许部分解析”的开关，而是一组按位或（|）组合的权限标志。

为什么需要这么细的粒度？考虑以下场景：

• 场景A（高安全需求）：你只希望解析出完全确定的键值对。对于{"name": "Alice", "age": 2，你只想要{"name": "Alice"}，因为"age": 2的值不完整，可能是25，也可能是2.5（浮点数），解析出来会误导用户。这时，你应该只设置Allow.OBJ，而不设置Allow.NUM。这样，解析器遇到不完整的数字2时，由于数字不被允许部分解析，它会认为整个"age": 2这个键值对是不安全的，从而在结果中丢弃它。
• 场景B（高实时性需求）：你希望尽可能快地展示任何信息。即使是{"name": "A，你也想先显示出键"name"和值"A"。这时，你可以设置Allow.OBJ | Allow.STR。解析器会补全字符串为"A"，并成功解析出对象。
• 场景C（特殊值处理）：LLM有时会输出Infinity，-Infinity，NaN（在JSON中通常不被直接支持，但某些解析器能处理）。如果流式输出中途停止在"score": Inf，你希望补全为"score": Infinity吗？这由Allow.INF或Allow.SPECIAL控制。

这种设计把“决策权”下放，让开发者根据具体业务场景的安全要求来配置，而不是库作者替你做决定。这是构建可靠Agent工具链中非常关键的一环。

注意：Allow.ALL是一个方便的默认值，但它意味着接受所有类型的不完整。在生产环境中，尤其是处理关键数据时，建议根据实际情况显式指定允许的类型，以规避因补全错误数据而导致的业务逻辑风险。

3. 核心API深度解析与实战要点

了解了设计思路，我们来深入看看怎么用它。安装很简单，pip install partial-json-parser就行。库本身零依赖，兼容Python 3.6+，类型提示做得也不错。

3.1 核心函数：loads与ensure_json

库提供了几个函数，最常用的是loads和ensure_json。

loads(json_string, allow_partial=Allow.ALL, parser=json.loads)这是主函数，行为类似json.loads，但多了一个allow_partial参数。它直接返回解析后的Python对象（字典、列表等）。

from partial_json_parser import loads, Allow # 示例1：允许对象和字符串不完整 result = loads('{"name": "Ali', Allow.OBJ | Allow.STR) print(result) # 输出：{'name': 'Ali'} # 解析器补全了字符串 `"Ali` -> `"Ali"`，并成功解析了对象。 # 示例2：只允许对象不完整，不允许字符串不完整 result = loads('{"name": "Ali', Allow.OBJ) # 注意，没有 Allow.STR print(result) # 输出：{} # 解析器发现 `"Ali` 是一个不完整的字符串，而字符串不被允许部分解析。 # 因此，它认为整个键值对 `"name": "Ali` 是不可靠的，选择丢弃这个不完整的成员。 # 由于对象内没有其他有效成员，最终返回空对象。 # 示例3：处理嵌套和数组 result = loads('[{"task": "write code", "status": "in_prog') print(result) # 输出：[{'task': 'write code', 'status': 'in_prog'}] # 默认 Allow.ALL，所以不完整的字符串 `"in_prog` 和对象 `{"task":...` 都被补全和解析了。

ensure_json(json_string, allow_partial=Allow.ALL)这个函数不进行解析，只负责“补全”。它返回一个语法上完整的JSON字符串。这在某些场景下非常有用，比如你需要将补全后的JSON字符串发送给另一个只接受完整JSON的API或前端。

from partial_json_parser import ensure_json, Allow partial_json = '{"id": 1, "tags": ["python", "web' complete_json = ensure_json(partial_json, Allow.OBJ | Allow.ARR | Allow.STR) print(complete_json) # 输出：'{"id": 1, "tags": ["python", "web"]}' # 注意，它补全了数组的闭合括号 `]` 和对象的闭合括号 `}`。

3.2 Allow标志位的组合策略与实战场景

如何选择Allow标志？下面是一个决策参考表：

实操心得：调试你的 Allow 策略一开始你可能会对解析结果感到困惑：“为什么我这个字符串没解析出来？” 这时候，交互式 playground 就派上用场了。通过pip install partial-json-parser[playground]安装后，在终端运行json-playground。你可以实时输入残缺的JSON，调整Allow标志，直观地看到补全后的字符串和解析结果。这是理解解析器行为最快的方式。

3.3 错误处理与“MalformedJSON”异常

库会抛出MalformedJSON异常。但要注意，这里的“Malformed”指的是在给定的allow_partial规则下，仍然无法进行有效补全或解析的语法错误。

from partial_json_parser import loads, MalformedJSON try: result = loads('{wrong}', Allow.OBJ) except MalformedJSON as e: print(f"解析失败: {e}") # 输出类似：解析失败: Malformed node or string on line 1 # 因为 `wrong` 不是一个有效的JSON值（不是字符串、数字、对象等），即使允许对象不完整，也无法处理。

这意味着，如果你设置了Allow.STR，那么"hello是合法的（可补全为"hello"）。但如果你没设置Allow.STR，或者输入是{hello（键名缺少引号），这属于JSON语法层面的根本错误，无论怎么设置Allow都会抛出异常。

注意：MalformedJSON异常是继承自ValueError的。在你的代码中，你可以选择捕获MalformedJSON来做特殊处理，或者更宽泛地捕获ValueError来涵盖所有JSON解析错误（包括标准json.loads可能抛出的错误）。

4. 在LLM流式应用中的集成实战

理论说再多，不如一行代码。我们来看如何将partial-json-parser集成到一个真实的LLM流式应用中。这里以使用OpenAI API和FastAPI构建一个简单的流式JSON端点为例。

4.1 场景构建：流式天气查询Agent

假设我们有一个Agent，用户问“上海天气如何？”，它需要调用外部天气API，并以结构化JSON流式返回。我们希望前端能实时看到如下构建过程：

{ "city": "上海", "temperature": 22, "condition": "晴", "humidity": 65 }

4.2 服务端实现（FastAPI + OpenAI）

# app.py import asyncio from typing import AsyncGenerator import json from fastapi import FastAPI, Response from fastapi.responses import StreamingResponse import openai from partial_json_parser import loads, Allow app = FastAPI() client = openai.AsyncOpenAI(api_key="your-api-key") # 请替换为你的API Key # 我们允许对象和字符串不完整。数字我们要求完整，因为温度、湿度是数值，部分解析（如"2"）可能错误。 ALLOW_PARTIAL = Allow.OBJ | Allow.STR async def simulate_llm_json_stream() -> AsyncGenerator[str, None]: """ 模拟一个LLM逐步生成JSON的过程。 在实际中，这里应该是调用OpenAI ChatCompletion并设置 stream=True。 """ # 模拟LLM的token流 chunks = [ '{', '"city": "', '上海', '", ', '"temperature": ', '22', ', ', '"condition": "', '晴', '", ', '"humidity": ', '65', '}' ] for chunk in chunks: yield chunk await asyncio.sleep(0.1) # 模拟网络延迟 @app.get("/stream-weather") async def stream_weather(): async def event_generator(): accumulated_text = "" async for chunk in simulate_llm_json_stream(): accumulated_text += chunk try: # 关键步骤：尝试解析累积的不完整JSON parsed_data = loads(accumulated_text, ALLOW_PARTIAL) # 将解析出的部分数据以SSE格式发送 # 我们发送整个当前解析出的对象，前端可以diff更新 yield f"data: {json.dumps(parsed_data, ensure_ascii=False)}\n\n" except Exception as e: # 如果解析失败（例如，在非常早期的阶段），可以忽略或发送空对象 # 在实际应用中，可能需要更精细的错误处理 pass # 流结束时，可以发送一个结束标记 yield "event: close\ndata: {}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # 对Nginx代理有用 } ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

4.3 前端实现（简单HTML + JavaScript）

<!DOCTYPE html> <html> <head> <title>流式JSON解析演示</title> </head> <body> <h2>实时天气信息</h2> <pre id="output">等待数据...</pre> <script> const eventSource = new EventSource('/stream-weather'); const outputEl = document.getElementById('output'); // 存储当前显示的数据 let currentData = {}; eventSource.onmessage = function(event) { if (event.data.trim() === '{}' && event.lastEventId === 'close') { eventSource.close(); console.log('流已结束'); return; } try { const newPartialData = JSON.parse(event.data); // 合并新解析出的部分数据到当前数据中 // 这是一个简单的深度合并，实际项目可能需要更复杂的策略 currentData = deepMerge(currentData, newPartialData); outputEl.textContent = JSON.stringify(currentData, null, 2); } catch (e) { console.error('解析SSE数据失败:', e); } }; eventSource.onerror = function(err) { console.error('EventSource错误:', err); eventSource.close(); }; // 简单的深度合并函数（仅用于演示） function deepMerge(target, source) { for (const key in source) { if (source[key] && typeof source[key] === 'object' && !Array.isArray(source[key])) { if (!target[key] || typeof target[key] !== 'object') { target[key] = {}; } deepMerge(target[key], source[key]); } else { target[key] = source[key]; } } return target; } </script> </body> </html>

这段代码是如何工作的？

1. 前端通过EventSource连接到/stream-weather端点。
2. 服务端模拟LLM，每次生成JSON的一个片段（chunk），并立即将其追加到累积字符串中。
3. 服务端使用partial_json_parser.loads尝试解析这个不完整的累积字符串。
4. 解析成功后，服务端将当前能确定的部分结果（一个字典）通过Server-Sent Events (SSE) 发送给前端。
5. 前端收到部分结果后，将其与之前收到的结果合并，并实时更新页面显示。

这样，用户就能看到JSON对象是如何从一个空对象{}，逐步变成{"city": "上海"}，再变成{"city": "上海", "temperature": 22}，直到最终完成的。体验非常流畅。

实操心得：性能与合并策略上面的例子中，每次收到chunk都尝试解析并全量发送当前结果。在实际高并发场景下，频繁的解析和序列化可能成为性能瓶颈。一个优化策略是“节流”（throttle），例如每收到3-5个chunk或每100毫秒才解析和发送一次。此外，前端的合并策略deepMerge在嵌套很深或数组结构下可能不够健壮，可以考虑使用像 Lodash 的_.merge或实现一个更智能的、能处理数组索引替换的合并器。

5. 高级用法、边界案例与排查技巧

5.1 使用自定义解析器

loads函数允许你传入一个自定义的parser，这为你提供了极大的灵活性。比如，你想使用更快的orjson库，或者你想在解析前后做一些自定义处理（如日志记录、数据清洗）。

import orjson from partial_json_parser import loads, Allow def my_parser(json_str: str): """使用orjson解析，并记录日志""" print(f"解析完整JSON: {json_str}") # orjson返回的是bytes，需要解码。它性能更好，但API与标准库略有不同。 return orjson.loads(json_str.encode('utf-8')) partial_str = '{"fast": true, "data": [1,2,' result = loads(partial_str, Allow.OBJ | Allow.ARR, parser=my_parser) print(result) # 输出：{'fast': True, 'data': [1, 2]} # 控制台会打印：解析完整JSON: {"fast": true, "data": [1,2,]} # 注意，补全器在数组末尾加了一个逗号，这是合法的JSON吗？在标准JSON中，尾随逗号是不允许的。 # 这里暴露了一个潜在问题：补全器生成的字符串，必须符合你自定义解析器的语法要求。

这里踩过一个坑：不是所有JSON解析器都遵循完全相同的语法。标准JSON（RFC 8259）不允许对象或数组末尾有逗号。但Python的json.loads允许吗？实际上，它不允许。然而，partial-json-parser的补全逻辑在某些情况下（为了简单）可能会产生尾随逗号。在上面的例子中，如果partial_str是'{"fast": true, "data": [1,2'，补全器为了闭合数组，会直接加上]，得到[1,2]，这是合法的。但如果像例子中已经有了逗号[1,2,，补全器可能只是闭合括号，得到[1,2,]，这对于标准json.loads是无效的，会引发JSONDecodeError。

重要提示：当你使用自定义解析器时，务必确保该解析器能处理partial-json-parser可能生成的所有中间格式。最安全的方法是始终使用库默认的json.loads，或者使用一个兼容性极强（如允许尾随逗号）的解析器，比如demjson3或simplejson（并设置特定参数）。在集成前，务必用边界案例测试。

5.2 处理边界案例与特殊值

LLM有时会输出一些非标准的“JSON”，比如JavaScript风格的Infinity，-Infinity，NaN。Python的json.loads默认无法解析它们。

from partial_json_parser import loads, Allow import math # 案例1：解析 Infinity (需要 Allow.INF 或 Allow.SPECIAL) result = loads('{"max_score": Inf', Allow.OBJ | Allow.INF) print(result) # 输出：{'max_score': inf} (Python的float('inf')) # 补全器将 `Inf` 补全为 `Infinity`，然后 `json.loads` 能解析吗？不能！ # 等等，这里有个关键点：`partial-json-parser` 的默认解析器能处理这些特殊值吗？ # 查看源码或测试发现，它似乎能处理。但为了保险，最好明确。 # 案例2：解析 NaN result = loads('{"ratio": Nan', Allow.OBJ | Allow.NAN) print(result) # 输出：{'ratio': nan} (Python的float('nan')) # 如果你需要将这些特殊值序列化为标准JSON字符串，后续可能需要处理。 final_json = json.dumps(result, allow_nan=False) # 这里会抛出 ValueError，因为inf/nan不是有效的JSON。 # 解决方案：在序列化前替换它们，或者使用允许NaN的序列化器（如simplejson）。

排查技巧：当遇到解析结果包含inf或nan时，你的下游系统可能无法处理。务必在业务逻辑中添加检查，或者在序列化时使用json.dumps(obj, allow_nan=False)来触发异常，以便进行数据清洗。

5.3 调试工具：fix函数

库提供了一个底层函数fix，它返回一个元组(consumed_slice, completion)。consumed_slice是输入字符串中被成功处理的部分，completion是为了使其完整而添加的后缀。这对于调试解析器的行为非常有帮助。

from partial_json_parser import fix, Allow input_str = '{"a": [1,2,' consumed, completion = fix(input_str, Allow.OBJ | Allow.ARR) print(f"输入: {input_str}") print(f"已处理部分: '{consumed}'") print(f"补全后缀: '{completion}'") print(f"完整JSON: '{consumed + completion}'") # 输出可能类似： # 输入: {"a": [1,2, # 已处理部分: '{"a": [1,2' # 补全后缀: ']}' # 完整JSON: '{"a": [1,2]}' # 注意，它没有保留尾随逗号，而是将其移除了。这避免了之前提到的尾随逗号问题。

通过fix函数，你可以清晰地看到解析器是如何“理解”你的输入，以及它决定如何补全的。当解析结果不符合预期时，这是第一手的调试信息。

6. 常见问题与排查技巧实录

在实际集成中，你肯定会遇到各种奇怪的问题。下面是我踩过的一些坑和解决方案。

最后一点个人体会：partial-json-parser是一个强大的工具，但它引入了一种“不确定性”。在追求实时性的同时，你必须想清楚：你的应用能容忍多大程度的“部分正确”？对于显示一个逐渐生成的句子，部分字符串是完全可以接受的。但对于一个关键的数字ID或状态码，等待其完整可能是更安全的选择。在设计Agent的输出Schema和前端展示逻辑时，将字段区分为“可流式部分”和“需完整后展示”两部分，并据此精心配置Allow标志，是构建健壮流式JSON应用的关键。这个库不是魔法，它是一把精准的手术刀，用好它需要对JSON和你的业务需求都有清晰的理解。