DSPy로 이메일에서 정보 추출하기

DSPy로 이메일에서 정보 추출하기

이 튜토리얼은 DSPy를 이용해 지능적인 이메일 처리 시스템을 만드는 방법을 보여줘요. 다양한 유형의 이메일에서 핵심 정보를 자동으로 뽑아내고, 의도를 분류하며, 데이터를 후속 처리가 가능한 구조로 정리하는 시스템을 만들어 볼 거예요.

출처: Extracting Information from Emails with DSPy

무엇을 만들게 되나요

이 튜토리얼을 마치면 DSPy 기반의 이메일 처리 시스템이 다음 일을 할 수 있어요.

• 이메일 유형 분류 (주문 확인, 지원 요청, 회의 초대 등)
• 핵심 엔티티 추출 (날짜, 금액, 제품명, 연락처 정보)
• 긴급도와 필요한 조치 판단
• 추출한 데이터를 일관된 형식으로 구조화
• 여러 이메일 형식을 견고하게 처리

준비 사항

• DSPy 모듈과 signature에 대한 기본 이해
• Python 3.9+ 설치
• OpenAI API 키 (또는 지원되는 다른 LLM 접근 권한)

설치와 설정

pip install dspy

권장사항: 내부에서 무슨 일이 벌어지는지 이해하려면 MLflow Tracing을 설정해 두세요.

MLflow DSPy 통합

MLflow는 DSPy와 기본적으로 통합되는 LLMOps 도구로, 설명 가능성과 실험 추적을 제공해요. 이 튜토리얼에서 MLflow를 쓰면 프롬프트와 최적화 진행 상황을 trace로 시각화해 DSPy의 동작을 더 잘 이해할 수 있어요. 다음 네 단계로 MLflow를 쉽게 설정할 수 있어요.

1. MLflow 설치

%pip install mlflow>=3.0.0

2. 별도 터미널에서 MLflow UI 시작

mlflow ui --port 5000 --backend-store-uri sqlite:///mlruns.db

3. 노트북을 MLflow에 연결

import mlflow

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("DSPy")

4. 추적 활성화

mlflow.dspy.autolog()

통합에 대해 더 자세히 알고 싶다면 MLflow DSPy 문서도 함께 보세요.

Step 1: 데이터 구조 정의하기

먼저 이메일에서 추출하고 싶은 정보의 유형을 정의해 볼게요.

import dspy
from typing import List, Optional, Literal
from datetime import datetime
from pydantic import BaseModel
from enum import Enum

class EmailType(str, Enum):
    ORDER_CONFIRMATION = "order_confirmation"
    SUPPORT_REQUEST = "support_request"
    MEETING_INVITATION = "meeting_invitation"
    NEWSLETTER = "newsletter"
    PROMOTIONAL = "promotional"
    INVOICE = "invoice"
    SHIPPING_NOTIFICATION = "shipping_notification"
    OTHER = "other"

class UrgencyLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class ExtractedEntity(BaseModel):
    entity_type: str
    value: str
    confidence: float

EmailType은 이메일이 주문 확인인지 지원 요청인지 등을, UrgencyLevel은 긴급도를 나타내요. ExtractedEntity는 뽑아낸 엔티티 하나의 유형·값·신뢰도를 담아요.

Step 2: DSPy signature 만들기

이제 이메일 처리 파이프라인용 signature를 정의할게요.

class ClassifyEmail(dspy.Signature):
    """Classify the type and urgency of an email based on its content."""

    email_subject: str = dspy.InputField(desc="The subject line of the email")
    email_body: str = dspy.InputField(desc="The main content of the email")
    sender: str = dspy.InputField(desc="Email sender information")

    email_type: EmailType = dspy.OutputField(desc="The classified type of email")
    urgency: UrgencyLevel = dspy.OutputField(desc="The urgency level of the email")
    reasoning: str = dspy.OutputField(desc="Brief explanation of the classification")

class ExtractEntities(dspy.Signature):
    """Extract key entities and information from email content."""

    email_content: str = dspy.InputField(desc="The full email content including subject and body")
    email_type: EmailType = dspy.InputField(desc="The classified type of email")

    key_entities: list[ExtractedEntity] = dspy.OutputField(desc="List of extracted entities with type, value, and confidence")
    financial_amount: Optional[float] = dspy.OutputField(desc="Any monetary amounts found (e.g., '$99.99')")
    important_dates: list[str] = dspy.OutputField(desc="List of important dates found in the email")
    contact_info: list[str] = dspy.OutputField(desc="Relevant contact information extracted")

class GenerateActionItems(dspy.Signature):
    """Determine what actions are needed based on the email content and extracted information."""

    email_type: EmailType = dspy.InputField()
    urgency: UrgencyLevel = dspy.InputField()
    email_summary: str = dspy.InputField(desc="Brief summary of the email content")
    extracted_entities: list[ExtractedEntity] = dspy.InputField(desc="Key entities found in the email")

    action_required: bool = dspy.OutputField(desc="Whether any action is required")
    action_items: list[str] = dspy.OutputField(desc="List of specific actions needed")
    deadline: Optional[str] = dspy.OutputField(desc="Deadline for action if applicable")
    priority_score: int = dspy.OutputField(desc="Priority score from 1-10")

class SummarizeEmail(dspy.Signature):
    """Create a concise summary of the email content."""

    email_subject: str = dspy.InputField()
    email_body: str = dspy.InputField()
    key_entities: list[ExtractedEntity] = dspy.InputField()

    summary: str = dspy.OutputField(desc="A 2-3 sentence summary of the email's main points")

ClassifyEmail signature는 이메일의 제목·본문·발신자를 보고 유형과 긴급도, 근거를 내놓고, ExtractEntities는 이메일 내용에서 엔티티·금액·중요 날짜·연락처를 뽑아요. GenerateActionItems는 필요한 조치를 결정하고, SummarizeEmail은 요약을 만들죠.

Step 3: 이메일 처리 모듈 만들기

이제 핵심 이메일 처리 모듈을 만들어 볼게요.

class EmailProcessor(dspy.Module):
    """A comprehensive email processing system using DSPy."""

    def __init__(self):
        super().__init__()

        # Initialize our processing components
        self.classifier = dspy.ChainOfThought(ClassifyEmail)
        self.entity_extractor = dspy.ChainOfThought(ExtractEntities)
        self.action_generator = dspy.ChainOfThought(GenerateActionItems)
        self.summarizer = dspy.ChainOfThought(SummarizeEmail)

    def forward(self, email_subject: str, email_body: str, sender: str = ""):
        """Process an email and extract structured information."""

        # Step 1: Classify the email
        classification = self.classifier(
            email_subject=email_subject,
            email_body=email_body,
            sender=sender
        )

        # Step 2: Extract entities
        full_content = f"Subject: {email_subject}\n\nFrom: {sender}\n\n{email_body}"
        entities = self.entity_extractor(
            email_content=full_content,
            email_type=classification.email_type
        )

        # Step 3: Generate summary
        summary = self.summarizer(
            email_subject=email_subject,
            email_body=email_body,
            key_entities=entities.key_entities
        )

        # Step 4: Determine actions
        actions = self.action_generator(
            email_type=classification.email_type,
            urgency=classification.urgency,
            email_summary=summary.summary,
            extracted_entities=entities.key_entities
        )

        # Step 5: Structure the results
        return dspy.Prediction(
            email_type=classification.email_type,
            urgency=classification.urgency,
            summary=summary.summary,
            key_entities=entities.key_entities,
            financial_amount=entities.financial_amount,
            important_dates=entities.important_dates,
            action_required=actions.action_required,
            action_items=actions.action_items,
            deadline=actions.deadline,
            priority_score=actions.priority_score,
            reasoning=classification.reasoning,
            contact_info=entities.contact_info
        )

forward는 4개의 ChainOfThought 모듈을 순서대로 실행해요. 이메일을 분류하고, 엔티티를 추출하고, 요약을 만든 뒤 조치를 결정해 dspy.Prediction으로 구조화된 결과를 돌려줘요. 각 단계가 이전 단계의 출력을 입력으로 받는 파이프라인 구조예요.

Step 4: 이메일 처리 시스템 실행하기

시스템을 테스트하기 위한 간단한 함수를 만들어 볼게요.

import os
def run_email_processing_demo():
    """Demonstration of the email processing system."""

    # Configure DSPy
    lm = dspy.LM(model='openai/gpt-4o-mini')
    dspy.configure(lm=lm)
    os.environ["OPENAI_API_KEY"] = "<YOUR OPENAI KEY>"

    # Create our email processor
    processor = EmailProcessor()

    # Sample emails for testing
    sample_emails = [
        {
            "subject": "Order Confirmation #12345 - Your MacBook Pro is on the way!",
            "body": """Dear John Smith,

Thank you for your order! We're excited to confirm that your order #12345 has been processed.

Order Details:
- MacBook Pro 14-inch (Space Gray)
- Order Total: $2,399.00
- Estimated Delivery: December 15, 2024
- Tracking Number: 1Z999AA1234567890

If you have any questions, please contact our support team at [email protected].

Best regards,
TechStore Team""",
            "sender": "[email protected]"
        },
        {
            "subject": "URGENT: Server Outage - Immediate Action Required",
            "body": """Hi DevOps Team,

We're experiencing a critical server outage affecting our production environment.

Impact: All users unable to access the platform
Started: 2:30 PM EST

Please join the emergency call immediately: +1-555-123-4567

This is our highest priority.

Thanks,
Site Reliability Team""",
            "sender": "[email protected]"
        },
        {
            "subject": "Meeting Invitation: Q4 Planning Session",
            "body": """Hello team,

You're invited to our Q4 planning session.

When: Friday, December 20, 2024 at 2:00 PM - 4:00 PM EST
Where: Conference Room A

Please confirm your attendance by December 18th.

Best,
Sarah Johnson""",
            "sender": "[email protected]"
        }
    ]

    # Process each email and display results
    print("🚀 Email Processing Demo")
    print("=" * 50)

    for i, email in enumerate(sample_emails):
        print(f"\n📧 EMAIL {i+1}: {email['subject'][:50]}...")

        # Process the email
        result = processor(
            email_subject=email["subject"],
            email_body=email["body"],
            sender=email["sender"]
        )

        # Display key results
        print(f"   📊 Type: {result.email_type}")
        print(f"   🚨 Urgency: {result.urgency}")
        print(f"   📝 Summary: {result.summary}")

        if result.financial_amount:
            print(f"   💰 Amount: ${result.financial_amount:,.2f}")

        if result.action_required:
            print(f"   ✅ Action Required: Yes")
            if result.deadline:
                print(f"   ⏰ Deadline: {result.deadline}")
        else:
            print(f"   ✅ Action Required: No")

# Run the demo
if __name__ == "__main__":
    run_email_processing_demo()

예상 출력

샘플 이메일 3개를 처리하면 이런 결과가 나와요.

• EMAIL 1 (주문 확인): 유형 order_confirmation, 긴급도 low. MacBook Pro 주문 #12345의 총액 $2,399.00, 예상 배송 2024-12-15가 포함된 요약을 만들어요. 필요한 조치는 없어요.
• EMAIL 2 (서버 장애): 유형 other, 긴급도 critical. SRE 팀이 2:30 PM EST에 시작된 심각한 서버 장애를 보고했고 DevOps 팀에 긴급 전화 참여를 요청했어요. 조치 필요, 마감은 즉시(Immediately).
• EMAIL 3 (회의 초대): 유형 meeting_invitation, 긴급도 medium. 2024-12-20 2:00~4:00 PM EST 회의실 A에서 열리는 Q4 계획 회의 초대 예요. 12월 18일까지 참석을 확인해야 하니 조치 필요.

다음 단계

• 이메일 유형 추가 및 분류 다듬기 (뉴스레터, 프로모션 등)
• 이메일 프로바이더 통합 추가 (Gmail API, Outlook, IMAP)
• 다른 LLM과 최적화 전략 실험
• 다국어 지원 추가로 국제 이메일 처리
• 최적화로 프로그램 성능 높이기

더 알아보기 (Learn more)

• ChainOfThought 모듈 — 추론 단계를 더하는 모듈
• Signature 소개 — 클래스 기반 signature 작성법