Building an Automatic Email‑Processing Agent with LangGraph 1.0 – A Hands‑On Guide

Scenario analysis

Automatic classification : detect email type (question, bug, billing, feature, complex) and urgency (low, medium, high, critical).

Intelligent handling : create a bug‑tracking ticket for bug reports; for other types query an internal knowledge base.

Reply generation : draft a targeted response based on classification, urgency and any retrieved context.

Review mechanism : route emails with high urgency or complex intent to a human‑review node before sending.

Agent state definition

class EmailClassification(TypedDict):
    intent: Literal['question', 'bug', 'billing', 'feature', 'complex']
    urgency: Literal['low', 'medium', 'high', 'critical']
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    email_content: str
    sender_email: str
    email_id: str
    classification: EmailClassification
    ticket_id: str | None
    search_results: list[str] | None
    customer_history: dict | None
    draft_response: str | None

Agent nodes

1. Read email (placeholder)

def read_email(state: EmailAgentState) -> EmailAgentState:
    """In production this would fetch the email via an API; here the content is passed directly."""
    pass

2. Classify intent and urgency

def classify_intent(state: EmailAgentState) -> EmailAgentState:
    structured_llm = llm.with_structured_output(EmailClassification)
    classification_prompt = f"""
    Analyze the email and classify it.
    Email: {state['email_content']}
    From: {state['sender_email']}
    Provide intent, urgency, topic and summary.
    """
    classification = structured_llm.invoke(classification_prompt)
    return {'classification': classification}

3. Knowledge‑base search (simulated)

def search_documentation(state: EmailAgentState) -> EmailAgentState:
    classification = state.get('classification', {})
    query = f"{classification.get('intent','')}{classification.get('topic','')}"
    try:
        search_results = ['search_result_1', 'search_result_2', 'search_result_3']
    except Exception:
        search_results = ['search interface unavailable']
    return {'search_results': search_results}

4. Bug‑tracking (simulated)

def bug_tracking(state: EmailAgentState) -> EmailAgentState:
    ticket_id = f"Bug-{uuid.uuid4()} fixed"
    return {'ticket_id': ticket_id}

5. Reply generation

def write_response(state: EmailAgentState) -> Command[Literal['human_review', 'send_reply']]:
    classification = state.get('classification', {})
    context_sections = []
    if state.get('search_results'):
        formatted_docs = "
".join([f"- {doc}" for doc in state['search_results']])
        context_sections.append(f"Related content:
{formatted_docs}")
    if state.get('customer_history'):
        tier = state['customer_history'].get('tier', 'standard')
        context_sections.append(f"Customer tier: {tier}")
    draft_prompt = f"""
    Write a 50‑word email reply:
    Email content: {state.get('email_content')}
    Email classification: {classification.get('intent','unknown')}
    Urgency: {classification.get('urgency','medium')}
    {'
'.join(context_sections)}
    """
    response = llm.invoke(draft_prompt)
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )
    goto = 'human_review' if needs_review else 'send_reply'
    return Command(update={'draft_response': response.content}, goto=goto)

6. Human review

def human_review(state: EmailAgentState) -> Command[Literal['send_reply', END]]:
    classification = state.get('classification', {})
    human_decision = interrupt({
        "Email ID": state['email_id'],
        "Original content": state['email_content'],
        "Draft reply": state['draft_response'],
        "Urgency": classification['urgency'],
        "Intent": classification['intent'],
        "Next": "Please approve sending this email"
    })
    if human_decision == 'approved':
        return Command(goto='send_reply')
    else:
        return Command(goto=END)

7. Send reply (simulated)

def send_reply(state: EmailAgentState):
    print('---Email sent successfully---')

Graph construction

builder = StateGraph(EmailAgentState)
builder.add_node('read_email', read_email)
builder.add_node('classify_intent', classify_intent)
builder.add_node('search_documentation', search_documentation)
builder.add_node('bug_tracking', bug_tracking)
builder.add_node('write_response', write_response)
builder.add_node('human_review', human_review)
builder.add_node('send_reply', send_reply)

builder.add_edge(START, 'read_email')
builder.add_edge('read_email', 'classify_intent')
builder.add_edge('classify_intent', 'search_documentation')
builder.add_edge('classify_intent', 'bug_tracking')
builder.add_edge('search_documentation', 'write_response')
builder.add_edge('bug_tracking', 'write_response')
builder.add_edge('send_reply', END)

memory = InMemorySaver()
app = builder.compile(checkpointer=memory)

Testing and validation

High‑priority bug report

Input state:

initial_state = {
    "email_content": "我遇到了一个紧急bug, 有客户重复订阅了一个产品",
    "sender_email": "[email protected]",
    "email_id": "email_123"
}
config = {"configurable": {"thread_id": "customer_123"}}
result = app.invoke(initial_state, config=config)

The classification node returns intent: bug and urgency: critical. A bug ticket is created, human_review is triggered, and after manual approval the reply is sent.

Routine greeting email

Input state:

initial_state = {
    "email_content": "你好呀，我是新同事苍井空",
    "sender_email": "[email protected]",
    "email_id": "email_123"
}
config = {"configurable": {"thread_id": "customer_123"}}
result = app.invoke(initial_state, config=config)

The workflow classifies the email as intent: question and urgency: low. No human review is required; the reply is generated and sent automatically.