wip align env variable in metadata agent

env.d/development/summary.dist

@@ -11,11 +11,10 @@ AWS_S3_SECRET_ACCESS_KEY="password"
 WHISPERX_BASE_URL="https://configure-your-url.com"
 WHISPERX_ASR_MODEL="large-v2"
 WHISPERX_API_KEY="your-secret-key"
-WHISPERX_DEFAULT_LANGUAGE="fr"
 
 LLM_BASE_URL="https://configure-your-url.com"
 LLM_API_KEY="dev-apikey"
-LLM_MODEL="albert-large"
+LLM_MODEL="Qwen/Qwen2.5-Coder-32B-Instruct-AWQ"
 
 WEBHOOK_API_TOKEN="secret"
 WEBHOOK_URL="https://configure-your-url.com"

src/agents/Dockerfile

@@ -21,4 +21,4 @@ COPY --from=builder /install /usr/local
 
 COPY . .
 
-CMD ["python", "multi-user-transcriber.py", "start"]
+CMD ["python", "multi_user_transcriber.py", "start"]

src/agents/metadata_extractor.py

@@ -0,0 +1,391 @@
+"""Metadata agent that extracts metadata from active room."""
+
+import asyncio
+import json
+import logging
+import os
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from io import BytesIO
+from typing import List, Optional
+
+from dotenv import load_dotenv
+from livekit import api, rtc
+from livekit.agents import (
+    Agent,
+    AgentSession,
+    AutoSubscribe,
+    JobContext,
+    JobProcess,
+    JobRequest,
+    RoomInputOptions,
+    RoomIO,
+    RoomOutputOptions,
+    WorkerOptions,
+    WorkerPermissions,
+    cli,
+    utils,
+)
+from livekit.plugins import silero
+from minio import Minio
+from minio.error import S3Error
+
+load_dotenv()
+
+logger = logging.getLogger("metadata-extractor")
+
+AGENT_NAME = os.getenv("ROOM_METADATA_EXTRACTOR_AGENT_NAME", "metadata-extractor")
+
+
+@dataclass
+class MetadataEvent:
+    """Wip."""
+
+    participant_id: str
+    type: str
+    timestamp: datetime
+    data: Optional[str] = None
+
+    def serialize(self) -> dict:
+        """Return a JSON-serializable dictionary representation of the event."""
+        data = asdict(self)
+        data["timestamp"] = self.timestamp.isoformat()
+        return data
+
+
+class VADAgent(Agent):
+    """Agent that monitors voice activity for a specific participant."""
+
+    def __init__(self, participant_identity: str, events: List):
+        """Wip."""
+        super().__init__(
+            instructions="not-needed",
+        )
+        self.participant_identity = participant_identity
+        self.events = events
+
+    async def on_enter(self) -> None:
+        """Initialize VAD monitoring for this participant."""
+
+        @self.session.on("user_state_changed")
+        def on_user_state(event):
+            timestamp = datetime.now(timezone.utc)
+
+            if event.new_state == "speaking":
+                event = MetadataEvent(
+                    participant_id=self.participant_identity,
+                    type="speech_start",
+                    timestamp=timestamp,
+                )
+                self.events.append(event)
+
+            elif event.old_state == "speaking":
+                event = MetadataEvent(
+                    participant_id=self.participant_identity,
+                    type="speech_end",
+                    timestamp=timestamp,
+                )
+                self.events.append(event)
+
+
+class MetadataAgent:
+    """Monitor and manage real-time metadata extraction from meeting rooms.
+
+    Oversees VAD (Voice Activity Detection) and participant metadata streams
+    to track and analyze real-time events, coordinating data collection across
+    participants for insights like speaking activity and engagement.
+    """
+
+    def __init__(self, ctx: JobContext, recording_id: str):
+        """Initialize metadata agent."""
+        self.minio_client = Minio(
+            endpoint=os.getenv("AWS_S3_ENDPOINT_URL"),
+            access_key=os.getenv("AWS_S3_ACCESS_KEY_ID"),
+            secret_key=os.getenv("AWS_S3_SECRET_ACCESS_KEY"),
+            secure=os.getenv("AWS_S3_SECURE_ACCESS", "False").lower() == "true",
+        )
+
+        # todo - raise error if none
+        self.bucket_name = os.getenv("AWS_STORAGE_BUCKET_NAME")
+
+        self.ctx = ctx
+        self._sessions: dict[str, AgentSession] = {}
+        self._tasks: set[asyncio.Task] = set()
+
+        self.output_filename = (
+            f"{os.getenv('AWS_S3_OUTPUT_FOLDER', 'metadata')}/{recording_id}-metadata.json"
+        )
+
+        # Storage for events
+        self.events = []
+        self.participants = {}
+
+        logger.info("MetadataAgent initialized")
+
+    def start(self):
+        """Start listening for participant connection events."""
+        self.ctx.room.on("participant_connected", self.on_participant_connected)
+        self.ctx.room.on("participant_disconnected", self.on_participant_disconnected)
+        self.ctx.room.on("participant_name_changed", self.on_participant_name_changed)
+
+        self.ctx.room.register_text_stream_handler("lk.chat", self.handle_chat_stream)
+
+        logger.info("Started listening for participant events")
+
+    async def on_chat_message_received(
+        self, reader: rtc.TextStreamReader, participant_identity: str
+    ):
+        """Wip."""
+        full_text = await reader.read_all()
+        logger.info(
+            "Received chat message from %s: '%s'", participant_identity, full_text
+        )
+
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant_identity,
+                type="chat_received",
+                timestamp=datetime.now(timezone.utc),
+                data=full_text,
+            )
+        )
+
+    def handle_chat_stream(self, reader, participant_identity):
+        """Wip."""
+        task = asyncio.create_task(
+            self.on_chat_message_received(reader, participant_identity)
+        )
+        self._tasks.add(task)
+        task.add_done_callback(lambda _: self._tasks.remove(task))
+
+    def save(self):
+        """Wip."""
+        logger.info("Persisting processed metadata output to disk…")
+
+        participants = []
+        for k, v in self.participants.items():
+            participants.append({"participantId": k, "name": v})
+
+        sorted_event = sorted(self.events, key=lambda e: e.timestamp)
+
+        payload = {
+            "events": [event.serialize() for event in sorted_event],
+            "participants": participants,
+        }
+
+        data = json.dumps(payload, indent=2).encode("utf-8")
+        stream = BytesIO(data)
+
+        try:
+            self.minio_client.put_object(
+                self.bucket_name,
+                self.output_filename,
+                stream,
+                length=len(data),
+                content_type="application/json",
+            )
+            logger.info(
+                "Uploaded speaker meeting metadata",
+            )
+        except S3Error:
+            logger.exception(
+                "Failed to upload meeting metadata",
+            )
+
+    async def aclose(self):
+        """Close all sessions and cleanup resources."""
+        logger.info("Closing all VAD monitoring sessions…")
+
+        await utils.aio.cancel_and_wait(*self._tasks)
+
+        await asyncio.gather(
+            *[self._close_session(session) for session in self._sessions.values()],
+            return_exceptions=True,
+        )
+
+        self.ctx.room.off("participant_connected", self.on_participant_connected)
+        self.ctx.room.off("participant_disconnected", self.on_participant_disconnected)
+        self.ctx.room.off("participant_name_changed", self.on_participant_name_changed)
+
+        logger.info("All VAD sessions closed")
+        self.save()
+
+    def on_participant_connected(self, participant: rtc.RemoteParticipant):
+        """Handle new participant connection by starting VAD monitoring."""
+        if participant.identity in self._sessions:
+            logger.debug("Session already exists for %s", participant.identity)
+            return
+
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant.identity,
+                type="participant_connected",
+                timestamp=datetime.now(timezone.utc),
+            )
+        )
+
+        self.participants[participant.identity] = participant.name
+
+        logger.info("New participant connected: %s", participant.identity)
+        task = asyncio.create_task(self._start_session(participant))
+        self._tasks.add(task)
+
+        def on_task_done(task: asyncio.Task):
+            try:
+                self._sessions[participant.identity] = task.result()
+            except Exception:
+                logger.exception("Failed to start session for %s", participant.identity)
+            finally:
+                self._tasks.discard(task)
+
+        task.add_done_callback(on_task_done)
+
+    def on_participant_disconnected(self, participant: rtc.RemoteParticipant):
+        """Handle participant disconnection by closing VAD monitoring."""
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant.identity,
+                type="participant_disconnected",
+                timestamp=datetime.now(timezone.utc),
+            )
+        )
+
+        session = self._sessions.pop(participant.identity, None)
+        if session is None:
+            logger.debug("No session found for %s", participant.identity)
+            return
+
+        logger.info("Participant disconnected: %s", participant.identity)
+        task = asyncio.create_task(self._close_session(session))
+        self._tasks.add(task)
+
+        def on_close_done(_):
+            self._tasks.discard(task)
+            logger.info(
+                "VAD session closed for %s (remaining sessions: %d)",
+                participant.identity,
+                len(self._sessions),
+            )
+
+        task.add_done_callback(on_close_done)
+
+    def on_participant_name_changed(self, participant: rtc.RemoteParticipant):
+        """Wip."""
+        logger.info("Participant's name changed: %s", participant.identity)
+        self.participants[participant.identity] = participant.name
+
+    async def _start_session(self, participant: rtc.RemoteParticipant) -> AgentSession:
+        """Create and start VAD monitoring session for participant."""
+        if participant.identity in self._sessions:
+            return self._sessions[participant.identity]
+
+        # Create session with VAD only - no STT, LLM, or TTS
+        session = AgentSession(
+            vad=self.ctx.proc.userdata["vad"],
+            turn_detection="vad",
+            user_away_timeout=30.0,
+        )
+
+        # Set up room IO to receive audio from this specific participant
+        room_io = RoomIO(
+            agent_session=session,
+            room=self.ctx.room,
+            participant=participant,
+            input_options=RoomInputOptions(
+                audio_enabled=True,
+                text_enabled=False,
+            ),
+            output_options=RoomOutputOptions(
+                audio_enabled=False,
+                transcription_enabled=False,
+            ),
+        )
+
+        await room_io.start()
+        await session.start(
+            agent=VADAgent(
+                participant_identity=participant.identity, events=self.events
+            )
+        )
+
+        return session
+
+    async def _close_session(self, session: AgentSession) -> None:
+        """Close and cleanup VAD monitoring session."""
+        try:
+            await session.drain()
+            await session.aclose()
+        except Exception:
+            logger.exception("Error closing session")
+
+
+async def entrypoint(ctx: JobContext):
+    """Initialize and run the multi-user VAD monitor."""
+    logger.info("Starting metadata agent in room: %s", ctx.room.name)
+    recording_id = ctx.job.metadata
+    vad_monitor = MetadataAgent(ctx, recording_id)
+    vad_monitor.start()
+
+    # Connect to room and subscribe to audio only
+    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
+
+    existing_participants = list(ctx.room.remote_participants.values())
+    for participant in existing_participants:
+        vad_monitor.on_participant_connected(participant)
+
+    async def cleanup():
+        logger.info("Shutting down VAD monitor...")
+        await vad_monitor.aclose()
+
+    ctx.add_shutdown_callback(cleanup)
+
+
+async def handle_job_request(job_req: JobRequest) -> None:
+    """Accept or reject the job request based on agent presence in the room."""
+    room_name = job_req.room.name
+    recording_id = job_req.job.metadata
+    agent_identity = f"{AGENT_NAME}-{room_name}"
+
+    async with api.LiveKitAPI() as lk:
+        try:
+            resp = await lk.room.list_participants(
+                list=api.ListParticipantsRequest(room=room_name)
+            )
+            already_present = any(
+                p.kind == rtc.ParticipantKind.PARTICIPANT_KIND_AGENT
+                and p.identity == agent_identity
+                for p in resp.participants
+            )
+            if already_present:
+                logger.info("Agent already in the room '%s' — reject", room_name)
+                await job_req.reject()
+            else:
+                logger.info(
+                    "Accept job for '%s' — identity=%s", room_name, agent_identity
+                )
+                await job_req.accept(identity=agent_identity, metadata=recording_id)
+        except Exception:
+            logger.exception("Error treating the job for '%s'", room_name)
+            await job_req.reject()
+
+
+def prewarm(proc: JobProcess):
+    """Preload voice activity detection model."""
+    proc.userdata["vad"] = silero.VAD.load()
+
+
+if __name__ == "__main__":
+    cli.run_app(
+        WorkerOptions(
+            entrypoint_fnc=entrypoint,
+            prewarm_fnc=prewarm,
+            request_fnc=handle_job_request,
+            agent_name=AGENT_NAME,
+            permissions=WorkerPermissions(
+                can_publish=False,
+                can_publish_data=False,
+                can_subscribe=True,
+                hidden=True,
+            ),
+        )
+    )

src/agents/pyproject.toml

@@ -1,13 +1,14 @@
 
 [project]
 name = "agents"
-version = "0.1.40"
+version = "0.1.39"
 requires-python = ">=3.12"
 dependencies = [
     "livekit-agents==1.2.6",
     "livekit-plugins-deepgram==1.2.6",
     "livekit-plugins-silero==1.2.6",
-    "python-dotenv==1.1.1"
+    "python-dotenv==1.1.1",
+    "minio==7.2.15"
 ]
 
 [project.optional-dependencies]

src/backend/core/api/viewsets.py

@@ -31,6 +31,10 @@ from core.recording.event.exceptions import (
 )
 from core.recording.event.notification import notification_service
 from core.recording.event.parsers import get_parser
+from core.recording.services.metadata_extractor import (
+    MetadataExtractorException,
+    MetadataExtractorService,
+)
 from core.recording.worker.exceptions import (
     RecordingStartError,
     RecordingStopError,
@@ -328,6 +332,15 @@ class RoomViewSet(
                 status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
             )
 
+        if (
+            settings.ROOM_METADATA_EXTRACTOR_ENABLED
+            and recording.mode == models.RecordingModeChoices.TRANSCRIPT
+        ):
+            try:
+                MetadataExtractorService().start(recording)
+            except MetadataExtractorException:
+                pass
+
         return drf_response.Response(
             {"message": f"Recording successfully started for room {room.slug}"},
             status=drf_status.HTTP_201_CREATED,

src/backend/core/recording/services/metadata_extractor.py

@@ -0,0 +1,97 @@
+"""Wip."""
+
+from logging import getLogger
+
+from django.conf import settings
+
+from asgiref.sync import async_to_sync
+from livekit.protocol.agent_dispatch import (
+    CreateAgentDispatchRequest,
+)
+
+from core import utils
+
+logger = getLogger(__name__)
+
+
+class MetadataExtractorException(Exception):
+    """Wip."""
+
+
+class MetadataExtractorService:
+    """Wip."""
+
+    @async_to_sync
+    async def start(self, recording):
+        """Wip."""
+
+        lkapi = utils.create_livekit_client()
+        room_id = str(recording.room.id)
+
+        try:
+            response = await lkapi.agent_dispatch.create_dispatch(
+                CreateAgentDispatchRequest(
+                    agent_name=settings.ROOM_METADATA_EXTRACTOR_AGENT_NAME,
+                    room=room_id,
+                    metadata=str(recording.id),
+                )
+            )
+        except Exception as e:
+            logger.exception(
+                "Failed to create metadata extractor agent for room %s", room_id
+            )
+            raise MetadataExtractorException(
+                "Failed to create metadata extractor agent"
+            ) from e
+        finally:
+            await lkapi.aclose()
+
+        dispatch_id = getattr(response, "id", None)
+
+        if not dispatch_id:
+            logger.error("LiveKit response missing dispatch ID for room %s", room_id)
+            raise MetadataExtractorException(
+                f"LiveKit did not return a dispatch_id for room {room_id}"
+            )
+
+        return dispatch_id
+
+    @async_to_sync
+    async def stop(self, recording):
+        """Wip."""
+
+        room_name = str(recording.room.id)
+        lkapi = utils.create_livekit_client()
+
+        try:
+            dispatches = await lkapi.agent_dispatch.list_dispatch(room_name=room_name)
+
+            dispatch_id = next(
+                (
+                    d.id
+                    for d in dispatches
+                    if d.agent_name == settings.ROOM_METADATA_EXTRACTOR_AGENT_NAME
+                ),
+                None,
+            )
+
+            if not dispatch_id:
+                logger.warning(
+                    "No metadata extractor agent found for room %s", room_name
+                )
+                return None
+
+            await lkapi.agent_dispatch.delete_dispatch(
+                dispatch_id=str(dispatch_id), room_name=room_name
+            )
+
+        except Exception as e:
+            logger.exception(
+                "Failed to stop metadata extractor agent dispatch for room %s",
+                room_name,
+            )
+            raise MetadataExtractorException(
+                f"Failed to stop metadata metadata extractor agent for room {room_name}"
+            ) from e
+        finally:
+            await lkapi.aclose()

src/backend/core/services/livekit_events.py

@@ -2,7 +2,6 @@
 
 # pylint: disable=no-member
 
-import re
 import uuid
 from enum import Enum
 from logging import getLogger
@@ -12,6 +11,10 @@ from django.conf import settings
 from livekit import api
 
 from core import models
+from core.recording.services.metadata_extractor import (
+    MetadataExtractorException,
+    MetadataExtractorService,
+)
 from core.recording.services.recording_events import (
     RecordingEventsError,
     RecordingEventsService,
@@ -93,17 +96,6 @@ class LiveKitEventsService:
         self.telephony_service = TelephonyService()
         self.recording_events = RecordingEventsService()
 
-        self._filter_regex = None
-        if settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX:
-            try:
-                self._filter_regex = re.compile(
-                    settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX
-                )
-            except re.error:
-                logger.exception(
-                    "Invalid LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX. Webhook filtering disabled."
-                )
-
     def receive(self, request):
         """Process webhook and route to appropriate handler."""
 
@@ -118,10 +110,6 @@ class LiveKitEventsService:
         except Exception as e:
             raise InvalidPayloadError("Invalid webhook payload") from e
 
-        if self._filter_regex and not self._filter_regex.search(data.room.name):
-            logger.info("Filtered webhook event for room '%s'", data.room.name)
-            return
-
         try:
             webhook_type = LiveKitWebhookEventType(data.event)
         except ValueError as e:
@@ -142,7 +130,7 @@ class LiveKitEventsService:
         """Handle 'egress_ended' event."""
 
         try:
-            recording = models.Recording.objects.get(
+            recording = models.Recording.objects.select_related("room").get(
                 worker_id=data.egress_info.egress_id
             )
         except models.Recording.DoesNotExist as err:
@@ -150,6 +138,15 @@ class LiveKitEventsService:
                 f"Recording with worker ID {data.egress_info.egress_id} does not exist"
             ) from err
 
+        if (
+            settings.ROOM_METADATA_EXTRACTOR_ENABLED
+            and recording.mode == models.RecordingModeChoices.TRANSCRIPT
+        ):
+            try:
+                MetadataExtractorService().stop(recording)
+            except MetadataExtractorException:
+                pass
+
         if (
             data.egress_info.status == api.EgressStatus.EGRESS_LIMIT_REACHED
             and recording.status == models.RecordingStatusChoices.ACTIVE

src/backend/core/tests/services/test_livekit_events.py

@@ -343,99 +343,3 @@ def test_receive_unsupported_event(mock_receive, service):
         UnsupportedEventTypeError, match="Unknown webhook type: unsupported_event"
     ):
         service.receive(mock_request)
-
-
-@mock.patch.object(api.WebhookReceiver, "receive")
-@mock.patch.object(LiveKitEventsService, "_handle_room_started")
-def test_receive_no_filter_processes_all_events(
-    mock_handle_room_started, mock_receive, mock_livekit_config, settings
-):
-    """Should process all events when filter regex is not configured."""
-    settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = None
-
-    mock_request = mock.MagicMock()
-    mock_request.headers = {"Authorization": "test_token"}
-    mock_request.body = b"{}"
-
-    mock_data = mock.MagicMock()
-    mock_data.room.name = "!JIfCxVLcKKkWrmVBOb:your-domain.com"
-    mock_data.event = "room_started"
-    mock_receive.return_value = mock_data
-
-    service = LiveKitEventsService()
-    service.receive(mock_request)
-
-    mock_handle_room_started.assert_called_once()
-
-
-@mock.patch.object(api.WebhookReceiver, "receive")
-@mock.patch.object(LiveKitEventsService, "_handle_room_started")
-def test_receive_invalid_filter_regex_processes_all_events(
-    mock_handle_room_started, mock_receive, mock_livekit_config, settings
-):
-    """Should process all events when filter regex is invalid (fail-safe)."""
-    settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = "(abc"
-
-    mock_request = mock.MagicMock()
-    mock_request.headers = {"Authorization": "test_token"}
-    mock_request.body = b"{}"
-
-    mock_data = mock.MagicMock()
-    mock_data.room.name = "!JIfCxVLcKKkWrmVBOb:your-domain.com"
-    mock_data.event = "room_started"
-    mock_receive.return_value = mock_data
-
-    service = LiveKitEventsService()
-    service.receive(mock_request)
-
-    mock_handle_room_started.assert_called_once()
-
-
-@mock.patch.object(api.WebhookReceiver, "receive")
-@mock.patch.object(LiveKitEventsService, "_handle_room_started")
-def test_receive_filter_drops_non_matching_events(
-    mock_handle_room_started, mock_receive, mock_livekit_config, settings
-):
-    """Should drop events when room name does not match filter regex."""
-    settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = (
-        r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
-    )
-
-    mock_request = mock.MagicMock()
-    mock_request.headers = {"Authorization": "test_token"}
-    mock_request.body = b"{}"
-
-    mock_data = mock.MagicMock()
-    mock_data.room.name = "!JIfCxVLcKKkWrmVBOb:your-domain.com"
-    mock_data.event = "room_started"
-    mock_receive.return_value = mock_data
-
-    service = LiveKitEventsService()
-    service.receive(mock_request)
-
-    mock_handle_room_started.assert_not_called()
-
-
-@mock.patch.object(api.WebhookReceiver, "receive")
-@mock.patch.object(LiveKitEventsService, "_handle_room_started")
-def test_receive_filter_processes_matching_events(
-    mock_handle_room_started, mock_receive, mock_livekit_config, settings
-):
-    """Should process events when room name matches filter regex."""
-    settings.LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = (
-        r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
-    )
-
-    mock_request = mock.MagicMock()
-    mock_request.headers = {"Authorization": "test_token"}
-    mock_request.body = b"{}"
-
-    mock_data = mock.MagicMock()
-    mock_data.room.name = str(uuid.uuid4())
-    mock_data.event = "room_started"
-    mock_receive.return_value = mock_data
-
-    service = LiveKitEventsService()
-    service.receive(mock_request)
-
-    mock_handle_room_started.assert_called_once()

src/backend/meet/settings.py

@@ -517,10 +517,6 @@ class Base(Configuration):
     LIVEKIT_VERIFY_SSL = values.BooleanValue(
         True, environ_name="LIVEKIT_VERIFY_SSL", environ_prefix=None
     )
-    # Regex to filter webhook events by room name. Only matching events are processed.
-    LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = values.Value(
-        None, environ_name="LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX", environ_prefix=None
-    )
     RESOURCE_DEFAULT_ACCESS_LEVEL = values.Value(
         "public", environ_name="RESOURCE_DEFAULT_ACCESS_LEVEL", environ_prefix=None
     )
@@ -672,6 +668,16 @@ class Base(Configuration):
         environ_prefix=None,
     )
 
+    # Metadata Extractor settings
+    ROOM_METADATA_EXTRACTOR_ENABLED = values.BooleanValue(
+        False, environ_name="ROOM_METADATA_EXTRACTOR_ENABLED", environ_prefix=None
+    )
+    ROOM_METADATA_EXTRACTOR_AGENT_NAME = values.Value(
+        "metadata-extractor",
+        environ_name="ROOM_METADATA_EXTRACTOR_AGENT_NAME",
+        environ_prefix=None,
+    )
+
     # External Applications
     APPLICATION_CLIENT_ID_LENGTH = values.PositiveIntegerValue(
         40,

src/backend/pyproject.toml

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "meet"
-version = "0.1.40"
+version = "0.1.39"
 authors = [{ "name" = "DINUM", "email" = "dev@mail.numerique.gouv.fr" }]
 classifiers = [
     "Development Status :: 5 - Production/Stable",

src/frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "meet",
-  "version": "0.1.40",
+  "version": "0.1.39",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "meet",
-      "version": "0.1.40",
+      "version": "0.1.39",
       "dependencies": {
         "@livekit/components-react": "2.9.13",
         "@livekit/components-styles": "1.1.6",

src/frontend/package.json

@@ -1,7 +1,7 @@
 {
   "name": "meet",
   "private": true,
-  "version": "0.1.40",
+  "version": "0.1.39",
   "type": "module",
   "scripts": {
     "dev": "panda codegen && vite",

src/frontend/src/features/rooms/livekit/hooks/useDeviceShortcut.ts

@@ -6,12 +6,12 @@ export const useDeviceShortcut = (kind: MediaDeviceKind) => {
     switch (kind) {
       case 'audioinput':
         return {
-          key: 'd',
+          key: 'e',
           ctrlKey: true,
         }
       case 'videoinput':
         return {
-          key: 'e',
+          key: 'd',
           ctrlKey: true,
         }
       default:

src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl

@@ -73,6 +73,7 @@ backend:
     ROOM_TELEPHONY_PHONE_NUMBER: '+33901020304'
     SSL_CERT_FILE: /usr/local/lib/python3.13/site-packages/certifi/cacert.pem
     ROOM_SUBTITLE_ENABLED: True
+    ROOM_METADATA_EXTRACTOR_ENABLED: True
 
 
   migrate:
@@ -155,7 +156,6 @@ summary:
     WHISPERX_API_KEY: your-secret-value
     WHISPERX_BASE_URL: https://configure-your-url.com
     WHISPERX_ASR_MODEL: large-v2
-    WHISPERX_DEFAULT_LANGUAGE: fr
     LLM_BASE_URL: https://configure-your-url.com
     LLM_API_KEY: your-secret-value
     LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
@@ -192,7 +192,6 @@ celeryTranscribe:
     WHISPERX_API_KEY: your-secret-value
     WHISPERX_BASE_URL: https://configure-your-url.com
     WHISPERX_ASR_MODEL: large-v2
-    WHISPERX_DEFAULT_LANGUAGE: fr
     LLM_BASE_URL: https://configure-your-url.com
     LLM_API_KEY: your-secret-value
     LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
@@ -230,7 +229,6 @@ celerySummarize:
     WHISPERX_API_KEY: your-secret-value
     WHISPERX_BASE_URL: https://configure-your-url.com
     WHISPERX_ASR_MODEL: large-v2
-    WHISPERX_DEFAULT_LANGUAGE: fr
     LLM_BASE_URL: https://configure-your-url.com
     LLM_API_KEY: your-secret-value
     LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
@@ -265,12 +263,23 @@ agents:
     LIVEKIT_API_KEY: {{ $key }}
     {{- end }}
     {{- end }}
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_OUTPUT_FOLDER: recordings
 
   image:
     repository: localhost:5001/meet-agents
     pullPolicy: Always
     tag: "latest"
 
+  command:
+    - "python"
+    - "metadata_extractor.py"
+    - "start"
+
   # Extra volume mounts to manage our local custom CA and avoid to disable ssl
   extraVolumeMounts:
     - name: certs

src/helm/meet/Chart.yaml

@@ -1,4 +1,4 @@
 apiVersion: v2
 type: application
 name: meet
-version: 0.0.13
+version: 0.0.13-beta.1

src/mail/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "mail_mjml",
-  "version": "0.1.40",
+  "version": "0.1.39",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "mail_mjml",
-      "version": "0.1.40",
+      "version": "0.1.39",
       "license": "MIT",
       "dependencies": {
         "@html-to/text-cli": "0.5.4",

src/mail/package.json

@@ -1,6 +1,6 @@
 {
   "name": "mail_mjml",
-  "version": "0.1.40",
+  "version": "0.1.39",
   "description": "An util to generate html and text django's templates from mjml templates",
   "type": "module",
   "dependencies": {

src/sdk/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "sdk",
-  "version": "0.1.40",
+  "version": "0.1.39",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "sdk",
-      "version": "0.1.40",
+      "version": "0.1.39",
       "license": "ISC",
       "workspaces": [
         "./library",

src/sdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "sdk",
-  "version": "0.1.40",
+  "version": "0.1.39",
   "author": "",
   "license": "ISC",
   "description": "",

src/summary/pyproject.toml

@@ -1,7 +1,7 @@
 
 [project]
 name = "summary"
-version = "0.1.40"
+version = "0.1.39"
 dependencies = [
     "fastapi[standard]>=0.105.0",
     "uvicorn>=0.24.0",

src/summary/summary/core/celery_worker.py

@@ -7,7 +7,7 @@ import os
 import tempfile
 import time
 from pathlib import Path
-from typing import Any, Mapping, Optional
+from typing import Optional
 
 import openai
 import sentry_sdk
@@ -22,8 +22,6 @@ 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 (
-    FORMAT_NEXT_STEPS,
-    FORMAT_PLAN,
     PROMPT_SYSTEM_CLEANING,
     PROMPT_SYSTEM_NEXT_STEP,
     PROMPT_SYSTEM_PART,
@@ -117,53 +115,24 @@ class LLMService:
             base_url=settings.llm_base_url, api_key=settings.llm_api_key
         )
 
-    def call(
-        self,
-        system_prompt: str,
-        user_prompt: str,
-        response_format: Optional[Mapping[str, Any]] = None,
-    ):
+    def call(self, system_prompt: str, user_prompt: str):
         """Call the LLM service.
 
         Takes a system prompt and a user prompt, and returns the LLM's response
         Returns None if the call fails.
         """
         try:
-            params: dict[str, Any] = {
-                "model": settings.llm_model,
-                "messages": [
+            response = self._client.chat.completions.create(
+                model=settings.llm_model,
+                messages=[
                     {"role": "system", "content": system_prompt},
                     {"role": "user", "content": user_prompt},
                 ],
-            }
-            if response_format is not None:
-                params["response_format"] = response_format
-
-            response = self._client.chat.completions.create(**params)
-
+            )
             return response.choices[0].message.content
-
         except Exception as e:
-            logger.exception("LLM call failed: %s", e)
-            raise LLMException("LLM call failed: {e}") from e
-
-
-def format_actions(llm_output: dict) -> str:
-    """Format the actions from the LLM output into a markdown list.
-
-    fomat:
-    - [ ] Action title Assignée à : assignee1, assignee2, Échéance : due_date
-    """
-    lines = []
-    for action in llm_output.get("actions", []):
-        title = action.get("title", "").strip()
-        assignees = ", ".join(action.get("assignees", [])) or "-"
-        due_date = action.get("due_date") or "-"
-        line = f"- [ ] {title} Assignée à : {assignees}, Échéance : {due_date}"
-        lines.append(line)
-    if lines:
-        return "### Prochaines étapes\n\n" + "\n".join(lines)
-    return ""
+            logger.error("LLM call failed: %s", e)
+            raise LLMException("LLM call failed.") from e
 
 
 def format_segments(transcription_data):
@@ -208,6 +177,25 @@ 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(
     bind=True,
     autoretry_for=[exceptions.HTTPError],
@@ -282,9 +270,7 @@ def process_audio_transcribe_summarize_v2(
         transcription_start_time = time.time()
         with open(temp_file_path, "rb") as audio_file:
             transcription = whisperx_client.audio.transcriptions.create(
-                model=settings.whisperx_asr_model,
-                file=audio_file,
-                language=settings.whisperx_default_language,
+                model=settings.whisperx_asr_model, file=audio_file
             )
             metadata_manager.track(
                 task_id,
@@ -347,25 +333,6 @@ def process_audio_transcribe_summarize_v2(
         logger.info("Summary generation not enabled for this user.")
 
 
-@signals.task_prerun.connect(sender=process_audio_transcribe_summarize_v2)
-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(sender=process_audio_transcribe_summarize_v2)
-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(sender=process_audio_transcribe_summarize_v2)
-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(
     bind=True,
     autoretry_for=[LLMException, Exception],
@@ -390,14 +357,13 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
 
     logger.info("TLDR generated")
 
-    parts = llm_service.call(
-        PROMPT_SYSTEM_PLAN, transcript, response_format=FORMAT_PLAN
-    )
+    parts = llm_service.call(PROMPT_SYSTEM_PLAN, transcript)
     logger.info("Plan generated")
 
-    res = json.loads(parts)
-    parts = res.get("titles", [])
-    logger.info("Parts to summarize: %s", parts)
+    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)
@@ -408,12 +374,7 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
 
     raw_summary = "\n\n".join(parts_summarized)
 
-    next_steps = llm_service.call(
-        PROMPT_SYSTEM_NEXT_STEP, transcript, response_format=FORMAT_NEXT_STEPS
-    )
-
-    next_steps = format_actions(json.loads(next_steps))
-
+    next_steps = llm_service.call(PROMPT_SYSTEM_NEXT_STEP, transcript)
     logger.info("Next steps generated")
 
     cleaned_summary = llm_service.call(PROMPT_SYSTEM_CLEANING, raw_summary)

src/summary/summary/core/config.py

@@ -39,8 +39,6 @@ class Settings(BaseSettings):
     whisperx_base_url: str = "https://api.openai.com/v1"
     whisperx_asr_model: str = "whisper-1"
     whisperx_max_retries: int = 0
-    # ISO 639-1 language code (e.g., "en", "fr", "es")
-    whisperx_default_language: Optional[str] = None
     llm_base_url: str
     llm_api_key: str
     llm_model: str

src/summary/summary/core/prompt.py

@@ -4,8 +4,12 @@ PROMPT_SYSTEM_TLDR = """Tu es un agent dont le rôle est de créer un TL;DR (ré
 ### 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 limiteras à 5 ou 6 sujets maximum. 
-L'introduction, ordre du jour, conclusion, etc. seront rajoutés a posteriori. Si il n'y a pas de sujets clairs, réponds "Général".
+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 limiteras à 5 ou 6 sujets maximum.
+L'introduction, ordre du jour, conclusion, etc. seront rajoutés a posteriori. Tu répondras 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 utiliseras 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épondras dans le format suivant :
@@ -19,53 +23,6 @@ Transcript complet :
 
 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épondras 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 utiliseras 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 de 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."""
-
-FORMAT_NEXT_STEPS = {
-    "type": "json_schema",
-    "json_schema": {
-        "name": "actions",
-        "schema": {
-            "type": "object",
-            "properties": {
-                "actions": {
-                    "type": "array",
-                    "items": {
-                        "type": "object",
-                        "properties": {
-                            "title": {"type": "string"},
-                            "assignees": {
-                                "type": "array",
-                                "items": {"type": "string"},
-                                "description": "Noms des personnes assignées",
-                            },
-                            "due_date": {
-                                "type": "string",
-                                "description": "Date d'échéance si mentionnée (si l'année nest pas précisée, ne pas l'ajouter)",
-                            },
-                        },
-                        "required": ["title", "assignees"],
-                        "additionalProperties": False,
-                    },
-                }
-            },
-            "required": ["actions"],
-            "additionalProperties": False,
-        },
-        "strict": True,
-    },
-}
-
-FORMAT_PLAN = {
-    "type": "json_schema",
-    "json_schema": {
-        "name": "Titles",
-        "schema": {
-            "type": "object",
-            "properties": {"titles": {"type": "array", "items": {"type": "string"}}},
-            "required": ["titles"],
-            "additionalProperties": False,
-        },
-        "strict": True,
-    },
-}
+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 utiliseras 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 de 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]"""