DSPy로 이메일에서 정보 추출하기
DSPy로 이메일에서 정보 추출하기
이 튜토리얼에서는 DSPy로 지능적인 이메일 처리 시스템을 만드는 방법을 보여드릴게요. 다양한 종류의 이메일에서 핵심 정보를 자동으로 추출하고, 의도를 분류하고, 이후 처리를 위해 데이터를 구조화하는 시스템을 함께 만들어 볼 거예요.
출처: 문서
본문
무엇을 만들까요
이 튜토리얼을 마치면 이런 능력을 가진 DSPy 기반 이메일 처리 시스템을 갖게 돼요:
- 이메일 유형 분류 (주문 확인, 지원 요청, 회의 초대 등)
- 핵심 개체 추출 (날짜, 금액, 제품명, 연락처 정보)
- 긴급도 수준과 필요한 조치 판단
- 추출된 데이터를 일관된 형식으로 구조화
- 여러 이메일 형식을 견고하게 처리
사전 준비 (Prerequisites)
- DSPy 모듈과 시그니처에 대한 기본적인 이해
- Python 3.9+ 설치
- OpenAI API 키 (또는 다른 지원되는 LLM 접근 권한)
설치 및 설정
pip install dspy
권장: MLflow Tracing을 설정해서 내부 동작을 이해해 보세요.
MLflow DSPy 통합
MLflow는 DSPy와 기본 통합을 제공하는 LLMOps 도구로, 설명 가능성(explainability)과 실험 추적(experiment tracking)을 지원해요. 이 튜토리얼에서는 MLflow로 프롬프트와 최적화 진행 상황을 트레이스로 시각화해 DSPy의 동작을 더 잘 이해할 수 있어요. 아래 4단계를 따라가면 쉽게 설정할 수 있어요.
- MLflow 설치
%pip install mlflow>=3.0.0
- 별도 터미널에서 MLflow UI 시작
mlflow ui --port 5000 --backend-store-uri sqlite:///mlruns.db
- 노트북을 MLflow에 연결
import mlflow
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("DSPy")
- 트레이싱 활성화
mlflow.dspy.autolog()
통합에 대한 더 자세한 내용은 MLflow DSPy 문서를 참고하세요.
1단계: 데이터 구조 정의하기
먼저 이메일에서 추출하고 싶은 정보의 유형을 정의할게요. Enum과 pydantic.BaseModel로 이메일 유형, 긴급도, 추출 개체의 틀을 잡아요.
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는 추출한 개체 하나하나의 틀을 나타내요.
2단계: DSPy 시그니처 만들기
이제 이메일 처리 파이프라인에 쓸 시그니처들을 정의해요. 각 시그니처는 입력과 출력 필드를 선언해서 "무엇을 받아 무엇을 내놓을지"를 명확히 해 줘요.
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")
여기서 dspy.OutputField의 desc는 LLM에게 각 필드를 어떻게 채워야 하는지 알려주는 지침 역할을 해요.
3단계: 이메일 처리 모듈 만들기
이제 네 개의 시그니처를 한 dspy.Module로 묶어서, 이메일 분류 → 개체 추출 → 요약 → 조치 판단으로 이어지는 파이프라인을 만들어요.
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는 각 단계의 결과를 다음 단계의 입력으로 넘겨주고, 마지막에 모든 결과를 하나의 dspy.Prediction으로 묶어 반환해요. 이렇게 하면 호출하는 쪽은 result.email_type, result.summary처럼 도트 접근으로 결과를 꺼낼 수 있어요.
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()
기대 출력 (Expected Output)
🚀 Email Processing Demo
==================================================
📧 EMAIL 1: Order Confirmation #12345 - Your MacBook Pro is on...
📊 Type: order_confirmation
🚨 Urgency: low
📝 Summary: The email confirms John Smith's order #12345 for a MacBook Pro 14-inch in Space Gray, totaling $2,399.00, with an estimated delivery date of December 15, 2024. It includes a tracking number and contact information for customer support.
💰 Amount: $2,399.00
✅ Action Required: No
📧 EMAIL 2: URGENT: Server Outage - Immediate Action Required...
📊 Type: other
🚨 Urgency: critical
📝 Summary: The Site Reliability Team has reported a critical server outage that began at 2:30 PM EST, preventing all users from accessing the platform. They have requested the DevOps Team to join an emergency call immediately to address the issue.
✅ Action Required: Yes
⏰ Deadline: Immediately
📧 EMAIL 3: Meeting Invitation: Q4 Planning Session...
📊 Type: meeting_invitation
🚨 Urgency: medium
📝 Summary: Sarah Johnson has invited the team to a Q4 planning session on December 20, 2024, from 2:00 PM to 4:00 PM EST in Conference Room A. Attendees are asked to confirm their participation by December 18th.
✅ Action Required: Yes
⏰ Deadline: December 18th
다음 단계
- 이메일 유형 추가 및 분류 정교화 (뉴스레터, 프로모션 등)
- 이메일 제공자 연동 추가 (Gmail API, Outlook, IMAP)
- 다양한 LLM과 최적화 전략 실험
- 국제 이메일 처리를 위한 다국어 지원 추가
- 프로그램 성능 향상을 위한 최적화