Performance & Optimisation¶

# Match first segment only
segment_text = segments[0]["transcription"]
embedding = model.encode(segment_text)  # Shape: (1024,)
search_results = qdrant.search(embedding)

Accuracy: 45%  (faible)
Time: 2s per speaker

# Pool ALL speaker segments
texts = [seg["transcription"] for seg in segments]
embeddings = model.encode(texts)  # Shape: (N, 1024)

# Mean pooling + L2 normalization
pooled = np.mean(embeddings, axis=0)
normalized = pooled / np.linalg.norm(pooled)

search_results = qdrant.search(normalized)

Accuracy: 78-82%  (+37%)
Time: 3s per speaker  (+1s, acceptable)

# Search all chunks, filter in Python
results = qdrant.search(
    collection_name=collection_name,
    query_vector=embedding,
    limit=20,
    score_threshold=0.0
)

# Filter in Python
filtered = [
    r for r in results
    if speaker_name in r.payload.get("all_participants", [])
]

Time: 10-15ms (fetch 20) + 5ms (filter) = 15-20ms

# Filter in Qdrant (native)
results = qdrant.search(
    collection_name=collection_name,
    query_vector=embedding,
    limit=3,
    score_threshold=0.5,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(
                key="all_participants",
                match=models.MatchAny(any=[speaker_name])
            )
        ]
    )
)

Time: 3-5ms (fetch + filter native)  (-70% latency)

class EmbeddingService:
    _model_cache = None

    @classmethod
    def get_model(cls):
        """
        Singleton pattern pour BGE-M3.

        Without caching:
            Model load time: 8-10s per request (lent)
            RAM: 1.2 GB × N requests = OOM

        With caching:
            Model load time: 8-10s once (optimal)
            RAM: 1.2 GB total (shared)
        """
        if cls._model_cache is None:
            cls._model_cache = FlagModel(
                "BAAI/bge-m3",
                use_fp16=True,  # Half precision (saves 50% RAM)
                cache_dir="/app/.cache/models"
            )
            logger.info("DEBUG: BGE-M3 model loaded and cached")

        return cls._model_cache

class MetadataExtractor:
    def __init__(self):
        self.cache = {}  # In-memory cache
        self.redis = redis.Redis()  # Persistent cache

    async def extract_metadata(
        self,
        text: str,
        cache_ttl: int = 7 * 24 * 3600  # 7 days
    ) -> Dict[str, Any]:
        """
        Cache LLM metadata par hash de texte.

        Scenario: Same CV uploaded for multiple transcripts
        Without caching: 3 API calls × $0.002 = $0.006
        With caching: 1 API call = $0.002 (66% savings)
        """
        # Generate cache key (text hash)
        cache_key = f"llm_metadata:{hashlib.sha256(text.encode()).hexdigest()}"

        # Check cache
        cached = await self.redis.get(cache_key)
        if cached:
            logger.info("DEBUG: LLM metadata cache HIT")
            return json.loads(cached)

        # Cache miss: Call LLM
        logger.info("DEBUG: LLM metadata cache MISS - calling API")
        metadata = await self._extract_with_llm(text)

        # Store in cache
        await self.redis.setex(
            cache_key,
            cache_ttl,
            json.dumps(metadata)
        )

        return metadata

# src/db.py
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,          # Max 10 concurrent connections
    max_overflow=20,       # +20 overflow (peak traffic)
    pool_timeout=30,       # Wait 30s for connection
    pool_recycle=3600,     # Recycle connections after 1h
    pool_pre_ping=True,    # Test connection before use
    echo=False             # No SQL logging (production)
)

# Performance logging
@log_performance
async def process_transcript(transcript_id: str, db: Session):
    start_time = time.time()

    try:
        # Phase 1: RAG Indexing
        phase_start = time.time()
        await rag_indexing(transcript_id, db)
        logger.info(
            f"PERF: RAG indexing - "
            f"Transcript: {transcript_id}, "
            f"Duration: {time.time() - phase_start:.2f}s"
        )

        # Phase 2: GPU Transcription
        phase_start = time.time()
        await gpu_transcription(transcript_id, db)
        logger.info(
            f"PERF: GPU transcription - "
            f"Transcript: {transcript_id}, "
            f"Duration: {time.time() - phase_start:.2f}s"
        )

        # Phase 3: Post-Processing
        phase_start = time.time()
        await post_processing(transcript_id, db)
        logger.info(
            f"PERF: Post-processing - "
            f"Transcript: {transcript_id}, "
            f"Duration: {time.time() - phase_start:.2f}s"
        )

        # Total
        logger.info(
            f"PERF: Total processing - "
            f"Transcript: {transcript_id}, "
            f"Duration: {time.time() - start_time:.2f}s"
        )

    except Exception as e:
        logger.error(
            f"PERF: Processing failed - "
            f"Transcript: {transcript_id}, "
            f"Duration: {time.time() - start_time:.2f}s, "
            f"Error: {str(e)}"
        )
        raise