LLM Prompting & Post-Processing¶

# POST http://localhost:8000/llm/submit
{
  "operation": "clean_transcription" | "identify_speakers",
  "payload": {
    "language": "fr",
    "segments": {...},
    "context": {...}
  }
}

# Response
{
  "task_id": "llm-task-abc123",
  "status": "pending"
}

# Redis Stream: llm:reply:{task_id}
{
  "task_id": "llm-task-abc123",
  "status": "completed" | "failed",
  "result": {...},
  "error": null | "error message"
}

CLEAN_TRANSCRIPTION_PROMPT = """
Tu es un expert en correction de transcriptions automatiques françaises.

CONTEXTE:
- Transcription automatique d'une réunion professionnelle
- Langue: {language}
- Participants identifiés: {identified_participants}
- Mots-clés du projet: {keywords}
- Glossaire: {glossary}

TRANSCRIPTION BRUTE:
---
{raw_transcription}
---

TÂCHE:
Corriger la transcription en appliquant les règles suivantes:

1. **Ponctuation**
   - Ajouter points (.) en fin de phrases
   - Ajouter virgules (,) pour séparer propositions
   - Ajouter points d'interrogation (?) pour questions
   - Ajouter points d'exclamation (!) si approprié

2. **Capitalisation**
   - Majuscule au début de chaque phrase
   - Majuscules pour noms propres (personnes, entreprises, lieux)
   - Majuscules pour acronymes (RAG, API, ONU)

3. **Noms propres**
   - Utiliser liste participants pour orthographe exacte
   - Exemple: "jean dupont" → "Jean Dupont"

4. **Acronymes et termes techniques**
   - Utiliser glossaire si disponible
   - Exemple: "rag" → "RAG", "api" → "API"
   - Exemple: "meuh tu" → "MeetNoo"

5. **Cohérence temporelle**
   - Vérifier conjugaison verbes
   - Corriger si nécessaire pour cohérence

RÈGLES CRITIQUES:
- NE PAS modifier le sens de la transcription
- NE PAS ajouter d'informations non présentes
- NE PAS supprimer d'informations
- CONSERVER la structure [Speaker Label]: texte
- CORRIGER uniquement forme, PAS le fond

FORMAT RÉPONSE (JSON strict):
```json
{
  "cleaned_transcription": {
    "SPEAKER_00": "Texte corrigé du speaker 00...",
    "SPEAKER_01": "Texte corrigé du speaker 01...",
    "SPEAKER_02": "Texte corrigé du speaker 02..."
  },
  "corrections_applied": [
    "Added punctuation: 12 periods, 8 commas, 3 question marks",
    "Fixed capitalization: 15 words",
    "Corrected acronyms: RAG (3x), API (2x)",
    "Corrected proper nouns: Jean Dupont, Marie Martin"
  ],
  "errors_corrected": [
    {
      "original": "meuh tu",
      "corrected": "MeetNoo",
      "type": "phonetic_error"
    },
    {
      "original": "rag",
      "corrected": "RAG",
      "type": "acronym"
    }
  ]
}

IDENTIFY_SPEAKERS_PROMPT = """
Tu es un expert en analyse de transcriptions pour identifier les participants.

CONTEXTE RAG:
- Participants potentiels: {potential_participants}
- Mots-clés projet: {keywords}
- Glossaire termes techniques: {glossary_terms}

SEGMENTS À IDENTIFIER:
---
{unidentified_segments}
---

TÂCHE:
Identifier chaque SPEAKER_XX en analysant:

1. **Auto-identifications explicites**
   - "Je suis [Nom]"
   - "En tant que [Rôle]"
   - "Je travaille chez [Entreprise]"

2. **Matching avec participants RAG**
   - Comparer contenu segments avec contexte participants
   - Exemple: "backend developer" → Jean Dupont (Lead Backend Developer)

3. **Cohérence thématique**
   - Qui parle de quoi
   - Exemple: Speaker qui parle de "diplomatie" → Kwame Mensah (Senior Diplomat)

4. **Keywords matching**
   - Comparer interventions avec keywords projet
   - Exemple: Parle de "RAG" + "embedding" → Probable ML Engineer

RÈGLES D'IDENTIFICATION:
- Confidence > 0.75 → Identifier avec nom réel
- Confidence < 0.75 → Laisser "Intervenant X" (ne pas deviner)
- NE JAMAIS inventer de noms absents du contexte RAG

CALCUL CONFIDENCE:
- Auto-identification explicite: 0.95
- Match contexte RAG + keywords: 0.80-0.90
- Match contexte RAG uniquement: 0.60-0.75
- Match thématique faible: 0.30-0.50
- Aucun indice: 0.10-0.20

FORMAT RÉPONSE (JSON strict):
```json
{
  "speaker_identifications": {
    "SPEAKER_00": {
      "identified_name": "Kwame Mensah",
      "confidence": 0.92,
      "reasoning": "Auto-identification 'je suis Kwame Mensah, Senior Diplomat à l'ONU' + match contexte RAG (Senior Diplomat, ONU)",
      "evidence": [
        "Explicit: 'je suis Kwame Mensah'",
        "Role match: 'Senior Diplomat'",
        "Company match: 'ONU'"
      ]
    },
    "SPEAKER_01": {
      "identified_name": "Intervenant 1",
      "confidence": 0.35,
      "reasoning": "Intervention courte, aucune auto-identification, aucun match RAG fort",
      "evidence": [
        "Thematic: mentioned 'RAG' (weak match avec ML Engineer)",
        "Too little data for confident identification"
      ]
    }
  },
  "summary": {
    "total_speakers": 2,
    "identified": 1,
    "pending": 1,
    "avg_confidence": 0.635
  }
}

async def process_llm_identifications(
    self,
    identifications: Dict[str, Any],
    pending_voiceprints: Dict[str, str],  # {speaker_label: voiceprint_lib_id}
    db: Session
):
    """
    Confirmer voiceprints pending si confidence > 0.75.
    """
    for speaker_label, data in identifications["speaker_identifications"].items():
        confidence = data["confidence"]
        identified_name = data["identified_name"]

        # Skip "Intervenant X" (low confidence)
        if "Intervenant" in identified_name:
            logger.info(
                f"DEBUG: Skipping confirmation for {speaker_label} - "
                f"Low confidence ({confidence:.2f})"
            )
            continue

        # Confirm if confidence > 0.75
        if confidence >= 0.75:
            voiceprint_lib_id = pending_voiceprints.get(speaker_label)

            if voiceprint_lib_id:
                # Update voiceprint status
                voiceprint = db.query(VoiceprintLibrary).filter(
                    VoiceprintLibrary.id == voiceprint_lib_id
                ).first()

                if voiceprint:
                    voiceprint.status = "confirmed"
                    voiceprint.identified_name = identified_name
                    voiceprint.match_source = "llm_inference"
                    voiceprint.updated_at = unix_timestamp()

                    db.commit()

                    logger.info(
                        f"DEBUG: Confirmed voiceprint {voiceprint_lib_id} - "
                        f"Name: {identified_name}, Confidence: {confidence:.2f}"
                    )

# BFF: src/services/redis_service.py
import redis.asyncio as aioredis

class RedisService:
    def __init__(self):
        self.client = None

    async def connect(self):
        self.client = await aioredis.from_url(
            "redis://localhost:6379/0",
            encoding="utf-8",
            decode_responses=True,
            max_connections=50,
            socket_connect_timeout=5,
            socket_keepalive=True
        )

        logger.info("DEBUG: Redis client connected")

    async def xadd(
        self,
        stream: str,
        data: Dict[str, str],
        maxlen: int = 1000
    ) -> str:
        """Add to stream with max length."""
        message_id = await self.client.xadd(
            stream,
            data,
            maxlen=maxlen,
            approximate=True
        )
        return message_id

    async def xread(
        self,
        streams: Dict[str, str],
        count: int = 1,
        block: int = 2000
    ) -> List[Any]:
        """Read from streams (blocking)."""
        messages = await self.client.xread(
            streams,
            count=count,
            block=block
        )
        return messages

# Cleaning et identification = tasks déterministes
# Pas de créativité nécessaire
temperature = 0.0

# Forcer JSON strict
prompt += "\n\nIMPORTANT: Réponds UNIQUEMENT avec le JSON, aucun texte avant ou après."

# Validation
try:
    result = json.loads(llm_response)
except json.JSONDecodeError:
    # Retry with explicit JSON instruction
    pass

# Add examples to prompt for better accuracy
EXAMPLES = """
EXEMPLE 1:
Input: "bonjour je suis jean dupont"
Output: "Bonjour, je suis Jean Dupont."

EXEMPLE 2:
Input: "le rag c est retrieval augmented generation"
Output: "Le RAG, c'est Retrieval-Augmented Generation."
"""

# Ne pas surcharger contexte
# MAX:
#   - 10 participants potentiels
#   - 20 keywords
#   - 10 glossary terms

# Si plus → Filtrer par pertinence RAG score

# src/prompts/llm_prompts.py
class LLMPrompts:
    # Version control
    VERSION = "1.0.0"

    # Clean transcription
    CLEAN_TRANSCRIPTION_V1 = """..."""

    # Identify speakers
    IDENTIFY_SPEAKERS_V1 = """..."""

    @classmethod
    def get_prompt(cls, operation: str, version: str = "v1") -> str:
        """Get prompt by operation and version."""
        prompts = {
            "clean_transcription": {
                "v1": cls.CLEAN_TRANSCRIPTION_V1
            },
            "identify_speakers": {
                "v1": cls.IDENTIFY_SPEAKERS_V1
            }
        }

        return prompts[operation][version]

class LLMPostProcessor:
    async def safe_llm_call(
        self,
        operation: str,
        payload: Dict[str, Any],
        fallback_func: Optional[Callable] = None
    ) -> Dict[str, Any]:
        """
        LLM call avec timeout et fallback.
        """
        try:
            # Submit task
            task_id = await self._submit_llm_task(operation, payload)

            # Poll with timeout
            result = await self._poll_llm_result(
                task_id,
                timeout=90
            )

            return result

        except TimeoutError:
            logger.error(
                f"DEBUG: LLM timeout after 90s - Operation: {operation}"
            )

            # Fallback if provided
            if fallback_func:
                logger.info("DEBUG: Using fallback function")
                return await fallback_func(payload)

            # No fallback: Re-raise
            raise

def regex_cleaning_fallback(raw_transcription: Dict[str, str]) -> Dict[str, str]:
    """
    Fallback cleaning si LLM timeout.

    Simple rules:
    - Capitalize first letter of each sentence
    - Add period at end if missing
    - Uppercase common acronyms (RAG, API, ONU)
    """
    cleaned = {}

    for speaker, text in raw_transcription.items():
        # Capitalize first letter
        text = text[0].upper() + text[1:] if text else text

        # Add period if missing
        if text and not text.endswith((".", "!", "?")):
            text += "."

        # Uppercase acronyms
        for acronym in ["rag", "api", "onu", "gpu", "cpu"]:
            text = re.sub(
                r"\b" + acronym + r"\b",
                acronym.upper(),
                text,
                flags=re.IGNORECASE
            )

        cleaned[speaker] = text

    return {
        "cleaned_transcription": cleaned,
        "corrections_applied": ["Basic regex cleaning (LLM fallback)"],
        "errors_corrected": []
    }

def keep_speakers_pending(
    speaker_labels: List[str]
) -> Dict[str, Any]:
    """
    Fallback si LLM identification timeout.

    Keep all speakers as "Intervenant X".
    """
    identifications = {}

    for i, label in enumerate(speaker_labels):
        identifications[label] = {
            "identified_name": f"Intervenant {i}",
            "confidence": 0.0,
            "reasoning": "LLM timeout - kept as pending",
            "evidence": []
        }

    return {
        "speaker_identifications": identifications,
        "summary": {
            "total_speakers": len(speaker_labels),
            "identified": 0,
            "pending": len(speaker_labels),
            "avg_confidence": 0.0
        }
    }