Реалізація AI-автоматизації протоколювання нарад

Автоматизація протоколювання зустрічей з AI Система автоматично транскрибує записи зустрічей, визначає учасників, виділяє рішення та завдання, формує структурований протокол. Інтегрується із Zoom, Google Meet, MS Teams, Slack, корпоративними системами управління завданнями. ### Архітектура пайплайна```

Запись встречи (аудио/видео) ↓ VAD + диаризация (pyannote.audio) ↓ Транскрибация (Whisper large-v3) ↓ NLP-обработка (GPT-4o / Claude) ↓ Извлечение: решения / задачи / ответственные / дедлайны ↓ Генерация протокола (Markdown / Notion / Jira) ### Транскрибація з діаризацієюpython import whisper from pyannote.audio import Pipeline import torch

class MeetingTranscriber: def init(self): self.whisper = whisper.load_model("large-v3", device="cuda") self.diarizer = Pipeline.from_pretrained( "pyannote/speaker-diarization-3.1", use_auth_token="HF_TOKEN" )

def transcribe_with_speakers(self, audio_path: str) -> list[dict]:
    # Диаризация
    diarization = self.diarizer(audio_path)
    segments_by_speaker = [
        {"speaker": turn.speaker, "start": turn.start, "end": turn.end}
        for turn, _, _ in diarization.itertracks(yield_label=True)
    ]

    # Транскрипция
    result = self.whisper.transcribe(audio_path, language="ru", word_timestamps=True)

    # Мёржим
    transcript = []
    for seg in result["segments"]:
        speaker = self._find_speaker(seg["start"], segments_by_speaker)
        transcript.append({
            "speaker": speaker,
            "start": seg["start"],
            "end": seg["end"],
            "text": seg["text"].strip()
        })

    return transcript

def _find_speaker(self, timestamp: float, diar_segments: list) -> str:
    for s in diar_segments:
        if s["start"] <= timestamp <= s["end"]:
            return s["speaker"]
    return "UNKNOWN"

### NLP-обробка та вилучення структуриpython from openai import AsyncOpenAI import json

client = AsyncOpenAI()

async def extract_meeting_structure(transcript: list[dict]) -> dict: # Форматируем транскрипт formatted = "\n".join([ f"[{seg['speaker']} | {int(seg['start']//60):02d}:{int(seg['start']%60):02d}] {seg['text']}" for seg in transcript ])

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "system",
        "content": """Ты — ассистент для протоколирования встреч.
        Проанализируй транскрипт и верни JSON:
        {
          "summary": "краткое резюме 2-3 предложения",
          "participants": ["SPEAKER_00 = Иван Петров", ...],
          "agenda_items": [{"topic": "...", "discussion": "..."}],
          "decisions": [{"decision": "...", "context": "..."}],
          "action_items": [{"task": "...", "owner": "...", "deadline": "..."}],
          "next_meeting": "дата/условие следующей встречи если обсуждалась"
        }"""
    }, {
        "role": "user",
        "content": f"Транскрипт встречи:\n\n{formatted[:8000]}"
    }],
    response_format={"type": "json_object"}
)

return json.loads(response.choices[0].message.content)

### Генерація протоколуpython def format_meeting_minutes(structure: dict, transcript: list[dict]) -> str: date = datetime.now().strftime("%d.%m.%Y") duration_min = int(transcript[-1]["end"] / 60) if transcript else 0

md = f"""## Протокол встречи от {date}

Продолжительность: {duration_min} минут Участники: {", ".join(structure.get("participants", []))}

Краткое резюме

{structure.get("summary", "")}

Принятые решения

""" for d in structure.get("decisions", []): md += f"- {d['decision']}\n {d.get('context', '')}\n\n"

md += "### Задачи\n\n"
md += "| Задача | Ответственный | Срок |\n|--------|--------------|------|\n"
for item in structure.get("action_items", []):
    md += f"| {item['task']} | {item.get('owner', '—')} | {item.get('deadline', '—')} |\n"

return md

### Інтеграції **Zoom:** через Zoom Recordings API - автоматично отримуємо mp4/m4a після закінчення зустрічі. **Google Meet:** через Google Workspace Events API + Cloud Storage для записів. **MS Teams:** через Microsoft Graph API `/communications/callRecords`.python @app.post("/webhook/zoom/recording") async def zoom_recording_webhook(payload: dict): if payload["event"] == "recording.completed": recording_url = payload["payload"]["object"]["recording_files"][0]["download_url"] meeting_id = payload["payload"]["object"]["uuid"] asyncio.create_task(process_meeting_recording(meeting_id, recording_url)) return {"status": "ok"} ### Експортpython class MinutesExporter: async def to_notion(self, minutes: str, database_id: str): ... async def to_confluence(self, minutes: str, space_key: str): ... async def to_jira_tasks(self, action_items: list, project_key: str): ... async def to_slack(self, summary: str, channel_id: str): ... async def to_email(self, minutes: str, recipients: list[str]): ...