표준적인 메일 머지 (mail merge) 도구들이 단순히 {first_name}과 같은 태그를 정적인 템플릿에 교체하는 것을 개인화라고 부르는 반면, 소규모 아웃리치 (outreach) 캠페인에는 훨씬 더 진정성 있는 접근 방식이 필요합니다. 회사 이름, 선택적인 연락처 정보, 그리고 특정 회사 관련 메모를 결합함으로써, 로컬 LLM (Large Language Model)으로 구동되는 시스템은 각 수신자마다 이메일 본문 전체를 자연스럽게 다시 작성할 수 있습니다. 이는 단순히 빈칸을 채운 플레이스홀더 (placeholder)를 제공하는 것이 아니라, 진정으로 맞춤화된 아웃리치를 전달합니다.

이것은 저의 로컬 이메일 에이전트 (email agent) 시리즈의 마지막 주요 모듈입니다. CSV에서 수신자를 불러오고, Llama 3.1을 사용하여 메시지를 개인화하며, Gmail을 통해 제어된 속도로 전송하고, 답장이 돌아오면 이를 분류하는 캠페인 엔진입니다.

이전 기사들을 놓치셨다면:

Part 1: From Inbox to Character: Building a Private, Local AI Email Agent

Part 2: How /search and /ask Work: Local Hybrid RAG with ChromaDB + SQLite FTS5

Part 3: LLM as Router: Intent Classification for a Local Telegram Email Agent

Part 4: Behind the Build: Hidden Headaches of a Local Gmail AI Agent

이 기사에서 저는 코드베이스의 캠페인 시스템을 단계별로 살펴볼 것입니다:

• 캠페인 제어를 위한 Telegram 명령어
• CSV 수신자 로딩
• Jinja2 프롬프트 렌더링 (prompt rendering)
• LLM 개인화
• SQLite 캠페인 상태
• 스로틀링 (throttling)을 적용한 Gmail 전송
• Gmail threadId를 통한 답장 매칭
• LLM 답장 분류
• 유용한 답장에 대한 Telegram 알림

캠페인 레이어 (The Campaign Layer)

캠페인 레이어 (The campaign layer)는 CSV와 템플릿을 개인화된 Gmail 메시지로 변환한 다음, 일반적인 이메일 수집 파이프라인 (email ingestion pipeline)을 통해 답장을 추적합니다.

캠페인 코드는 주로 다음 세 개의 파일에 나뉘어 저장되어 있습니다:

webservice/src/email_service/services/cmd_campaign.py
webservice/src/email_service/services/campaign_engine.py
webservice/src/email_service/services/campaign_sender.py

이러한 분리는 의도된 것입니다:

• cmd_campaign.py는 Telegram 명령어를 파싱 (parse)하고 명령어 인자 (arguments)를 검증합니다.
• campaign_engine.py는 캠페인을 생성하고, 수신자를 로드하며, LLM을 호출하고, 메시지를 미리 보기 (preview) 하며, 답장을 분류합니다.
• campaign_sender.py는 백그라운드에서 실행되며, 실행 중인 캠페인을 찾고, 다음 준비된 수신자에게 메시지를 보내며, 응답이 없는 (ghosted) 수신자를 확인합니다.

캠페인 모듈은 앱의 나머지 부분과 동일한 디자인 패턴을 사용합니다. Telegram은 오직 제어 인터페이스 (control surface) 역할만 수행하며, 실제 작업은 그 뒤에 있는 서비스 모듈 (service modules)에서 이루어집니다.

사용자 흐름 (The User Flow)

캠페인은 설정에서 시작하여 개인화, 속도 제한 전송 (throttled sending), 그리고 답장 분류 단계로 진행됩니다.

명령어 흐름은 다음과 같습니다:

campaign create outreach cover_letter.txt Role at {company_name}
campaign load outreach recipients.csv
campaign personalize outreach
...

이 명령어들은 앞에 슬래시 (/)를 붙이거나 붙이지 않고 Telegram을 통해 전송할 수 있습니다. 라우터 (router)는 campaign을 직접적인 복합 명령어 네임스페이스 (compound command namespace)로 취급하므로, campaign preview outreach는 LLM 경로를 거치지 않고 캠페인 핸들러 (campaign handler)로 바로 이동합니다.

자연어 (Natural language) 또한 작동할 수 있습니다. 의도 분류기 (intent classifier)는 campaign_create, campaign_load, campaign_personalize, campaign_preview, campaign_start, campaign_pause, campaign_resume, campaign_status, campaign_results와 같은 캠페인 의도 (campaign intents)를 알고 있습니다.

직접 명령 경로 (direct command path)는 더 빠르고 예측 가능합니다. 자연어 경로 (natural language path)는 다음과 같이 입력할 때 유용합니다:

start the outreach campaign

LLM은 해당 요청을 분류하고, 캠페인 이름을 추출하여 동일한 기본 핸들러 (underlying handler)로 전달합니다.

캠페인 생성 (Campaign Creation)

텔레그램 명령 핸들러 (Telegram command handler)는 단순하고 작습니다:

def campaign_create(args: list[str]) -> str:
    if len(args) < 2:
        return (
...

엔진은 설정된 캠페인 디렉토리 (campaign directory)에 템플릿 파일이 존재하는지 확인합니다:

template_path = settings.campaigns_dir / template_file
if not template_path.exists():
    return f"Template not found: {template_path}"

그 다음 Campaign 행 (row)을 생성합니다:

campaign = Campaign(
    name=name,
    template_file=template_file,
...

기본 캠페인 디렉토리는 다음과 같습니다:

campaigns_dir: Path = Path("campaigns")

이를 통해 캠페인 템플릿, CSV 파일, 그리고 선택적인 첨부 파일들을 코드와 분리하여 관리합니다.

CSV에서 수신자 로드하기 (Loading Recipients from CSV)

CSV 로더 (CSV loader)는 경로를 직접 받거나, 캠페인 디렉토리 내의 파일명을 받습니다:

csv_file = Path(csv_path)
if not csv_file.exists():
    csv_file = settings.campaigns_dir / csv_path
...

예상되는 CSV 필드 (fields)는 간단합니다:

company_name,to_address,contact_name,company_info
Acme Cloud,hr@example.com,Jane Smith,"Cloud infrastructure provider"
Northstar Labs,jobs@example.com,,"AI tooling for internal operations"

실제로 company_name과 to_address는 최소한으로 유용한 필드입니다. contact_name과 company_info는 선택 사항이지만, 개인화 (personalization)를 더 좋게 만드는 요소들입니다.

로더는 각 행을 CampaignRecipient로 저장합니다:

수신자에게 할당하는 것은 다음과 같습니다:

recipient = CampaignRecipient(
    campaign_id=campaign.id,
    company_name=row.get("company_name", ""),
...

그다음 캠페인 카운터를 업데이트합니다:

campaign.total_recipients = count
session.commit()

개인화 프롬프트 (Personalization Prompt)

템플릿은 수신자 컨텍스트가 포함된 프롬프트로 렌더링되고, 그 후 로컬 LLM이 개인화된 제목과 본문을 반환합니다.

개인화 단계는 템플릿 파일을 읽는 것으로 시작됩니다:

template_path = settings.campaigns_dir / campaign.template_file
template_body = template_path.read_text(encoding="utf-8")
subject_template = campaign.subject_template

그다음, 각 대기 중인 수신자는 personalize.j2를 통해 렌더링됩니다:

Personalize this email template for a specific recipient.
You are Sable, assisting with personalized outreach. Keep the result natural and specific.

...

실제 호출은 프로젝트의 다른 모든 곳에서 사용되는 것과 동일한 패턴입니다:

prompt = _personalize.render(
    template_body=template_body,
    company_name=rd["company_name"],
...

그다음 행이 업데이트됩니다:

r.personalized_subject = parsed.get("personalized_subject", "")
r.personalized_body = parsed.get("personalized_body", "")
r.status = "personalized"
...

`status =

for i, r in enumerate(recipients, 1):
lines.append(f"--- #{i}: {r.company_name} ({r.to_address}) ---")
lines.append(f"Subject: {r.personalized_subject}")
...

이것은 Telegram이 검토 표면이기 때문에 의도적으로 일반 텍스트입니다.

데이터베이스 상태 (Database State)

캠페인 테이블은 캠페인 수준 카운터와 수신자별 상태를 분리합니다.

데이터베이스 모델은 하나의 캠페인 행(campaign row)과 여러 수신자 행(recipient rows)을 가집니다:

class Campaign(Base):
    __tablename__ = "campaigns"

...

수신자 행은 개인화 출력 및 답장 추적 데이터를 저장합니다:

class CampaignRecipient(Base):
    __tablename__ = "campaign_recipients"

...

실제 수신자 상태 흐름은 다음과 같습니다:

pending -> personalized -> sent -> classified

이후 reply_classification 필드는 최종 상태가 어떤 종류인지 알려줍니다:

interview | rejection | follow_up | automated | ghosted

모델 주석에는 별도의 replied 상태를 위한 공간도 있지만, 현재 구현에서는 일치하는 답장을 직접 classified로 이동시킵니다.

속도 제한 전송 (Throttled Sending)

발신자는 FastAPI 앱이 시작될 때 자동으로 시작됩니다:

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
...

발신자 자체는 데몬 스레드입니다:

def start_sender():
    global _thread
    if _thread and _thread.is_alive():
...

각 루프에서, 실행 중인 캠페인을 처리하고 ghosted 수신자를 확인합니다:

def _sender_loop():
    while not _stop_event.is_set():
        try:
...

발신자는 `status ==

recipient = (
    session.query(CampaignRecipient)
    .filter_by(campaign_id=campaign_id, status="personalized")
...

발신자는 status == "personalized"인 수신자를 조회합니다.

개인화된 수신자가 더 이상 남아있지 않으면 캠페인이 완료됩니다:

if not recipient:
    campaign = session.query(Campaign).get(campaign_id)
    if campaign:
...

그렇지 않으면 Gmail 클라이언트를 통해 이메일을 발송합니다:

sent = gmail_client.send_email(
    service,
    to=to_address,
...

그 다음 Gmail에서 반환된 ID를 저장합니다:

r.status = "sent"
r.sent_at = datetime.now()
r.gmail_message_id = sent.get("id")
...

전송 속도 조절(Throttle)은 send_rate에 의해 제어됩니다:

delay = 3600 / send_rate

현재 모델의 기본값은 시간당 50통의 이메일이며, 이는 발송 사이에 72초의 간격을 둡니다:

3600 / 50 = 72

발신자는 단순한 time.sleep(...) 대신 _stop_event.wait(timeout=delay)를 통해 대기하므로, 종료(shutdown) 시 대기 상태를 빠르게 중단할 수 있습니다.

답장 추적 (Reply Tracking)

캠페인 답장은 별도의 편지함에서 처리되는 대신, 표준 Gmail 수집 경로를 통해 처리되며 스레드 ID (thread ID)를 기준으로 매칭됩니다.

캠페인 이메일이 발송되면 Gmail은 threadId를 반환합니다. 캠페인 발신기는 이를 수신자 행의 gmail_thread_id로 저장합니다.

나중에 일반적인 처리 파이프라인을 통해 이메일이 들어오면, email_processor.py가 들어온 Gmail 스레드를 확인합니다:

if request.thread_id:
    try:
        campaign_engine.check_campaign_reply(
...

캠페인 엔진은 일치하는 수신자를 찾습니다: