🔧(summary) fix featureflag summary and adding variables in settings

fixing the checking of the summary-enabled featureflag
cleaning the prints
adding queues name in the settings
renaming the transcribe worker to original name
removing useless variable declarations

src/summary/compose.yaml

@@ -17,9 +17,18 @@ services:
     depends_on:
       - redis
   celery_worker:
-    container_name: celery_worker
+    container_name: celery_worker_transcribe
     build: .
-    command: celery -A summary.core.celery_worker worker --pool=solo --loglevel=debug
+    command: celery -A summary.core.celery_worker worker --pool=solo --loglevel=debug -Q transcribe_queue
+    volumes:
+      - .:/app
+    depends_on:
+      - redis
+      - app
+  celery_summarize_worker:
+    container_name: celery_worker_summarize
+    build: .
+    command: celery -A summary.core.celery_worker worker --pool=solo --loglevel=debug -Q summarize_queue
     volumes:
       - .:/app
     depends_on:

src/summary/summary/api/route/tasks.py

@@ -6,12 +6,13 @@ from typing import Optional
 from celery.result import AsyncResult
 from fastapi import APIRouter
 from pydantic import BaseModel
-
+from summary.core.config import get_settings
 from summary.core.celery_worker import (
     process_audio_transcribe_summarize,
     process_audio_transcribe_summarize_v2,
 )
 
+settings = get_settings()
 
 class TaskCreation(BaseModel):
     """Task data."""
@@ -45,7 +46,8 @@ async def create_task(request: TaskCreation):
                 request.room,
                 request.recording_date,
                 request.recording_time,
-            ]
+            ],           
+            queue=settings.transcribe_queue,
         )
 
     return {"id": task.id, "message": "Task created"}

src/summary/summary/core/analytics.py

@@ -48,6 +48,24 @@ class Analytics:
         except Exception as e:
             raise AnalyticsException("Failed to capture analytics event") from e
 
+    def is_feature_enabled(
+        self, feature_name: str, distinct_id: str = None
+    ) -> bool:
+        """
+        Check if a feature flag is enabled in PostHog for a dinstinct_id.
+        """
+        if self.is_disabled:
+            return False
+
+        try:
+            logger.info(
+                f"Check feature flag {feature_name} for user {distinct_id}"
+            )
+            return self._client.feature_enabled(feature_name, distinct_id)
+        except Exception as e:
+            logger.error(f"Error checking feature flag {feature_name}: {e}")
+            return False
+
 
 @lru_cache
 def get_analytics():

src/summary/summary/core/celery_app.py

@@ -0,0 +1,23 @@
+from celery import Celery
+from celery import signals
+import sentry_sdk
+from summary.core.config import get_settings
+import summary.core.celery_signals
+
+settings = get_settings()
+
+celery = Celery(
+    __name__,
+    broker=settings.celery_broker_url,
+    backend=settings.celery_result_backend,
+    broker_connection_retry_on_startup=True,
+)
+
+celery.config_from_object("summary.core.celery_config")
+
+if settings.sentry_dsn and settings.sentry_is_enabled:
+
+    @signals.celeryd_init.connect
+    def init_sentry(**_kwargs):
+        """Initialize sentry."""
+        sentry_sdk.init(dsn=settings.sentry_dsn, enable_tracing=True)

src/summary/summary/core/celery_signals.py

@@ -0,0 +1,26 @@
+from summary.core.analytics import MetadataManager, get_analytics
+from celery import signals
+from summary.core.config import get_settings
+
+analytics = get_analytics()
+settings = get_settings()
+metadata_manager = MetadataManager()
+
+
+@signals.task_prerun.connect
+def task_started(task_id=None, task=None, args=None, **kwargs):
+    """Signal handler called before task execution begins."""
+    task_args = args or []
+    metadata_manager.create(task_id, task_args)
+
+
+@signals.task_retry.connect
+def task_retry_handler(request=None, reason=None, einfo=None, **kwargs):
+    """Signal handler called when task execution retries."""
+    metadata_manager.retry(request.id)
+
+
+@signals.task_failure.connect
+def task_failure_handler(task_id, exception=None, **kwargs):
+    """Signal handler called when task execution fails permanently."""
+    metadata_manager.capture(task_id, settings.posthog_event_failure)

src/summary/summary/core/celery_summarize_worker.py

@@ -0,0 +1,123 @@
+"""Celery summarize workers."""
+
+from celery import Celery, signals
+from celery.utils.log import get_task_logger
+import openai
+from requests import Session
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+
+from summary.core.prompt import (
+    PROMPT_SYSTEM_PLAN,
+    PROMPT_SYSTEM_TLDR,
+    PROMPT_SYSTEM_PART,
+    PROMPT_USER_PART,
+    PROMPT_SYSTEM_CLEANING,
+    PROMPT_SYSTEM_NEXT_STEP,
+)
+from summary.core.config import get_settings
+from summary.core.celery_app import celery
+
+settings = get_settings()
+logger = get_task_logger(__name__)
+
+
+def create_retry_session():
+    """Create an HTTP session configured with retry logic."""
+    session = Session()
+    retries = Retry(
+        total=settings.webhook_max_retries,
+        backoff_factor=settings.webhook_backoff_factor,
+        status_forcelist=settings.webhook_status_forcelist,
+        allowed_methods={"POST"},
+    )
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    return session
+
+
+def post_with_retries(url, data):
+    """Send POST request with automatic retries."""
+    session = create_retry_session()
+    session.headers.update({"Authorization": f"Bearer {settings.webhook_api_token}"})
+    try:
+        response = session.post(url, json=data)
+        response.raise_for_status()
+        return response
+    finally:
+        session.close()
+
+
+def LLM_call(client, system_prompt, user_prompt, retry=2):
+
+    data = {
+        "model": settings.resume_llm_model,
+        "messages": [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_prompt},
+        ],
+    }
+
+    try:
+        response = client.chat.completions.create(**data)
+        return response.choices[0].message.content
+    except Exception as e:
+        logger.error("LLM call failed: %s", e)
+        return False
+
+
+@celery.task(
+    bind=True, autoretry_for=[Exception], max_retries=3, queue=settings.summarize_queue
+)
+def summarize_transcription(self, transcript: str, email: str, sub: str, title: str):
+    logger.info("Starting summarization task")
+
+    logger.info("Initiating summarize client")
+
+    client_summary = openai.OpenAI(
+        base_url=settings.resume_endpoint, api_key=settings.resume_api_key
+    )
+
+    tldr = LLM_call(client_summary, PROMPT_SYSTEM_TLDR, transcript)
+
+    logger.info("TLDR generated")
+
+    parts = LLM_call(client_summary, PROMPT_SYSTEM_PLAN, transcript)
+    logger.info("Plan generated")
+
+    parts = parts.split("\n")
+    parts = [x for x in parts if x.strip() != ""]
+    logger.info("Empty parts removed")
+
+    parts_summarized = []
+    for part in parts:
+        prompt_user_part = PROMPT_USER_PART.format(part=part, transcript=transcript)
+        logger.info("Summarizing part: %s", part)
+        parts_summarized.append(
+            LLM_call(
+                client_summary, PROMPT_SYSTEM_PART, prompt_user_part.format(part=part)
+            )
+        )
+
+    logger.info("Parts summarized")
+
+    raw_summary = "\n\n".join(parts_summarized)
+
+    next_steps = LLM_call(client_summary, PROMPT_SYSTEM_NEXT_STEP, transcript)
+    logger.info("Next steps generated")
+    cleaned_summary = LLM_call(client_summary, PROMPT_SYSTEM_CLEANING, raw_summary)
+    logger.info("Summary cleaned")
+    summary = tldr + "\n\n" + cleaned_summary + "\n\n" + next_steps
+
+    data = {
+        "title": title + " - Summary",
+        "content": summary,
+        "email": email,
+        "sub": sub,
+    }
+
+    logger.debug("Submitting webhook to %s", settings.webhook_url)
+
+    response = post_with_retries(settings.webhook_url, data)
+
+    logger.info("Webhook submitted successfully. Status: %s", response.status_code)
+    logger.debug("Response body: %s", response.text)

src/summary/summary/core/celery_worker.py

@@ -19,33 +19,18 @@ from requests import Session, exceptions
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
-from summary.core.analytics import MetadataManager, get_analytics
 from summary.core.config import get_settings
 from summary.core.prompt import get_instructions
+from summary.core.celery_summarize_worker import summarize_transcription
+from summary.core.celery_app import celery
+from summary.core.celery_signals import metadata_manager
+from summary.core.analytics import get_analytics
 
 settings = get_settings()
 analytics = get_analytics()
 
-metadata_manager = MetadataManager()
-
 logger = get_task_logger(__name__)
 
-celery = Celery(
-    __name__,
-    broker=settings.celery_broker_url,
-    backend=settings.celery_result_backend,
-    broker_connection_retry_on_startup=True,
-)
-
-celery.config_from_object("summary.core.celery_config")
-
-if settings.sentry_dsn and settings.sentry_is_enabled:
-
-    @signals.celeryd_init.connect
-    def init_sentry(**_kwargs):
-        """Initialize sentry."""
-        sentry_sdk.init(dsn=settings.sentry_dsn, enable_tracing=True)
-
 
 DEFAULT_EMPTY_TRANSCRIPTION = """
 **Aucun contenu audio n’a été détecté dans votre transcription.**
@@ -137,25 +122,6 @@ def post_with_retries(url, data):
         session.close()
 
 
-@signals.task_prerun.connect
-def task_started(task_id=None, task=None, args=None, **kwargs):
-    """Signal handler called before task execution begins."""
-    task_args = args or []
-    metadata_manager.create(task_id, task_args)
-
-
-@signals.task_retry.connect
-def task_retry_handler(request=None, reason=None, einfo=None, **kwargs):
-    """Signal handler called when task execution retries."""
-    metadata_manager.retry(request.id)
-
-
-@signals.task_failure.connect
-def task_failure_handler(task_id, exception=None, **kwargs):
-    """Signal handler called when task execution fails permanently."""
-    metadata_manager.capture(task_id, settings.posthog_event_failure)
-
-
 @celery.task(max_retries=settings.celery_max_retries)
 def process_audio_transcribe_summarize(filename: str, email: str, sub: str):
     """Process an audio file by transcribing it and generating a summary.
@@ -209,7 +175,7 @@ def process_audio_transcribe_summarize(filename: str, email: str, sub: str):
 
     instructions = get_instructions(transcription)
     summary_response = openai_client.chat.completions.create(
-        model=settings.openai_llm_model, messages=instructions
+        model=settings.resume_llm_model, messages=instructions
     )
 
     summary = summary_response.choices[0].message.content
@@ -236,6 +202,7 @@ def process_audio_transcribe_summarize(filename: str, email: str, sub: str):
     bind=True,
     autoretry_for=[exceptions.HTTPError],
     max_retries=settings.celery_max_retries,
+    queue=settings.transcribe_queue,
 )
 def process_audio_transcribe_summarize_v2(
     self,
@@ -353,5 +320,10 @@ def process_audio_transcribe_summarize_v2(
     logger.debug("Response body: %s", response.text)
 
     metadata_manager.capture(task_id, settings.posthog_event_success)
-
-    # TODO - integrate summarize the transcript and create a new document.
+    if not analytics.is_feature_enabled("summary-enabled", distinct_id=sub):
+        logger.info("Summary generation skipped (feature flag disabled).")
+    else:
+        logger.info("Queuing summary generation task.")
+        summarize_transcription.apply_async(
+            args=[formatted_transcription, email, sub, title], queue=settings.summarize_queue
+        )

src/summary/summary/core/config.py

@@ -35,8 +35,10 @@ class Settings(BaseSettings):
     openai_api_key: str
     openai_base_url: str = "https://api.openai.com/v1"
     openai_asr_model: str = "whisper-1"
-    openai_llm_model: str = "gpt-4o"
+    resume_llm_model: str = "Qwen/Qwen2.5-Coder-32B-Instruct-AWQ"
     openai_max_retries: int = 0
+    resume_api_key: str
+    resume_endpoint: str
 
     # Webhook-related settings
     webhook_max_retries: int = 2
@@ -66,7 +68,9 @@ class Settings(BaseSettings):
     task_tracker_redis_url: str = "redis://redis/0"
     task_tracker_prefix: str = "task_metadata:"
 
-
+    # Queue redis
+    summarize_queue: str = "summarize_queue"
+    transcribe_queue: str = "transcribe_queue"
 @lru_cache
 def get_settings():
     """Load and cache application settings."""

src/summary/summary/core/prompt.py

@@ -50,3 +50,34 @@ def get_instructions(transcript):
         },
         {"role": "user", "content": prompt},
     ]
+
+
+PROMPT_SYSTEM_TLDR = """Tu es un agent dont le rôle est de créer un TL;DR (résumé très concis) d'un compte rendu de réunion. Tu utilisera un style synthétique administratif, à la troisième personne, sans affect. Tu recevras en entrée le transcript. Ta tâche est de rédiger un résumé concis et structuré, en te concentrant uniquement sur les informations essentielles et pertinentes. Tu répondras en un paragraphe structuré (3 à 6 phrases), sans rien ajouter d'autre. Tu répondras dans le format suivant sans rien ajouter d'autre:
+### Résumé TL;DR
+
+[Résumé concis et structuré]"""
+
+PROMPT_SYSTEM_PLAN = """Ta tâche est de diviser le contenu du transcript en sujets concrets correspondant aux grands axes discutés durant la réunion. Ne crée pas de catégories génériques. Les titres doivent être courts, précis et représentatifs des échanges. Veille à ce que chaque sujet soit distinct et qu’aucun thème ne soit répété. Tu te limitera à 5 ou 6 sujets maximum.
+L'introducion, ordre du jour, conclusion etc seront rajoutés a posteriori. Tu répondra dans le format suivant sans rien ajouter d'autre:
+"Titre du sujet 1
+Titre du sujet 2
+Titre du sujet 3
+..."
+"""
+
+PROMPT_SYSTEM_PART = """Tu es un agent dont le rôle est de créer une partie du résumé d'un compte rendu de réunion. Tu utilisera un style synthétique administratif, à la troisième personne, sans affect. Tu recevras en entrée le transcript, et le titre du sujet correspondant. Ta tâche est de rédiger un résumé concis de cette partie et uniquement cette partie, en te concentrant uniquement sur les informations essentielles et pertinentes. Le résumé de chaque partie doit tenir en 4 à 6 phrases maximum, sans entrer dans les détails mineurs. Tu répondra dans le format suivant :
+    ### Titre du sujet [Traduire ce titre selon la langue du transcript]
+    [Résumé concis et structuré de la partie du transcript]
+    """
+
+PROMPT_USER_PART = """Titre de la partie à résumer : {part}
+
+Transcript complet :
+{transcript}"""
+
+PROMPT_SYSTEM_CLEANING = """Tu es un agent dont le rôle est de nettoyer un résumé de compte rendu de réunion. Tu recevras en entrée le résumé brut, potentiellement avec des erreurs de formatage, des incohérences ou des redondances. Ta tâche est de corriger les erreurs de formatage, d'améliorer la clarté et la cohérence du texte, et de t'assurer que le résumé est bien structuré et facile à lire. Ton but principal est de retirer les redondances et les répétitions. Assure la cohérence entre les titres, et homogénéise le style d’écriture entre les parties. Supprime les doublons d’informations entre les parties si présents. Si certaines parties sont plus secondaires, tu peux les fusionner ou les réduire en 1 à 2 phrases. Mets en avant les points centraux qui ont fait l’objet de décisions ou d’actions. Tu répondra uniquement avec le résumé sans rien ajouter d'autre"""
+
+PROMPT_SYSTEM_NEXT_STEP = """Tu es un agent dont le rôle est d'extraire les prochaines étapes d'un transcript de réunion. Tu utilisera un style synthétique administratif, à la troisième personne, sans affect. Tu recevras en entrée le transcript. Ta tâche est d'identifier et lister toutes les actions à entreprendre, en indiquant la ou les personnes assignées et en précisant les échéances si elles sont mentionnées. Ne retiens que les actions concrètes et à venir. Ignore les remarques générales ou les constats sans suite. Les actions doivent suivre ce format strict :
+### Prochaines étapes
+
+- [ ] [Action à effectuer] Assignée à : [Nom], Échéance : [Date si mentionnée]"""