Building Production Agentic Meeting Transcription Systems: Complete Tutorial with Real-Time AI Agents, RAG Integration, and Enterprise Deployment

Modern enterprises waste billions of dollars annually on meetings that produce no actionable outcomes. Teams spend hours discussing critical decisions, only to have disagreements later about "what was actually decided." Action items get lost between note-taking apps and email threads. Context from previous meetings disappears, forcing teams to repeatedly cover the same ground.

What if every meeting automatically generated accurate transcripts, extracted action items, and maintained searchable context across all your organization's meeting history?

This tutorial builds a production-ready agentic meeting transcription system that processes audio in real-time, uses RAG (Retrieval Augmented Generation) to incorporate historical context, and deploys multi-agent workflows to analyze meetings automatically. You'll implement:

• Real-time transcription agent using OpenAI Whisper
• Speaker diarization agent identifying who said what
• RAG integration with vector stores for historical meeting context
• Analysis agents for summarization and action item extraction
• LangGraph orchestration coordinating multiple AI agents
• Production deployment with Docker and Kubernetes

The complete code repository demonstrates enterprise patterns: error handling, observability, resource management, and horizontal scaling. This isn't a toy demo—it's production architecture you can deploy tomorrow.

Repository: github.com/CrashBytes/ByteSizedExamples/tree/main/agentic-meeting-transcription-tutorial

Architecture Overview: Multi-Agent Meeting Processing

Our system employs a hierarchical multi-agent architecture where specialized agents handle distinct aspects of meeting processing. This separation of concerns enables independent scaling, easier testing, and clear separation of responsibilities.

The Agent Hierarchy

Layer 1: Input Processing Agents

• Transcription Agent: Converts audio to text using Whisper
• Diarization Agent: Identifies speakers and timestamps

Layer 2: Context Retrieval Agents

• Vector Store Agent: Searches historical meetings
• Context Assembly Agent: Builds relevant context for analysis

Layer 3: Analysis Agents

• Summarization Agent: Generates meeting summaries at multiple detail levels
• Action Items Agent: Extracts and structures action items
• Decision Tracking Agent: Identifies and records decisions made

Layer 4: Orchestration

• LangGraph Coordinator: Manages agent workflows, handles state, coordinates execution

This hierarchy allows agents to work in parallel where possible (transcription and diarization can run simultaneously) while maintaining sequential dependencies where necessary (analysis requires completed transcription).

Technology Stack

Core AI/ML:

• OpenAI Whisper (speech-to-text)
• Pyannote.audio (speaker diarization)
• LangChain (agent framework)
• LangGraph (workflow orchestration)
• OpenAI GPT-4 (analysis and summarization)

Vector Store:

• Qdrant (vector database)
• SentenceTransformers (embeddings)

Backend:

• FastAPI (REST API and WebSocket)
• PostgreSQL (structured data)
• Redis (job queue and caching)

Frontend:

• Next.js 14 (React framework)
• TailwindCSS (styling)
• WebSocket (real-time updates)

Deployment:

• Docker (containerization)
• Kubernetes (orchestration)
• Prometheus + Grafana (monitoring)

Part 1: Building the Transcription Agent

The transcription agent converts audio to text using OpenAI's Whisper model. We'll implement real-time streaming transcription that processes audio as it arrives rather than waiting for the complete recording.

Audio Streaming with WebSocket

First, implement WebSocket audio streaming to handle real-time audio input:

# agents/audio_stream.py
import asyncio
import numpy as np
from fastapi import WebSocket
from typing import AsyncGenerator
import logging

logger = logging.getLogger(__name__)

class AudioStreamManager:
    """Manages real-time audio streaming via WebSocket"""

    def __init__(self, sample_rate: int = 16000, chunk_duration: float = 2.0):
        self.sample_rate = sample_rate
        self.chunk_size = int(sample_rate * chunk_duration)
        self.buffer = bytearray()

    async def stream_audio(
        self,
        websocket: WebSocket
    ) -> AsyncGenerator[np.ndarray, None]:
        """
        Stream audio chunks from WebSocket connection

        Yields:
            Audio chunks as numpy arrays ready for processing
        """
        try:
            while True:
                # Receive audio data from client
                data = await websocket.receive_bytes()
                self.buffer.extend(data)

                # Process complete chunks
                while len(self.buffer) >= self.chunk_size * 2:  # 2 bytes per sample
                    # Extract chunk
                    chunk_bytes = self.buffer[:self.chunk_size * 2]
                    self.buffer = self.buffer[self.chunk_size * 2:]

                    # Convert to numpy array
                    audio_chunk = np.frombuffer(chunk_bytes, dtype=np.int16)
                    audio_float = audio_chunk.astype(np.float32) / 32768.0

                    yield audio_float

        except Exception as e:
            logger.error(f"Audio streaming error: {e}")
            raise

Whisper Transcription Agent

Implement the transcription agent using Whisper for high-quality speech-to-text:

# agents/transcription_agent.py
import whisper
import torch
from typing import Dict, Optional
import logging

logger = logging.getLogger(__name__)

class TranscriptionAgent:
    """Agent for converting speech to text using Whisper"""

    def __init__(
        self,
        model_size: str = "base",
        device: Optional[str] = None,
        language: str = "en"
    ):
        """
        Initialize Whisper transcription agent

        Args:
            model_size: Whisper model size (tiny, base, small, medium, large)
            device: Computing device (cuda, cpu, or auto-detect)
            language: Target language for transcription
        """
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.model = whisper.load_model(model_size, device=self.device)
        self.language = language

        logger.info(f"Transcription agent initialized with {model_size} model on {self.device}")

    async def transcribe_chunk(
        self,
        audio: np.ndarray,
        temperature: float = 0.0
    ) -> Dict[str, any]:
        """
        Transcribe audio chunk to text

        Args:
            audio: Audio data as numpy array
            temperature: Sampling temperature (0 = deterministic)

        Returns:
            Dictionary with transcription results
        """
        try:
            # Run Whisper transcription
            result = self.model.transcribe(
                audio,
                language=self.language,
                temperature=temperature,
                no_speech_threshold=0.6,
                logprob_threshold=-1.0
            )

            return {
                "text": result["text"].strip(),
                "language": result["language"],
                "segments": result["segments"],
                "confidence": self._calculate_confidence(result)
            }

        except Exception as e:
            logger.error(f"Transcription error: {e}")
            return {
                "text": "",
                "error": str(e),
                "confidence": 0.0
            }

    def _calculate_confidence(self, result: Dict) -> float:
        """Calculate average confidence from segment probabilities"""
        if not result.get("segments"):
            return 0.0

        confidences = [
            segment.get("avg_logprob", -1.0)
            for segment in result["segments"]
        ]

        # Convert log probabilities to confidence score
        avg_logprob = sum(confidences) / len(confidences)
        confidence = min(1.0, max(0.0, (avg_logprob + 1.0)))

        return confidence

Real-Time Transcription Pipeline

Combine streaming and transcription into a pipeline:

# agents/transcription_pipeline.py
from typing import AsyncGenerator, Dict
import asyncio

class TranscriptionPipeline:
    """Pipeline combining audio streaming and transcription"""

    def __init__(self, transcription_agent: TranscriptionAgent):
        self.agent = transcription_agent

    async def process_stream(
        self,
        audio_stream: AsyncGenerator[np.ndarray, None]
    ) -> AsyncGenerator[Dict, None]:
        """
        Process audio stream and yield transcription results

        Args:
            audio_stream: Generator yielding audio chunks

        Yields:
            Transcription results for each chunk
        """
        async for audio_chunk in audio_stream:
            # Transcribe chunk
            result = await self.agent.transcribe_chunk(audio_chunk)

            if result["text"]:
                yield {
                    "timestamp": asyncio.get_event_loop().time(),
                    "text": result["text"],
                    "confidence": result["confidence"],
                    "language": result.get("language", "en")
                }

Part 2: Speaker Diarization Agent

Speaker diarization identifies "who spoke when" in meeting recordings. We'll use Pyannote.audio, which provides state-of-the-art diarization capabilities.

Diarization Agent Implementation

# agents/diarization_agent.py
from pyannote.audio import Pipeline
from pyannote.core import Annotation, Segment
import torch
from typing import Dict, List
import logging

logger = logging.getLogger(__name__)

class DiarizationAgent:
    """Agent for speaker diarization using Pyannote"""

    def __init__(
        self,
        auth_token: str,
        device: Optional[str] = None,
        num_speakers: Optional[int] = None
    ):
        """
        Initialize diarization agent

        Args:
            auth_token: Hugging Face auth token for Pyannote models
            device: Computing device (cuda or cpu)
            num_speakers: Expected number of speakers (optional)
        """
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.num_speakers = num_speakers

        # Load Pyannote pipeline
        self.pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1",
            use_auth_token=auth_token
        ).to(torch.device(self.device))

        logger.info(f"Diarization agent initialized on {self.device}")

    async def diarize(
        self,
        audio_file: str,
        min_speakers: int = 1,
        max_speakers: int = 10
    ) -> Dict[str, any]:
        """
        Perform speaker diarization on audio file

        Args:
            audio_file: Path to audio file
            min_speakers: Minimum expected speakers
            max_speakers: Maximum expected speakers

        Returns:
            Diarization results with speaker segments
        """
        try:
            # Run diarization
            diarization = self.pipeline(
                audio_file,
                num_speakers=self.num_speakers,
                min_speakers=min_speakers,
                max_speakers=max_speakers
            )

            # Process results into structured format
            segments = self._process_diarization(diarization)

            return {
                "speakers": list(set(seg["speaker"] for seg in segments)),
                "segments": segments,
                "num_speakers": len(set(seg["speaker"] for seg in segments))
            }

        except Exception as e:
            logger.error(f"Diarization error: {e}")
            return {
                "speakers": [],
                "segments": [],
                "error": str(e)
            }

    def _process_diarization(
        self,
        diarization: Annotation
    ) -> List[Dict]:
        """Convert Pyannote annotation to structured segments"""
        segments = []

        for turn, _, speaker in diarization.itertracks(yield_label=True):
            segments.append({
                "speaker": speaker,
                "start": turn.start,
                "end": turn.end,
                "duration": turn.end - turn.start
            })

        return sorted(segments, key=lambda x: x["start"])

Combining Transcription and Diarization

Merge transcription text with speaker information:

# agents/transcript_assembler.py
from typing import Dict, List

class TranscriptAssembler:
    """Combines transcription and diarization into attributed transcript"""

    @staticmethod
    def merge_transcripts(
        transcription: Dict,
        diarization: Dict
    ) -> List[Dict]:
        """
        Merge transcription segments with speaker labels

        Args:
            transcription: Transcription results with segments
            diarization: Diarization results with speaker segments

        Returns:
            List of segments with text and speaker attribution
        """
        merged = []

        for trans_seg in transcription.get("segments", []):
            trans_start = trans_seg["start"]
            trans_end = trans_seg["end"]
            trans_text = trans_seg["text"]

            # Find overlapping speaker segment
            speaker = "Unknown"
            max_overlap = 0

            for diar_seg in diarization.get("segments", []):
                overlap = TranscriptAssembler._calculate_overlap(
                    trans_start, trans_end,
                    diar_seg["start"], diar_seg["end"]
                )

                if overlap > max_overlap:
                    max_overlap = overlap
                    speaker = diar_seg["speaker"]

            merged.append({
                "speaker": speaker,
                "start": trans_start,
                "end": trans_end,
                "text": trans_text,
                "confidence": trans_seg.get("confidence", 0.0)
            })

        return merged

    @staticmethod
    def _calculate_overlap(
        start1: float, end1: float,
        start2: float, end2: float
    ) -> float:
        """Calculate temporal overlap between two segments"""
        overlap_start = max(start1, start2)
        overlap_end = min(end1, end2)
        overlap = max(0, overlap_end - overlap_start)

        duration1 = end1 - start1
        return overlap / duration1 if duration1 > 0 else 0

Part 3: RAG Integration for Meeting Context

RAG (Retrieval Augmented Generation) enhances meeting analysis by providing relevant context from historical meetings. We'll implement vector storage and semantic search across meeting history.

Vector Store Setup

# agents/vector_store.py
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer
from typing import List, Dict
import uuid
import logging

logger = logging.getLogger(__name__)

class MeetingVectorStore:
    """Vector store for meeting transcripts and context"""

    def __init__(
        self,
        qdrant_url: str = "localhost:6333",
        collection_name: str = "meetings",
        embedding_model: str = "all-MiniLM-L6-v2"
    ):
        """
        Initialize vector store for meetings

        Args:
            qdrant_url: Qdrant server URL
            collection_name: Collection name for meeting vectors
            embedding_model: SentenceTransformer model name
        """
        self.client = QdrantClient(url=qdrant_url)
        self.collection_name = collection_name
        self.encoder = SentenceTransformer(embedding_model)
        self.embedding_dim = self.encoder.get_sentence_embedding_dimension()

        self._initialize_collection()

    def _initialize_collection(self):
        """Create Qdrant collection if it doesn't exist"""
        try:
            self.client.get_collection(self.collection_name)
            logger.info(f"Using existing collection: {self.collection_name}")
        except:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.embedding_dim,
                    distance=Distance.COSINE
                )
            )
            logger.info(f"Created new collection: {self.collection_name}")

    async def store_meeting(
        self,
        meeting_id: str,
        transcript: List[Dict],
        metadata: Dict
    ):
        """
        Store meeting transcript in vector store

        Args:
            meeting_id: Unique meeting identifier
            transcript: List of transcript segments
            metadata: Meeting metadata (date, participants, etc.)
        """
        points = []

        for idx, segment in enumerate(transcript):
            # Create embedding for segment text
            embedding = self.encoder.encode(segment["text"]).tolist()

            # Create point for storage
            point = PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload={
                    "meeting_id": meeting_id,
                    "segment_index": idx,
                    "speaker": segment.get("speaker", "Unknown"),
                    "text": segment["text"],
                    "timestamp": segment.get("start", 0),
                    "metadata": metadata
                }
            )
            points.append(point)

        # Batch upload to Qdrant
        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )

        logger.info(f"Stored {len(points)} segments for meeting {meeting_id}")

Context Retrieval Agent

# agents/context_retrieval_agent.py
from typing import List, Dict
import logging

logger = logging.getLogger(__name__)

class ContextRetrievalAgent:
    """Agent for retrieving relevant meeting context"""

    def __init__(self, vector_store: MeetingVectorStore):
        self.vector_store = vector_store

    async def retrieve_context(
        self,
        query: str,
        limit: int = 5,
        score_threshold: float = 0.7
    ) -> List[Dict]:
        """
        Retrieve relevant context from historical meetings

        Args:
            query: Search query
            limit: Maximum number of results
            score_threshold: Minimum similarity score

        Returns:
            List of relevant meeting segments
        """
        # Generate query embedding
        query_vector = self.vector_store.encoder.encode(query).tolist()

        # Search vector store
        results = self.vector_store.client.search(
            collection_name=self.vector_store.collection_name,
            query_vector=query_vector,
            limit=limit,
            score_threshold=score_threshold
        )

        # Format results
        context = []
        for result in results:
            context.append({
                "text": result.payload["text"],
                "speaker": result.payload["speaker"],
                "meeting_id": result.payload["meeting_id"],
                "score": result.score,
                "metadata": result.payload.get("metadata", {})
            })

        logger.info(f"Retrieved {len(context)} context segments for query: {query}")
        return context

Part 4: Analysis Agents with LangChain

Analysis agents process completed transcripts to extract insights, generate summaries, and identify action items. We'll implement these using LangChain for structured LLM interactions.

Summarization Agent

# agents/summarization_agent.py
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import HumanMessage, SystemMessage
from typing import Dict, List
import logging

logger = logging.getLogger(__name__)

class SummarizationAgent:
    """Agent for generating meeting summaries at multiple detail levels"""

    def __init__(
        self,
        model_name: str = "gpt-4",
        temperature: float = 0.3
    ):
        self.llm = ChatOpenAI(
            model_name=model_name,
            temperature=temperature
        )

    async def summarize(
        self,
        transcript: List[Dict],
        context: List[Dict] = None,
        detail_level: str = "medium"
    ) -> Dict[str, str]:
        """
        Generate meeting summary

        Args:
            transcript: Meeting transcript segments
            context: Historical context from RAG
            detail_level: Summary detail (brief, medium, detailed)

        Returns:
            Dictionary with summaries at different levels
        """
        # Format transcript for summarization
        transcript_text = self._format_transcript(transcript)
        context_text = self._format_context(context) if context else ""

        # Generate summaries at different levels
        summaries = {}

        if detail_level in ["brief", "all"]:
            summaries["brief"] = await self._generate_brief_summary(
                transcript_text, context_text
            )

        if detail_level in ["medium", "all"]:
            summaries["medium"] = await self._generate_medium_summary(
                transcript_text, context_text
            )

        if detail_level in ["detailed", "all"]:
            summaries["detailed"] = await self._generate_detailed_summary(
                transcript_text, context_text
            )

        return summaries

    async def _generate_brief_summary(
        self,
        transcript: str,
        context: str
    ) -> str:
        """Generate brief 2-3 sentence summary"""
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=(
                "You are a meeting summarization expert. "
                "Generate a brief 2-3 sentence summary of the meeting covering "
                "the main topic and key outcomes."
            )),
            HumanMessage(content=(
                f"Meeting transcript:\n{transcript}\n\n"
                f"Historical context:\n{context}\n\n"
                "Provide a brief summary:"
            ))
        ])

        response = await self.llm.apredict_messages(prompt.format_messages())
        return response.content

    def _format_transcript(self, transcript: List[Dict]) -> str:
        """Format transcript segments for LLM consumption"""
        formatted = []
        for segment in transcript:
            speaker = segment.get("speaker", "Unknown")
            text = segment.get("text", "")
            formatted.append(f"{speaker}: {text}")
        return "\n".join(formatted)

Action Items Extraction Agent

# agents/action_items_agent.py
from langchain.chat_models import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
import logging

logger = logging.getLogger(__name__)

class ActionItem(BaseModel):
    """Structured action item"""
    description: str = Field(description="Description of the action item")
    assignee: Optional[str] = Field(description="Person assigned to the action")
    due_date: Optional[str] = Field(description="Due date if mentioned")
    priority: str = Field(description="Priority level: high, medium, or low")
    context: str = Field(description="Relevant context from the meeting")

class ActionItemsList(BaseModel):
    """List of action items"""
    items: List[ActionItem] = Field(description="List of action items")

class ActionItemsAgent:
    """Agent for extracting structured action items from meetings"""

    def __init__(
        self,
        model_name: str = "gpt-4",
        temperature: float = 0.2
    ):
        self.llm = ChatOpenAI(
            model_name=model_name,
            temperature=temperature
        )
        self.parser = PydanticOutputParser(pydantic_object=ActionItemsList)

    async def extract_action_items(
        self,
        transcript: List[Dict]
    ) -> List[ActionItem]:
        """
        Extract action items from meeting transcript

        Args:
            transcript: Meeting transcript segments

        Returns:
            List of structured action items
        """
        transcript_text = self._format_transcript(transcript)

        prompt = f"""
        Analyze the following meeting transcript and extract all action items.

        For each action item, identify:
        - Clear description of what needs to be done
        - Who is assigned (if mentioned)
        - Due date (if mentioned)
        - Priority level (high, medium, low)
        - Relevant context from the discussion

        Transcript:
        {transcript_text}

        {self.parser.get_format_instructions()}
        """

        response = await self.llm.apredict(prompt)
        result = self.parser.parse(response)

        logger.info(f"Extracted {len(result.items)} action items")
        return result.items

Part 5: LangGraph Orchestration

LangGraph coordinates multiple agents into a coherent workflow, managing state and handling complex execution patterns.

Meeting Processing Workflow

# orchestration/meeting_workflow.py
from langgraph.graph import StateGraph, END
from typing import Dict, List, TypedDict
import logging

logger = logging.getLogger(__name__)

class MeetingState(TypedDict):
    """State for meeting processing workflow"""
    audio_file: str
    transcript: List[Dict]
    diarization: Dict
    attributed_transcript: List[Dict]
    context: List[Dict]
    summaries: Dict[str, str]
    action_items: List[Dict]
    status: str
    error: Optional[str]

class MeetingWorkflow:
    """LangGraph workflow for orchestrating meeting processing"""

    def __init__(
        self,
        transcription_agent,
        diarization_agent,
        context_agent,
        summarization_agent,
        action_items_agent
    ):
        self.transcription_agent = transcription_agent
        self.diarization_agent = diarization_agent
        self.context_agent = context_agent
        self.summarization_agent = summarization_agent
        self.action_items_agent = action_items_agent

        self.workflow = self._build_workflow()

    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow"""
        workflow = StateGraph(MeetingState)

        # Add nodes
        workflow.add_node("transcribe", self._transcribe_node)
        workflow.add_node("diarize", self._diarize_node)
        workflow.add_node("merge", self._merge_node)
        workflow.add_node("retrieve_context", self._context_node)
        workflow.add_node("summarize", self._summarize_node)
        workflow.add_node("extract_actions", self._actions_node)

        # Define edges
        workflow.set_entry_point("transcribe")
        workflow.add_edge("transcribe", "diarize")
        workflow.add_edge("diarize", "merge")
        workflow.add_edge("merge", "retrieve_context")
        workflow.add_edge("retrieve_context", "summarize")
        workflow.add_edge("summarize", "extract_actions")
        workflow.add_edge("extract_actions", END)

        return workflow.compile()

    async def _transcribe_node(self, state: MeetingState) -> MeetingState:
        """Transcription node"""
        try:
            result = await self.transcription_agent.transcribe_file(
                state["audio_file"]
            )
            state["transcript"] = result["segments"]
            state["status"] = "transcribed"
        except Exception as e:
            state["error"] = f"Transcription failed: {str(e)}"
            logger.error(state["error"])
        return state

    async def process_meeting(self, audio_file: str) -> MeetingState:
        """
        Process meeting through complete workflow

        Args:
            audio_file: Path to meeting audio file

        Returns:
            Final workflow state with all results
        """
        initial_state = MeetingState(
            audio_file=audio_file,
            transcript=[],
            diarization={},
            attributed_transcript=[],
            context=[],
            summaries={},
            action_items=[],
            status="pending",
            error=None
        )

        final_state = await self.workflow.ainvoke(initial_state)
        return final_state

Part 6: FastAPI Backend Service

The production backend exposes our agentic system via REST and WebSocket APIs.

Complete FastAPI Application

# main.py
from fastapi import FastAPI, WebSocket, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Dict, Optional
import uvicorn
import logging

# Import agents and orchestration
from agents import (
    TranscriptionAgent,
    DiarizationAgent,
    MeetingVectorStore,
    ContextRetrievalAgent,
    SummarizationAgent,
    ActionItemsAgent
)
from orchestration import MeetingWorkflow

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize FastAPI app
app = FastAPI(
    title="Agentic Meeting Transcription API",
    description="Production API for AI-powered meeting transcription and analysis",
    version="1.0.0"
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize agents (configure with environment variables in production)
transcription_agent = TranscriptionAgent(model_size="base")
diarization_agent = DiarizationAgent(auth_token=os.getenv("PYANNOTE_TOKEN"))
vector_store = MeetingVectorStore(qdrant_url=os.getenv("QDRANT_URL"))
context_agent = ContextRetrievalAgent(vector_store)
summarization_agent = SummarizationAgent()
action_items_agent = ActionItemsAgent()

# Initialize workflow
workflow = MeetingWorkflow(
    transcription_agent=transcription_agent,
    diarization_agent=diarization_agent,
    context_agent=context_agent,
    summarization_agent=summarization_agent,
    action_items_agent=action_items_agent
)

# Request/Response models
class ProcessMeetingRequest(BaseModel):
    audio_url: str
    meeting_metadata: Optional[Dict] = None

class ProcessMeetingResponse(BaseModel):
    meeting_id: str
    status: str
    transcript: List[Dict]
    summaries: Dict[str, str]
    action_items: List[Dict]

# API endpoints
@app.post("/api/meetings/process", response_model=ProcessMeetingResponse)
async def process_meeting(request: ProcessMeetingRequest):
    """
    Process meeting audio through complete agentic workflow

    Returns:
        Complete meeting analysis including transcript, summaries, and action items
    """
    try:
        # Process meeting through workflow
        result = await workflow.process_meeting(request.audio_url)

        if result["error"]:
            raise HTTPException(status_code=500, detail=result["error"])

        # Store in vector database for future context
        meeting_id = str(uuid.uuid4())
        await vector_store.store_meeting(
            meeting_id=meeting_id,
            transcript=result["attributed_transcript"],
            metadata=request.meeting_metadata or {}
        )

        return ProcessMeetingResponse(
            meeting_id=meeting_id,
            status=result["status"],
            transcript=result["attributed_transcript"],
            summaries=result["summaries"],
            action_items=result["action_items"]
        )

    except Exception as e:
        logger.error(f"Meeting processing error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.websocket("/ws/transcribe")
async def websocket_transcribe(websocket: WebSocket):
    """
    Real-time transcription via WebSocket

    Accepts audio stream and returns transcription chunks in real-time
    """
    await websocket.accept()

    try:
        stream_manager = AudioStreamManager()
        audio_stream = stream_manager.stream_audio(websocket)
        pipeline = TranscriptionPipeline(transcription_agent)

        async for result in pipeline.process_stream(audio_stream):
            await websocket.send_json(result)

    except Exception as e:
        logger.error(f"WebSocket error: {e}")
        await websocket.close(code=1011, reason=str(e))

@app.get("/api/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "agents": {
            "transcription": "ready",
            "diarization": "ready",
            "summarization": "ready"
        }
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Part 7: Production Deployment with Docker and Kubernetes

Deploy the system to production using containers and orchestration.

Dockerfile

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    ffmpeg \
    libsndfile1 \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Download Whisper models
RUN python -c "import whisper; whisper.load_model('base')"

# Expose port
EXPOSE 8000

# Run application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose for Local Development

# docker-compose.yml
version: '3.8'

services:
  api:
    build: .
    ports:
      - '8000:8000'
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - PYANNOTE_TOKEN=${PYANNOTE_TOKEN}
      - QDRANT_URL=http://qdrant:6333
      - POSTGRES_URL=postgresql://user:password@postgres:5432/meetings
    depends_on:
      - qdrant
      - postgres
      - redis
    volumes:
      - ./data:/app/data

  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - '6333:6333'
    volumes:
      - qdrant_data:/qdrant/storage

  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: meetings
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'

volumes:
  qdrant_data:
  postgres_data:

Kubernetes Deployment

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: meeting-transcription-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: meeting-api
  template:
    metadata:
      labels:
        app: meeting-api
    spec:
      containers:
        - name: api
          image: crashbytes/meeting-transcription:latest
          ports:
            - containerPort: 8000
          env:
            - name: OPENAI_API_KEY
              valueFrom:
                secretKeyRef:
                  name: api-secrets
                  key: openai-api-key
            - name: QDRANT_URL
              value: 'http://qdrant-service:6333'
          resources:
            requests:
              memory: '2Gi'
              cpu: '1000m'
            limits:
              memory: '4Gi'
              cpu: '2000m'
          livenessProbe:
            httpGet:
              path: /api/health
              port: 8000
            initialDelaySeconds: 30
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /api/health
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: meeting-api-service
spec:
  selector:
    app: meeting-api
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000
  type: LoadBalancer

Conclusion: Production-Ready Agentic Systems

You now have a complete production-ready agentic meeting transcription system featuring:

✅ Real-time transcription with OpenAI Whisper
✅ Speaker diarization identifying who spoke when
✅ RAG integration providing historical meeting context
✅ Multi-agent analysis generating summaries and action items
✅ LangGraph orchestration coordinating complex workflows
✅ Production deployment with Docker and Kubernetes

The complete code repository includes:

• All agent implementations with error handling
• FastAPI service with REST and WebSocket APIs
• Docker containerization
• Kubernetes manifests for production deployment
• Comprehensive tests and monitoring setup
• Documentation and deployment guides

Repository: github.com/CrashBytes/ByteSizedExamples/tree/main/agentic-meeting-transcription-tutorial

This architecture scales to enterprise requirements while remaining maintainable and observable. The separation of concerns through specialized agents enables independent scaling, testing, and improvement of each component.

Deploy this system to transform how your organization handles meetings—eliminating manual note-taking, ensuring accurate records, and automatically extracting actionable insights from every conversation.

---

Next Steps

Enhancements to consider:

1. Add sentiment analysis agents for detecting meeting tone
2. Implement decision tracking agents for capturing commitments
3. Build integration agents for syncing with Slack, email, and project management tools
4. Add multilingual support for international teams
5. Implement automated follow-up scheduling based on action items

Resources:

• LangChain documentation: python.langchain.com
• LangGraph tutorials: langchain-ai.github.io/langgraph
• Whisper documentation: github.com/openai/whisper
• Pyannote audio: github.com/pyannote/pyannote-audio
• Qdrant vector database: qdrant.tech

Related Tutorials:

• "Building Enterprise RAG Systems: Production Architecture and Deployment"
• "Multi-Agent AI Systems: LangGraph Patterns and Best Practices"
• "Real-Time AI Applications: WebSocket Architecture at Scale"
• "Vector Databases for Production AI: Qdrant, Pinecone, and Weaviate Compared"