From 486c7371568183b520fe5fc09ef0e266600dda71 Mon Sep 17 00:00:00 2001 From: Martin Guitteny <“martin.guitteny@centralesupelec.fr”> Date: Fri, 5 Sep 2025 16:58:52 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=A7(summary)=20fix=20featureflag=20sum?= =?UTF-8?q?mary=20and=20adding=20variables=20in=20settings=20fixing=20the?= =?UTF-8?q?=20checking=20of=20the=20summary-enabled=20featureflag=20cleani?= =?UTF-8?q?ng=20the=20prints=20adding=20queues=20name=20in=20the=20setting?= =?UTF-8?q?s=20renaming=20the=20transcribe=20worker=20to=20original=20name?= =?UTF-8?q?=20removing=20useless=20variable=20declarations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/summary/compose.yaml | 2 +- src/summary/summary/api/route/tasks.py | 5 +-- src/summary/summary/core/analytics.py | 22 ++++--------- .../summary/core/celery_summarize_worker.py | 32 ++++++------------- src/summary/summary/core/celery_worker.py | 10 +++--- src/summary/summary/core/config.py | 6 ++-- 6 files changed, 28 insertions(+), 49 deletions(-) diff --git a/src/summary/compose.yaml b/src/summary/compose.yaml index 1bbed33f..75490b81 100644 --- a/src/summary/compose.yaml +++ b/src/summary/compose.yaml @@ -16,7 +16,7 @@ services: ".env" depends_on: - redis - celery_transcribe_worker: + celery_worker: container_name: celery_worker_transcribe build: . command: celery -A summary.core.celery_worker worker --pool=solo --loglevel=debug -Q transcribe_queue diff --git a/src/summary/summary/api/route/tasks.py b/src/summary/summary/api/route/tasks.py index e9490f7b..31eed711 100644 --- a/src/summary/summary/api/route/tasks.py +++ b/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.""" @@ -46,7 +47,7 @@ async def create_task(request: TaskCreation): request.recording_date, request.recording_time, ], - queue="transcribe_queue", + queue=settings.transcribe_queue, ) return {"id": task.id, "message": "Task created"} diff --git a/src/summary/summary/core/analytics.py b/src/summary/summary/core/analytics.py index 9dd360b6..5ed9e18d 100644 --- a/src/summary/summary/core/analytics.py +++ b/src/summary/summary/core/analytics.py @@ -48,32 +48,22 @@ class Analytics: except Exception as e: raise AnalyticsException("Failed to capture analytics event") from e - def feature_enabled( - self, feature_name: str, distinct_id: str = None, email: str = None + def is_feature_enabled( + self, feature_name: str, distinct_id: str = None ) -> bool: """ - Vérifie si un feature flag est activé dans PostHog. - distinct_id optionnel pour les flags ciblés. + Check if a feature flag is enabled in PostHog for a dinstinct_id. """ if self.is_disabled: return False - if email: - self.capture( - "user_identified", - distinct_id=distinct_id, - properties={"$set": {"email": email}}, - ) - logger.info(f"[Analytics] Identify user {distinct_id} with email {email}") try: logger.info( - f"[Analytics] Check feature flag {feature_name} for user {distinct_id}" + f"Check feature flag {feature_name} for user {distinct_id}" ) - return self._client.is_feature_enabled(feature_name, distinct_id) + return self._client.feature_enabled(feature_name, distinct_id) except Exception as e: - logger.warning( - f"[Analytics] Impossible de vérifier le flag {feature_name}: {e}" - ) + logger.error(f"Error checking feature flag {feature_name}: {e}") return False diff --git a/src/summary/summary/core/celery_summarize_worker.py b/src/summary/summary/core/celery_summarize_worker.py index 68664cc3..7918abfb 100644 --- a/src/summary/summary/core/celery_summarize_worker.py +++ b/src/summary/summary/core/celery_summarize_worker.py @@ -50,7 +50,7 @@ def post_with_retries(url, data): def LLM_call(client, system_prompt, user_prompt, retry=2): data = { - "model": "Qwen/Qwen2.5-Coder-32B-Instruct-AWQ", + "model": settings.resume_llm_model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, @@ -59,38 +59,29 @@ def LLM_call(client, system_prompt, user_prompt, retry=2): try: response = client.chat.completions.create(**data) - print(response) return response.choices[0].message.content except Exception as e: - if retry > 0: - print(f"Erreur lors de l'appel à l'API : {e}. Nouvelle tentative...") - return LLM_call(system_prompt, user_prompt, retry=retry - 1) - else: - print(f"Erreur lors de l'appel à l'API : {e}") - return False + logger.error("LLM call failed: %s", e) + return False @celery.task( - bind=True, autoretry_for=[Exception], max_retries=3, queue="summarize_queue" + 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") - prompt_system_TLDR = PROMPT_SYSTEM_TLDR - prompt_system_plan = PROMPT_SYSTEM_PLAN - prompt_system_part = PROMPT_SYSTEM_PART - logger.info("Initiating summarize client") client_summary = openai.OpenAI( base_url=settings.resume_endpoint, api_key=settings.resume_api_key ) - out = LLM_call(client_summary, prompt_system_TLDR, transcript) + tldr = LLM_call(client_summary, PROMPT_SYSTEM_TLDR, transcript) logger.info("TLDR generated") - parts = LLM_call(client_summary, prompt_system_plan, transcript) + parts = LLM_call(client_summary, PROMPT_SYSTEM_PLAN, transcript) logger.info("Plan generated") parts = parts.split("\n") @@ -103,7 +94,7 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title: logger.info("Summarizing part: %s", part) parts_summarized.append( LLM_call( - client_summary, prompt_system_part, prompt_user_part.format(part=part) + client_summary, PROMPT_SYSTEM_PART, prompt_user_part.format(part=part) ) ) @@ -111,14 +102,11 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title: raw_summary = "\n\n".join(parts_summarized) - prompt_system_next_steps = PROMPT_SYSTEM_NEXT_STEP - prompt_system_cleaning = PROMPT_SYSTEM_CLEANING - - next_steps = LLM_call(client_summary, prompt_system_next_steps, transcript) + 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) + cleaned_summary = LLM_call(client_summary, PROMPT_SYSTEM_CLEANING, raw_summary) logger.info("Summary cleaned") - summary = out + "\n\n" + cleaned_summary + "\n\n" + next_steps + summary = tldr + "\n\n" + cleaned_summary + "\n\n" + next_steps data = { "title": title + " - Summary", diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index e6ff2fa5..e00e00e6 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -175,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 @@ -202,7 +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="transcribe_queue", + queue=settings.transcribe_queue, ) def process_audio_transcribe_summarize_v2( self, @@ -320,12 +320,10 @@ def process_audio_transcribe_summarize_v2( logger.debug("Response body: %s", response.text) metadata_manager.capture(task_id, settings.posthog_event_success) - - if not analytics.feature_enabled("summary-enabled", distinct_id=sub, email=email): - print("Summary generation skipped (feature flag disabled).") + 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="summarize_queue" + args=[formatted_transcription, email, sub, title], queue=settings.summarize_queue ) diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index 75e96d3b..149c9749 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -35,7 +35,7 @@ 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 @@ -68,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."""