feat: Add JSON parsers infrastructure (Phase 1)

- Create json_utils.py with parse_json_dict() and parse_json_list() functions
- Add comprehensive unit tests (49 tests, all passing)
- Add deprecation warnings to string_utils.py for pipe-delimited parsing
- Follows PRD Phase 1: Parser Infrastructure

This is the first step in replacing pipe-delimited LLM output format with
structured JSON format as per PRD_STANDARDIZE_LIST_FORMATS.md
This commit is contained in:
ppanchigar
2026-02-02 16:44:29 -06:00
parent 5d1024e9dc
commit f633922381
3 changed files with 633 additions and 0 deletions
+345
View File
@@ -0,0 +1,345 @@
"""
Unit tests for JSON utils module.
Tests the new JSON parsing functions that replace pipe-delimited output format.
"""
import pytest
from src.utils.json_utils import (
parse_json_dict,
parse_json_list,
parse_json_dict_or_list,
)
class TestParseJsonDict:
"""Tests for parse_json_dict function."""
def test_simple_dict(self):
"""Test parsing a simple JSON dictionary."""
output = '{"field": "value", "count": 42}'
result = parse_json_dict(output)
assert result == {"field": "value", "count": 42}
def test_dict_with_explanation_before(self):
"""Test parsing JSON dict with explanatory text before it."""
output = 'Here is my answer: {"field": "value", "count": 42}'
result = parse_json_dict(output)
assert result == {"field": "value", "count": 42}
def test_dict_with_explanation_after(self):
"""Test parsing JSON dict with explanatory text after it."""
output = '{"key": "value"} This is some explanation text.'
result = parse_json_dict(output)
assert result == {"key": "value"}
def test_dict_with_explanation_both_sides(self):
"""Test parsing JSON dict with text on both sides."""
output = 'Analysis: The answer is {"result": "success"} as shown above.'
result = parse_json_dict(output)
assert result == {"result": "success"}
def test_nested_dict(self):
"""Test parsing nested JSON dictionary."""
output = '{"outer": {"inner": "value"}, "list": [1, 2, 3]}'
result = parse_json_dict(output)
assert result == {"outer": {"inner": "value"}, "list": [1, 2, 3]}
def test_dict_with_special_characters(self):
"""Test parsing dict with special characters in strings."""
output = '{"field": "value with \\"quotes\\"", "other": "value"}'
result = parse_json_dict(output)
assert result == {"field": 'value with "quotes"', "other": "value"}
def test_dict_with_newlines(self):
"""Test parsing dict with newlines in the JSON."""
output = '''
{
"field": "value",
"count": 42
}
'''
result = parse_json_dict(output)
assert result == {"field": "value", "count": 42}
def test_multiple_dicts_returns_last(self):
"""Test that when multiple dicts present, the last one is returned."""
output = '{"first": 1} and then {"second": 2}'
result = parse_json_dict(output)
assert result == {"second": 2}
def test_dict_with_null_values(self):
"""Test parsing dict with null values."""
output = '{"field": null, "other": "value"}'
result = parse_json_dict(output)
assert result == {"field": None, "other": "value"}
def test_dict_with_boolean_values(self):
"""Test parsing dict with boolean values."""
output = '{"is_valid": true, "is_empty": false}'
result = parse_json_dict(output)
assert result == {"is_valid": True, "is_empty": False}
def test_dict_with_numeric_values(self):
"""Test parsing dict with various numeric types."""
output = '{"int": 42, "float": 3.14, "negative": -10}'
result = parse_json_dict(output)
assert result == {"int": 42, "float": 3.14, "negative": -10}
def test_empty_dict(self):
"""Test parsing an empty dictionary."""
output = "{}"
result = parse_json_dict(output)
assert result == {}
def test_dict_with_unicode(self):
"""Test parsing dict with unicode characters."""
output = '{"field": "café", "emoji": "😀"}'
result = parse_json_dict(output)
assert result == {"field": "café", "emoji": "😀"}
def test_no_dict_raises_error(self):
"""Test that ValueError is raised when no dict is found."""
output = "This is just text without any JSON"
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_only_list_raises_error(self):
"""Test that ValueError is raised when only a list is found."""
output = '["item1", "item2"]'
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_malformed_json_raises_error(self):
"""Test that ValueError is raised for malformed JSON."""
output = '{"field": "value"' # Missing closing brace
with pytest.raises(ValueError, match="No valid JSON dictionary found"):
parse_json_dict(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_dict(123)
def test_dict_with_list_values(self):
"""Test parsing dict with list values."""
output = '{"items": ["a", "b", "c"], "counts": [1, 2, 3]}'
result = parse_json_dict(output)
assert result == {"items": ["a", "b", "c"], "counts": [1, 2, 3]}
class TestParseJsonList:
"""Tests for parse_json_list function."""
def test_simple_list(self):
"""Test parsing a simple JSON list."""
output = '["value1", "value2", "value3"]'
result = parse_json_list(output)
assert result == ["value1", "value2", "value3"]
def test_list_with_explanation_before(self):
"""Test parsing JSON list with explanatory text before it."""
output = 'The values are: ["value1", "value2", "value3"]'
result = parse_json_list(output)
assert result == ["value1", "value2", "value3"]
def test_list_with_explanation_after(self):
"""Test parsing JSON list with explanatory text after it."""
output = '["item1", "item2"] as you can see above'
result = parse_json_list(output)
assert result == ["item1", "item2"]
def test_list_with_explanation_both_sides(self):
"""Test parsing JSON list with text on both sides."""
output = 'Results: ["YES"] based on analysis'
result = parse_json_list(output)
assert result == ["YES"]
def test_list_of_numbers(self):
"""Test parsing list of numbers."""
output = "[1, 2, 3, 4, 5]"
result = parse_json_list(output)
assert result == [1, 2, 3, 4, 5]
def test_list_of_mixed_types(self):
"""Test parsing list with mixed types."""
output = '["string", 42, true, null]'
result = parse_json_list(output)
assert result == ["string", 42, True, None]
def test_nested_list(self):
"""Test parsing nested JSON list."""
output = '[["a", "b"], ["c", "d"]]'
result = parse_json_list(output)
assert result == [["a", "b"], ["c", "d"]]
def test_list_of_dicts(self):
"""Test parsing list containing dictionaries."""
output = '[{"field": "value1"}, {"field": "value2"}]'
result = parse_json_list(output)
assert result == [{"field": "value1"}, {"field": "value2"}]
def test_list_with_newlines(self):
"""Test parsing list with newlines in the JSON."""
output = '''
[
"item1",
"item2",
"item3"
]
'''
result = parse_json_list(output)
assert result == ["item1", "item2", "item3"]
def test_multiple_lists_returns_last(self):
"""Test that when multiple lists present, the last one is returned."""
output = '["first"] and then ["second"]'
result = parse_json_list(output)
assert result == ["second"]
def test_empty_list(self):
"""Test parsing an empty list."""
output = "[]"
result = parse_json_list(output)
assert result == []
def test_single_item_list(self):
"""Test parsing list with single item."""
output = '["N/A"]'
result = parse_json_list(output)
assert result == ["N/A"]
def test_no_list_raises_error(self):
"""Test that ValueError is raised when no list is found."""
output = "This is just text without any JSON"
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_only_dict_raises_error(self):
"""Test that ValueError is raised when only a dict is found."""
output = '{"field": "value"}'
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_malformed_json_raises_error(self):
"""Test that ValueError is raised for malformed JSON."""
output = '["item1", "item2"' # Missing closing bracket
with pytest.raises(ValueError, match="No valid JSON list found"):
parse_json_list(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_list([1, 2, 3])
class TestParseJsonDictOrList:
"""Tests for parse_json_dict_or_list function."""
def test_returns_dict(self):
"""Test that it returns a dict when dict is present."""
output = '{"field": "value"}'
result = parse_json_dict_or_list(output)
assert result == {"field": "value"}
assert isinstance(result, dict)
def test_returns_list(self):
"""Test that it returns a list when list is present."""
output = '["value1", "value2"]'
result = parse_json_dict_or_list(output)
assert result == ["value1", "value2"]
assert isinstance(result, list)
def test_returns_last_when_both_present(self):
"""Test that it returns the last JSON object when both types present."""
output = '{"first": 1} and then ["second"]'
result = parse_json_dict_or_list(output)
assert result == ["second"]
assert isinstance(result, list)
def test_returns_last_dict_when_multiple_dicts(self):
"""Test that it returns the last dict when multiple dicts present."""
output = '{"first": 1} and {"second": 2}'
result = parse_json_dict_or_list(output)
assert result == {"second": 2}
def test_no_json_raises_error(self):
"""Test that ValueError is raised when no JSON is found."""
output = "No JSON here"
with pytest.raises(ValueError, match="No valid JSON object found"):
parse_json_dict_or_list(output)
def test_non_string_input_raises_error(self):
"""Test that TypeError is raised for non-string input."""
with pytest.raises(TypeError, match="Expected string input"):
parse_json_dict_or_list(None)
class TestEdgeCases:
"""Tests for edge cases and special scenarios."""
def test_json_with_curly_braces_in_strings(self):
"""Test parsing JSON containing braces in string values."""
output = '{"template": "Use {variable} here"}'
result = parse_json_dict(output)
assert result == {"template": "Use {variable} here"}
def test_json_with_brackets_in_strings(self):
"""Test parsing JSON containing brackets in string values."""
output = '["[optional]", "[required]"]'
result = parse_json_list(output)
assert result == ["[optional]", "[required]"]
def test_json_with_escaped_quotes(self):
"""Test parsing JSON with escaped quotes in strings."""
output = '{"quote": "He said \\"hello\\""}'
result = parse_json_dict(output)
assert result == {"quote": 'He said "hello"'}
def test_json_with_backslashes(self):
"""Test parsing JSON with backslashes."""
output = '{"path": "C:\\\\Users\\\\file.txt"}'
result = parse_json_dict(output)
assert result == {"path": "C:\\Users\\file.txt"}
def test_very_long_string_values(self):
"""Test parsing JSON with very long string values."""
long_text = "x" * 10000
output = f'{{"text": "{long_text}"}}'
result = parse_json_dict(output)
assert result == {"text": long_text}
def test_deeply_nested_structure(self):
"""Test parsing deeply nested JSON structure."""
output = '{"a": {"b": {"c": {"d": "value"}}}}'
result = parse_json_dict(output)
assert result == {"a": {"b": {"c": {"d": "value"}}}}
def test_json_in_markdown_code_block(self):
"""Test parsing JSON within markdown code blocks."""
output = '''
```json
{"field": "value"}
```
'''
result = parse_json_dict(output)
assert result == {"field": "value"}
def test_json_with_reasoning_text(self):
"""Test realistic LLM output with reasoning and JSON."""
output = """
Based on the analysis, I found the following information:
1. The exhibit clearly states the LOB
2. Multiple programs are mentioned
Here is my answer:
{"LOB": ["Medicare", "Duals"], "PROGRAM": ["MA", "MASNP"]}
"""
result = parse_json_dict(output)
assert result == {"LOB": ["Medicare", "Duals"], "PROGRAM": ["MA", "MASNP"]}
def test_json_with_explanation_in_chinese(self):
"""Test parsing with non-English explanatory text."""
output = '这是答案: {"field": "value"}'
result = parse_json_dict(output)
assert result == {"field": "value"}
+262
View File
@@ -0,0 +1,262 @@
"""
JSON parsing utilities for LLM responses.
This module provides robust parsers for extracting JSON dictionaries and lists
from raw LLM output, replacing the deprecated pipe-delimited format.
"""
import json
import logging
import re
from typing import Any
logger = logging.getLogger(__name__)
def parse_json_dict(raw_llm_output: str) -> dict[str, Any]:
"""
Extract and parse a JSON dictionary from raw LLM output.
This function searches for the last valid JSON dictionary (object) in the
raw output string and returns it as a Python dict. It handles common cases
where the LLM includes explanatory text before or after the JSON.
Args:
raw_llm_output: Raw string output from the LLM, potentially containing
explanatory text along with JSON dictionary.
Returns:
dict: The parsed JSON dictionary.
Raises:
ValueError: If no valid JSON dictionary is found in the output.
TypeError: If raw_llm_output is not a string.
Examples:
>>> output = 'Here is my answer: {"field": "value", "count": 42}'
>>> parse_json_dict(output)
{'field': 'value', 'count': 42}
>>> output = '{"key": "value"} This is some explanation text.'
>>> parse_json_dict(output)
{'key': 'value'}
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
found_dicts = []
i = 0
while i < len(raw_llm_output):
if raw_llm_output[i] == "{":
start = i
stack = ["{"]
j = i + 1
while j < len(raw_llm_output):
# Skip over string literals to avoid counting braces inside strings
if raw_llm_output[j] == '"':
j += 1
while j < len(raw_llm_output):
if raw_llm_output[j] == "\\" and j + 1 < len(raw_llm_output):
j += 2 # Skip escaped character
continue
if raw_llm_output[j] == '"':
break
j += 1
elif raw_llm_output[j] in "{[":
stack.append(raw_llm_output[j])
elif raw_llm_output[j] in "}]":
if not stack:
break
open_bracket = stack.pop()
if (open_bracket == "{" and raw_llm_output[j] != "}") or (
open_bracket == "[" and raw_llm_output[j] != "]"
):
break
if not stack:
candidate = raw_llm_output[start : j + 1]
try:
obj = json.loads(candidate)
# Only keep if it's a dictionary
if isinstance(obj, dict):
found_dicts.append(obj)
except json.JSONDecodeError:
pass
i = j # Move past this object
break
j += 1
i += 1
if found_dicts:
# Return the last valid dictionary found
return found_dicts[-1]
raise ValueError(
f"No valid JSON dictionary found in LLM output. "
f"Output preview: {raw_llm_output[:200]}..."
)
def parse_json_list(raw_llm_output: str) -> list[Any]:
"""
Extract and parse a JSON list (array) from raw LLM output.
This function searches for the last valid JSON list in the raw output
string and returns it as a Python list. It handles common cases where
the LLM includes explanatory text before or after the JSON.
Args:
raw_llm_output: Raw string output from the LLM, potentially containing
explanatory text along with JSON list.
Returns:
list: The parsed JSON list.
Raises:
ValueError: If no valid JSON list is found in the output.
TypeError: If raw_llm_output is not a string.
Examples:
>>> output = 'The values are: ["value1", "value2", "value3"]'
>>> parse_json_list(output)
['value1', 'value2', 'value3']
>>> output = '["item1", "item2"] as you can see above'
>>> parse_json_list(output)
['item1', 'item2']
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
found_lists = []
i = 0
while i < len(raw_llm_output):
if raw_llm_output[i] == "[":
start = i
stack = ["["]
j = i + 1
while j < len(raw_llm_output):
# Skip over string literals to avoid counting brackets inside strings
if raw_llm_output[j] == '"':
j += 1
while j < len(raw_llm_output):
if raw_llm_output[j] == "\\" and j + 1 < len(raw_llm_output):
j += 2 # Skip escaped character
continue
if raw_llm_output[j] == '"':
break
j += 1
elif raw_llm_output[j] in "{[":
stack.append(raw_llm_output[j])
elif raw_llm_output[j] in "}]":
if not stack:
break
open_bracket = stack.pop()
if (open_bracket == "{" and raw_llm_output[j] != "}") or (
open_bracket == "[" and raw_llm_output[j] != "]"
):
break
if not stack:
candidate = raw_llm_output[start : j + 1]
try:
obj = json.loads(candidate)
# Only keep if it's a list
if isinstance(obj, list):
found_lists.append(obj)
except json.JSONDecodeError:
pass
i = j # Move past this object
break
j += 1
i += 1
if found_lists:
# Return the last valid list found
return found_lists[-1]
raise ValueError(
f"No valid JSON list found in LLM output. "
f"Output preview: {raw_llm_output[:200]}..."
)
def parse_json_dict_or_list(raw_llm_output: str) -> dict[str, Any] | list[Any]:
"""
Extract and parse either a JSON dictionary or list from raw LLM output.
This is a convenience function that attempts to parse the output as either
a dictionary or a list, returning whichever is found last in the output.
Useful when the expected output format may vary.
Args:
raw_llm_output: Raw string output from the LLM.
Returns:
dict | list: The parsed JSON object (dictionary or list).
Raises:
ValueError: If no valid JSON dictionary or list is found in the output.
TypeError: If raw_llm_output is not a string.
"""
if not isinstance(raw_llm_output, str):
raise TypeError(
f"Expected string input, got {type(raw_llm_output).__name__}"
)
found_objects = []
i = 0
while i < len(raw_llm_output):
if raw_llm_output[i] in "{[":
start = i
stack = [raw_llm_output[i]]
j = i + 1
while j < len(raw_llm_output):
# Skip over string literals
if raw_llm_output[j] == '"':
j += 1
while j < len(raw_llm_output):
if raw_llm_output[j] == "\\" and j + 1 < len(raw_llm_output):
j += 2
continue
if raw_llm_output[j] == '"':
break
j += 1
elif raw_llm_output[j] in "{[":
stack.append(raw_llm_output[j])
elif raw_llm_output[j] in "}]":
if not stack:
break
open_bracket = stack.pop()
if (open_bracket == "{" and raw_llm_output[j] != "}") or (
open_bracket == "[" and raw_llm_output[j] != "]"
):
break
if not stack:
candidate = raw_llm_output[start : j + 1]
try:
obj = json.loads(candidate)
found_objects.append(obj)
except json.JSONDecodeError:
pass
i = j
break
j += 1
i += 1
if found_objects:
return found_objects[-1]
raise ValueError(
f"No valid JSON object found in LLM output. "
f"Output preview: {raw_llm_output[:200]}..."
)
+26
View File
@@ -35,6 +35,12 @@ def extract_text_from_delimiters(
Returns:
- str: The extracted answer or "N/A" if no match is found.
.. deprecated:: 2026.02
Using Delimiter.PIPE with this function is deprecated and will be removed
in a future version. Prompts should return JSON format instead of
pipe-delimited strings. Use `json_utils.parse_json_list()` or
`json_utils.parse_json_dict()` for parsing JSON responses.
"""
if not isinstance(raw_output, str):
raise TypeError(
@@ -50,6 +56,13 @@ def extract_text_from_delimiters(
# Define a pattern based on the delimiter type
if delimiter == Delimiter.PIPE:
warnings.warn(
"Delimiter.PIPE is deprecated for LLM output parsing. "
"Prompts should return JSON format instead. "
"Use json_utils.parse_json_list() or json_utils.parse_json_dict() instead.",
DeprecationWarning,
stacklevel=2,
)
pattern = PIPE_PATTERN
elif delimiter == Delimiter.BACKTICK:
pattern = BACKTICK_PATTERN
@@ -191,7 +204,20 @@ def universal_json_load(s: str):
Extract and parse all highest-level valid JSON objects from a string.
Returns the last valid top-level JSON object found (dict, list, etc).
Raises ValueError if no valid JSON objects are found. Does not clean up syntax.
.. deprecated:: 2026.02
This function is deprecated and will be removed in a future version.
Use `json_utils.parse_json_dict()` for dictionary outputs or
`json_utils.parse_json_list()` for list outputs instead.
The new parsers provide type-specific parsing with better error handling.
"""
warnings.warn(
"universal_json_load is deprecated and will be removed in a future version. "
"Use json_utils.parse_json_dict() or json_utils.parse_json_list() instead.",
DeprecationWarning,
stacklevel=2,
)
if not isinstance(s, str):
raise TypeError(f"Expected string input, got {type(s).__name__}")
found = []