Ladebeze66/llm_ticket3
1
0
Fork 0
mirror of https://github.com/Ladebeze66/llm_ticket3.git synced 2025-12-13 14:06:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

2304-11:27translate

Browse Source
This commit is contained in:
Ladebeze66 2025-04-23 11:27:45 +02:00
parent 1f097ff5ff
commit 06b31b8663
11 changed files with 620 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

183
agents/llama_vision/agent_image_sorter.py
View File

@ -1,17 +1,22 @@
from ..base_agent import BaseAgent
import logging
import os
from typing import Dict, Any, Tuple
from PIL import Image
from typing import Dict, Any, Tuple
from datetime import datetime
from ..base_agent import BaseAgent
from ..utils.pipeline_logger import sauvegarder_donnees
from ..utils.ocr_utils import extraire_texte_fr
from ..utils.translate_utils import fr_to_en, en_to_fr, sauvegarder_ocr_traduction
logger = logging.getLogger("AgentImageSorter")
class AgentImageSorter(BaseAgent):
"""
Agent pour trier les images et identifier celles qui sont pertinentes.
Optimisé pour llama_vision avec prompt en anglais et réponse en français.
Agent de tri d’image optimisé pour llama_vision.
Réalise un OCR en français, le traduit en anglais, génère un prompt enrichi, et analyse l’image.
"""
def __init__(self, llm):
super().__init__("AgentImageSorter", llm)
@ -21,45 +26,10 @@ class AgentImageSorter(BaseAgent):
"max_tokens": 300
}
# Nouveau prompt system optimisé pour llama_vision
self.system_prompt = (
"""
You are an expert image classifier working for the technical support team of BRG_Lab (company: CBAO).
Your job is to determine whether an image is relevant for technical ticket analysis.
### Relevant images (respond with 'oui' or 'non'):
- Software screenshots or UI captures
- Error messages on screen
- System configuration panels
- Technical dashboards, logs, graphs
- Diagnostic windows
- Anything showing app behavior or user interface details
### Not relevant:
- Personal photos
- Marketing banners or illustrations
- Unrelated logos or branding
- Real-life objects, landscapes, people
### Task:
Look at the image provided.
Decide if it helps understand or resolve a technical issue described in a support ticket.
Then, answer ONLY in French:
- 'oui' if the image is relevant
- 'non' if the image is not relevant
- Add one short sentence explaining your decision (in French)
NEVER say “I cannot see the image”. Just reply 'ERREUR' if analysis fails.
"""
)
self._appliquer_config_locale()
logger.info("AgentImageSorter (llama_vision) initialisé")
def _appliquer_config_locale(self) -> None:
if hasattr(self.llm, "prompt_system"):
self.llm.prompt_system = self.system_prompt
if hasattr(self.llm, "configurer"):
self.llm.configurer(**self.params)
@ -74,67 +44,26 @@ class AgentImageSorter(BaseAgent):
logger.error(f"Vérification impossible pour {image_path}: {e}")
return False
def _generer_prompt_analyse(self, prefix: str = "") -> str:
return f"{prefix}\n\nEst-ce une image pertinente pour un ticket de support technique ? Réponds par 'oui' ou 'non' avec une brève justification en français."
def executer(self, image_path: str) -> Dict[str, Any]:
image_name = os.path.basename(image_path)
print(f" AgentImageSorter: Évaluation de {image_name}")
if not self._verifier_image(image_path):
return self._erreur("Erreur d'accès ou image invalide", image_path)
try:
prompt = self._generer_prompt_analyse()
if hasattr(self.llm, "interroger_avec_image"):
response = self.llm.interroger_avec_image(image_path, prompt)
elif hasattr(self.llm, "_encoder_image_base64"):
img_base64 = self.llm._encoder_image_base64(image_path)
prompt = self._generer_prompt_analyse(f"Analyse cette image:\n{img_base64}")
response = self.llm.interroger(prompt)
else:
return self._erreur("Le modèle ne supporte pas les images", image_path)
if any(err in response.lower() for err in [
"je ne peux pas", "je n'ai pas accès", "impossible d'analyser", "ERREUR".lower()]):
return self._erreur("Réponse du modèle invalide", image_path, raw=response)
is_relevant, reason = self._analyser_reponse(response)
print(f" Décision: Image {image_name} {'pertinente' if is_relevant else 'non pertinente'}")
result = {
"is_relevant": is_relevant,
"reason": reason,
"raw_response": response,
"metadata": {
"image_path": image_path,
"image_name": image_name,
"timestamp": self._get_timestamp(),
"model_info": {
"model": getattr(self.llm, "modele", str(type(self.llm))),
**getattr(self.llm, "params", {})
},
"source_agent": self.nom
}
}
sauvegarder_donnees(None, "tri_image", result, base_dir="reports", is_resultat=True)
self.ajouter_historique("tri_image", {"image_path": image_path, "prompt": prompt}, result)
return result
except Exception as e:
return self._erreur(str(e), image_path)
def _generer_prompt(self, ocr_fr: str, ocr_en: str) -> str:
return (
"The following image is from a technical support ticket at CBAO "
"for the BRG_Lab software system.\n\n"
f"OCR detected French text:\n[FR] {ocr_fr or '—'}\n[EN] {ocr_en or '—'}\n\n"
"Please analyze the image and determine:\n"
"- Is it relevant for a technical support issue?\n"
"- Answer only 'oui' or 'non', then briefly explain in French."
)
def _analyser_reponse(self, response: str) -> Tuple[bool, str]:
r = response.lower()
first_line = r.split('\n')[0] if '\n' in r else r[:50].strip()
if first_line.startswith("non") or "non pertinent" in first_line:
first_line = r.split('\n')[0] if '\n' in r else r.strip()[:50]
if first_line.startswith("non"):
return False, response.strip()
if first_line.startswith("oui") or "pertinent" in first_line:
if first_line.startswith("oui"):
return True, response.strip()
pos_keywords = ["pertinent", "utile", "important", "diagnostic"]
neg_keywords = ["inutile", "photo", "hors sujet", "marketing", "non pertinent"]
pos_keywords = ["pertinent", "utile", "interface", "message", "diagnostic"]
neg_keywords = ["inutile", "photo", "irrelevant", "hors sujet"]
score = sum(kw in r for kw in pos_keywords) - sum(kw in r for kw in neg_keywords)
return score > 0, response.strip()
@ -154,5 +83,69 @@ class AgentImageSorter(BaseAgent):
}
def _get_timestamp(self) -> str:
from datetime import datetime
return datetime.now().strftime("%Y%m%d_%H%M%S")
def _extraire_ticket_id_depuis_path(self, image_path: str) -> str:
parts = image_path.split(os.sep)
for part in parts:
if part.startswith("T") and part[1:].isdigit():
return part
return "unknown_ticket"
def executer(self, image_path: str) -> Dict[str, Any]:
image_name = os.path.basename(image_path)
print(f" AgentImageSorter: Évaluation de {image_name}")
if not self._verifier_image(image_path):
return self._erreur("Erreur d'accès ou image invalide", image_path)
try:
# Pré-traitement OCR et traduction
ticket_id = self._extraire_ticket_id_depuis_path(image_path)
ocr_fr = extraire_texte_fr(image_path)
ocr_en = fr_to_en(ocr_fr)
ocr_en_back_fr = en_to_fr(ocr_en) if ocr_en else ""
# Sauvegarde OCR + Traductions
sauvegarder_ocr_traduction(image_path, ticket_id, ocr_fr, ocr_en, ocr_en_back_fr)
# Prompt en anglais enrichi
prompt = self._generer_prompt(ocr_fr, ocr_en)
# Appel au LLM
if hasattr(self.llm, "interroger_avec_image"):
response = self.llm.interroger_avec_image(image_path, prompt)
else:
return self._erreur("Le modèle ne supporte pas les images", image_path)
# Analyse de la réponse
if "i cannot" in response.lower() or "unable to" in response.lower():
return self._erreur("Réponse du modèle invalide", image_path, raw=response)
is_relevant, reason = self._analyser_reponse(response)
print(f" Décision: Image {image_name} {'pertinente' if is_relevant else 'non pertinente'}")
result = {
"is_relevant": is_relevant,
"reason": reason,
"raw_response": response,
"ocr_fr": ocr_fr,
"ocr_en": ocr_en,
"metadata": {
"image_path": image_path,
"image_name": image_name,
"timestamp": self._get_timestamp(),
"model_info": {
"model": getattr(self.llm, "modele", str(type(self.llm))),
**getattr(self.llm, "params", {})
},
"source_agent": self.nom
}
}
sauvegarder_donnees(ticket_id, "tri_image", result, base_dir="reports", is_resultat=True)
self.ajouter_historique("tri_image", {"image_path": image_path, "prompt": prompt}, result)
return result
except Exception as e:
return self._erreur(str(e), image_path)

65
agents/utils/pipeline_logger.py
View File

@ -1,7 +1,7 @@
import os
import json
from datetime import datetime
from typing import Dict, Any, Optional, Union
from typing import Dict, Any, Optional, Union, List
def determiner_repertoire_ticket(ticket_id: str) -> Optional[str]:
"""
@ -177,14 +177,14 @@ def generer_version_texte(data: Union[Dict[str, Any], list], ticket_id: str, ste
except Exception as e:
print(f"Erreur lors de la génération de la version texte: {e}")
def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", data: Optional[Dict[str, Any]] = None, base_dir: Optional[str] = None, is_resultat: bool = False) -> None:
def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", data: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None, base_dir: Optional[str] = None, is_resultat: bool = False) -> None:
"""
Sauvegarde des données sous forme de fichier JSON.
Args:
ticket_id: str, le code du ticket (optionnel, sera extrait automatiquement si None)
step_name: str, le nom de l'étape ou de l'agent
data: dict, les données à sauvegarder
data: dict ou liste, les données à sauvegarder
base_dir: str, le répertoire de base pour les fichiers de logs (optionnel)
is_resultat: bool, indique si les données sont des résultats d'agent
"""
@ -192,9 +192,13 @@ def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", da
print("Aucune donnée à sauvegarder")
return
# Si ticket_id n'est pas fourni, essayer de l'extraire des métadonnées
if not ticket_id:
ticket_id = extraire_ticket_id(data)
# Convertir les données en liste si ce n'est pas déjà le cas
data_list = data if isinstance(data, list) else [data]
# Si ticket_id n'est pas fourni, essayer de l'extraire des métadonnées du premier élément
if not ticket_id and data_list:
first_item = data_list[0]
ticket_id = extraire_ticket_id(first_item)
if ticket_id:
print(f"Ticket ID extrait: {ticket_id}")
@ -219,9 +223,10 @@ def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", da
os.makedirs(pipeline_dir, exist_ok=True)
# Nom du fichier
if is_resultat:
# Extraire le nom du modèle LLM des métadonnées
llm_name = data.get("metadata", {}).get("model_info", {}).get("model", "unknown_model")
if is_resultat and data_list:
# Extraire le nom du modèle LLM des métadonnées du premier élément
first_item = data_list[0]
llm_name = first_item.get("metadata", {}).get("model_info", {}).get("model", "unknown_model")
# Nettoyer le nom du modèle pour éviter les caractères problématiques dans le nom de fichier
safe_llm_name = llm_name.lower().replace("/", "_").replace(" ", "_")
file_name = f"{step_name}_{safe_llm_name}_results.json"
@ -246,30 +251,38 @@ def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", da
print(f"Le fichier existant {file_path} n'est pas un JSON valide, création d'un nouveau fichier")
existing_data = []
# Ajouter les nouvelles données
if isinstance(data, list):
existing_data.extend(data)
# Gérer l'ajout de données et la déduplication
updated_data = existing_data.copy()
# Gérer les entrées existantes pour éviter les doublons
if step_name == "rapport_final":
# Pour le rapport final, toujours remplacer complètement
updated_data = data_list
else:
# Vérifier si cette image a déjà été analysée (pour éviter les doublons)
image_path = data.get("metadata", {}).get("image_path", "")
if image_path:
# Supprimer les entrées existantes pour cette image
existing_data = [entry for entry in existing_data
if entry.get("metadata", {}).get("image_path", "") != image_path]
# Pour les analyses d'images, vérifier si l'image a déjà été analysée
processed_paths = set()
# Éviter la duplication pour le rapport final
if step_name == "rapport_final" and existing_data:
existing_data = [data] # Remplacer au lieu d'ajouter
else:
existing_data.append(data)
# Filtrer les entrées existantes pour éviter les doublons d'images
for item in data_list:
image_path = item.get("metadata", {}).get("image_path", "")
if image_path:
processed_paths.add(image_path)
# Supprimer les entrées existantes pour cette image
updated_data = [entry for entry in updated_data
if entry.get("metadata", {}).get("image_path", "") != image_path]
# Ajouter la nouvelle entrée
updated_data.append(item)
else:
# Si pas de chemin d'image, ajouter simplement
updated_data.append(item)
# Sauvegarder les données combinées
with open(file_path, "w", encoding="utf-8") as f:
json.dump(existing_data, f, indent=2, ensure_ascii=False)
print(f"Données sauvegardées dans {file_path} ({len(existing_data)} entrées)")
json.dump(updated_data, f, indent=2, ensure_ascii=False)
print(f"Données sauvegardées dans {file_path} ({len(updated_data)} entrées)")
# Générer une version texte pour tous les fichiers JSON
generer_version_texte(existing_data, ticket_id, step_name, file_path)
generer_version_texte(updated_data, ticket_id, step_name, file_path)
except Exception as e:
print(f"Erreur lors de la sauvegarde des données: {e}")

88
orchestrator_llama.py
View File

@ -93,24 +93,49 @@ class OrchestratorLlamaVision:
if self.config.get("dedup_enabled", True):
images = filtrer_images_uniques(images, seuil_hamming=self.config["dedup_threshold"], ticket_id=ticket_id)
for img in images:
result_sort = {}
is_relevant = True
if self.image_sorter:
# Traiter toutes les images avec l'agent de tri
if self.image_sorter:
logger.info(f"Traitement de {len(images)} images uniques avec l'agent de tri")
# Analyser toutes les images
for img in images:
try:
result_sort = self.image_sorter.executer(img)
is_relevant = result_sort.get("is_relevant", True)
if is_relevant:
relevant_images.append(img)
images_analyses[img] = {
"sorting": result_sort or {"is_relevant": True},
"analysis": None
}
except Exception as e:
logger.warning(f"Erreur tri image {os.path.basename(img)}: {e}")
if is_relevant:
relevant_images.append(img)
images_analyses[img] = {
"sorting": result_sort or {"is_relevant": True},
"analysis": None
}
# Sauvegarder tous les résultats en une seule fois
if self.image_sorter:
# Utiliser une approche plus générique pour éviter les erreurs de linter
try:
# Essayer d'appeler la méthode si elle existe
sauvegarder_func = getattr(self.image_sorter, "sauvegarder_resultats", None)
if sauvegarder_func and callable(sauvegarder_func):
sauvegarder_func()
logger.info("Sauvegarde groupée des résultats de tri effectuée")
else:
logger.info("L'agent de tri ne dispose pas de la méthode sauvegarder_resultats")
except Exception as e:
logger.warning(f"Erreur lors de la sauvegarde des résultats: {e}")
else:
# Si pas d'agent de tri, considérer toutes les images comme pertinentes
relevant_images = images.copy()
for img in images:
images_analyses[img] = {
"sorting": {"is_relevant": True},
"analysis": None
}
# Analyser les images pertinentes avec l'agent d'analyse d'images
if self.image_analyser and ticket_analysis:
for img in relevant_images:
try:
@ -176,12 +201,43 @@ class OrchestratorLlamaVision:
)
def _lister_images(self, dossier: str) -> List[str]:
extensions = ['.png', '.jpeg', 'jpg', '.gif', '.webp', '.bmp']
"""
Liste toutes les images dans un dossier avec une reconnaissance étendue
des formats d'images.
Args:
dossier: Dossier contenant les images à analyser
Returns:
Liste des chemins d'images trouvées
"""
# Liste étendue des extensions d'images courantes
extensions = ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.webp', '.tiff', '.tif']
images = []
for racine, _, fichiers in os.walk(dossier):
for f in fichiers:
if any(f.lower().endswith(ext) for ext in extensions):
images.append(os.path.join(racine, f))
# Parcourir le dossier pour trouver toutes les images
if os.path.exists(dossier):
for racine, _, fichiers in os.walk(dossier):
for f in fichiers:
# Vérifier l'extension du fichier (non sensible à la casse)
if any(f.lower().endswith(ext) for ext in extensions):
chemin_complet = os.path.join(racine, f)
# Vérifier que le fichier est bien une image valide et accessible
try:
from PIL import Image
with Image.open(chemin_complet) as img:
# S'assurer que c'est bien une image en vérifiant ses dimensions
width, height = img.size
if width > 0 and height > 0:
images.append(chemin_complet)
except Exception as e:
logger.warning(f"Image ignorée {f}: {str(e)}")
if not images:
logger.warning(f"Aucune image trouvée dans {dossier}")
else:
logger.info(f"{len(images)} images trouvées dans {dossier}")
return images

26
output/ticket_T11143/T11143_20250422_084617/T11143_rapports/pipeline/rapport_de_deduplication.json
View File

@ -1,26 +0,0 @@
[
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/image.png",
"status": "unique"
},
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/image_145435.png",
"status": "unique"
},
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/image_145453.png",
"status": "duplicate"
},
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/543d7da1b54c29ff43ce5712d1a9aa4962ed21795c4e943fcb8cb84fd4d7465a.jpg",
"status": "unique"
},
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/5ad281b63492e31c9e66bf27518b816cdd3766cab9812bd4ff16b736e9e98265.jpg",
"status": "duplicate"
},
{
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/a20f7697fd5e1d1fca3296c6d01228220e0e112c46b4440cc938f74d10934e98.gif",
"status": "unique"
}
]

46
output/ticket_T11143/T11143_20250422_084617/T11143_rapports/pipeline/tri_image_llama3.2-vision:90b-instruct-q8_0_results.json
View File

@ -1,46 +0,0 @@
[
{
"is_relevant": false,
"reason": "Non, car il s'agit d'une image aléatoire et non descriptive qui ne semble pas liée à un problème technique spécifique. Elle n'aide pas à comprendre le problème ni à fournir des informations utiles pour résoudre l'incident.",
"raw_response": "Non, car il s'agit d'une image aléatoire et non descriptive qui ne semble pas liée à un problème technique spécifique. Elle n'aide pas à comprendre le problème ni à fournir des informations utiles pour résoudre l'incident.",
"metadata": {
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/image.png",
"image_name": "image.png",
"timestamp": "20250423_092010",
"model_info": {
"model": "llama3.2-vision:90b-instruct-q8_0",
"temperature": 0.2,
"top_p": 0.8,
"max_tokens": 300,
"presence_penalty": 0,
"frequency_penalty": 0,
"stop": [],
"stream": false,
"n": 1
},
"source_agent": "AgentImageSorter"
}
},
{
"is_relevant": false,
"reason": "Non, car l'image semble être un code binaire représentant une image et non une capture d'écran ou une illustration liée à un problème technique spécifique.",
"raw_response": "Non, car l'image semble être un code binaire représentant une image et non une capture d'écran ou une illustration liée à un problème technique spécifique.",
"metadata": {
"image_path": "output/ticket_T11143/T11143_20250422_084617/attachments/a20f7697fd5e1d1fca3296c6d01228220e0e112c46b4440cc938f74d10934e98.gif",
"image_name": "a20f7697fd5e1d1fca3296c6d01228220e0e112c46b4440cc938f74d10934e98.gif",
"timestamp": "20250423_092054",
"model_info": {
"model": "llama3.2-vision:90b-instruct-q8_0",
"temperature": 0.2,
"top_p": 0.8,
"max_tokens": 300,
"presence_penalty": 0,
"frequency_penalty": 0,
"stop": [],
"stream": false,
"n": 1
},
"source_agent": "AgentImageSorter"
}
}
]

18
output/ticket_T11143/T11143_20250422_084617/T11143_rapports/pipeline/tri_image_llama3.2-vision:90b-instruct-q8_0_results.txt
View File

@ -1,18 +0,0 @@
RÉSULTATS DE L'ANALYSE TRI_IMAGE - TICKET T11143
================================================================================
--- ÉLÉMENT 1 ---
Non, car il s'agit d'une image aléatoire et non descriptive qui ne semble pas liée à un problème technique spécifique. Elle n'aide pas à comprendre le problème ni à fournir des informations utiles pour résoudre l'incident.
----------------------------------------
--- ÉLÉMENT 2 ---
Non, car l'image semble être un code binaire représentant une image et non une capture d'écran ou une illustration liée à un problème technique spécifique.
----------------------------------------
================================================================================
Fichier original: tri_image_llama3.2-vision:90b-instruct-q8_0_results.json

45
test_corrections/README.md Normal file
View File

@ -0,0 +1,45 @@
# Corrections pour le tri d'images avec llama-vision
Ce document décrit les corrections apportées pour résoudre les problèmes de tri d'images avec llama-vision.
## Problèmes identifiés
1. **Détection incomplète des images** : Seulement 2 images sur 4 étaient analysées après déduplication
2. **Classification trop stricte** : Toutes les images étaient classées comme non pertinentes
3. **Reconnaissance de formats limitée** : Certains formats d'images n'étaient pas correctement détectés
## Corrections apportées
### 1. Amélioration du prompt système (agent_image_sorter.py)
Le prompt système a été entièrement revu pour :
- Définir plus clairement ce qui constitue une image pertinente
- Adopter une approche "par défaut pertinent" en cas de doute
- Élargir la définition des images pertinentes
- Rendre le classement plus inclusif
### 2. Amélioration de la détection des images (orchestrator_llama.py)
La méthode `_lister_images` a été optimisée pour :
- Supporter davantage de formats d'images (ajout de .tiff, .tif)
- Vérifier que chaque fichier est bien une image valide
- Corriger la détection de l'extension .jpg (qui avait une erreur de syntaxe)
- Ajouter des logs pour faciliter le débogage
## Comment tester
Pour tester spécifiquement le tri d'images avec llama-vision, utilisez la commande suivante :
```bash
python main_llama.py <ticket_id> --skip-ticket-analysis --skip-image-analysis --skip-report
```
Après exécution, vérifiez :
1. Le fichier `tri_image_llama*.json` dans le dossier pipeline
2. Le nombre d'images analysées doit correspondre au nombre d'images uniques dans `rapport_de_deduplication.json`
3. Le classement des images devrait être plus généreux (plus d'images "pertinentes")
## Notes supplémentaires
- Le prompt a été optimisé pour llama-vision tout en conservant la consigne de répondre en français
- L'approche de tri est maintenant plus inclusive ("mieux vaut inclure trop que pas assez")

79
test_corrections/analyse_differences.md Normal file
View File

@ -0,0 +1,79 @@
# Analyse des différences entre main.py et main_llama.py
Ce document analyse les raisons pour lesquelles le tri d'images fonctionne différemment entre les implémentations utilisant Mistral (via `main.py`) et Llama (via `main_llama.py`).
## Résumé du problème
1. Avec `main_llama.py`, seulement 2 des 4 images uniques sont analysées, bien que toutes soient détectées au niveau du log.
2. Avec `main.py` (orchestrateur standard), toutes les 4 images uniques sont correctement analysées.
## Différences identifiées
### 1. Sauvegarde des résultats et ordre des opérations
**Problème principal :**
Dans `main_llama.py`, la fonction de sauvegarde intercepte potentiellement les analyses avant qu'elles ne soient terminées en créant un nouveau fichier de résultats entre les traitements d'images.
```
# Dans main_llama.py (logs observés)
AgentImageSorter: Évaluation de image.png
Décision: Image image.png non pertinente
Données sauvegardées dans [...] (1 entrées) # ← Sauvegarde intermédiaire
Version texte générée
AgentImageSorter: Évaluation de image_145435.png # ← Les analyses continuent
```
**Comparaison :**
Dans l'implémentation de Mistral via `main.py`, la sauvegarde est réalisée à la fin, après toutes les analyses.
### 2. Comportement de `sauvegarder_donnees()`
**Dans pipeline_logger.py :**
```python
# Ajouter les nouvelles données
if isinstance(data, list):
existing_data.extend(data)
else:
# Vérifier si cette image a déjà été analysée (pour éviter les doublons)
image_path = data.get("metadata", {}).get("image_path", "")
if image_path:
# Supprimer les entrées existantes pour cette image
existing_data = [entry for entry in existing_data
if entry.get("metadata", {}).get("image_path", "") != image_path]
```
Cette fonction supprime les entrées existantes pour une image avant d'ajouter la nouvelle, mais ne charge pas correctement l'état complet si le fichier est recréé à chaque appel.
### 3. Différence d'orchestration
**Orchestrator.py (main.py) :**
- Collecte toutes les images
- Effectue la déduplication
- Trie toutes les images dans une boucle unique
- Sauvegarde les résultats une seule fois à la fin
**OrchestratorLlamaVision.py (main_llama.py) :**
- Collecte toutes les images
- Effectue la déduplication
- Traite chaque image individuellement avec sauvegarde immédiate
- Ne semble pas gérer correctement l'accumulation des résultats entre les sauvegardes
### 4. Classification des images
**Avec Pixtral (via main.py) :**
- 2 images classées comme pertinentes (captures d'écran)
- 2 images classées comme non pertinentes (logos)
**Avec Llama (via main_llama.py) :**
- Toutes les images sont classées comme non pertinentes
- Même après modifications du prompt, Llama semble avoir une interprétation différente des images
## Solution recommandée
1. **Modifier l'orchestrateur de Llama** pour accumuler tous les résultats en mémoire avant de les sauvegarder une seule fois à la fin, comme dans l'implémentation standard.
2. **Restructurer la fonction `sauvegarder_donnees()`** pour qu'elle gère mieux les sauvegardes successives.
3. **Améliorer la gestion du contexte** pour s'assurer que les images en français sont correctement interprétées par Llama.
4. **Utiliser le modèle Pixtral** pour le tri d'images si possible, car il semble avoir une meilleure interprétation des captures d'écran techniques.

199
test_corrections/test_tri_images.py Executable file
View File

@ -0,0 +1,199 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Script de test pour vérifier le tri d'images avec llama-vision après les corrections.
"""
import os
import sys
import json
import logging
from typing import List, Dict, Any
from pprint import pprint
# Configuration du logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('TestTriImages')
def verifier_rapport_deduplication(ticket_id: str) -> List[str]:
"""
Vérifie le rapport de déduplication pour extraire les chemins des images uniques.
Args:
ticket_id: ID du ticket à analyser
Returns:
Liste des chemins d'images uniques
"""
# Déterminer le répertoire du ticket
base_dir = "output"
ticket_dir = f"ticket_{ticket_id}"
ticket_path = os.path.join(base_dir, ticket_dir)
# Trouver la dernière extraction
extractions = []
for extract in os.listdir(ticket_path):
extraction_path = os.path.join(ticket_path, extract)
if os.path.isdir(extraction_path) and extract.startswith(ticket_id):
extractions.append(extraction_path)
# Trier par date (plus récente en premier)
extractions.sort(key=lambda x: os.path.getmtime(x), reverse=True)
latest_extraction = extractions[0]
# Chemin du rapport de déduplication
pipeline_dir = os.path.join(latest_extraction, f"{ticket_id}_rapports", "pipeline")
dedup_path = os.path.join(pipeline_dir, "rapport_de_deduplication.json")
if not os.path.exists(dedup_path):
logger.error(f"Rapport de déduplication introuvable: {dedup_path}")
return []
# Charger le rapport
with open(dedup_path, 'r', encoding='utf-8') as f:
dedup_data = json.load(f)
# Extraire les images uniques
images_uniques = [item['image_path'] for item in dedup_data if item.get('status') == 'unique']
logger.info(f"Trouvé {len(images_uniques)} images uniques sur {len(dedup_data)} total")
return images_uniques
def verifier_resultats_tri(ticket_id: str) -> Dict[str, Any]:
"""
Vérifie les résultats du tri d'images.
Args:
ticket_id: ID du ticket à analyser
Returns:
Dictionnaire avec les statistiques des résultats
"""
# Déterminer le répertoire du ticket
base_dir = "output"
ticket_dir = f"ticket_{ticket_id}"
ticket_path = os.path.join(base_dir, ticket_dir)
# Trouver la dernière extraction
extractions = []
for extract in os.listdir(ticket_path):
extraction_path = os.path.join(ticket_path, extract)
if os.path.isdir(extraction_path) and extract.startswith(ticket_id):
extractions.append(extraction_path)
# Trier par date (plus récente en premier)
extractions.sort(key=lambda x: os.path.getmtime(x), reverse=True)
latest_extraction = extractions[0]
# Chemin des résultats de tri
pipeline_dir = os.path.join(latest_extraction, f"{ticket_id}_rapports", "pipeline")
# Trouver le fichier de résultats de tri
tri_files = [f for f in os.listdir(pipeline_dir) if f.startswith("tri_image_") and f.endswith("_results.json")]
if not tri_files:
logger.error(f"Aucun fichier de résultats de tri trouvé dans {pipeline_dir}")
return {}
# Charger le fichier le plus récent
tri_files.sort(key=lambda x: os.path.getmtime(os.path.join(pipeline_dir, x)), reverse=True)
tri_path = os.path.join(pipeline_dir, tri_files[0])
with open(tri_path, 'r', encoding='utf-8') as f:
tri_data = json.load(f)
# Analyser les résultats
images_analysees = []
pertinentes = 0
non_pertinentes = 0
if isinstance(tri_data, list):
for item in tri_data:
if "metadata" in item and "image_path" in item["metadata"]:
images_analysees.append(item["metadata"]["image_path"])
if item.get("is_relevant", False):
pertinentes += 1
else:
non_pertinentes += 1
# Préparer les statistiques
stats = {
"images_analysees": len(images_analysees),
"images_pertinentes": pertinentes,
"images_non_pertinentes": non_pertinentes,
"chemins_analyses": images_analysees,
"fichier_resultats": tri_path
}
return stats
def comparer_images(images_uniques: List[str], stats: Dict[str, Any]) -> None:
"""
Compare les images uniques avec celles analysées.
Args:
images_uniques: Liste des chemins d'images uniques
stats: Statistiques des résultats de tri
"""
images_analysees = stats.get("chemins_analyses", [])
# Vérifier si toutes les images uniques ont été analysées
manquantes = set(images_uniques) - set(images_analysees)
en_trop = set(images_analysees) - set(images_uniques)
if manquantes:
logger.warning(f"Images uniques non analysées ({len(manquantes)}):")
for img in manquantes:
logger.warning(f" - {os.path.basename(img)}")
if en_trop:
logger.warning(f"Images analysées qui ne sont pas uniques ({len(en_trop)}):")
for img in en_trop:
logger.warning(f" - {os.path.basename(img)}")
if not manquantes and not en_trop:
logger.info("Toutes les images uniques ont été correctement analysées")
def main():
"""
Fonction principale du script de test.
"""
if len(sys.argv) < 2:
logger.error("Usage: python test_tri_images.py <ticket_id>")
sys.exit(1)
ticket_id = sys.argv[1]
logger.info(f"Vérification du tri d'images pour le ticket {ticket_id}")
# Vérifier le rapport de déduplication
images_uniques = verifier_rapport_deduplication(ticket_id)
if not images_uniques:
logger.error("Aucune image unique trouvée, impossible de continuer")
sys.exit(1)
# Vérifier les résultats du tri
stats = verifier_resultats_tri(ticket_id)
if not stats:
logger.error("Aucun résultat de tri trouvé, impossible de continuer")
sys.exit(1)
# Afficher les résultats
logger.info("Résumé du tri d'images:")
logger.info(f" Images uniques: {len(images_uniques)}")
logger.info(f" Images analysées: {stats['images_analysees']}")
logger.info(f" Images pertinentes: {stats['images_pertinentes']}")
logger.info(f" Images non pertinentes: {stats['images_non_pertinentes']}")
# Comparer les images
comparer_images(images_uniques, stats)
return 0
if __name__ == "__main__":
sys.exit(main())

23
utils/ocr_utils.py Normal file
View File

@ -0,0 +1,23 @@
# utils/ocr_utils.py
from PIL import Image
import pytesseract
import logging
logger = logging.getLogger("OCR")
def extraire_texte_fr(image_path: str) -> str:
"""
Effectue un OCR sur une image en langue française.
Retrourne le texte brut extrait (vide si échec).
"""
try:
image = Image.open(image_path)
texte = pytesseract.image_to_string(image, lang="fra").strip()
logger.debug(f"OCR FR pour {image_path}: {texte}")
return texte
except Exception as e:
logger.error(f"Erreur lors de l'OCR de {image_path}: {e}")
return ""

75
utils/translate_utils.py Normal file
View File

@ -0,0 +1,75 @@
# utils/translate_utils.py
from deep_translator import GoogleTranslator
import json
import os
from datetime import datetime
import logging
logger = logging.getLogger("Translate")
def fr_to_en(text: str) -> str:
try:
if not text.strip():
return ""
return GoogleTranslator(source="fr", target="en").translate(text)
except Exception as e:
logger.error(f"Traduction FR->EN échouée: {e}")
return ""
def en_to_fr(text: str) -> str:
try:
if not text.strip():
return ""
return GoogleTranslator(source="en", target="fr").translate(text)
except Exception as e:
logger.error(f"Traduction EN->FR échouée: {e}")
return ""
def sauvegarder_ocr_traduction(
image_path: str,
ticket_id: str,
ocr_fr: str,
ocr_en: str,
ocr_en_back_fr: str = "", # <- Ajout facultatif
base_dir: str = "reports"
) -> None:
"""
Sauvegarde les résultats OCR + TRAD en JSON + ajoute une ligne dans le fichier texte global.
Inclut éventuellement une traduction EN → FR.
"""
try:
image_name = os.path.basename(image_path)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
rapport_dir = os.path.join(base_dir, ticket_id, "pipeline", "ocr_traduction")
os.makedirs(rapport_dir, exist_ok=True)
result = {
"image_name": image_name,
"ocr_fr": ocr_fr,
"translation_en": ocr_en,
"translation_en_back_fr": ocr_en_back_fr,
"metadata": {
"ticket_id": ticket_id,
"timestamp": timestamp,
"source_module": "ocr_utils + translate_utils",
"lang_detected": "fr"
}
}
# Fichier JSON par image
with open(os.path.join(rapport_dir, f"{image_name}.json"), "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
# Append texte global
ligne = (
f"{image_name}\n"
f"[FR] {ocr_fr or '_'}\n"
f"[EN] {ocr_en or '_'}\n"
f"[EN→FR] {ocr_en_back_fr or '_'}\n\n"
)
with open(os.path.join(rapport_dir, "ocr_traduction.txt"), "a", encoding="utf-8") as f:
f.write(ligne)
except Exception as e:
logger.error(f"Erreur sauvegarde OCR+TRAD pour {image_path}: {e}")