Workflow RAG - Enrichissement Contextuel¶

similarity = cosine_similarity(new_voiceprint, stored_voiceprint)

if similarity > 0.85:
    identified_name = stored_voiceprint.identified_name
else:
    auto_save_pending()

class VoiceprintMatcher:
    def __init__(self, db: Session, threshold: float = 0.85):
        self.db = db
        self.threshold = threshold

    def match_speaker(
        self,
        voiceprint_audio_512d: List[float],
        user_id: str
    ) -> Optional[Dict[str, Any]]:
        """
        Match voiceprint contre bibliothèque user.

        Returns:
            Match dict si similarity > threshold, sinon None
        """
        # Fetch all confirmed voiceprints for this user
        voiceprints = self.db.query(VoiceprintLibrary).filter(
            VoiceprintLibrary.user_id == user_id,
            VoiceprintLibrary.status == "confirmed"
        ).all()

        if not voiceprints:
            return None

        # Compute similarities
        best_match = None
        best_score = 0.0

        for vp in voiceprints:
            similarity = self._cosine_similarity(
                voiceprint_audio_512d,
                json.loads(vp.voiceprint_audio_512d)
            )

            if similarity > best_score:
                best_score = similarity
                best_match = vp

        # Threshold check
        if best_score >= self.threshold:
            return {
                "voiceprint_lib_id": best_match.id,
                "identified_name": best_match.identified_name,
                "email": best_match.email,
                "phone": best_match.phone,
                "company": best_match.company,
                "similarity": best_score,
                "match_source": "voiceprint_audio"
            }

        return None

    @staticmethod
    def _cosine_similarity(a: List[float], b: List[float]) -> float:
        """Cosine similarity optimisé."""
        a_np = np.array(a)
        b_np = np.array(b)
        return float(
            np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np))
        )

async def _enrich_identified_speaker(
    self,
    speaker_name: str,
    speaker_segments: List[Dict],
    collection_name: str
) -> Dict[str, Optional[str]]:
    """
    Enrichir speaker déjà identifié avec métadonnées RAG.

    Steps:
    1. Mean pooling de TOUS les segments du speaker
    2. Search Qdrant avec nested filter (all_participants.name = speaker_name)
    3. Extract metadata: email, phone, company, department
    """
    # Step 1: Mean Pooling
    query_embedding = await self.embedding_service.mean_pool_speaker_segments(
        segments=speaker_segments,
        min_segment_length=10  # Filter "euh", "hmm"
    )

    # Step 2: Qdrant Search with Nested Filter
    search_results = await self.qdrant_service.search_similar_chunks(
        collection_name=collection_name,
        query_vector=query_embedding,
        top_k=3,
        participant_filter=speaker_name,  # CRITICAL: Filter by name
        score_threshold=0.5
    )

    # Step 3: Extract Metadata
    enrichment = {
        "role": None,
        "email": None,
        "phone": None,
        "company": None,
        "department": None
    }

    for result in search_results:
        payload = result["payload"]
        participants = payload.get("participants", [])

        for p in participants:
            if isinstance(p, dict) and p.get("name") == speaker_name:
                # Found matching participant
                enrichment.update({
                    "role": p.get("role"),
                    "email": p.get("email"),
                    "phone": p.get("phone"),
                    "company": p.get("company"),
                    "department": p.get("department")
                })

                # Update voiceprint library (permanent metadata)
                await self._update_voiceprint_metadata(
                    voiceprint_lib_id=voiceprint_lib_id,
                    metadata=enrichment,
                    db=db
                )

                return enrichment

    return enrichment

async def _extract_potential_speakers(
    self,
    speaker_segments: List[Dict],
    collection_name: str
) -> Dict[str, List]:
    """
    Extraire context pour speaker non identifié.

    Steps:
    1. Mean pooling segments
    2. Search Qdrant (NO filter - search everything)
    3. Extract: potential_speakers, keywords, glossary
    """
    # Step 1: Mean Pooling
    query_embedding = await self.embedding_service.mean_pool_speaker_segments(
        segments=speaker_segments,
        min_segment_length=10
    )

    # Step 2: Qdrant Search (no participant filter)
    search_results = await self.qdrant_service.search_similar_chunks(
        collection_name=collection_name,
        query_vector=query_embedding,
        top_k=3,
        participant_filter=None,  # No filter
        score_threshold=0.0
    )

    # Step 3: Extract Context
    extraction = {
        "participants": [],
        "keywords": [],
        "glossary_terms": []
    }

    for result in search_results:
        payload = result["payload"]

        # Extract participants (with type filtering - bug fix)
        all_participants = payload.get("all_participants", [])
        for participant in all_participants:
            if isinstance(participant, dict):
                name = participant.get("name")
                if name and name != "null":
                    extraction["participants"].append(name)
            elif isinstance(participant, str):
                if participant != "null":
                    extraction["participants"].append(participant)

        # Extract mentioned_participants (with type filtering)
        mentioned = payload.get("mentioned_participants", [])
        for participant in mentioned:
            if isinstance(participant, dict):
                name = participant.get("name")
                if name and name != "null":
                    extraction["participants"].append(name)
            elif isinstance(participant, str):
                if participant != "null":
                    extraction["participants"].append(participant)

        # Extract keywords
        keywords = payload.get("keywords", [])
        extraction["keywords"].extend(keywords)

        # Extract glossary
        glossary = payload.get("glossary", {})
        extraction["glossary_terms"].extend(glossary.keys())

    # Deduplicate (safe - only strings after filtering)
    extraction["participants"] = list(set(extraction["participants"]))
    extraction["keywords"] = list(set(extraction["keywords"]))
    extraction["glossary_terms"] = list(set(extraction["glossary_terms"]))

    return extraction

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

CONTEXTE:
- Transcription brute d'une réunion (français)
- Participants identifiés: {participants}
- Mots-clés projet: {keywords}

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

TÂCHE:
Corriger la transcription en appliquant:

1. **Ponctuation correcte** (points, virgules, majuscules)
2. **Acronymes** (utiliser glossaire si disponible)
3. **Noms propres** (participants, entreprises, lieux)
4. **Cohérence temporelle** (verbes au bon temps)

RÈGLES:
- NE PAS modifier le sens
- NE PAS ajouter d'informations
- Conserver structure [Speaker]: texte

FORMAT RÉPONSE (JSON):
```json
{
  "cleaned_transcription": {
    "SPEAKER_00": "Texte corrigé...",
    "SPEAKER_01": "Texte corrigé..."
  },
  "corrections_applied": [
    "Fixed punctuation (12 commas, 8 periods)",
    "Corrected acronym 'RAG' with glossary"
  ]
}

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

CONTEXTE:
- Participants potentiels (RAG): {potential_participants}
- Mots-clés: {keywords}
- Glossaire: {glossary_terms}

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

TÂCHE:
Identifier chaque SPEAKER_XX en analysant:
1. Le **contenu** de ses interventions
2. **Participants potentiels** du contexte RAG
3. **Auto-identifications** ("Je suis X", "En tant que Y")
4. **Cohérence thématique** (qui parle de quoi)

RÈGLES:
- Confidence > 0.75 → Identification valide
- Confidence < 0.75 → Laisser "Intervenant X"
- NE PAS inventer de noms

FORMAT RÉPONSE (JSON):
```json
{
  "speaker_identifications": {
    "SPEAKER_00": {
      "identified_name": "Jean Dupont",
      "confidence": 0.85,
      "reasoning": "S'est présenté comme lead developer, parle de backend"
    },
    "SPEAKER_01": {
      "identified_name": "Intervenant 1",
      "confidence": 0.30,
      "reasoning": "Pas assez d'indices"
    }
  }
}

# Confirm pending voiceprints si confidence > 0.75
for speaker_label, data in identifications.items():
    if data["confidence"] >= 0.75:
        await confirm_pending_voiceprint(
            voiceprint_lib_id=pending_voiceprints[speaker_label],
            identified_name=data["identified_name"],
            match_source="llm_inference",
            db=db
        )

class EmbeddingService:
    def mean_pool_speaker_segments(
        self,
        segments: List[Dict[str, str]],
        min_segment_length: int = 10
    ) -> List[float]:
        """
        Mean pooling avec filtrage bruit.

        Steps:
        1. Filter segments < min_length ("euh", "hmm")
        2. Encode all valid segments
        3. Mean pooling
        4. L2 normalization (CRITICAL for cosine)
        """
        # Step 1: Filter
        valid_texts = [
            seg["transcription"]
            for seg in segments
            if len(seg["transcription"].strip()) >= min_segment_length
        ]

        if not valid_texts:
            raise ValueError("No valid segments for pooling")

        # Step 2: Encode
        embeddings = self.model.encode(
            valid_texts,
            normalize_embeddings=False,  # Don't normalize yet
            convert_to_numpy=True
        )
        # Shape: (N, 1024)

        # Step 3: Mean Pooling
        pooled = np.mean(embeddings, axis=0)
        # Shape: (1024,)

        # Step 4: L2 Normalization
        norm = np.linalg.norm(pooled)
        if norm == 0:
            raise ValueError("Zero norm after pooling")

        normalized = pooled / norm

        # Validation
        final_norm = np.linalg.norm(normalized)
        logger.info(
            f"DEBUG: Pooled {len(valid_texts)} segments, "
            f"final_norm={final_norm:.6f}"
        )

        # Must be 1.0 for cosine similarity
        assert abs(final_norm - 1.0) < 0.0001, f"Norm must be 1.0, got {final_norm}"

        return normalized.tolist()

async def auto_save_pending_voiceprint(
    self,
    speaker_label: str,
    voiceprint_512d: List[float],
    user_id: str,
    transcript_id: str,
    db: Session
) -> str:
    """
    Auto-save voiceprint non matché en status 'pending'.

    Returns:
        voiceprint_lib_id
    """
    # Generate text voiceprint (1024d) from speaker segments
    speaker_segments = [
        seg for seg in segments
        if seg["speaker"] == speaker_label
    ]

    voiceprint_1024d = await self.embedding_service.mean_pool_speaker_segments(
        segments=speaker_segments,
        min_segment_length=10
    )

    # Create pending voiceprint
    voiceprint = VoiceprintLibrary(
        id=generate_uuid(),
        user_id=user_id,
        transcript_id=transcript_id,
        speaker_label=speaker_label,

        voiceprint_audio_512d=json.dumps(voiceprint_512d),
        audio_model="pyannote-audio",

        voiceprint_text_1024d=json.dumps(voiceprint_1024d),
        text_model="BAAI/bge-m3",

        status="pending",
        identified_name=None,
        match_source="unknown",

        first_seen_at=datetime.utcnow(),
        last_seen_at=datetime.utcnow(),

        created_at=unix_timestamp(),
        updated_at=unix_timestamp()
    )

    db.add(voiceprint)
    db.commit()

    logger.info(
        f"DEBUG: Auto-saved pending voiceprint - "
        f"ID: {voiceprint.id}, speaker: {speaker_label}"
    )

    return voiceprint.id

async def confirm_pending_voiceprint(
    self,
    voiceprint_lib_id: str,
    identified_name: str,
    match_source: str,
    db: Session
):
    """
    Confirmer voiceprint pending après identification LLM.
    """
    voiceprint = db.query(VoiceprintLibrary).filter(
        VoiceprintLibrary.id == voiceprint_lib_id
    ).first()

    if not voiceprint:
        raise ValueError(f"Voiceprint {voiceprint_lib_id} not found")

    voiceprint.status = "confirmed"
    voiceprint.identified_name = identified_name
    voiceprint.match_source = match_source
    voiceprint.last_seen_at = datetime.utcnow()
    voiceprint.updated_at = unix_timestamp()

    db.commit()

    logger.info(
        f"DEBUG: Confirmed voiceprint {voiceprint_lib_id} - "
        f"Name: {identified_name}, source: {match_source}"
    )