34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
from collections import deque
|
|
from typing import Any
|
|
|
|
from app.domain.schemas import TranscriptChunk
|
|
|
|
|
|
class ContextBuilder:
|
|
"""Maintains a Redis-compatible meeting context; uses memory for offline tests."""
|
|
|
|
def __init__(self, window_size: int = 12) -> None:
|
|
self.window_size = window_size
|
|
self._transcript: deque[TranscriptChunk] = deque(maxlen=window_size)
|
|
self._participants: set[str] = set()
|
|
self._recent_decisions: deque[dict[str, Any]] = deque(maxlen=10)
|
|
|
|
def add_chunk(self, chunk: TranscriptChunk) -> None:
|
|
self._transcript.append(chunk)
|
|
self._participants.add(chunk.speaker)
|
|
|
|
def add_decision(self, decision: dict[str, Any]) -> None:
|
|
self._recent_decisions.append(decision)
|
|
|
|
def snapshot(self, app_state: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return {
|
|
"meeting": {"id": "demo-meeting", "title": "Wedding Planning Review"},
|
|
"participants": sorted(self._participants),
|
|
"transcript_window": [chunk.model_dump(mode="json") for chunk in self._transcript],
|
|
"recent_ai_decisions": list(self._recent_decisions),
|
|
"application_state": app_state or {},
|
|
}
|
|
|
|
def transcript(self) -> list[TranscriptChunk]:
|
|
return list(self._transcript)
|