🎯 왜 멀티에이전트가 필요할까?

🏗️ 멀티에이전트 시스템 구조 이해하기

📐 기본 아키텍처

🔄 에이전트 간 통신 방식

💻 실전! 멀티에이전트 시스템 구축하기

🛠️ 필요한 도구들

📝 Step 1: 음성인식 에이전트 만들기

import openai
from pathlib import Path

class TranscriptionAgent:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.name = "음성인식 에이전트"
    
    def transcribe_audio(self, audio_file_path):
        """
        음성 파일을 텍스트로 변환
        """
        print(f"🎤 {self.name}: 음성 파일 분석 시작...")
        
        with open(audio_file_path, "rb") as audio_file:
            transcript = openai.Audio.transcribe(
                model="whisper-1",
                file=audio_file,
                language="ko"  # 한국어 지정
            )
        
        print(f"✅ {self.name}: 변환 완료!")
        return transcript['text']
    
    def transcribe_with_timestamps(self, audio_file_path):
        """
        타임스탬프와 함께 변환 (화자 구분에 유용)
        """
        with open(audio_file_path, "rb") as audio_file:
            transcript = openai.Audio.transcribe(
                model="whisper-1",
                file=audio_file,
                response_format="verbose_json",
                language="ko"
            )
        
        # 세그먼트별로 정리
        segments = []
        for segment in transcript['segments']:
            segments.append({
                'start': segment['start'],
                'end': segment['end'],
                'text': segment['text']
            })
        
        return segments

# 사용 예시
agent = TranscriptionAgent(api_key="your-api-key")
text = agent.transcribe_audio("meeting_audio.mp3")
print(text)

✍️ Step 2: 회의록 작성 에이전트 만들기

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

class MeetingMinutes(BaseModel):
    """회의록 데이터 모델"""
    title: str = Field(description="회의 제목")
    date: str = Field(description="회의 날짜")
    attendees: List[str] = Field(description="참석자 목록")
    agenda_items: List[dict] = Field(description="안건별 내용")
    decisions: List[str] = Field(description="결정 사항")
    action_items: List[dict] = Field(description="액션 아이템")

class MinutesWriterAgent:
    def __init__(self, api_key):
        self.llm = ChatOpenAI(
            model="gpt-4",
            temperature=0.3,  # 창의성보다 정확성 중시
            openai_api_key=api_key
        )
        self.name = "회의록 작성 에이전트"
        self.parser = PydanticOutputParser(pydantic_object=MeetingMinutes)
    
    def create_minutes(self, transcript_text):
        """
        텍스트를 구조화된 회의록으로 변환
        """
        print(f"✍️ {self.name}: 회의록 작성 중...")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """당신은 전문 회의록 작성자입니다.
            주어진 회의 내용을 분석하여 체계적인 회의록을 작성하세요.
            
            다음 형식을 따라주세요:
            {format_instructions}
            
            중요한 점:
            1. 안건별로 명확하게 구분하세요
            2. 결정사항과 논의사항을 구분하세요
            3. 액션아이템은 담당자와 기한을 포함하세요
            4. 객관적이고 간결하게 작성하세요"""),
            ("user", "회의 내용:\n{transcript}")
        ])
        
        chain = prompt | self.llm
        
        response = chain.invoke({
            "transcript": transcript_text,
            "format_instructions": self.parser.get_format_instructions()
        })
        
        # 파싱
        minutes = self.parser.parse(response.content)
        
        print(f"✅ {self.name}: 회의록 작성 완료!")
        return minutes
    
    def format_to_markdown(self, minutes: MeetingMinutes):
        """
        회의록을 마크다운 형식으로 변환
        """
        md = f"""# {minutes.title}

## 📅 회의 정보
- **날짜**: {minutes.date}
- **참석자**: {', '.join(minutes.attendees)}

## 📋 안건 및 논의 내용
"""
        for i, item in enumerate(minutes.agenda_items, 1):
            md += f"\n### {i}. {item['title']}\n"
            md += f"{item['content']}\n"
        
        md += "\n## ✅ 결정 사항\n"
        for i, decision in enumerate(minutes.decisions, 1):
            md += f"{i}. {decision}\n"
        
        md += "\n## 🎯 액션 아이템\n"
        for item in minutes.action_items:
            md += f"- [ ] {item['task']} (담당: {item['assignee']}, 기한: {item['deadline']})\n"
        
        return md

# 사용 예시
writer = MinutesWriterAgent(api_key="your-api-key")
minutes = writer.create_minutes(transcript_text)
markdown = writer.format_to_markdown(minutes)
print(markdown)

📊 Step 3: 요약 및 분석 에이전트 만들기

from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document

class SummaryAgent:
    def __init__(self, api_key):
        self.llm = ChatOpenAI(
            model="gpt-4",
            temperature=0.2,
            openai_api_key=api_key
        )
        self.name = "요약 분석 에이전트"
    
    def create_executive_summary(self, minutes_text):
        """
        경영진용 요약본 생성 (핵심만!)
        """
        print(f"📊 {self.name}: 경영진 요약본 작성 중...")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """당신은 경영진을 위한 요약 전문가입니다.
            회의록을 읽고 다음 내용을 포함한 간결한 요약을 작성하세요:
            
            1. 회의의 핵심 목적 (1-2문장)
            2. 주요 결정사항 (3-5개 bullet points)
            3. 중요한 액션아이템 (우선순위 높은 것만)
            4. 리스크나 이슈 (있다면)
            
            전체 길이는 200단어 이내로 작성하세요."""),
            ("user", "{minutes}")
        ])
        
        chain = prompt | self.llm
        summary = chain.invoke({"minutes": minutes_text})
        
        return summary.content
    
    def extract_key_insights(self, minutes_text):
        """
        회의에서 도출된 인사이트 추출
        """
        print(f"💡 {self.name}: 인사이트 추출 중...")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """회의 내용을 분석하여 다음을 추출하세요:
            
            1. 새로운 아이디어나 제안
            2. 발견된 문제점이나 개선점
            3. 향후 기회 요소
            4. 주의해야 할 리스크
            
            각 항목을 명확하게 구분하여 작성하세요."""),
            ("user", "{minutes}")
        ])
        
        chain = prompt | self.llm
        insights = chain.invoke({"minutes": minutes_text})
        
        return insights.content
    
    def analyze_sentiment(self, minutes_text):
        """
        회의 분위기 분석
        """
        print(f"😊 {self.name}: 회의 분위기 분석 중...")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """회의 내용의 전반적인 분위기를 분석하세요:
            
            - 긍정적/부정적/중립적 분위기
            - 참석자들의 참여도
            - 의견 일치도
            - 전반적인 에너지 레벨
            
            간단하게 요약해주세요."""),
            ("user", "{minutes}")
        ])
        
        chain = prompt | self.llm
        sentiment = chain.invoke({"minutes": minutes_text})
        
        return sentiment.content

# 사용 예시
summary_agent = SummaryAgent(api_key="your-api-key")
executive_summary = summary_agent.create_executive_summary(markdown)
insights = summary_agent.extract_key_insights(markdown)
sentiment = summary_agent.analyze_sentiment(markdown)

print("=== 경영진 요약 ===")
print(executive_summary)
print("\n=== 주요 인사이트 ===")
print(insights)
print("\n=== 회의 분위기 ===")
print(sentiment)

🔗 Step 4: 에이전트들을 연결하기

import asyncio
from datetime import datetime
import json

class MeetingMinutesOrchestrator:
    """
    모든 에이전트를 조율하는 오케스트레이터
    """
    def __init__(self, api_key):
        self.transcription_agent = TranscriptionAgent(api_key)
        self.writer_agent = MinutesWriterAgent(api_key)
        self.summary_agent = SummaryAgent(api_key)
        self.results = {}
    
    async def process_meeting(self, audio_file_path, output_dir="./output"):
        """
        전체 프로세스를 실행
        """
        print("🚀 회의록 자동화 시스템 시작!\n")
        start_time = datetime.now()
        
        # Step 1: 음성 인식
        print("=" * 50)
        transcript = self.transcription_agent.transcribe_audio(audio_file_path)
        self.results['transcript'] = transcript
        print(f"📝 변환된 텍스트 길이: {len(transcript)} 글자\n")
        
        # Step 2: 회의록 작성
        print("=" * 50)
        minutes = self.writer_agent.create_minutes(transcript)
        markdown = self.writer_agent.format_to_markdown(minutes)
        self.results['minutes'] = markdown
        print(f"📄 회의록 작성 완료\n")
        
        # Step 3: 요약 및 분석 (병렬 처리)
        print("=" * 50)
        print("📊 요약 및 분석 시작 (병렬 처리)...")
        
        summary_tasks = [
            self.summary_agent.create_executive_summary(markdown),
            self.summary_agent.extract_key_insights(markdown),
            self.summary_agent.analyze_sentiment(markdown)
        ]
        
        # 비동기로 동시 실행
        exec_summary, insights, sentiment = await asyncio.gather(*summary_tasks)
        
        self.results['executive_summary'] = exec_summary
        self.results['insights'] = insights
        self.results['sentiment'] = sentiment
        
        # Step 4: 결과 저장
        print("\n" + "=" * 50)
        print("💾 결과 저장 중...")
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        # 전체 회의록 저장
        with open(f"{output_dir}/minutes_{timestamp}.md", "w", encoding="utf-8") as f:
            f.write(markdown)
        
        # 요약본 저장
        summary_content = f"""# 회의 요약 보고서
        
## 📋 경영진 요약
{exec_summary}

## 💡 주요 인사이트
{insights}

## 😊 회의 분위기
{sentiment}
"""
        with open(f"{output_dir}/summary_{timestamp}.md", "w", encoding="utf-8") as f:
            f.write(summary_content)
        
        # JSON 형식으로도 저장
        with open(f"{output_dir}/data_{timestamp}.json", "w", encoding="utf-8") as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
        
        end_time = datetime.now()
        duration = (end_time - start_time).total_seconds()
        
        print(f"✅ 모든 작업 완료!")
        print(f"⏱️ 총 소요 시간: {duration:.2f}초")
        print(f"📁 결과 파일 위치: {output_dir}/")
        
        return self.results

# 실제 사용
async def main():
    orchestrator = MeetingMinutesOrchestrator(api_key="your-api-key")
    results = await orchestrator.process_meeting(
        audio_file_path="meeting_recording.mp3",
        output_dir="./meeting_outputs"
    )
    
    print("\n🎉 회의록 자동화 완료!")
    print("생성된 파일:")
    print("- 전체 회의록 (Markdown)")
    print("- 요약 보고서 (Markdown)")
    print("- 데이터 파일 (JSON)")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

🎨 고급 기능 추가하기

👥 화자 구분 (Speaker Diarization)

from pyannote.audio import Pipeline
import torch

class SpeakerDiarizationAgent:
    def __init__(self, hf_token):
        """
        Hugging Face의 pyannote 모델 사용
        """
        self.pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization",
            use_auth_token=hf_token
        )
        self.name = "화자 구분 에이전트"
    
    def identify_speakers(self, audio_file_path):
        """
        오디오에서 화자를 구분
        """
        print(f"👥 {self.name}: 화자 분석 중...")
        
        # 화자 구분 실행
        diarization = self.pipeline(audio_file_path)
        
        # 결과 정리
        speakers_timeline = []
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            speakers_timeline.append({
                'start': turn.start,
                'end': turn.end,
                'speaker': speaker,
                'duration': turn.end - turn.start
            })
        
        print(f"✅ {len(set([s['speaker'] for s in speakers_timeline]))}명의 화자 감지!")
        return speakers_timeline
    
    def merge_with_transcript(self, transcript_segments, speaker_timeline):
        """
        음성 인식 결과와 화자 정보를 결합
        """
        merged = []
        
        for segment in transcript_segments:
            # 해당 시간대의 화자 찾기
            segment_time = (segment['start'] + segment['end']) / 2
            
            speaker = "Unknown"
            for sp in speaker_timeline:
                if sp['start'] <= segment_time <= sp['end']:
                    speaker = sp['speaker']
                    break
            
            merged.append({
                'start': segment['start'],
                'end': segment['end'],
                'speaker': speaker,
                'text': segment['text']
            })
        
        return merged
    
    def format_conversation(self, merged_data):
        """
        대화 형식으로 포맷팅
        """
        conversation = []
        current_speaker = None
        current_text = []
        
        for item in merged_data:
            if item['speaker'] != current_speaker:
                if current_speaker is not None:
                    conversation.append({
                        'speaker': current_speaker,
                        'text': ' '.join(current_text)
                    })
                current_speaker = item['speaker']
                current_text = [item['text']]
            else:
                current_text.append(item['text'])
        
        # 마지막 화자 추가
        if current_speaker is not None:
            conversation.append({
                'speaker': current_speaker,
                'text': ' '.join(current_text)
            })
        
        return conversation

# 사용 예시
diarization_agent = SpeakerDiarizationAgent(hf_token="your-hf-token")
speaker_timeline = diarization_agent.identify_speakers("meeting.mp3")
merged = diarization_agent.merge_with_transcript(transcript_segments, speaker_timeline)
conversation = diarization_agent.format_conversation(merged)

🔍 실시간 처리 (Real-time Processing)

import asyncio
from queue import Queue
import threading

class RealtimeProcessor:
    def __init__(self, api_key):
        self.transcription_queue = Queue()
        self.processing_queue = Queue()
        self.results_queue = Queue()
        
        self.transcription_agent = TranscriptionAgent(api_key)
        self.writer_agent = MinutesWriterAgent(api_key)
        
        self.is_running = False
    
    def start_streaming(self, audio_stream):
        """
        실시간 오디오 스트림 처리 시작
        """
        self.is_running = True
        
        # 3개의 워커 스레드 시작
        threading.Thread(target=self._transcription_worker, daemon=True).start()
        threading.Thread(target=self._processing_worker, daemon=True).start()
        threading.Thread(target=self._results_worker, daemon=True).start()
        
        print("🎙️ 실시간 처리 시작!")
        
        # 오디오 청크를 큐에 추가
        for audio_chunk in audio_stream:
            if not self.is_running:
                break
            self.transcription_queue.put(audio_chunk)
    
    def _transcription_worker(self):
        """
        음성 인식 워커
        """
        while self.is_running:
            if not self.transcription_queue.empty():
                audio_chunk = self.transcription_queue.get()
                
                # 음성을 텍스트로 변환
                text = self.transcription_agent.transcribe_audio(audio_chunk)
                
                # 다음 단계로 전달
                self.processing_queue.put(text)
    
    def _processing_worker(self):
        """
        텍스트 처리 워커
        """
        accumulated_text = []
        
        while self.is_running:
            if not self.processing_queue.empty():
                text = self.processing_queue.get()
                accumulated_text.append(text)
                
                # 일정량 쌓이면 처리
                if len(accumulated_text) >= 5:  # 5개 청크마다
                    combined_text = ' '.join(accumulated_text)
                    
                    # 중간 요약 생성
                    summary = self._create_interim_summary(combined_text)
                    
                    self.results_queue.put({
                        'type': 'interim_summary',
                        'content': summary,
                        'timestamp': datetime.now()
                    })
                    
                    accumulated_text = []
    
    def _results_worker(self):
        """
        결과 출력 워커
        """
        while self.is_running:
            if not self.results_queue.empty():
                result = self.results_queue.get()
                
                # 실시간으로 결과 표시
                print(f"\n⚡ 실시간 업데이트 [{result['timestamp']}]")
                print(f"📝 {result['content']}\n")
    
    def _create_interim_summary(self, text):
        """
        중간 요약 생성
        """
        prompt = f"""다음 회의 내용의 핵심을 2-3문장으로 요약하세요:

{text}

요약:"""
        
        # 간단한 요약 생성
        response = self.writer_agent.llm.predict(prompt)
        return response
    
    def stop(self):
        """
        처리 중지
        """
        self.is_running = False
        print("⏹️ 실시간 처리 중지")

# 사용 예시
processor = RealtimeProcessor(api_key="your-api-key")

# 가상의 오디오 스트림 (실제로는 마이크나 화상회의 플랫폼에서 받아옴)
def audio_stream_generator():
    # 실제로는 실시간 오디오를 청크로 나눠서 yield
    for i in range(10):
        yield f"audio_chunk_{i}.mp3"
        time.sleep(5)  # 5초마다 청크 생성

# 실행
processor.start_streaming(audio_stream_generator())

📱 다양한 플랫폼 연동

import requests
from datetime import datetime, timedelta

class ZoomIntegration:
    def __init__(self, api_key, api_secret):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = "https://api.zoom.us/v2"
    
    def get_recent_recordings(self, days=7):
        """
        최근 녹화 파일 가져오기
        """
        from_date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
        to_date = datetime.now().strftime("%Y-%m-%d")
        
        endpoint = f"{self.base_url}/users/me/recordings"
        params = {
            'from': from_date,
            'to': to_date
        }
        
        response = requests.get(
            endpoint,
            headers=self._get_headers(),
            params=params
        )
        
        return response.json()['meetings']
    
    def download_recording(self, recording_url, output_path):
        """
        녹화 파일 다운로드
        """
        response = requests.get(
            recording_url,
            headers=self._get_headers(),
            stream=True
        )
        
        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        return output_path
    
    def _get_headers(self):
        # JWT 토큰 생성 로직
        # (실제로는 Zoom SDK 사용 권장)
        return {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders

class EmailNotifier:
    def __init__(self, smtp_server, smtp_port, email, password):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.email = email
        self.password = password
    
    def send_minutes(self, recipients, subject, minutes_file, summary):
        """
        회의록을 이메일로 발송
        """
        msg = MIMEMultipart()
        msg['From'] = self.email
        msg['To'] = ', '.join(recipients)
        msg['Subject'] = subject
        
        # 이메일 본문
        body = f"""
안녕하세요,

회의록이 자동으로 생성되었습니다.

=== 회의 요약 ===
{summary}

자세한 내용은 첨부 파일을 확인해주세요.

감사합니다.
"""
        msg.attach(MIMEText(body, 'plain'))
        
        # 회의록 파일 첨부
        with open(minutes_file, 'rb') as f:
            part = MIMEBase('application', 'octet-stream')
            part.set_payload(f.read())
            encoders.encode_base64(part)
            part.add_header(
                'Content-Disposition',
                f'attachment; filename= {minutes_file}'
            )
            msg.attach(part)
        
        # 이메일 발송
        with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
            server.starttls()
            server.login(self.email, self.password)
            server.send_message(msg)
        
        print(f"📧 이메일 발송 완료: {', '.join(recipients)}")

# 사용 예시
notifier = EmailNotifier(
    smtp_server="smtp.gmail.com",
    smtp_port=587,
    email="your-email@gmail.com",
    password="your-password"
)

notifier.send_minutes(
    recipients=["team@company.com"],
    subject="[자동생성] 2024년 1월 프로젝트 회의록",
    minutes_file="minutes_20240115.md",
    summary=executive_summary
)

from notion_client import Client

class NotionIntegration:
    def __init__(self, token):
        self.client = Client(auth=token)
    
    def create_meeting_page(self, database_id, minutes_data):
        """
        Notion 데이터베이스에 회의록 페이지 생성
        """
        # 페이지 속성 설정
        properties = {
            "제목": {
                "title": [
                    {
                        "text": {
                            "content": minutes_data['title']
                        }
                    }
                ]
            },
            "날짜": {
                "date": {
                    "start": minutes_data['date']
                }
            },
            "참석자": {
                "multi_select": [
                    {"name": attendee} 
                    for attendee in minutes_data['attendees']
                ]
            },
            "상태": {
                "select": {
                    "name": "완료"
                }
            }
        }
        
        # 페이지 내용 (블록)
        children = []
        
        # 요약 섹션
        children.append({
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "📋 회의 요약"}}]
            }
        })
        
        children.append({
            "object": "block",
            "type": "paragraph",
            "paragraph": {
                "rich_text": [{"type": "text", "text": {"content": minutes_data['summary']}}]
            }
        })
        
        # 안건 섹션
        children.append({
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "📝 안건 및 논의사항"}}]
            }
        })
        
        for item in minutes_data['agenda_items']:
            children.append({
                "object": "block",
                "type": "heading_3",
                "heading_3": {
                    "rich_text": [{"type": "text", "text": {"content": item['title']}}]
                }
            })
            children.append({
                "object": "block",
                "type": "paragraph",
                "paragraph": {
                    "rich_text": [{"type": "text", "text": {"content": item['content']}}]
                }
            })
        
        # 액션 아이템 섹션
        children.append({
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "✅ 액션 아이템"}}]
            }
        })
        
        for action in minutes_data['action_items']:
            children.append({
                "object": "block",
                "type": "to_do",
                "to_do": {
                    "rich_text": [{
                        "type": "text",
                        "text": {
                            "content": f"{action['task']} (담당: {action['assignee']}, 기한: {action['deadline']})"
                        }
                    }],
                    "checked": False
                }
            })
        
        # 페이지 생성
        page = self.client.pages.create(
            parent={"database_id": database_id},
            properties=properties,
            children=children
        )
        
        print(f"✅ Notion 페이지 생성 완료: {page['url']}")
        return page['url']

# 사용 예시
notion = NotionIntegration(token="your-notion-token")
page_url = notion.create_meeting_page(
    database_id="your-database-id",
    minutes_data=minutes_data
)
print(f"📄 회의록 링크: {page_url}")

🎯 실전 활용 사례

💼 사례 1: 스타트업 A사의 일일 스탠드업

🏭 사례 2: 제조업 B사의 품질 회의

🎓 사례 3: 대학 C의 연구 미팅

⚠️ 주의사항과 한계점

🔒 보안 및 프라이버시

🎯 정확도 문제

💰 비용 관리

🔧 기술적 한계

🚀 미래 전망과 발전 방향

🤖 더 똑똑한 에이전트

🌐 플랫폼 통합

🎨 개인화와 커스터마이징

📊 예측과 인사이트

🛠️ 직접 만들어보기: 완전한 예제

📦 프로젝트 구조

meeting-minutes-automation/
│
├── agents/
│   ├── __init__.py
│   ├── transcription_agent.py      # 음성인식
│   ├── diarization_agent.py        # 화자 구분
│   ├── writer_agent.py              # 회의록 작성
│   ├── summary_agent.py             # 요약 분석
│   └── action_agent.py              # 액션 아이템 추출
│
├── integrations/
│   ├── __init__.py
│   ├── zoom_integration.py          # Zoom 연동
│   ├── notion_integration.py        # Notion 연동
│   ├── email_notifier.py            # 이메일 발송
│   └── slack_notifier.py            # Slack 알림
│
├── utils/
│   ├── __init__.py
│   ├── audio_processor.py           # 오디오 처리
│   ├── text_processor.py            # 텍스트 처리
│   └── file_manager.py              # 파일 관리
│
├── config/
│   ├── config.yaml                  # 설정 파일
│   └── prompts.yaml                 # 프롬프트 템플릿
│
├── main.py                          # 메인 실행 파일
├── requirements.txt                 # 의존성 패키지
└── README.md                        # 문서

⚙️ 설정 파일 (config.yaml)

# config/config.yaml

# API 키 설정
api_keys:
  openai: "your-openai-key"
  huggingface: "your-hf-key"
  notion: "your-notion-key"
  zoom: "your-zoom-key"

# 모델 설정
models:
  transcription: "whisper-1"
  llm: "gpt-4"
  llm_temperature: 0.3
  
# 처리 옵션
processing:
  chunk_size: 300  # 초 단위
  parallel_processing: true
  max_workers: 4
  
# 출력 설정
output:
  formats: ["markdown", "pdf", "json"]
  save_audio: false
  save_transcript: true
  
# 알림 설정
notifications:
  email:
    enabled: true
    recipients: ["team@company.com"]
  slack:
    enabled: true
    channel: "#meeting-minutes"
  notion:
    enabled: true
    database_id: "your-database-id"
    
# 보안 설정
security:
  encrypt_files: true
  auto_delete_after_days: 90
  access_control: true

🎯 메인 실행 파일 (main.py)

#!/usr/bin/env python3
"""
회의록 자동화 시스템 - 메인 실행 파일
"""

import asyncio
import argparse
from pathlib import Path
import yaml
from datetime import datetime

from agents.transcription_agent import TranscriptionAgent
from agents.diarization_agent import SpeakerDiarizationAgent
from agents.writer_agent import MinutesWriterAgent
from agents.summary_agent import SummaryAgent
from agents.action_agent import ActionItemAgent

from integrations.email_notifier import EmailNotifier
from integrations.slack_notifier import SlackNotifier
from integrations.notion_integration import NotionIntegration

from utils.audio_processor import AudioProcessor
from utils.file_manager import FileManager

class MeetingMinutesSystem:
    """
    멀티에이전트 회의록 자동화 시스템
    """
    
    def __init__(self, config_path="config/config.yaml"):
        # 설정 로드
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = yaml.safe_load(f)
        
        # 에이전트 초기화
        self._init_agents()
        
        # 통합 서비스 초기화
        self._init_integrations()
        
        # 유틸리티 초기화
        self.audio_processor = AudioProcessor()
        self.file_manager = FileManager(self.config['output'])
        
        print("✅ 시스템 초기화 완료!")
    
    def _init_agents(self):
        """에이전트 초기화"""
        api_key = self.config['api_keys']['openai']
        hf_token = self.config['api_keys']['huggingface']
        
        self.transcription_agent = TranscriptionAgent(api_key)
        self.diarization_agent = SpeakerDiarizationAgent(hf_token)
        self.writer_agent = MinutesWriterAgent(api_key)
        self.summary_agent = SummaryAgent(api_key)
        self.action_agent = ActionItemAgent(api_key)
        
        print("🤖 에이전트 초기화 완료")
    
    def _init_integrations(self):
        """통합 서비스 초기화"""
        if self.config['notifications']['email']['enabled']:
            self.email_notifier = EmailNotifier(self.config)
        
        if self.config['notifications']['slack']['enabled']:
            self.slack_notifier = SlackNotifier(self.config)
        
        if self.config['notifications']['notion']['enabled']:
            self.notion = NotionIntegration(
                self.config['api_keys']['notion']
            )
        
        print("🔗 통합 서비스 초기화 완료")
    
    async def process_meeting(self, audio_file, meeting_info=None):
        """
        회의 처리 메인 함수
        
        Args:
            audio_file: 오디오 파일 경로
            meeting_info: 회의 정보 (제목, 날짜, 참석자 등)
        """
        print("\n" + "="*60)
        print("🚀 회의록 자동화 시작!")
        print("="*60 + "\n")
        
        start_time = datetime.now()
        results = {}
        
        try:
            # Step 1: 오디오 전처리
            print("📁 Step 1: 오디오 파일 처리")
            processed_audio = self.audio_processor.preprocess(audio_file)
            print("✅ 오디오 전처리 완료\n")
            
            # Step 2: 음성 인식 + 화자 구분 (병렬 처리)
            print("🎤 Step 2: 음성 인식 및 화자 구분")
            transcript_task = self.transcription_agent.transcribe_with_timestamps(
                processed_audio
            )
            speaker_task = self.diarization_agent.identify_speakers(
                processed_audio
            )
            
            transcript, speakers = await asyncio.gather(
                transcript_task,
                speaker_task
            )
            
            # 음성 인식 결과와 화자 정보 병합
            merged_transcript = self.diarization_agent.merge_with_transcript(
                transcript,
                speakers
            )
            conversation = self.diarization_agent.format_conversation(
                merged_transcript
            )
            
            results['transcript'] = conversation
            print("✅ 음성 인식 및 화자 구분 완료\n")
            
            # Step 3: 회의록 작성
            print("✍️ Step 3: 회의록 작성")
            minutes = await self.writer_agent.create_minutes(
                conversation,
                meeting_info
            )
            markdown = self.writer_agent.format_to_markdown(minutes)
            results['minutes'] = markdown
            print("✅ 회의록 작성 완료\n")
            
            # Step 4: 분석 작업 (병렬 처리)
            print("📊 Step 4: 요약 및 분석")
            analysis_tasks = [
                self.summary_agent.create_executive_summary(markdown),
                self.summary_agent.extract_key_insights(markdown),
                self.summary_agent.analyze_sentiment(markdown),
                self.action_agent.extract_action_items(markdown)
            ]
            
            exec_summary, insights, sentiment, action_items = await asyncio.gather(
                *analysis_tasks
            )
            
            results['executive_summary'] = exec_summary
            results['insights'] = insights
            results['sentiment'] = sentiment
            results['action_items'] = action_items
            print("✅ 분석 완료\n")
            
            # Step 5: 파일 저장
            print("💾 Step 5: 결과 저장")
            saved_files = self.file_manager.save_results(results, meeting_info)
            print(f"✅ 파일 저장 완료: {len(saved_files)}개 파일\n")
            
            # Step 6: 알림 발송 (병렬 처리)
            print("📢 Step 6: 알림 발송")
            notification_tasks = []
            
            if self.config['notifications']['email']['enabled']:
                notification_tasks.append(
                    self.email_notifier.send_minutes(
                        results,
                        saved_files['markdown']
                    )
                )
            
            if self.config['notifications']['slack']['enabled']:
                notification_tasks.append(
                    self.slack_notifier.post_summary(
                        exec_summary,
                        saved_files['markdown']
                    )
                )
            
            if self.config['notifications']['notion']['enabled']:
                notification_tasks.append(
                    self.notion.create_meeting_page(
                        self.config['notifications']['notion']['database_id'],
                        results
                    )
                )
            
            if notification_tasks:
                await asyncio.gather(*notification_tasks)
            
            print("✅ 알림 발송 완료\n")
            
            # 완료 메시지
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()
            
            print("="*60)
            print("🎉 모든 작업 완료!")
            print(f"⏱️ 총 소요 시간: {duration:.2f}초")
            print(f"📁 저장된 파일: {saved_files}")
            print("="*60)
            
            return results
            
        except Exception as e:
            print(f"❌ 오류 발생: {str(e)}")
            raise
    
    async def process_realtime(self, audio_stream):
        """
        실시간 처리 (스트리밍)
        """
        print("🎙️ 실시간 처리 모드 시작")
        
        # 실시간 처리 로직
        # (이전 예제의 RealtimeProcessor 활용)
        pass

def main():
    """메인 함수"""
    parser = argparse.ArgumentParser(
        description="회의록 자동화 시스템"
    )
    parser.add_argument(
        'audio_file',
        help="처리할 오디오 파일 경로"
    )
    parser.add_argument(
        '--title',
        help="회의 제목",
        default="회의"
    )
    parser.add_argument(
        '--date',
        help="회의 날짜 (YYYY-MM-DD)",
        default=datetime.now().strftime("%Y-%m-%d")
    )
    parser.add_argument(
        '--attendees',
        help="참석자 (쉼표로 구분)",
        default=""
    )
    parser.add_argument(
        '--config',
        help="설정 파일 경로",
        default="config/config.yaml"
    )
    
    args = parser.parse_args()
    
    # 회의 정보 구성
    meeting_info = {
        'title': args.title,
        'date': args.date,
        'attendees': [a.strip() for a in args.attendees.split(',') if a.strip()]
    }
    
    # 시스템 초기화 및 실행
    system = MeetingMinutesSystem(config_path=args.config)
    
    # 비동기 실행
    asyncio.run(system.process_meeting(args.audio_file, meeting_info))

if __name__ == "__main__":
    main()

🚀 실행 방법

# 가상환경 생성
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 패키지 설치
pip install -r requirements.txt

# 설정 파일 수정
# config/config.yaml 파일에서 API 키 등을 설정

# 가장 간단한 실행
python main.py meeting_recording.mp3

# 회의 정보와 함께 실행
python main.py meeting_recording.mp3 \
    --title "2024 Q1 전략 회의" \
    --date "2024-01-15" \
    --attendees "김철수, 이영희, 박민수"

# 커스텀 설정 파일 사용
python main.py meeting_recording.mp3 \
    --config my_config.yaml

output/
├── minutes_20240115_143022.md      # 전체 회의록
├── summary_20240115_143022.md      # 요약본
├── data_20240115_143022.json       # JSON 데이터
└── transcript_20240115_143022.txt  # 원본 텍스트

💡 재능넷에서 전문가 찾기

🎓 학습 리소스

🎬 마무리하며

🚀 다음 단계