Files
resolutionflow/backend/app/services/llm_utils.py
chihlasm 10cf5f45eb refactor: consolidate LLM JSON parsing into shared llm_utils module
Extracted duplicate _strip_markdown_fences / _parse_llm_json functions
from 7 files into app/services/llm_utils.py. Two shared functions:
- strip_markdown_fences(): fence stripping only
- parse_llm_json(): fence stripping + JSON parse + error logging

Files updated: flowpilot_engine, knowledge_flywheel, session_to_flow_service,
ai_tree_generator_service, ai_fix_service, ai_chat_service, kb_conversion_service

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-21 03:25:25 +00:00

40 lines
1.2 KiB
Python

"""Shared utilities for parsing LLM responses."""
import json
import logging
from typing import Any
logger = logging.getLogger(__name__)
def strip_markdown_fences(text: str) -> str:
"""Strip markdown code fences from LLM output, returning raw content.
Use this when you need just the stripping without JSON parsing
(e.g., when the caller has its own error handling for json.loads).
"""
text = text.strip()
if text.startswith("```"):
lines = text.split("\n")
lines = [line for line in lines if not line.strip().startswith("```")]
text = "\n".join(lines).strip()
return text
def parse_llm_json(raw_text: str) -> dict[str, Any]:
"""Parse JSON from LLM response, handling common quirks.
Strips markdown code fences (```json ... ``` or ``` ... ```) if present,
then parses the remaining text as JSON.
Raises:
ValueError: If the text is not valid JSON after fence stripping.
"""
text = strip_markdown_fences(raw_text)
try:
return json.loads(text)
except json.JSONDecodeError as e:
logger.warning("LLM JSON parse failed: %s — raw: %.300s", e, text)
raise ValueError(f"Invalid JSON from LLM: {e}") from e