75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
|
"""Response processing: cleaning, JSON parsing, and reply sending."""
|
||
|
|
|
||
|
|
import re
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
from typing import Optional, List, Tuple
|
||
|
|
|
||
|
|
logger = logging.getLogger('Beta')
|
||
|
|
|
||
|
|
|
||
|
|
def clean_response_text(raw_reply: str, json_str: Optional[str]) -> str:
|
||
|
|
"""Nettoie le texte de réponse LLM en supprimant les artefacts JSON et techniques."""
|
||
|
|
clean = raw_reply
|
||
|
|
|
||
|
|
# Retirer le bloc JSON identifié
|
||
|
|
if json_str:
|
||
|
|
clean = clean.replace(json_str, "").strip()
|
||
|
|
|
||
|
|
# Suppression des balises hallucinées (<|channel|>, thought, etc.)
|
||
|
|
clean = re.sub(r'<[^>]+?>', '', clean)
|
||
|
|
|
||
|
|
# Nettoyage des résidus de formatage
|
||
|
|
clean = clean.replace('```json', '').replace('```', '').strip()
|
||
|
|
|
||
|
|
# Suppression des clés orphelines si le modèle a foiré le JSON
|
||
|
|
orphan_pattern = (
|
||
|
|
r'[\u201c\u201d\u2018\u2019"\'](thought|action|response)'
|
||
|
|
r'[\u201c\u201d\u2018\u2019"\']\s*:\s*'
|
||
|
|
r'[\u201c\u201d\u2018\u2019"\'].*?'
|
||
|
|
r'[\u201c\u201d\u2018\u2019"\'],?'
|
||
|
|
)
|
||
|
|
clean = re.sub(orphan_pattern, '', clean, flags=re.DOTALL | re.IGNORECASE)
|
||
|
|
|
||
|
|
# Retrait ultime des accolades
|
||
|
|
if clean.strip().startswith('{') or clean.strip().endswith('}'):
|
||
|
|
clean = clean.strip('{}').strip()
|
||
|
|
|
||
|
|
return clean.strip()
|
||
|
|
|
||
|
|
|
||
|
|
def parse_json_actions(json_str: Optional[str]) -> Tuple[list, list, bool]:
|
||
|
|
"""Parse le JSON extrait et retourne (actions_list, responses_list, is_explicit_none)."""
|
||
|
|
if not json_str:
|
||
|
|
return [], [], False
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Normalisation JSON (guillemets typographiques)
|
||
|
|
json_str_norm = (
|
||
|
|
json_str
|
||
|
|
.replace('\u201c', '"')
|
||
|
|
.replace('\u201d', '"')
|
||
|
|
.replace('\u2018', "'")
|
||
|
|
.replace('\u2019', "'")
|
||
|
|
)
|
||
|
|
data = json.loads(json_str_norm)
|
||
|
|
actions_list = data if isinstance(data, list) else [data]
|
||
|
|
|
||
|
|
responses = []
|
||
|
|
is_explicit_none = False
|
||
|
|
|
||
|
|
for action_data in actions_list:
|
||
|
|
norm_data = {k.lower(): v for k, v in action_data.items()}
|
||
|
|
|
||
|
|
if norm_data.get('action', '').upper() == 'NONE' and not norm_data.get('response'):
|
||
|
|
is_explicit_none = True
|
||
|
|
|
||
|
|
if norm_data.get('response'):
|
||
|
|
resp_text = re.sub(r'<[^>]+?>', '', str(norm_data.get('response'))).strip()
|
||
|
|
if resp_text:
|
||
|
|
responses.append(resp_text)
|
||
|
|
|
||
|
|
return actions_list, responses, is_explicit_none
|
||
|
|
except Exception as e:
|
||
|
|
logger.debug(f"JSON parsing error: {e}")
|
||
|
|
return [], [], False
|