File size: 3,035 Bytes
a613e88 | 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 | import json
import re
from typing import Any
def safe_invoke(llm, prompt: str) -> str:
"""
Safely invoke any LangChain LLM and always return a string.
Handles:
- AIMessage (ChatOpenAI, etc.)
- raw string outputs
- dict / list responses
"""
if not prompt or not isinstance(prompt, str):
raise ValueError("Prompt must be a non-empty string")
try:
response = llm.invoke(prompt)
# Case 1: Chat models (AIMessage)
if hasattr(response, "content"):
return str(response.content)
# Case 2: Already string
if isinstance(response, str):
return response
# Case 3: dict / list → stringify safely
return json.dumps(response, indent=2)
except Exception as e:
return f"[LLM_ERROR] {str(e)}"
# ---------------------------------------------------
def safe_json_parse(text: str) -> dict:
"""
Try to parse LLM output into JSON safely.
If fails, return fallback structure.
"""
if not isinstance(text, str):
return {
"error": "Invalid JSON from model",
"raw_output": text,
}
candidates = []
stripped = text.strip()
candidates.append(stripped)
fence_match = re.search(r"```(?:json)?\s*(.*?)\s*```", stripped, re.DOTALL | re.IGNORECASE)
if fence_match:
candidates.append(fence_match.group(1).strip())
first_object = stripped.find("{")
last_object = stripped.rfind("}")
if first_object != -1 and last_object != -1 and last_object > first_object:
candidates.append(stripped[first_object:last_object + 1].strip())
first_array = stripped.find("[")
last_array = stripped.rfind("]")
if first_array != -1 and last_array != -1 and last_array > first_array:
candidates.append(stripped[first_array:last_array + 1].strip())
seen = set()
for candidate in candidates:
if not candidate or candidate in seen:
continue
seen.add(candidate)
try:
return json.loads(candidate)
except Exception:
continue
return {
"error": "Invalid JSON from model",
"raw_output": text
}
# ---------------------------------------------------
def safe_merge(*args: Any) -> str:
"""
Safely merge multiple inputs into one string.
Handles:
- None
- dict
- list
- string
"""
merged_parts = []
for arg in args:
if arg is None:
continue
if isinstance(arg, (dict, list)):
merged_parts.append(json.dumps(arg, indent=2))
else:
merged_parts.append(str(arg))
return "\n".join(merged_parts)
# ---------------------------------------------------
def safe_input(text: Any, fallback: str) -> str:
"""
Normalize optional user inputs (chat / feelings).
"""
if text is None:
return fallback
if isinstance(text, str) and text.strip() == "":
return fallback
return str(text)
|