File size: 4,531 Bytes
d06fa57
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8956871
d06fa57
8956871
 
 
 
 
 
d06fa57
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
# Copyright (c) 2026-present, Yumeow. Licensed under the MIT License.
"""Template-guided JSON parsing for LLM outputs."""

from __future__ import annotations

import ast
import json
import re
from collections.abc import Mapping
from typing import Any

import yaml


def parse_json_with_template(text: str, template: Any) -> Any:
    """Extract a JSON payload from model text and coerce its structure to match ``template``.

    Args:
        text: Raw text output potentially containing fenced or inline JSON.
        template: Example Python structure (dict, list, or primitive) defining expected types and keys.

    Returns:
        Any: Parsed and type-coerced Python object matching ``template``.
    """
    errors = []
    for candidate in _candidate_texts(text):
        for loader in (_loads_json, _loads_python_literal, _loads_yaml):
            try:
                return _coerce_to_template(loader(candidate), template)
            except Exception as exc:
                errors.append(exc)
    raise ValueError(f"Could not parse response with template. Last errors: {errors[-3:]}")


def _candidate_texts(text: str) -> list[str]:
    text = text.strip()
    candidates = []

    fenced_blocks = re.findall(r"```(?:json|python)?\s*(.*?)\s*```", text, flags=re.S | re.I)
    candidates.extend(block.strip() for block in fenced_blocks if block.strip())

    candidates.extend(_balanced_snippets(text))
    if text:
        candidates.append(text)

    deduped = []
    for candidate in candidates:
        if candidate not in deduped:
            deduped.append(candidate)
    return deduped


def _balanced_snippets(text: str) -> list[str]:
    snippets = []
    stack = []
    start = None
    quote = None
    escaped = False

    pairs = {"{": "}", "[": "]"}
    opens = set(pairs)
    closes = set(pairs.values())

    for idx, char in enumerate(text):
        if quote:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == quote:
                quote = None
            continue

        if char in ("'", '"'):
            quote = char
            continue

        if char in opens:
            if not stack:
                start = idx
            stack.append(pairs[char])
        elif char in closes and stack:
            if char != stack[-1]:
                stack.clear()
                start = None
                continue
            stack.pop()
            if not stack and start is not None:
                snippets.append(text[start : idx + 1])
                start = None

    return snippets


def _loads_json(text: str) -> Any:
    return json.loads(text)


def _loads_python_literal(text: str) -> Any:
    return ast.literal_eval(text)


def _loads_yaml(text: str) -> Any:
    return yaml.safe_load(text)


def _coerce_to_template(value: Any, template: Any) -> Any:
    if template is Any:
        return value

    if isinstance(template, Mapping):
        if not isinstance(value, Mapping):
            raise TypeError(f"Expected mapping, got {type(value).__name__}")
        return {key: _coerce_to_template(value[key], child) for key, child in template.items()}

    if isinstance(template, list):
        if not isinstance(value, list):
            raise TypeError(f"Expected list, got {type(value).__name__}")
        if len(template) == 0:
            return value
        if len(template) != 1:
            if len(value) != len(template):
                raise ValueError(f"Expected list of length {len(template)}, got {len(value)}")
            return [_coerce_to_template(item, child) for item, child in zip(value, template)]
        return [_coerce_to_template(item, template[0]) for item in value]

    if template is float:
        return _coerce_float(value)
    if template is int:
        return int(value)
    if template is str:
        return str(value)
    if template is bool:
        return _coerce_bool(value)
    if isinstance(template, type):
        return template(value)

    return value


def _coerce_float(value: Any) -> float:
    if isinstance(value, str):
        value = value.strip()
        if value.endswith("%"):
            return float(value[:-1]) / 100.0
    return float(value)


def _coerce_bool(value: Any) -> bool:
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"true", "yes", "y", "1"}:
            return True
        if lowered in {"false", "no", "n", "0"}:
            return False
    return bool(value)