Error Correction Memory¶

class TranscriptionError(Base):
    """
    Error Correction Memory pour RAG v3.
    Stocke les erreurs de transcription reportées par utilisateurs.
    """

    __tablename__ = "transcription_errors"

    # Identifiers
    id = Column(String(36), primary_key=True)
    user_id = Column(String(36), ForeignKey("users.id"), nullable=False, index=True)
    transcript_id = Column(String(36), ForeignKey("transcripts.uuid"), nullable=True)
    segment_id = Column(String(36), nullable=True)

    # Error pattern
    wrong_text = Column(String(500), nullable=False)
    correct_text = Column(String(500), nullable=False)

    # Pattern tracking
    frequency = Column(Integer, default=1, nullable=False)
    context = Column(String(100), nullable=True)  # tech, medical, legal, business

    # Metadata
    confidence = Column(Float, nullable=True)  # 0-1
    first_seen = Column(String(50), nullable=True)
    last_seen = Column(String(50), nullable=True)

    # Timestamps
    created_at = Column(String(50), nullable=False)
    updated_at = Column(String(50), nullable=False)

def save_error_pattern(
    wrong_text: str,
    correct_text: str,
    user_id: str,
    db: Session,
    transcript_id: Optional[str] = None,
    segment_id: Optional[str] = None,
    context: Optional[str] = None,
    confidence: Optional[float] = None
) -> str

error_memory.save_error_pattern(
    wrong_text="cooper nettys",
    correct_text="kubernetes",
    user_id="user_123",
    db=db,
    context="tech",
    confidence=0.95  # Correction manuelle = haute confiance
)

def retrieve_error_patterns(
    segment_texts: List[str],
    user_id: str,
    db: Session,
    context: Optional[str] = None,
    min_frequency: int = 2,
    limit: int = 20
) -> List[Dict]

[
    {
        "wrong_text": "cooper nettys",
        "correct_text": "kubernetes",
        "frequency": 15,
        "context": "tech",
        "confidence": 0.95
    },
    {
        "wrong_text": "doc air",
        "correct_text": "Docker",
        "frequency": 8,
        "context": "tech",
        "confidence": 0.92
    }
]

# Endpoint: PATCH /api/transcripts/{id}/segments
@router.patch("/{transcript_id}/segments")
async def update_segment(
    transcript_id: str,
    segment_data: dict,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    segment = get_segment_from_db(segment_data["segment_id"])

    if "transcription" in segment_data:
        old_text = segment.transcription
        new_text = segment_data["transcription"]

        # 1. Sauvegarder l'original (première modification)
        if segment.original_transcription is None:
            segment.original_transcription = old_text

        # 2. Sauvegarder le pattern d'erreur
        if old_text and new_text and old_text != new_text:
            error_memory.save_error_pattern(
                wrong_text=old_text,
                correct_text=new_text,
                user_id=current_user.id,
                db=db,
                transcript_id=transcript_id,
                segment_id=segment.id,
                context="tech",  # Détecté automatiquement ou choisi par user
                confidence=0.95  # Correction manuelle = haute confiance
            )
            logger.info(f"Error pattern saved for segment {segment.id}")

        # 3. Mettre à jour le texte
        segment.transcription = new_text
        db.commit()

    return {"status": "updated"}

# Récupérer les patterns d'erreur de l'utilisateur
error_patterns = self.error_memory.retrieve_error_patterns(
    segment_texts=segment_texts,
    user_id=user_id,
    db=db,
    context="tech",
    min_frequency=2,
    limit=15
)

# Construire le contexte pour le prompt
if error_patterns:
    pattern_lines = [
        f"- '{p['wrong_text']}' → '{p['correct_text']}' (fréquence: {p['frequency']})"
        for p in error_patterns[:10]
    ]
    error_patterns_info = (
        f"\n\nERREURS FREQUENTES DETECTEES (PRIORITE HAUTE - corrigez TOUJOURS):\n"
        + "\n".join(pattern_lines)
    )

# Injecter dans le prompt LLM
prompt = f"""Tu es un assistant de nettoyage de transcriptions automatiques.

Tâche: Corriger et améliorer la transcription suivante (sortie Whisper).
{error_patterns_info}
{context_info}
{speaker_info}

Transcription à corriger:
{full_transcription}
"""

# Analyze first 5 segments to determine context
context = await llm_service.detect_context(segments[:5])
# Returns: "tech", "medical", "legal", "business", "general"

tech_keywords = ["API", "Docker", "kubernetes", "Python", "React"]
medical_keywords = ["patient", "diagnostic", "traitement", "radiographie"]

if any(kw in transcription for kw in tech_keywords):
    context = "tech"
elif any(kw in transcription for kw in medical_keywords):
    context = "medical"
else:
    context = "general"

# Métriques business
total_patterns = db.query(func.count(TranscriptionError.id)).scalar()
avg_frequency = db.query(func.avg(TranscriptionError.frequency)).scalar()
top_errors = db.query(TranscriptionError).order_by(TranscriptionError.frequency.desc()).limit(10)

# WER improvement estimation
corrections_applied = len([p for p in patterns if p['frequency'] >= 3])
estimated_wer_gain = corrections_applied * 0.002  # ~0.2% per common pattern