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>
40 lines
1.2 KiB
Python
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
|