Building Email Triage Agent with AI Agents 2026

Introduction

AI agents that can read, categorize, and draft email responses are transforming how developers work. This guide shows you how to build a production-ready Email Triage Agent using ReAct + Gmail API.

What We're Building

A Email Triage Agent that can:

Understand complex requests

Break them into sub-tasks

Execute tasks autonomously

Handle errors and retry

Produce consistent, high-quality output

Architecture

User Request
    ↓
[Email Triage Agent Orchestrator]
    ↓
[Task Planning] → [Tool Selection] → [Execution]
    ↓                                      ↓
[Validation] ←──────────────────── [Results]
    ↓
Final Output

Implementation with LangGraph

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
import json
State definition
class EmailTriageAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    task: str
    sub_tasks: List[str]
    completed_tasks: List[str]
    final_output: str | None
    iterations: int
Define tools for read, categorize, and draft email responses
@tool
def analyze_task(task: str) -> str:
    """Break down a complex task into sub-tasks."""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(f"Break this into 3-5 specific, actionable sub-tasks: {task}")
    return response.content
tools = [analyze_task, execute_sub_task, validate_output]
Initialize LLM with tools
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
llm_with_tools = llm.bind_tools(tools)
Agent node
def agent_node(state: EmailTriageAgentState):
    if state.get("iterations", 0) > 8:
        return {"final_output": "Max iterations reached", "iterations": 9}
    
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1
    }
Tool execution node
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
def should_continue(state: EmailTriageAgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
        return "tools"
    return "end"
Build graph
workflow = StateGraph(EmailTriageAgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
agent = workflow.compile()

Using the Agent

python
from langchain_core.messages import HumanMessage
def run_email_triage_agent(request: str) -> str:
    """Run the Email Triage Agent on a user request."""
    
    initial_state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent.invoke(initial_state)
    
    # Extract the final answer
    last_message = result["messages"][-1]
    return last_message.content
Usage
output = run_email_triage_agent(
    "Create a comprehensive plan to read, categorize, and draft email responses"
)
print(output)

Adding Memory with Persistence

python
from langgraph.checkpoint.sqlite import SqliteSaver
def run_with_memory(request: str, session_id: str) -> str:
    config = {"configurable": {"thread_id": session_id}}
    
    state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent_with_memory.invoke(state, config=config)
    return result["messages"][-1].content
First interaction
response1 = run_with_memory("Start read, categorize, and draft email responses", session_id="session-001")
Follow-up (agent remembers context)
response2 = run_with_memory("Continue from where we left off", session_id="session-001")

Production: FastAPI Service

python
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
app = FastAPI(title="Email Triage Agent Service")
@app.post("/agent/run")
async def run_agent(request: AgentRequest):
    if request.stream:
        async def stream_response():
            async for event in agent.astream_events(
                {"messages": [HumanMessage(content=request.task)], 
                 "task": request.task, "sub_tasks": [], 
                 "completed_tasks": [], "final_output": None, "iterations": 0},
                version="v2"
            ):
                if event["event"] == "on_chat_model_stream":
                    content = event["data"]["chunk"].content
                    if content:
                        yield content
        
        return StreamingResponse(stream_response(), media_type="text/plain")
    
    result = run_email_triage_agent(request.task)
    return {"result": result, "session_id": request.session_id}
@app.get("/health")
async def health():
    return {"status": "healthy", "agent": "Email Triage Agent"}

Monitoring Agent Performance

python
from dataclasses import dataclass
from datetime import datetime
import statistics
metrics_store: List[AgentMetrics] = []
def run_with_metrics(task: str) -> tuple[str, AgentMetrics]:
    import time
    start = time.time()
    
    try:
        result = run_email_triage_agent(task)
        success = True
    except Exception as e:
        result = f"Error: {e}"
        success = False
    
    duration = (time.time() - start) * 1000
    
    # Note: iterations would come from actual state in production
    metrics = AgentMetrics(
        task=task[:50],
        iterations=3,  
        duration_ms=duration,
        success=success,
        output_length=len(result)
    )
    metrics_store.append(metrics)
    return result, metrics
def print_metrics_report():
    if not metrics_store:
        return
    
    successful = [m for m in metrics_store if m.success]
    durations = [m.duration_ms for m in metrics_store]
    
    print(f"Total runs: {len(metrics_store)}")
    print(f"Success rate: {len(successful)/len(metrics_store):.1%}")
    print(f"Avg duration: {statistics.mean(durations):.0f}ms")
    print(f"p95 duration: {sorted(durations)[int(len(durations)*0.95)]:.0f}ms")

Best Practices

Conclusion

Building Email Triage Agent with AI agents enables autonomous read, categorize, and draft email responses. The LangGraph implementation provides the right balance of control and flexibility for production use.

Start with a simple proof of concept, add persistence, then scale up as confidence grows.

---

*Email Triage Agent implementation using ReAct + Gmail API | May 2026*