Skip to content
Module 3 · Services Deep DiveIntermediate6 min read · 1,236 words

Lesson 08c: Multi-Agent Email Service

"One agent analyzes, one agent replies, one agent reviews - collaboration beats solo performance."

Introduction

Multi-agent systems decompose complex tasks into specialized roles. Instead of one "do everything" agent, we have three specialized agents working together:

  1. Analysis Agent - Extracts structured data
  2. Reply Agent - Generates professional response
  3. Reviewer Agent - Quality checks everything

File Overview

Location: backend/app/services/multi_agent_email_service.py
Size: 3,233 bytes
Functions: 3 agents
Pattern: Agent orchestration

Complete File Walkthrough

Imports and Setup

python
import json
import os
from dotenv import load_dotenv
from openai import OpenAI
from services.openai_service import analyze_email
 
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

Note: Creates separate OpenAI client (could be optimized to share with openai_service)

Agent 1: run_analysis_agent()

Purpose

Extract structured information from email.

Complete Code

python
def run_analysis_agent(email_text: str) -> dict:
    """
    Analysis Agent:
    Reads the email and extracts summary, intent, priority,
    deadline, action items, and suggested reply.
    """
    return analyze_email(email_text)

Design Pattern

Delegation Pattern: This agent delegates to openai_service.analyze_email()

Why wrap it?

  • Semantic clarity: "run_analysis_agent" describes role
  • Future extensibility: Could add pre/post-processing
  • Consistency: All agents follow same naming pattern

What it returns:

python
{
  "summary": "Database migration request",
  "intent": "request_action",
  "priority": "high",
  "deadline": "Friday",
  "action_items": ["Prepare scripts", "Test in staging"],
  "suggested_reply": "We'll handle this..."
}

Agent 2: run_reply_agent()

Purpose

Generate professional reply based on analysis.

Complete Code

python
def run_reply_agent(email_text: str, analysis: dict) -> str:

Parameters:

  • email_text: Original email (for context)
  • analysis: Output from analysis agent

Returns: str - Professional email reply

System Prompt Design

python
system_prompt = """
You are a professional email reply agent.
 
Your job:
- Write a clear professional reply.
- Use the analysis provided.
- Do not promise anything unrealistic.
- If a deadline is mentioned, acknowledge it carefully.
- Keep the reply concise and business-friendly.
Return only the email reply text.
"""

Key constraints:

  • ✅ Professional tone
  • ✅ Use provided analysis
  • ⚠️ No unrealistic promises (safety)
  • ⚠️ Careful with deadlines (liability)
  • ✅ Concise format

Why these rules matter:

python
# Bad (without constraints)
"We'll have this done by tomorrow!" 
# ❌ Unrealistic promise
 
# Good (with constraints)
"We'll review the requirements and provide a timeline by end of day."
# ✅ Professional, no commitments

User Prompt Construction

python
user_prompt = f"""
Original email:
{email_text}
 
Analysis:
{json.dumps(analysis, indent=2)}
 
Write the best professional reply:
"""

Structure:

  1. Context: Original email (understand sender's tone)
  2. Data: Analysis results (structured information)
  3. Instruction: Clear task

Why include original email?

  • Maintain context and tone
  • Reference specific points
  • Personalize response

Temperature Setting

python
temperature=0.3,

Why 0.3? (Creative enough but consistent)

  • Email replies need personality (not robotic)
  • But also need consistency (professional standard)
  • Balance between 0.2 (too rigid) and 0.5 (too variable)

Complete Example

Input:

Email: "Hi team, can you migrate our database to PostgreSQL by Friday? 
This is urgent."
 
Analysis: {
  "priority": "high",
  "deadline": "Friday",
  "action_items": ["Migrate database"]
}

Output:

Hi,
 
Thank you for reaching out regarding the database migration to PostgreSQL.
 
We understand this is a high-priority request with a Friday deadline. 
Our team will:
 
1. Review the current database structure today
2. Prepare a migration plan and backup strategy
3. Provide a realistic timeline by end of day
 
We'll keep you updated on our progress and flag any potential challenges early.
 
Best regards,
The Team

Notice:

  • ✅ Acknowledges urgency
  • ✅ No hard commitment to Friday
  • ✅ Proposes realistic steps
  • ✅ Professional tone

Agent 3: run_reviewer_agent()

Purpose

Quality assurance - check if analysis, tasks, and reply are correct and safe.

Complete Code

python
def run_reviewer_agent(
    email_text: str,
    analysis: dict,
    suggested_reply: str,
    tasks_created: list[dict],
) -> dict:

Parameters: Everything from previous agents
Returns: Quality assessment with approval

Reviewer System Prompt

python
system_prompt = """
You are a quality reviewer agent for an AI email workflow.
 
Review the analysis, tasks, and suggested reply.
 
Return ONLY valid JSON with this exact structure:
{
  "approved": true,
  "quality_score": 0.0,
  "issues": ["issue 1", "issue 2"],
  "recommendation": "short recommendation"
}
 
Rules:
- approved must be true only if the output is safe and useful.
- quality_score must be between 0 and 1.
- issues should be an empty list if there are no issues.
- Do not include markdown.
- Return only JSON.
"""

What reviewer checks:

  • ✅ Does reply address the email?
  • ✅ Are tasks relevant?
  • ✅ Is tone appropriate?
  • ✅ Any safety issues?
  • ✅ Missing information?

Review Criteria

Quality Score Interpretation:

  • 0.9-1.0: Excellent, approved
  • 0.7-0.8: Good, minor issues
  • 0.5-0.6: Acceptable, needs review
  • <0.5: Rejected, redo required

Temperature = 0 (Maximum Consistency)

python
temperature=0,

Why zero?

  • Reviewer makes binary decisions (approve/reject)
  • Needs maximum consistency
  • No creativity required
  • Deterministic evaluation

JSON Parsing with Validation

python
raw_text = response.choices[0].message.content
 
try:
    result = json.loads(raw_text)
except json.JSONDecodeError:
    raise ValueError(f"Reviewer Agent did not return valid JSON: {raw_text}")
 
return {
    "approved": bool(result.get("approved")),
    "quality_score": float(result.get("quality_score", 0)),
    "issues": result.get("issues", []),
    "recommendation": result.get("recommendation", ""),
}

Defensive programming:

  • ✅ Try-except for malformed JSON
  • ✅ Type coercion (bool(), float())
  • ✅ Default values with .get()
  • ✅ Clear error messages

Complete Multi-Agent Workflow

Orchestration (in routers/agents.py)

python
# Step 1: Analysis
analysis = run_analysis_agent(email_text)
 
# Step 2: Tasks (if enabled)
tasks_created = create_tasks_from_analysis(analysis)
 
# Step 3: Reply
suggested_reply = run_reply_agent(email_text, analysis)
 
# Step 4: Review
review = run_reviewer_agent(
    email_text,
    analysis,
    suggested_reply,
    tasks_created
)
 
# Step 5: Decision
if review["approved"] and review["quality_score"] > 0.7:
    return {"status": "approved", "reply": suggested_reply}
else:
    return {"status": "needs_review", "issues": review["issues"]}

Workflow Diagram

Rendering diagram…

Advantages of Multi-Agent Pattern

1. Separation of Concerns

Each agent has ONE job:

  • Analysis: Extract data
  • Reply: Generate text
  • Reviewer: Check quality

2. Easier to Debug

python
# Single-agent (hard to debug)
result = process_email(email)  # Black box
 
# Multi-agent (easy to debug)
analysis = run_analysis_agent(email)  # Check here
print(analysis)  # ✅ Data looks good
 
reply = run_reply_agent(email, analysis)  # Check here
print(reply)  # ❌ Reply too formal
 
# Found problem: Reply agent prompt needs adjustment

3. Independent Improvement

  • Improve reply quality without touching analysis
  • Add more review criteria without changing reply
  • Swap agents independently

4. Parallel Execution (Future)

python
# Sequential (current)
analysis = run_analysis_agent(email)  # 1.5s
reply = run_reply_agent(email, analysis)  # 2.0s
# Total: 3.5s
 
# Parallel (future optimization)
with ThreadPoolExecutor() as executor:
    analysis_future = executor.submit(run_analysis_agent, email)
    # Reply can start as soon as analysis completes
# Potential total: 2.5s (overlap possible)

Disadvantages & Tradeoffs

1. Higher Latency

3 LLM calls vs 1 LLM call = 3x latency

2. Higher Cost

3 API calls vs 1 API call = 3x cost

3. More Complex

More functions, more testing, more potential failures

When to use single-agent:

  • Simple tasks
  • Cost-sensitive
  • Latency-critical

When to use multi-agent:

  • Complex workflows
  • Quality-critical
  • Need auditability

Key Takeaways

🎯 Three specialized agents > one generalist 🎯 Analysis → Reply → Review pipeline 🎯 Temperature varies by agent role (0 for reviewer, 0.3 for reply) 🎯 Each agent has focused prompt and clear output 🎯 Reviewer provides quality gate with score 🎯 Pattern enables independent improvement

Interview Questions

Q: Why use 3 agents instead of 1?
A: Separation of concerns, easier debugging, independent improvement, quality gates.

Q: Why temperature=0 for reviewer?
A: Reviewer makes binary decisions, needs consistency, no creativity required.

Q: How would you optimize latency?
A: Parallel execution where possible, caching for repeated emails, async processing.

👉 Next: Document Service →

Lesson 08c Complete! Services: 3/5 (60%)

PreviousLesson 08b - Monitoring ServiceNextLesson 08d - Document Service