AI Automation: Build Intelligent Workflows That Work 24/7

IF email contains "invoice" THEN move to folder

Understand email intent → Categorize appropriately →
Draft response if needed → Flag for human review if complex

New Email Arrives
    ↓
AI Analyzes Content
    ↓
├── Spam/Marketing → Archive
├── Urgent → Notify immediately
├── Routine inquiry → Draft response
└── Complex → Flag for review

def process_email(email):
    analysis = ai.analyze(f"""
    Analyze this email and return JSON:
    - category: spam/marketing/urgent/routine/complex
    - summary: one sentence
    - action_needed: yes/no
    - suggested_response: if routine inquiry

    Email:
    From: {email.sender}
    Subject: {email.subject}
    Body: {email.body}
    """)

    if analysis.category == "routine" and analysis.suggested_response:
        create_draft(email, analysis.suggested_response)
    elif analysis.category == "urgent":
        send_notification(analysis.summary)

def process_invoice(document):
    extracted = ai.extract(f"""
    Extract from this invoice:
    - vendor_name
    - invoice_number
    - date
    - line_items (description, quantity, price)
    - total
    - due_date

    Return as JSON.

    Document text:
    {document.text}
    """)

    # Validate and save to accounting system
    if validate_invoice(extracted):
        save_to_quickbooks(extracted)
        return "Processed successfully"
    else:
        flag_for_review(document, extracted)
        return "Flagged for review"

Customer Query
    ↓
AI First Response (instant)
    ↓
├── Resolved → Close ticket
├── Needs info → Ask clarifying questions
└── Complex → Route to human + provide context

Blog Post Published
    ↓
AI Generates:
├── Twitter thread (5-7 tweets)
├── LinkedIn post (professional tone)
├── Email newsletter snippet
├── YouTube script outline
└── Instagram carousel text

def generate_daily_briefing(topics):
    # Gather information
    news = fetch_news(topics)
    social = fetch_social_mentions(topics)
    competitors = fetch_competitor_updates()

    # AI synthesis
    briefing = ai.generate(f"""
    Create a daily briefing covering:

    1. Top News (3-5 items)
    {news}

    2. Social Media Highlights
    {social}

    3. Competitor Updates
    {competitors}

    Format: Executive summary (2 paragraphs) followed by
    bullet points for each section.
    """)

    return briefing

Trigger → Process → AI Analysis → Action

Trigger → AI Decides → Multiple Possible Actions

Goal → AI Plans → Execute → Evaluate → Repeat

from openai import OpenAI

client = OpenAI(
    api_key="your-key",
    base_url="https://api.crazyrouter.com/v1"
)

def analyze_email(email_content):
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Fast and cheap for classification
        messages=[
            {
                "role": "system",
                "content": """You are an email assistant. Analyze emails and return JSON:
                {
                    "priority": "high/medium/low",
                    "category": "inquiry/complaint/order/spam/other",
                    "sentiment": "positive/neutral/negative",
                    "action_required": true/false,
                    "summary": "one sentence summary",
                    "suggested_reply": "draft reply if appropriate, null otherwise"
                }"""
            },
            {"role": "user", "content": email_content}
        ],
        response_format={"type": "json_object"}
    )
    return json.loads(response.choices[0].message.content)

import imaplib
import email

def fetch_new_emails():
    mail = imaplib.IMAP4_SSL('imap.gmail.com')
    mail.login('your@email.com', 'app-password')
    mail.select('inbox')

    _, messages = mail.search(None, 'UNSEEN')

    emails = []
    for num in messages[0].split():
        _, msg = mail.fetch(num, '(RFC822)')
        email_body = email.message_from_bytes(msg[0][1])
        emails.append({
            'from': email_body['from'],
            'subject': email_body['subject'],
            'body': get_body(email_body)
        })

    return emails

def process_inbox():
    emails = fetch_new_emails()

    for email in emails:
        analysis = analyze_email(f"""
        From: {email['from']}
        Subject: {email['subject']}
        Body: {email['body']}
        """)

        if analysis['priority'] == 'high':
            send_notification(f"Urgent: {analysis['summary']}")

        if analysis['suggested_reply']:
            create_draft(email, analysis['suggested_reply'])

        # Log for review
        log_analysis(email, analysis)

import schedule
import time

schedule.every(5).minutes.do(process_inbox)

while True:
    schedule.run_pending()
    time.sleep(1)

# Instead of 10 separate calls
for email in emails:
    analyze(email)

# Batch into one call
analyze_batch(emails)  # 10x cheaper

# Cache FAQ responses
faq_cache = {}

def get_response(query):
    query_hash = hash(normalize(query))
    if query_hash in faq_cache:
        return faq_cache[query_hash]

    response = ai.generate(query)
    faq_cache[query_hash] = response
    return response

# Bad: Send entire email thread
analyze(full_thread)  # 10,000 tokens

# Good: Send relevant parts
analyze(extract_latest_message(full_thread))  # 500 tokens

def robust_ai_call(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                timeout=30
            )
            return response.choices[0].message.content

        except RateLimitError:
            time.sleep(2 ** attempt)
        except APIError as e:
            log_error(e)
            if attempt == max_retries - 1:
                return fallback_response(prompt)

    return fallback_response(prompt)

def process_with_review(item):
    ai_result = ai.process(item)

    if ai_result.confidence < 0.8:
        # Queue for human review
        add_to_review_queue(item, ai_result)
        return "Pending review"

    if ai_result.is_high_stakes:
        # Require approval
        request_approval(item, ai_result)
        return "Pending approval"

    # Auto-process
    execute_action(ai_result)
    return "Completed"

New Lead Submitted
    ↓
AI Analyzes:
- Company size
- Industry fit
- Budget signals
- Urgency indicators
    ↓
Score: High/Medium/Low
    ↓
├── High → Immediate sales notification
├── Medium → Add to nurture sequence
└── Low → Marketing automation only

Ticket Created
    ↓
AI Extracts:
- Issue type
- Severity
- Product area
- Customer tier
    ↓
Route to:
├── Billing → Finance team
├── Technical → Engineering
├── General → Support queue
└── VIP → Priority handling

User Content Submitted
    ↓
AI Checks:
- Policy violations
- Spam indicators
- Quality score
    ↓
├── Clear violation → Auto-remove
├── Borderline → Human review
└── Approved → Publish