No-Code Agentic Workflows

"""Python equivalent of the n8n "email triage with approval" workflow

Each function below matches one node in the visual workflow:
  Trigger -> Classify (AI) -> Route (Switch) -> Draft reply (AI)
          -> Human approval -> Send / Log

Seeing the same logic as code helps you understand what the visual nodes do,
and shows when moving from no-code to code makes sense.

Illustrative: classify() and draft_reply() stand in for LLM nodes.
"""
from datetime import datetime

NEW_EMAILS = [
    {"from": "student@university.edu", "subject": "Can I join the crash course late?", "body": "I missed week 1."},
    {"from": "promo@shop.com", "subject": "50% off laptops", "body": "Today only!"},
    {"from": "dean@university.edu", "subject": "Urgent: room change", "body": "Workshop moves to Building 22."},
]

def trigger():                        # n8n: Gmail / IMAP trigger node
    return NEW_EMAILS

def classify(email):                  # n8n: AI Agent or Text Classifier node
    text = (email["subject"] + " " + email["body"]).lower()
    if "urgent" in text:
        return "urgent"
    if any(w in text for w in ("sale", "% off", "promo")):
        return "spam"
    return "question"

def draft_reply(email):               # n8n: AI Agent node with a reply prompt
    return f"Hello, thanks for your message about '{email['subject']}'. We will get back to you shortly."

def human_approval(draft):            # n8n: "Send and wait for response" approval step
    return "approve"                  # simulated: a person clicked Approve

def send(email, text):                # n8n: Gmail "send" node
    return f"sent to {email['from']}"

LOG = []
def log(email, category, outcome):    # n8n: Google Sheets / database node
    LOG.append({"time": datetime(2026, 1, 1, 9, 0).isoformat(), "from": email["from"], "category": category, "outcome": outcome})

for email in trigger():
    category = classify(email)
    if category == "spam":                                   # n8n: Switch node, branch 1
        log(email, category, "archived")
    elif category == "urgent":                               # branch 2
        log(email, category, "forwarded to on-call staff")
    else:                                                    # branch 3
        draft = draft_reply(email)
        decision = human_approval(draft)
        log(email, category, send(email, draft) if decision == "approve" else "rejected by reviewer")

for row in LOG:
    print(row)

"""Python equivalent of a visual "document Q&A" (RAG) flow

Visual builders such as Langflow, Flowise, and Dify ship templates for
retrieval-augmented generation (RAG). The flow is the same in each tool:

  Load documents -> Split into chunks -> Embed -> Store in a vector index
  Question -> Embed -> Retrieve top chunks -> Prompt + LLM -> Answer with sources

Illustrative: we "embed" text as word-count vectors and use cosine similarity
so this runs without downloads. Real flows use an embedding model and a
vector database.
"""
import math
import re
from collections import Counter

DOCS = {
    "syllabus.txt": "The crash course has ten modules. Module 5 covers the Model Context Protocol (MCP). "
                    "Module 10 covers governance, guardrails and human oversight.",
    "faq.txt": "Notebooks open in Google Colab. You need basic Python. No paid API key is required "
               "because examples use a stand-in model.",
}

def split(text):                                    # "Text splitter" node (one chunk per sentence)
    return [s for s in re.split(r"(?<=[.!?])\s+", text) if s]

def embed(text):                                    # "Embeddings" node
    return Counter(re.findall(r"[a-z0-9]+", text.lower()))

def cosine(a, b):
    dot = sum(a[k] * b[k] for k in a)
    return dot / ((math.sqrt(sum(v * v for v in a.values())) * math.sqrt(sum(v * v for v in b.values()))) or 1)

INDEX = [(name, chunk, embed(chunk)) for name, text in DOCS.items() for chunk in split(text)]  # "Vector store" node

def retrieve(question, k=2):                        # "Retriever" node
    q = embed(question)
    return sorted(INDEX, key=lambda item: cosine(q, item[2]), reverse=True)[:k]

def answer(question):                               # "Prompt" + "LLM" nodes (stand-in)
    hits = retrieve(question)
    context = "\n".join(f"[{name}] {chunk}" for name, chunk, _ in hits)
    prompt = f"Answer using only this context:\n{context}\n\nQuestion: {question}"
    best_name, best_chunk, _ = hits[0]
    return prompt, f"{best_chunk} (source: {best_name})"

for q in ["Which module covers MCP?", "Do I need an API key?"]:
    prompt, reply = answer(q)
    print("PROMPT SENT TO THE LLM:\n" + prompt)
    print("ANSWER:", reply, "\n" + "-" * 60)