Workshop

Building
Agentic AI Systems
from Scratch

One Problem, Three Frameworks

LangChain  ·  LangGraph  ·  CrewAI

Instructor: Ruslan Magana Vsevolodovna  |  ruslanmv.com

What You Will Learn

Course Outline

What Is an Agent?

A system that controls the flow, not just the output.

The agent decides what to do next — a regular LLM just gives you one answer.

The Spectrum: LLM → CoT → Agent

It is not binary. There is a spectrum of intelligence — understanding it helps you choose the right level for each problem.

Rule of thumb: if the task needs external data or has business impact → use an agent. Pure text → LLM or CoT is enough.

The ReAct Pattern

Reason + Act — the loop at the heart of modern agents

Reasoning is explicit, auditable, and debuggable — every step is logged.

Multi-Agent Systems

Multiple agents, each with a specialised role, collaborating to solve a problem

Memory & Context: How Agents Remember

Three layers of memory — each extends how far the agent can reach.

In this course we use short-term memory. For production, add checkpointing and RAG for knowledge that exceeds the context window.

MCP & A2A: Connecting Agents

Two protocols power the agentic ecosystem. Understanding the difference is critical.

Orchestration Patterns

Choosing the right pattern is often more important than choosing the right framework.

In this course: ReAct Loop (LangChain), DAG (LangGraph), Sequential (CrewAI).

Key Takeaway

An agent is a system that reasons, acts, and iterates toward a goal using tools and state, rather than producing a single response. It manages memory through context windows, checkpoints, and vector stores. It connects to tools via MCP and to other agents via A2A. The orchestration pattern — sequential, DAG, ReAct, hierarchical, or routing — determines how much control vs. flexibility the system has.

The Problem: Email Triage

Classify incoming emails and route them to the right action.

The Pipeline

Shared by all three approaches

Clone & Install

git clone https://github.com/ruslanmv/agentic-ai-concepts.git
cd agentic-ai-concepts

python3 -m venv .venv
source .venv/bin/activate

make install          # pip install -r requirements.txt
export OPENAI_API_KEY="sk-..."

Verify the setup:

make test             # 80 offline tests — no API key needed
make evaluate         # baseline evaluation against golden dataset

Project Structure

agentic-ai-concepts/
├── src/
│   ├── schema.py              # Pydantic models
│   ├── preprocessing.py       # PII masking
│   ├── fallback.py            # Keyword fallback
│   ├── routing.py             # Label → action
│   ├── tools.py               # Agent tools
│   ├── evaluate.py            # Metrics + prod gate
│   ├── langchain_agent.py     # Approach 1
│   ├── langgraph_agent.py     # Approach 2
│   └── crewai_agent.py        # Approach 3
├── data/golden_dataset.csv    # 30 labelled emails
├── tests/                     # 80 offline + 12 integration
└── examples/
    ├── run_all.py
    └── compare_frameworks.py  # Side-by-side verdict

Module 1 — Schema

The contract all three frameworks must produce.

class EmailLabel(str, Enum):
    PHISHING = "phishing"
    SPAM     = "spam"
    INVOICE  = "invoice"
    MEETING  = "meeting"
    SUPPORT  = "support"
    OTHER    = "other"

class EmailClassification(BaseModel):
    label: EmailLabel
    confidence: confloat(ge=0.0, le=1.0)
    rationale: str
    indicators: List[str] = []
    requires_human_review: bool = False

Pydantic enforces the contract — invalid LLM output fails fast.

Module 2 — PII Preprocessing

Replace sensitive data before sending to the LLM.

_PII_PATTERNS = OrderedDict(
    SSN=re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    CREDIT_CARD=re.compile(r"\b(?:\d[ -]*?){13,16}\b"),
    EMAIL=re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
    IBAN=re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b"),
    PHONE=re.compile(r"\b(\+?\d[\d\s\-\(\)]{7,}\d)\b"),
)

def mask_pii(text: str) -> str:
    for tag, pattern in _PII_PATTERNS.items():
        text = pattern.sub(f"[{tag}]", text)
    return text

>>> mask_pii("Contact [email protected], SSN 123-45-6789")
'Contact [EMAIL], SSN [SSN]'

Module 3 — Keyword Fallback

Deterministic safety net — always have a fallback for any probabilistic component.

_KEYWORD_MAP = {
    EmailLabel.PHISHING: ["verify", "password", "urgent", "suspend", 
                          "locked", "click", "link", "account"],
    EmailLabel.INVOICE:  ["invoice", "payment", "remittance", "iban", "vat"],
    EmailLabel.MEETING:  ["meeting", "calendar", "invite", "zoom", "agenda"],
    EmailLabel.SUPPORT:  ["ticket", "issue", "bug", "incident", "outage"],
    EmailLabel.SPAM:     ["unsubscribe", "promotion", "deal", "free", "win"],
}

def keyword_fallback(subject, body) -> EmailClassification:
    # Count keyword hits per category → pick highest
    # Confidence capped at 0.8 (honest about limitations)
    ...

Module 4 — Routing

Map classification → downstream action. Human review always takes priority.

class RouteAction(str, Enum):
    HUMAN_REVIEW = "queue_for_human_review"
    AP_QUEUE     = "send_to_ap_queue"
    CALENDAR     = "create_calendar_suggestion"
    TICKET       = "create_support_ticket"
    QUARANTINE   = "quarantine"
    INBOX        = "inbox"

def route(classification: EmailClassification) -> RouteAction:
    if classification.requires_human_review:
        return RouteAction.HUMAN_REVIEW    # always takes priority
    return _LABEL_TO_ACTION.get(classification.label, RouteAction.INBOX)

Why Tools Matter

Four Tools

Simulated databases for offline testing. In production, swap for real API (Application Programming Interface) calls.

Tool Example: scan_urls

@tool
def scan_urls(email_body: str) -> str:
    """Scan all URLs in an email body for malicious indicators."""
    url_pattern = re.compile(r"https?://[^\s<>\"']+")
    urls = url_pattern.findall(email_body)

    if not urls:
        return "No URLs found. URL risk: NONE."

    results = []
    for url in urls:
        is_malicious = any(
            re.search(p, url.lower())
            for p in _MALICIOUS_URL_PATTERNS
        )
        if is_malicious:
            results.append(f"  {url} → MALICIOUS (risk: 0.9)")
        ...

    return f"Found {len(urls)} URL(s):\n" + "\n".join(results)

The @tool decorator makes it callable by the LLM agent.

Different Emails → Different Tools

The ReAct Loop in Action

Building the Agent

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from src.tools import ALL_TOOLS

def _build_agent():
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

    agent = create_react_agent(
        model=llm,
        tools=ALL_TOOLS,      # 4 tools from src/tools.py
        prompt=SYSTEM_PROMPT,
    )
    return agent

That's it — create_react_agent handles the entire Reason → Act → Observe loop.

System Prompt

SYSTEM_PROMPT = """
You are an enterprise email triage agent.
Classify emails by gathering evidence using your tools.

## Workflow
1. Analyse the email for signals
2. Use tools to gather evidence:
   - URLs or suspicious → scan_urls
   - Sender domain → check_sender_reputation
   - Sender email → lookup_known_contacts
   - Invoice number → check_invoice_registry
3. Classify: phishing/spam/invoice/meeting/support/other
4. Return JSON: label, confidence, rationale, indicators

## Rules
- Phishing → always requires_human_review = true
- Base confidence on tool evidence, not gut feeling
"""

Guardrails

Applied after the agent produces a classification.

CONFIDENCE_THRESHOLD = 0.6

def apply_guardrails(classification, subject, body):
    # Hard rule: phishing → always flag, cap confidence
    if classification.label == EmailLabel.PHISHING:
        classification.requires_human_review = True
        classification.confidence = min(classification.confidence, 0.85)
        return classification

    # Soft rule: low confidence → deterministic fallback
    if classification.confidence < CONFIDENCE_THRESHOLD:
        return keyword_fallback(subject, body)

    return classification

Run It

make run-langchain
# or: python -m src.langchain_agent

The tools_used field is the audit trail — you know exactly why the agent decided.

Architecture

Typed State

class GraphState(TypedDict):
    """Immutable state flowing through the graph."""
    subject: str
    body: str
    masked_body: str
    classification: Optional[EmailClassification]
    action: Optional[str]

Node Functions

Each node is independently testable.

def preprocess_node(state: GraphState) -> dict:
    sanitised = preprocess_email(state["subject"], state["body"])
    return {"masked_body": sanitised["body"]}

def classify_node(state: GraphState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
    classifier = llm.with_structured_output(EmailClassification)
    result = (prompt | classifier).invoke({
        "subject": state["subject"], "body": state["masked_body"],
    })
    return {"classification": result}

def guardrails_node(state: GraphState) -> dict:
    result = state["classification"]
    if result.label == EmailLabel.PHISHING:
        result.requires_human_review = True
        result.confidence = min(result.confidence, 0.85)
    return {"classification": result}

Building the Graph

from langgraph.graph import END, StateGraph

def build_graph():
    graph = StateGraph(GraphState)

    # Nodes
    graph.add_node("preprocess", preprocess_node)
    graph.add_node("classify",   classify_node)
    graph.add_node("guardrails", guardrails_node)
    graph.add_node("fallback",   fallback_node)
    graph.add_node("route",      route_node)

    # Edges
    graph.set_entry_point("preprocess")
    graph.add_edge("preprocess", "classify")
    graph.add_edge("classify",   "guardrails")
    graph.add_conditional_edges("guardrails",
        decide_after_guardrails,
        {"fallback": "fallback", "route": "route"})
    graph.add_edge("fallback", "route")
    graph.add_edge("route", END)

    return graph.compile()

Why LangGraph for Enterprise

Three Agents, One Crew

Agent Definition

from crewai import Agent, Crew, Process, Task

classifier_agent = Agent(
    role="Email Classifier",
    goal="Classify the email into one of: phishing, spam, "
         "invoice, meeting, support, other.",
    backstory="You are a senior email triage specialist "
              "at a Fortune 500 company.",
    verbose=False,
    allow_delegation=False,
)

# ... risk_agent, policy_agent defined similarly

crew = Crew(
    agents=[classifier_agent, risk_agent, policy_agent],
    tasks=[classify_task, risk_task, policy_task],
    process=Process.sequential,
)
result = crew.kickoff()

Trade-offs

Best for complex problems that genuinely need collaborative multi-step reasoning across different roles.

Test Strategy

Running Tests

# Offline only (fast, no API key)
make test

# Everything including live LLM
make test-all

The Golden Dataset

data/golden_dataset.csv — 30 hand-labelled emails

Hard samples: BEC wire transfer, legit security alert, ambiguous reply threads.

Precision vs Recall

F1 = harmonic mean of both. It balances precision and recall.

Production Readiness Gate

MIN_WEIGHTED_F1      = 0.70   # Overall performance
MIN_PHISHING_RECALL  = 0.80   # Must catch ≥ 80% of phishing
MIN_PHISHING_PREC    = 0.60   # Must not over-flag

Three checks. All must pass.

Baseline Result: Keyword Fallback

make evaluate

Keywords alone aren't enough. We need the LLM agents.

Run the Comparison

export OPENAI_API_KEY="sk-..."
make compare

Runs all 30 golden samples through all 4 approaches.

# Or directly:
python examples/compare_frameworks.py --all

Side-by-Side Results

Numbers may vary slightly due to LLM non-determinism.

The Verdict by Criterion

🏆 Recommendation

CrewAI is the right choice when you need collaborative multi-step reasoning across truly different roles — but overkill for single classification.

Production Architecture

Saving & CI/CD (Continuous Integration / Continuous Delivery) Integration

# In your CI pipeline:
python examples/compare_frameworks.py --all \
    --output data/eval_results/comparison.json

# Parse the result:
import json
data = json.load(open('data/eval_results/comparison.json'))
winner = data.get('recommended')
print(f'Recommended: {winner}')
for a in data['approaches']:
    gate = '✓' if a['passed_production_gate'] else '✗'
    print(f"  {gate} {a['approach']}: F1={a['weighted_f1']:.4f}")

What Makes an Agent Production-Ready

The One-Paragraph Summary

An agent is a system that reasons, acts, and iterates toward a goal using tools and state, rather than producing a single response. Multi-agent systems decompose complex workflows across specialised roles. LangChain enables ReAct-style agents with dynamic tool selection, LangGraph provides deterministic state-machine orchestration for enterprise workflows, and CrewAI enables collaborative multi-agent designs. Agentic systems trade simplicity for control, safety, and auditability.

What to Build Next

Thank You!

📦 github.com/ruslanmv/agentic-ai-concepts

📝 Full tutorial: docs/blog.md

🌐 ruslanmv.com

If this course helped, ⭐ the repo and share with your team.