Class LenientJsonMessageFormatter

masfactory/core/message/json.py:9–68

Best-effort JSON formatter tolerant to fenced blocks and noisy text.

from .base import StatelessFormatter

class LenientJsonMessageFormatter(StatelessFormatter):
    """Best-effort JSON formatter tolerant to fenced blocks and noisy text."""

    def __init__(self):
        super().__init__()
        self._is_input_formatter = True
        self._is_output_formatter = True
        self._agent_introducer = """
        !IMPORTANT!:Your response must contain a single JSON object. Avoid extra text. If you must, wrap the JSON in a ```json code block.
        """

    def format(self, message: str) -> dict:  # type: ignore[override]
        """Parse one JSON object from possibly noisy model output."""
        # Remove <think>...</think> tags and their content.
        if isinstance(message, str):
            message = re.sub(r"<think>.*?</think>", "", message, flags=re.DOTALL).strip()

        if isinstance(message, dict):
            return message

        if not isinstance(message, str):
            message = str(message)

        candidates: list[str] = []

        # Prefer fenced JSON blocks when present.
        for m in re.finditer(r"```(?:json)?\s*(.*?)```", message, flags=re.DOTALL | re.IGNORECASE):
            inner = (m.group(1) or "").strip()
            if inner:
                candidates.append(inner)

        # Fallback: take the widest {...} span.
        start = message.find("{")
        end = message.rfind("}")
        if start != -1 and end != -1 and end > start:
            candidates.append(message[start : end + 1].strip())

        # Last resort: try the whole message.
        candidates.append(message.strip())

        last_error: Exception | None = None
        for cand in candidates:
            if not cand:
                continue
            try:
                parsed = json.loads(cand)
            except Exception as e:
                last_error = e
                try:
                    parsed = ast.literal_eval(cand)
                except Exception as e2:
                    last_error = e2
                    continue
            if isinstance(parsed, dict):
                return parsed

        raise ValueError(f"LenientJsonMessageFormatter: failed to parse JSON object: {last_error}")