1. 为什么我们需要json-repair工具?
在处理大模型API返回数据时,开发者最头疼的问题之一就是JSON格式不规范。大模型(如GPT、Claude等)生成的JSON响应常常存在以下问题:
- 键名缺少引号(如
{name:"value"}而非{"name":"value"}) - 字符串中使用单引号而非双引号
- 尾部多余的逗号(如
{"a":1,}) - 注释残留(如
/* comment */) - 控制字符未转义
这些问题会导致标准JSON解析器(如Python内置的json模块)直接抛出JSONDecodeError。传统解决方案是使用正则表达式修补,但面对复杂嵌套结构时,正则往往力不从心。
2. json-repair的核心技术解析
2.1 智能语法分析算法
json-repair采用基于状态机的解析策略,其核心工作流程如下:
- 词法分析:将输入文本分解为tokens(标识符、字符串、数字等)
- 容错解析:允许非标准JSON语法(如无引号键名)
- AST构建:生成抽象语法树时自动补全缺失结构
- 规范生成:输出符合RFC 8259标准的JSON字符串
关键代码片段展示其容错能力:
# 处理无引号键名 def parse_unquoted_key(lexer): chars = [] while lexer.peek().isalnum() or lexer.peek() in ('_', '-'): chars.append(lexer.next()) return ''.join(chars)2.2 典型修复场景实测
| 测试用例 | 异常JSON | 修复结果 |
|---|---|---|
| 键名无引号 | {name:"Alice"} | {"name":"Alice"} |
| 单引号字符串 | {'name':'Bob'} | {"name":"Bob"} |
| 尾部逗号 | {"nums":[1,2,]} | {"nums":[1,2]} |
| JS注释 | {"a":1/*test*/} | {"a":1} |
| NaN/Infinity | {"val":NaN} | {"val":null} |
注意:根据JSON规范,NaN和Infinity会被转为null而非字符串
3. 完整集成方案
3.1 安装与基础使用
pip install json-repair处理大模型API响应的典型代码:
import json from json_repair import repair_json llm_response = "{name:'GPT-4', features:[few-shot, reasoning]}" try: data = json.loads(llm_response) # 这里会报错 except json.JSONDecodeError: repaired = repair_json(llm_response) data = json.loads(repaired) # 成功解析3.2 高级配置参数
通过JsonRepair类进行精细控制:
from json_repair import JsonRepair repairer = JsonRepair( allow_unquoted_keys=True, # 允许无引号键名 allow_single_quotes=True, # 允许单引号字符串 allow_trailing_commas=True, # 允许尾部逗号 allow_comments=False, # 是否保留注释 nan_to_null=True # NaN转null ) repaired = repairer.repair(broken_json)4. 性能优化与生产实践
4.1 流式处理大响应
对于大模型返回的超长JSON(>1MB),建议使用流式修复:
from json_repair import StreamingRepair stream = get_llm_response_stream() # 假设是生成器 repairer = StreamingRepair() for chunk in stream: repaired_chunk = repairer.process(chunk) # 处理片段... final_json = repairer.finalize() # 获取完整修复结果4.2 与流行框架集成
FastAPI中间件示例:
from fastapi import Request from json_repair import repair_json async def repair_middleware(request: Request, call_next): response = await call_next(request) if 'application/json' in response.headers.get('content-type',''): body = await response.body() try: json.loads(body) # 先尝试标准解析 except json.JSONDecodeError: repaired = repair_json(body.decode()) response.body = repaired.encode() return response5. 疑难问题排查指南
5.1 常见错误场景
二进制数据混淆:
- 现象:修复后出现乱码
- 方案:先确认是否为纯文本JSON,必要时做base64检测
模板字符串冲突:
- 现象:
${variable}被误认为语法错误 - 方案:设置
allow_template_literals=True
- 现象:
日期格式混淆:
- 现象:
2023-01-01被拆分为减法运算 - 方案:启用
date_strings=True选项
- 现象:
5.2 性能监控指标
通过装饰器记录修复耗时:
import time from functools import wraps def log_repair_time(func): @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = (time.perf_counter() - start) * 1000 print(f"JSON repair took {elapsed:.2f}ms | Length: {len(args[0])}") return result return wrapper # 使用示例 @log_repair_time def safe_loads(json_str): try: return json.loads(json_str) except json.JSONDecodeError: return json.loads(repair_json(json_str))6. 替代方案对比
| 工具 | 优点 | 缺点 |
|---|---|---|
json5 | 支持完整JSON5规范 | 无法处理严重损坏的结构 |
demjson | 历史悠久 | 已停止维护,Python3兼容性问题 |
yaml.safe_load | 支持更宽松语法 | 存在安全风险,性能较差 |
json-repair | 专注修复能力 | 不支持JSON5全部特性 |
对于大模型应用,推荐组合方案:
def robust_json_parse(text): try: return json.loads(text) except json.JSONDecodeError: try: return json5.loads(text) # 先尝试JSON5 except: return json.loads(repair_json(text)) # 最后启用修复7. 实际项目经验分享
在大规模日志分析系统中的实践心得:
- 预处理过滤:先用简单正则过滤明显非JSON内容(如HTML错误页面)
- 分级修复:轻微错误用
json5,复杂错误再用json-repair - 结果验证:修复后必须验证数据完整性,特别是数值类型
- 性能热点:超过10MB的JSON建议先拆分处理
典型错误处理模式:
def process_llm_response(response): if not response.strip().startswith(('{', '[')): response = extract_json_from_text(response) # 自定义提取逻辑 try: data = json.loads(response) except json.JSONDecodeError as e: logger.warning(f"JSON decode error: {e}") repaired = repair_json(response) try: data = json.loads(repaired) except Exception as e: logger.error(f"Repair failed: {e}") data = {"error": str(e), "raw": response[:1000]} validate_data_structure(data) # 自定义校验 return data