Pipeline de Transcription - Workflow Complet¶

# 1. Validate audio format
allowed_formats = ['.mp3', '.wav', '.m4a', '.flac', '.ogg', '.mp4', '.mkv']
if not audio_file.filename.endswith(tuple(allowed_formats)):
    raise HTTPException(400, "Unsupported audio format")

# 2. Upload to S3
s3_key = f"users/{user_id}/audio/{transcript_id}/{filename}"
s3_url = await s3_client.upload_file(audio_file, s3_key)

# 3. Create transcript record
transcript = Transcript(
    id=transcript_id,
    user_id=user_id,
    title="Réunion Panel Citoyen",
    status="processing",
    audio_url=s3_url,
    created_at=unix_timestamp()
)
db.add(transcript)
db.commit()

# Collection naming: isolation par user + transcript
collection_name = f"user_{user_id}_transcript_{transcript_id}"

# Create with BGE-M3 dimensions
await qdrant_service.create_collection(
    collection_name=collection_name,
    vector_size=1024,
    distance=Distance.COSINE
)

# Save collection reference
transcript.qdrant_collection_name = collection_name
db.commit()

# 1. Extract text
text = await text_extraction_service.extract(file, file_type="pdf")
# Output: "Jean Dupont - Lead Developer\nEmail: jean@company.com\n..."

# 2. LLM Metadata Extraction (AVANT chunking)
metadata = await llm_metadata_extractor.extract(text, filename)
# Output: ParticipantMetadata(
#   name="Jean Dupont",
#   role="Lead Developer",
#   email="jean@company.com",
#   company="TechCorp"
# )

# 3. Semantic Chunking
chunks = await semantic_chunking_service.chunk(
    text=text,
    metadata=metadata,
    max_chunk_size=2000
)
# Output: [
#   {"text": "Jean Dupont - Lead...", "chunk_index": 0, "metadata": {...}},
#   {"text": "Responsibilities: ...", "chunk_index": 1, "metadata": {...}}
# ]

# 4. Generate Embeddings
chunks_with_embeddings = await embedding_service.encode_batch(chunks)
# Adds: "embedding": [0.023, -0.156, ..., 0.012] (1024d)

# 5. Analyze Chunks (per-chunk metadata)
for chunk in chunks_with_embeddings:
    chunk["mentioned_participants"] = extract_mentioned(chunk["text"], metadata)
    chunk["chunk_type"] = classify_chunk_type(chunk["text"], metadata)
    chunk["keyword_associations"] = extract_keywords(chunk["text"])

# 6. Upsert to Qdrant
await qdrant_service.upsert_chunks(
    collection_name=collection_name,
    chunks=chunks_with_embeddings
)

# BFF calls MeetNoo
response = await http_client.post(
    f"{MEETNOO_SERVICES_URL}/api/v1/pipeline/start",
    headers={"X-Pipeline-Key": PIPELINE_API_KEY},
    json={
        "tenant_id": user_id,
        "file_url": s3_key,  # or full s3:// URL
        "project_id": transcript_id,
        "config": {
            "chunk_duration_seconds": 900,
            "similarity_threshold": 0.75,
            "transcription_model": "large-v3",
            "llm_model": "Qwen/Qwen2.5-3B-Instruct"
        }
    },
    timeout=10
)

meetnoo_transcript_id = response.json()["transcription_id"]

# Save mapping
transcript.meetnoo_transcript_id = meetnoo_transcript_id
db.commit()

# BFF consumes events in background task
async def consume_pipeline_events():
    while True:
        messages = redis_client.xreadgroup(
            groupname="smart-trans-group",
            consumername="consumer-1",
            streams={"pipeline:events": ">"},
            count=10,
            block=5000
        )

        for stream, messages_list in messages:
            for message_id, data in messages_list:
                txn_id = data[b'txn_id'].decode()
                stage = data[b'stage'].decode()
                status = data[b'status'].decode()
                progress = int(data[b'progress'].decode())

                # Map meetnoo_transcript_id to our transcript_id
                transcript = db.query(Transcript).filter(
                    Transcript.meetnoo_transcript_id == txn_id
                ).first()

                if not transcript:
                    continue

                # Update progress in DB
                await update_transcript_progress(
                    db,
                    transcript.id,
                    progress,
                    f"gpu_{stage}",
                    f"GPU Processing: {stage}",
                    status=status
                )

                # Trigger post-processing when completed
                if stage == "pipeline" and status == "completed":
                    await trigger_post_processing(transcript.id)

                # ACK message
                redis_client.xack("pipeline:events", "smart-trans-group", message_id)

async def post_process_transcript(transcript_id: str, db: Session):
    """
    Post-processing après GPU transcription.

    Workflow:
    1. Fetch results from MeetNoo
    2. For each speaker:
       - Priority 1: Voiceprint matching
       - Priority 2: RAG enrichment/extraction
       - Priority 3: LLM processing
    3. Save enriched segments
    """
    # Fetch MeetNoo results
    meetnoo_results = await fetch_meetnoo_results(transcript.meetnoo_transcript_id)

    # Initialize speaker identification
    voiceprint_results = {}
    rag_context_global = {
        "all_participants": [],
        "keywords": [],
        "glossary_terms": []
    }

    # Group segments by speaker
    segments_by_speaker = group_by_speaker(meetnoo_results["segments"])

    # Process each speaker
    for speaker_label, speaker_segments in segments_by_speaker.items():
        speaker_voiceprint = get_speaker_voiceprint(
            meetnoo_results["speakers"],
            speaker_label
        )

        # PRIORITY 1: Voiceprint Matching
        match_result = await voiceprint_matcher.match_speaker(
            voiceprint_audio_512d=speaker_voiceprint,
            user_id=transcript.user_id,
            db=db
        )

        if match_result:
            # Match found
            voiceprint_results[speaker_label] = match_result
            is_identified = True
            identified_name = match_result["identified_name"]
        else:
            # No match - auto-save pending
            voiceprint_lib_id = await auto_save_pending_voiceprint(
                speaker_label=speaker_label,
                voiceprint_512d=speaker_voiceprint,
                user_id=transcript.user_id,
                transcript_id=transcript_id,
                db=db
            )
            is_identified = False
            identified_name = f"Intervenant {speaker_label.split('_')[1]}"

        # PRIORITY 2: RAG Enrichment/Extraction
        rag_results = await rag_service.process_speaker(
            speaker_segments=speaker_segments,
            collection_name=transcript.qdrant_collection_name,
            is_identified=is_identified,
            identified_name=identified_name if is_identified else None
        )

        if is_identified:
            # Enrich metadata
            voiceprint_results[speaker_label].update(rag_results["enrichment"])
        else:
            # Extract context for LLM
            rag_context_global["all_participants"].extend(
                rag_results["extraction"]["participants"]
            )
            rag_context_global["keywords"].extend(
                rag_results["extraction"]["keywords"]
            )
            rag_context_global["glossary_terms"].extend(
                rag_results["extraction"]["glossary_terms"]
            )

    # PRIORITY 3: LLM Processing

    # 3.1 Clean transcription (ALL segments)
    clean_transcription await llm_post_processor.clean_transcription(
        raw_transcription={seg["speaker"]: seg["text"] for seg in meetnoo_results["segments"]},
        participants=rag_context_global["all_participants"],
        keywords=rag_context_global["keywords"],
        language="fr"
    )

    # 3.2 Identify speakers (ONLY pending)
    pending_speakers = {
        label: segments
        for label, result in voiceprint_results.items()
        if result.get("match_source") == "pending"
    }

    if pending_speakers:
        speaker_identifications = await llm_post_processor.identify_speakers(
            unidentified_segments={
                label: " ".join([seg["text"] for seg in segments])
                for label, segments in pending_speakers.items()
            },
            potential_participants=rag_context_global["all_participants"],
            keywords=rag_context_global["keywords"],
            glossary_terms=rag_context_global["glossary_terms"]
        )

        # Confirm pending voiceprints
        for speaker_label, identification in speaker_identifications.items():
            if identification["confidence"] > 0.75:
                await confirm_pending_voiceprint(
                    voiceprint_lib_id=voiceprint_results[speaker_label]["voiceprint_lib_id"],
                    identified_name=identification["identified_name"],
                    db=db
                )
                voiceprint_results[speaker_label].update({
                    "identified_name": identification["identified_name"],
                    "match_source": "llm_inference",
                    "match_confidence": identification["confidence"]
                })

    # Save enriched segments
    enriched_segments = []
    for segment in meetnoo_results["segments"]:
        speaker_label = segment["speaker"]
        speaker_info = voiceprint_results.get(speaker_label, {})

        enriched_segment = EnrichedSegment(
            id=generate_uuid(),
            transcript_id=transcript_id,
            segment_index=segment["index"],
            speaker_label=speaker_label,
            identified_name=speaker_info.get("identified_name"),
            role=speaker_info.get("role"),
            voiceprint_lib_id=speaker_info.get("voiceprint_lib_id"),
            text=segment["text"],
            clean_text=clean_transcription.get(speaker_label),
            start_ms=int(segment["start"] * 1000),
            end_ms=int(segment["end"] * 1000),
            confidence=speaker_info.get("match_confidence", 0.0),
            match_source=speaker_info.get("match_source", "pending"),
            context_used=True,
            metadata={
                "role": speaker_info.get("role"),
                "specialties": speaker_info.get("specialties"),
                "source_document": speaker_info.get("source_document")
            },
            created_at=unix_timestamp(),
            updated_at=unix_timestamp()
        )
        enriched_segments.append(enriched_segment)

    db.bulk_save_objects(enriched_segments)

    # Update transcript status
    transcript.status = "completed"
    transcript.updated_at = unix_timestamp()
    db.commit()

    return {
        "status": "completed",
        "segments_count": len(enriched_segments),
        "speakers_identified": len([r for r in voiceprint_results.values() if r.get("match_source") != "pending"])
    }

# INSERT enriched_segments batch (33 records)
db.bulk_save_objects(enriched_segments)
db.commit()

logger.info(f"Saved {len(enriched_segments)} enriched segments for transcript {transcript_id}")

transcript.status = "completed"
transcript.total_segments = len(enriched_segments)
transcript.total_speakers = len(set([seg.speaker_label for seg in enriched_segments]))
transcript.duration_ms = meetnoo_results["duration_ms"]
transcript.updated_at = unix_timestamp()
db.commit()

# Emit SSE event to frontend
await sse_manager.emit(
    user_id=transcript.user_id,
    event="transcript.completed",
    data={
        "transcript_id": transcript_id,
        "status": "completed",
        "total_segments": transcript.total_segments,
        "total_speakers": transcript.total_speakers
    }
)

def update_transcript_progress(
    db: Session,
    transcript_id: str,
    percentage: int,
    step_key: str,
    step_name: str,
    substep_message: str = None,
    current_segment: int = None,
    total_segments: int = None
):
    """
    Update transcript progress (visible in frontend progress bar).
    """
    progress = TranscriptProgress(
        id=generate_uuid(),
        transcript_id=transcript_id,
        percentage=percentage,
        step_key=step_key,
        step_name=step_name,
        substep_message=substep_message,
        current_segment=current_segment,
        total_segments=total_segments,
        created_at=unix_timestamp()
    )
    db.add(progress)
    db.commit()

    # Emit SSE
    sse_manager.emit(
        user_id=transcript.user_id,
        event="transcript.progress",
        data={
            "transcript_id": transcript_id,
            "percentage": percentage,
            "step_name": step_name
        }
    )