Compare commits

..

12 Commits

Author SHA1 Message Date
lebaudantoine 1ce80cb57e 🩹(doc) fix wrong endpoint path
Applications to application in the application/token endpoint.
Spotted by external contributor.
2025-10-22 05:01:41 +02:00
Ghislain LE MEUR 59d4c2583b 🐛(auth) fix LiveKit token authentication field mismatch
Fixes "Invalid LiveKit token" errors caused by field mismatch between
token generation and authentication lookup.

Previously:
- generate_token() used user.sub as token identity
- LiveKitTokenAuthentication tried to retrieve user via user.id field
- This failed when sub was not a UUID (e.g., from LemonLDAP OIDC provider)

Now:
- generate_token() continues using user.sub (canonical OIDC identifier)
- LiveKitTokenAuthentication correctly looks up by sub field
- Both sides now consistently use the same field

This ensures compatibility with all RFC 7519-compliant OIDC providers,
regardless of their sub claim format.
2025-10-20 04:57:02 +02:00
Ghislain LE MEUR 4b80b4ac9f 🔖(helm) release chart 0.0.14
Fix missing image and command attributes for celery workers
2025-10-17 12:18:31 +02:00
Ghislain LE MEUR 96d7a8875b 🐛(helm) add default commands for celery workers
Without explicit commands in values.yaml,
celeryTranscribe and  celerySummarize pods
were using the Dockerfile's default CMD (uvicorn),
which started the REST API instead of Celery workers.

This fix adds default commands to values.yaml for both services,
ensuring they run as Celery workers processing their respective
queues (transcribe-queue and summarize-queue).
2025-10-17 12:18:31 +02:00
Ghislain LE MEUR dc177b69d8 🐛(summary) add image
Add missing image attributes for summary, celerySummarize and celeryTranscribe
2025-10-17 12:18:31 +02:00
Martin Guitteny 36b2156c7b ️(summary) change formating from prompt to response_format
Add ability to use response_format in call function in order to
have better result with albert-large model
Use reponse_format for next steps and plan generation
2025-10-13 12:07:54 +02:00
lebaudantoine ec94d613fa 🔖(minor) bump release to 0.1.40
- enhance technical documentation
- introduce external-api and service account
- fix inverted keyboard shortcuts
- allow configuring whisperX language (still wip)
- filter livekit event when sharing a single livekit instance
2025-10-12 17:13:09 +02:00
lebaudantoine 70d9d55227 🔖(helm) release chart 0.0.13
This chart exposes an external API from the backend pod.
Currently, it does not include conditional addition of the external API route.
This functionality will be added later.
2025-10-12 17:05:58 +02:00
lebaudantoine 5c74ace0d8 🐛(backend) filter LiveKit events by room name regex to exclude Tchap
Add configurable room name regex filtering to exclude Tchap events from shared
LiveKit server webhooks, preventing backend spam from unrelated application
events while maintaining UUID-based room processing for visio.

Those unrelated application events are spamming the sentry.

Acknowledges this is a pragmatic solution trading proper namespace
prefixing for immediate spam reduction with minimal refactoring impact
leaving prefix-based approach for future improvement.
2025-10-12 16:57:44 +02:00
lebaudantoine f0939b6f7c 🐛(summary) scope metadata manager signals to transcription tasks only
Restrict metadata manager signal triggers to transcription-specific Celery
tasks to prevent exceptions when new summary worker executes tasks
not designed for metadata operations, reducing false-positive Sentry errors.
2025-10-10 14:00:00 +02:00
lebaudantoine aecc48f928 🔧(summary) add configurable language settings for WhisperX transcription
Make WhisperX language detection configurable through FastAPI settings
to handle empty audio start scenarios where automatic detection fails and
incorrectly defaults to English despite 99% French usage.

Quick fix acknowledging long-term solution should allow dynamic
per-recording language selection configured by users through web
interface rather than global server settings.
2025-10-10 13:55:53 +02:00
lebaudantoine 4353db4a5f 🐛(frontend) fix inverted keyboard shortcuts for video and microphone
Correct accidentally swapped keyboard shortcuts between video and
microphone toggle controls introduced during device component
refactoring, restoring expected shortcut behavior reported by users.
2025-10-09 22:44:14 +02:00
27 changed files with 311 additions and 605 deletions
+3 -3
View File
@@ -7,7 +7,7 @@ info:
#### Authentication Flow
1. Exchange application credentials for a JWT token via `/external-api/v1.0/applications/token`.
1. Exchange application credentials for a JWT token via `/external-api/v1.0/application/token`.
2. Use the JWT token in the `Authorization: Bearer <token>` header for all subsequent requests.
3. Tokens are scoped and allow applications to act on behalf of specific users.
@@ -40,7 +40,7 @@ tags:
description: Room management operations
paths:
/applications/token:
/application/token:
post:
tags:
- Authentication
@@ -283,7 +283,7 @@ components:
scheme: bearer
bearerFormat: JWT
description: |
JWT token obtained from the `/applications/token` endpoint.
JWT token obtained from the `/application/token` endpoint.
Include in requests as: `Authorization: Bearer <token>`
schemas:
+2 -1
View File
@@ -11,10 +11,11 @@ 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="Qwen/Qwen2.5-Coder-32B-Instruct-AWQ"
LLM_MODEL="albert-large"
WEBHOOK_API_TOKEN="secret"
WEBHOOK_URL="https://configure-your-url.com"
+1 -1
View File
@@ -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"]
-391
View File
@@ -1,391 +0,0 @@
"""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,
),
)
)
+2 -3
View File
@@ -1,14 +1,13 @@
[project]
name = "agents"
version = "0.1.39"
version = "0.1.40"
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",
"minio==7.2.15"
"python-dotenv==1.1.1"
]
[project.optional-dependencies]
-13
View File
@@ -31,10 +31,6 @@ 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,
@@ -332,15 +328,6 @@ 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,
+1 -1
View File
@@ -30,7 +30,7 @@ class LiveKitTokenAuthentication(authentication.BaseAuthentication):
raise exceptions.AuthenticationFailed("Token missing user identity")
try:
user = UserModel.objects.get(id=user_id)
user = UserModel.objects.get(sub=user_id)
except UserModel.DoesNotExist:
user = AnonymousUser()
@@ -1,97 +0,0 @@
"""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()
+17 -14
View File
@@ -2,6 +2,7 @@
# pylint: disable=no-member
import re
import uuid
from enum import Enum
from logging import getLogger
@@ -11,10 +12,6 @@ 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,
@@ -96,6 +93,17 @@ 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."""
@@ -110,6 +118,10 @@ 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:
@@ -130,7 +142,7 @@ class LiveKitEventsService:
"""Handle 'egress_ended' event."""
try:
recording = models.Recording.objects.select_related("room").get(
recording = models.Recording.objects.get(
worker_id=data.egress_info.egress_id
)
except models.Recording.DoesNotExist as err:
@@ -138,15 +150,6 @@ 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
@@ -343,3 +343,99 @@ 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()
+4 -10
View File
@@ -517,6 +517,10 @@ 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
)
@@ -668,16 +672,6 @@ 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,
+1 -1
View File
@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "meet"
version = "0.1.39"
version = "0.1.40"
authors = [{ "name" = "DINUM", "email" = "dev@mail.numerique.gouv.fr" }]
classifiers = [
"Development Status :: 5 - Production/Stable",
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "meet",
"version": "0.1.39",
"version": "0.1.40",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "meet",
"version": "0.1.39",
"version": "0.1.40",
"dependencies": {
"@livekit/components-react": "2.9.13",
"@livekit/components-styles": "1.1.6",
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "meet",
"private": true,
"version": "0.1.39",
"version": "0.1.40",
"type": "module",
"scripts": {
"dev": "panda codegen && vite",
@@ -6,12 +6,12 @@ export const useDeviceShortcut = (kind: MediaDeviceKind) => {
switch (kind) {
case 'audioinput':
return {
key: 'e',
key: 'd',
ctrlKey: true,
}
case 'videoinput':
return {
key: 'd',
key: 'e',
ctrlKey: true,
}
default:
@@ -73,7 +73,6 @@ 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:
@@ -156,6 +155,7 @@ 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,6 +192,7 @@ 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
@@ -229,6 +230,7 @@ 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
@@ -263,23 +265,12 @@ 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
+1 -1
View File
@@ -1,4 +1,4 @@
apiVersion: v2
type: application
name: meet
version: 0.0.13-beta.1
version: 0.0.14
+41 -2
View File
@@ -428,6 +428,13 @@ posthog:
## @section summary
summary:
## @param summary.image.repository Repository to use to pull meet's summary container image
## @param summary.image.tag meet's summary container tag
## @param summary.image.pullPolicy summary container image pull policy
image:
repository: lasuite/meet-summary
pullPolicy: IfNotPresent
tag: "latest"
## @param summary.dpAnnotations Annotations to add to the summary Deployment
dpAnnotations: {}
@@ -527,11 +534,27 @@ summary:
## @section celeryTranscribe
celeryTranscribe:
## @param celeryTranscribe.image.repository Repository to use to pull meet's celeryTranscribe container image
## @param celeryTranscribe.image.tag meet's celeryTranscribe container tag
## @param celeryTranscribe.image.pullPolicy celeryTranscribe container image pull policy
image:
repository: lasuite/meet-summary
pullPolicy: IfNotPresent
tag: "latest"
## @param celeryTranscribe.dpAnnotations Annotations to add to the celeryTranscribe Deployment
dpAnnotations: {}
## @param celeryTranscribe.command Override the celeryTranscribe container command
command: []
command:
- "celery"
- "-A"
- "summary.core.celery_worker"
- "worker"
- "--pool=solo"
- "--loglevel=info"
- "-Q"
- "transcribe-queue"
## @param celeryTranscribe.args Override the celeryTranscribe container args
args: []
@@ -620,11 +643,27 @@ celeryTranscribe:
## @section celerySummarize
celerySummarize:
## @param celerySummarize.image.repository Repository to use to pull meet's celerySummarize container image
## @param celerySummarize.image.tag meet's celerySummarize container tag
## @param celerySummarize.image.pullPolicy celerySummarize container image pull policy
image:
repository: lasuite/meet-summary
pullPolicy: IfNotPresent
tag: "latest"
## @param celerySummarize.dpAnnotations Annotations to add to the celerySummarize Deployment
dpAnnotations: {}
## @param celerySummarize.command Override the celerySummarize container command
command: []
command:
- "celery"
- "-A"
- "summary.core.celery_worker"
- "worker"
- "--pool=solo"
- "--loglevel=info"
- "-Q"
- "summarize-queue"
## @param celerySummarize.args Override the celerySummarize container args
args: []
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "mail_mjml",
"version": "0.1.39",
"version": "0.1.40",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "mail_mjml",
"version": "0.1.39",
"version": "0.1.40",
"license": "MIT",
"dependencies": {
"@html-to/text-cli": "0.5.4",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "mail_mjml",
"version": "0.1.39",
"version": "0.1.40",
"description": "An util to generate html and text django's templates from mjml templates",
"type": "module",
"dependencies": {
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "sdk",
"version": "0.1.39",
"version": "0.1.40",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "sdk",
"version": "0.1.39",
"version": "0.1.40",
"license": "ISC",
"workspaces": [
"./library",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "sdk",
"version": "0.1.39",
"version": "0.1.40",
"author": "",
"license": "ISC",
"description": "",
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "summary"
version = "0.1.39"
version = "0.1.40"
dependencies = [
"fastapi[standard]>=0.105.0",
"uvicorn>=0.24.0",
+73 -34
View File
@@ -7,7 +7,7 @@ import os
import tempfile
import time
from pathlib import Path
from typing import Optional
from typing import Any, Mapping, Optional
import openai
import sentry_sdk
@@ -22,6 +22,8 @@ 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,
@@ -115,24 +117,53 @@ class LLMService:
base_url=settings.llm_base_url, api_key=settings.llm_api_key
)
def call(self, system_prompt: str, user_prompt: str):
def call(
self,
system_prompt: str,
user_prompt: str,
response_format: Optional[Mapping[str, Any]] = None,
):
"""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:
response = self._client.chat.completions.create(
model=settings.llm_model,
messages=[
params: dict[str, Any] = {
"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.error("LLM call failed: %s", e)
raise LLMException("LLM call failed.") from 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 ""
def format_segments(transcription_data):
@@ -177,25 +208,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(
bind=True,
autoretry_for=[exceptions.HTTPError],
@@ -270,7 +282,9 @@ 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
model=settings.whisperx_asr_model,
file=audio_file,
language=settings.whisperx_default_language,
)
metadata_manager.track(
task_id,
@@ -333,6 +347,25 @@ 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],
@@ -357,13 +390,14 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
logger.info("TLDR generated")
parts = llm_service.call(PROMPT_SYSTEM_PLAN, transcript)
parts = llm_service.call(
PROMPT_SYSTEM_PLAN, transcript, response_format=FORMAT_PLAN
)
logger.info("Plan generated")
parts = parts.split("\n")
parts = [x for x in parts if x.strip() != ""]
logger.info("Empty parts removed")
res = json.loads(parts)
parts = res.get("titles", [])
logger.info("Parts to summarize: %s", parts)
parts_summarized = []
for part in parts:
prompt_user_part = PROMPT_USER_PART.format(part=part, transcript=transcript)
@@ -374,7 +408,12 @@ 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)
next_steps = llm_service.call(
PROMPT_SYSTEM_NEXT_STEP, transcript, response_format=FORMAT_NEXT_STEPS
)
next_steps = format_actions(json.loads(next_steps))
logger.info("Next steps generated")
cleaned_summary = llm_service.call(PROMPT_SYSTEM_CLEANING, raw_summary)
+2
View File
@@ -39,6 +39,8 @@ 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
+52 -9
View File
@@ -4,12 +4,8 @@ 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 quaucun 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_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 quaucun 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_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 :
@@ -23,6 +19,53 @@ 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 dinformations 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 lobjet de décisions ou dactions. 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. Les actions doivent suivre ce format strict :
### Prochaines étapes
- [ ] [Action à effectuer] Assignée à : [Nom], Échéance : [Date si mentionnée]"""
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,
},
}