Components of AI Agents

"""Perception: turning raw inputs into structured observations

An agent cannot act on raw text, images, or sensor values directly. The
perception step cleans and structures them into an "observation" that the
reasoning step can use. Here we parse a messy user request and a sensor feed.

Illustrative: a real agent would often use an LLM or a vision model for this
step. The goal is the same: produce a clean, structured observation.
"""
import json
import re
from dataclasses import asdict, dataclass, field


@dataclass
class Observation:
    intent: str
    entities: dict = field(default_factory=dict)
    sensors: dict = field(default_factory=dict)
    warnings: list = field(default_factory=list)


def perceive(user_text: str, sensor_feed: dict) -> Observation:
    text = user_text.lower().strip()

    # Very small intent detector (an LLM classifier would go here).
    if any(w in text for w in ("book", "reserve")):
        intent = "book_room"
    elif any(w in text for w in ("weather", "temperature", "hot")):
        intent = "check_conditions"
    else:
        intent = "unknown"

    entities = {}
    if m := re.search(r"(\d+)\s*(people|persons|students)", text):
        entities["group_size"] = int(m.group(1))
    if m := re.search(r"\b(\d{1,2})\s*(am|pm)\b", text):
        hour = int(m.group(1)) % 12 + (12 if m.group(2) == "pm" else 0)
        entities["hour"] = hour
    if m := re.search(r"building\s*(\d+)", text):
        entities["building"] = m.group(1)

    # Validate sensor readings instead of trusting them blindly.
    sensors, warnings = {}, []
    for name, value in sensor_feed.items():
        if value is None:
            warnings.append(f"sensor '{name}' returned no data")
        elif name == "temperature_c" and not -10 <= value <= 60:
            warnings.append(f"temperature {value} C looks wrong, ignoring it")
        else:
            sensors[name] = value

    return Observation(intent, entities, sensors, warnings)


obs = perceive(
    "Hi! Please book a room in Building 22 for 6 students at 3 pm",
    {"temperature_c": 41, "occupancy_pct": None, "noise_db": 48},
)
print(json.dumps(asdict(obs), indent=2))

"""Reasoning: weighing evidence to reach a decision

Reasoning connects observations and goals to a conclusion. LLM-based agents
often reason in natural language ("chain of thought"). Here we make each
reasoning step explicit so you can read the trace and check it.

Illustrative: the rules below stand in for what an LLM would infer.
"""
observation = {
    "intent": "book_room",
    "group_size": 6,
    "hour": 15,
    "rooms": [
        {"name": "B22-101", "capacity": 4, "free_at": [9, 10, 15]},
        {"name": "B22-204", "capacity": 8, "free_at": [13, 15, 16]},
        {"name": "B22-310", "capacity": 20, "free_at": [15]},
    ],
}


def reason(obs: dict) -> tuple[str, list[str]]:
    trace = []
    size, hour = obs["group_size"], obs["hour"]
    trace.append(f"Goal: find a room for {size} people at {hour}:00.")

    fits = [r for r in obs["rooms"] if r["capacity"] >= size]
    trace.append(f"Rooms large enough: {[r['name'] for r in fits]}.")

    free = [r for r in fits if hour in r["free_at"]]
    trace.append(f"Of those, free at {hour}:00: {[r['name'] for r in free]}.")

    if not free:
        trace.append("No room satisfies both constraints, so I should ask the user.")
        return "ask_user", trace

    # Prefer the smallest room that fits, to keep big rooms for big groups.
    best = min(free, key=lambda r: r["capacity"])
    trace.append(f"Choose the smallest suitable room: {best['name']} (capacity {best['capacity']}).")
    return best["name"], trace


decision, trace = reason(observation)
for i, step in enumerate(trace, 1):
    print(f"Thought {i}: {step}")
print("\nDecision:", decision)

"""Planning: breaking a goal into ordered, executable steps

Planning turns a goal into a sequence of steps that respects dependencies
(some steps must happen before others). This example declares tasks with
their prerequisites and computes a valid order with a topological sort.

Illustrative: an LLM planner would propose the tasks and dependencies in
structured form; the ordering logic below is what makes the plan executable.
"""
from graphlib import TopologicalSorter

goal = "Organize a one-day Agentic AI workshop"

# task -> set of tasks that must finish first
tasks = {
    "pick_date": set(),
    "book_room": {"pick_date"},
    "invite_speakers": {"pick_date"},
    "open_registration": {"book_room", "invite_speakers"},
    "prepare_colab_notebooks": {"invite_speakers"},
    "send_reminders": {"open_registration"},
    "run_workshop": {"send_reminders", "prepare_colab_notebooks"},
}

sorter = TopologicalSorter(tasks)
sorter.prepare()

print(f"Goal: {goal}\n")
stage = 1
while sorter.is_active():
    ready = sorted(sorter.get_ready())          # tasks whose prerequisites are done
    print(f"Stage {stage} (can run in parallel): {', '.join(ready)}")
    for task in ready:
        sorter.done(task)
    stage += 1

# Replanning: if a step fails, only the steps that depend on it must change.
failed = "book_room"
affected = {t for t, deps in tasks.items() if failed in deps}
print(f"\nIf '{failed}' fails, re-plan these dependent steps first: {sorted(affected)}")

"""Action and tools: calling functions from structured model output

Modern LLMs can "call" tools: you describe each tool with a name, a short
description, and a JSON schema for its arguments. The model replies with a
structured request (tool name + arguments), your code runs the function,
and the result goes back to the model. This is often called function calling.

Illustrative: fake_llm() returns the kind of JSON a real model would return.
"""
import json
import math

# 1) Tools are ordinary Python functions...
def get_weather(city: str) -> dict:
    data = {"Dhahran": {"temp_c": 38, "sky": "clear"}, "Riyadh": {"temp_c": 41, "sky": "dusty"}}
    return data.get(city, {"error": f"no data for {city}"})

def calculator(expression: str) -> float:
    allowed = {"sqrt": math.sqrt, "pi": math.pi}
    return eval(expression, {"__builtins__": {}}, allowed)  # restricted; see the governance module

# 2) ...plus a description the model can read (a JSON schema per tool).
TOOLS = {
    "get_weather": {
        "function": get_weather,
        "description": "Current weather for a city",
        "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
    },
    "calculator": {
        "function": calculator,
        "description": "Evaluate a math expression",
        "parameters": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]},
    },
}

def fake_llm(user_message: str) -> str:
    """Stands in for a model that decides which tool to call."""
    if "weather" in user_message.lower():
        return json.dumps({"tool": "get_weather", "arguments": {"city": "Dhahran"}})
    return json.dumps({"tool": "calculator", "arguments": {"expression": "sqrt(2) * 10"}})

# 3) The action step: validate the request, then execute it.
def act(model_output: str):
    call = json.loads(model_output)
    name, args = call["tool"], call["arguments"]
    if name not in TOOLS:
        return {"error": f"unknown tool {name}"}
    missing = [p for p in TOOLS[name]["parameters"]["required"] if p not in args]
    if missing:
        return {"error": f"missing arguments: {missing}"}
    return TOOLS[name]["function"](**args)

for question in ["What's the weather in Dhahran?", "What is ten times the square root of two?"]:
    request = fake_llm(question)
    print("User:      ", question)
    print("Model asks:", request)
    print("Tool result:", act(request), "\n")

"""Memory: short-term context and long-term recall

Agents use two kinds of memory:
  * Short-term memory: the recent conversation that fits in the model's context.
  * Long-term memory: facts stored outside the model and retrieved when relevant
    (semantic facts, episodic past events, procedural "how-to" knowledge).

Long-term memory is usually retrieved by similarity search over embeddings.
To keep this runnable anywhere, we use simple word-overlap similarity instead.
"""
from collections import deque


class ShortTermMemory:
    """Keeps only the last N turns, like a limited context window."""
    def __init__(self, max_turns: int = 3):
        self.turns = deque(maxlen=max_turns)

    def add(self, role: str, text: str):
        self.turns.append(f"{role}: {text}")

    def as_context(self) -> str:
        return "\n".join(self.turns)


class LongTermMemory:
    """Stores facts with a type and retrieves the most similar ones."""
    def __init__(self):
        self.items = []

    def store(self, text: str, kind: str):
        self.items.append({"text": text, "kind": kind})

    @staticmethod
    def _similarity(a: str, b: str) -> float:
        wa, wb = set(a.lower().split()), set(b.lower().split())
        return len(wa & wb) / (len(wa | wb) or 1)       # Jaccard similarity

    def retrieve(self, query: str, k: int = 2):
        ranked = sorted(self.items, key=lambda m: self._similarity(query, m["text"]), reverse=True)
        return [m for m in ranked[:k] if self._similarity(query, m["text"]) > 0]


stm = ShortTermMemory(max_turns=3)
ltm = LongTermMemory()
ltm.store("the user prefers meetings in the morning", "semantic")
ltm.store("last week the user booked room B22-204 for the ai club", "episodic")
ltm.store("to book a room, check capacity first then availability", "procedural")
ltm.store("the cafeteria closes at 8 pm", "semantic")

for role, text in [("user", "hi"), ("agent", "hello! how can I help?"),
                   ("user", "book a room for the ai club"), ("agent", "sure, for when?")]:
    stm.add(role, text)

print("Short-term memory (only the last 3 turns survive):")
print(stm.as_context())

query = "book a room for the ai club in the morning"
print(f"\nLong-term memories relevant to: '{query}'")
for m in ltm.retrieve(query, k=3):
    print(f"  [{m['kind']}] {m['text']}")