AI Agents: Mastering 3 Essential Patterns (Reflexive). Part 3 of 3

#ai #programming #python #agents

The code for these patterns is available on Github. [Repo]

The Reflexive Pattern – When There's No Time to Think

In Article 2 (ReAct), we turned our agent into a meticulous "Investigator." The guy would sit down, analyze, plan, Google search, and then answer. That's great for complex tasks, but it has a huge bug in the real world: it's slow.

Think of it this way: if you touch a hot stove, your brain doesn't start a debate: "Hmm, I detect a temperature of 200°C, this will cause protein denaturation, I suggest moving the biceps...". Haha, no. Your spinal cord takes command and pulls your hand away before you even feel the pain.

The Reflexive Pattern (sometimes called Semantic Router) is exactly that: the "reflex action" of Artificial Intelligence.

The Use Case: The Security Sentinel

To understand this, imagine we build a system that reads your server logs in real-time. Among thousands of normal visits, suddenly this comes in:
"GET /search?q=' OR 1=1;--"

If you use a ReAct agent here, you're dead. The agent would think:

"Wow, an unusual pattern. I'm going to use the search tool to understand what 'SQL Injection' is. Ah, I see it's dangerous. Proceeding to block."

By the time it finished "thinking" that (10-15 seconds), your database is already leaked on the Dark Web.

Here is where the Reflexive Agent shines:

Stimulus: Sees the malicious string.
Reflex: Executes block_ip().
Time: < 1 second. No doubts. No loops.

Pattern Anatomy: Cutting the Red Wire

The architecture here is radically linear.

Unlike ReAct, here we explicitly eliminate the ability to reason.

Let's look at an example below...

import os
import sys
import logging
import traceback
from typing import List
from dotenv import load_dotenv, find_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

# 1. Global Logging and Error Handling Configuration
LOG_DIR = os.path.join(os.path.dirname(__file__), "log")
LOG_FILE = os.path.join(LOG_DIR, "logs.txt")

if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

def global_exception_handler(exctype, value, tb):
    """Captures unhandled exceptions and records them in the log."""
    error_msg = "".join(traceback.format_exception(exctype, value, tb))
    logger.error(f"Unhandled exception:\n{error_msg}")
    sys.__excepthook__(exctype, value, tb)

sys.excepthook = global_exception_handler

# 2. Environment Variables Loading
env_path = find_dotenv()
if env_path:
    load_dotenv(env_path)
    logger.info(f".env file loaded from: {env_path}")
else:
    logger.warning(".env file not found")

# 3. Action Tool Definitions (Simulated)
def block_ip(ip: str, reason: str) -> str:
    """
    Blocks a malicious IP address immediately.

    Args:
        ip (str): The IP address to block.
        reason (str): The reason for blocking (e.g., SQL Injection attempt).
    """
    msg = f"⛔ BLOCKING IP [{ip}]: {reason}"
    logger.warning(msg)
    return msg

def escalate_to_admin(log_id: str, severity: str) -> str:
    """
    Escalates a critical issue to a human administrator.

    Args:
        log_id (str): Unique identifier for the log or event.
        severity (str): Severity level (Critical, Urgent, etc.).
    """
    msg = f"⚠️ ADMIN ALERT: Log [{log_id}] - Severity [{severity}]"
    logger.critical(msg)
    return msg

def log_normal_traffic(source: str) -> str:
    """
    Registers legitimate and safe traffic.

    Args:
        source (str): Traffic source (IP or user).
    """
    msg = f"✅ Normal traffic from [{source}]"
    logger.info(msg)
    return msg

# 4. Agno Agent Configuration (Reflexive Pattern)
model_id = os.getenv("BASE_MODEL", "gpt-4o-mini")

agent = Agent(
    model=OpenAIChat(id=model_id),
    tools=[block_ip, escalate_to_admin, log_normal_traffic],
    instructions=[
        "You are an automated security system (Reflexive Pattern).",
        "Your job is to classify incoming logs and execute the corresponding tool IMMEDIATELY.",
        "Action rules:",
        "- If you detect SQL Injection, XSS, or known attacks -> Use 'block_ip'.",
        "- If you detect server errors (500), service downs, or timeouts -> Use 'escalate_to_admin'.",
        "- If traffic seems legitimate, successful login, or normal navigation -> Use 'log_normal_traffic'.",
        "DO NOT explain your decision. DO NOT greet. Just execute the tool directly in response to the stimulus.",
        "Process the log and terminate execution."
    ]
)

# 5. Flow Simulation (The Stream)
def main():
    logger.info("Starting Security Sentinel Agent (Reflexive)...")
    print("--- Security Sentinel - Reflexive Pattern ---")

    # Extended list of simulated logs
    logs = [
        "User 'admin' logged in successfully from 192.168.1.10",
        "GET /search?q=' OR 1=1;-- (SQL Injection attempt from 203.0.113.5)",
        "Service 'PaymentGateway' timed out after 3000ms (Error 500)",
        "User 'jdoe' updated profile picture",
        "<script>alert('hacked')</script> sent via form from 88.22.11.44",
        "DB_ERROR: Connection pool exhausted (LogID: 4452)",
        "Bot crawler identified: Googlebot/2.1 from 66.249.66.1",
        "Multiple failed login attempts for user 'sysadmin' from 45.33.22.11",
        "HealthCheck: All systems operational - latency 12ms",
        "Unauthorized access attempt to /etc/passwd from 172.16.0.50"
    ]

    print(f"\nProcessing {len(logs)} logs in the 'stream'...\n")

    for i, log_entry in enumerate(logs, 1):
        try:
            print(f"[{i}] Stimulus: {log_entry}")
            logger.info(f"Processing log {i}: {log_entry}")

            # In reflexive mode, we expect the agent to execute the tool immediately.
            # We use show_tool_calls=True to demonstrate the action.
            agent.print_response(log_entry, show_tool_calls=True)
            print("-" * 50)

        except Exception as e:
            logger.error(f"Error processing log {i}: {str(e)}")
            print(f"An error occurred in event {i}: {e}")

if __name__ == "__main__":
    main()

Restrictive Instructions (Negative Constraints): In the system prompt (instructions), we explicitly forbid the agent to think or chat ("DO NOT explain", "Just execute"). It is programmed solely to fire tools.
Direct Mapping: There are no feedback loops. The flow is linear: Incoming Log → Classification → Tool Execution.
Model Selection: gpt-4o-mini is configured. By using a smaller and faster model, we prioritize latency and processing volume.

Why is this Pattern so Cool? (Pros)

Absurd Speed (Minimal Latency): By not generating "thought" text, the response is almost instantaneous.
Eats whatever you throw at it (High Throughput): It's perfect for processing thousands of emails, live chats, or monitoring alerts per minute.
It's Predictable (Determinism): By removing the "creativity," the agent behaves almost like a classic script. Given X, it always does Y.
Cheap: Fewer output tokens and smaller models = ridiculous API bill.

Where does it fail? (Cons)

Goldfish Memory (Contextual Blindness): The agent reacts to what is in front of it right now. If an attack is split into three steps over 10 minutes, this agent won't connect the dots.
It's Stubborn (Rigidity): If it encounters something ambiguous that isn't in its rules, it tends to fail or hallucinate a wrong category because we force it to choose quickly.
Black Box: Unlike ReAct, there are no "Thought" logs here to tell you why it blocked a user. It just did it, and that's it.

Comparison for Developers

Feature	ReAct (Level 2)	Reflexive (Level 3)
Analogy	Sherlock Holmes	A Nightclub Bouncer
Priority	Quality and Reasoning	Speed and Volume
Loop	Yes (Think-Act-Observe)	No (Fire and Forget)
Ideal Model	Top Models (GPT-4o, Claude)	Light Models (Mini, Haiku)

Tips from the Trenches

If you are going to put this into production, be careful with these three things:

Beware the "Ban Hammer": Since this agent shoots first and asks questions later, you are going to have false positives. My advice: make the action flag_for_review() instead of delete_account(), unless you are 100% sure.
Be strict with the Prompt (Negative Constraints): You have to use negative constraints.
Bad prompt: "Classify this text."
Good prompt: "Classify this text. DO NOT explain your reason. DO NOT generate introductory text. Only return the JSON."
The Hybrid Architecture: The Reflexive pattern is usually the "Infantry." You put this agent on the front line to filter out 80% of the easy noise. If it detects something it doesn't understand or is ambiguous, then (and only then) does it escalate the ticket to a ReAct agent (smarter and more expensive) to investigate. That way you have the best of both worlds.