{
"cells": [
{
"cell_type": "markdown",
"id": "1aee0a89",
"metadata": {},
"source": [
"# Chapter 5: Foundational Cognitive Architectures\n",
"\n",
"**Book:** *30 Agents Every AI Engineer Must Build* \n",
"**Author:** Imran Ahmad \n",
"**Publisher:** Packt Publishing, 2026 \n",
"**Chapter Pages:** 117–144\n",
"\n",
"> *\"Information is the resolution of uncertainty.\"* \n",
"> — **Claude Shannon**, Father of Information Theory\n",
"\n",
"---\n",
"\n",
"## Introduction\n",
"\n",
"This chapter examines the three foundational cognitive architectures that empower intelligent, autonomous agents to operate in complex, dynamic environments. These architectures serve as the structural backbone of modern AI systems, simulating essential aspects of human cognition — **decision-making**, **planning**, and **memory** — to enable agents to tackle real-world problems with increasing sophistication.\n",
"\n",
"Each architecture plays a distinct role in shaping intelligent behavior. Together, they enable AI systems to evolve from simple, reactive scripts into adaptive, autonomous workflows capable of handling intricate tasks over extended time horizons. Understanding these foundational patterns is crucial because they form the backbone of virtually every intelligent system in production today.\n",
"\n",
"### The Three Foundational Architectures\n",
"\n",
"| Agent Type | Focus | Chapter Section | Book Pages |\n",
"|---|---|---|---|\n",
"| **Autonomous Decision-Making Agent** | Real-time perception → cognition → action loop | Section 5.1 | pp. 118–132 |\n",
"| **Planning Agent** | Hierarchical task decomposition and dynamic execution | Section 5.2 | pp. 131–137 |\n",
"| **Memory-Augmented Agent** | Working, episodic, and semantic memory for continuity | Section 5.3 | pp. 135–144 |\n",
"\n",
"### Key Concepts at a Glance\n",
"\n",
"| Concept | Description |\n",
"|---------|-------------|\n",
"| **Cognitive Loop** | Perception → Cognition → Action → Learning — the continuous cycle enabling autonomy (Figure 5.1) |\n",
"| **Strategy Scoring** | Weighted multi-axis framework selecting between full autonomous, immediate escalation, and guided resolution |\n",
"| **Task DAGs** | Dependency-aware directed acyclic graphs for billing, outage, and generic workflows |\n",
"| **Hierarchical Decomposition** | Breaking high-level goals into phased, actionable subtasks (Figure 5.2) |\n",
"| **Memory Triad** | Working memory (session), episodic memory (history), semantic memory (facts) — see Figure 5.3 |\n",
"| **Integrated Architecture** | All three agents combined in a single scenario (Figure 5.4, Table 5.2) |\n"
]
},
{
"cell_type": "markdown",
"id": "ada619cf",
"metadata": {},
"source": [
"## Table of Contents\n",
"\n",
"1. [Environment Setup](#1.-Environment-Setup)\n",
"2. [Autonomous Decision-Making Agent](#2.-Autonomous-Decision-Making-Agent) (Section 5.1, pp. 118-132)\n",
" - Perception Module\n",
" - Cognition Module (Strategy Scoring)\n",
" - Action Module (Task DAG Execution)\n",
" - Safety Checks & Escalation\n",
" - Full Agent Class & Case Study\n",
"3. [Planning Agent](#3.-Planning-Agent) (Section 5.2, pp. 131-137)\n",
" - Hierarchical Decomposition\n",
" - Dynamic Execution & Monitoring\n",
" - Marketing Campaign Demo\n",
"4. [Memory-Augmented Agent](#4.-Memory-Augmented-Agent) (Section 5.3, pp. 135-144)\n",
" - Three Memory Types\n",
" - Healthcare Assistant Case Study\n",
"5. [Comparative Analysis](#5.-Comparative-Analysis) (Section 5.4, pp. 141-142)\n",
"6. [Engineering Best Practices & Wrap-Up](#6.-Engineering-Best-Practices-&-Wrap-Up) (Section 5.5, pp. 142-144)"
]
},
{
"cell_type": "markdown",
"id": "b2c84c48",
"metadata": {},
"source": [
"---\n",
"## 1. Environment Setup\n",
"\n",
"Import standard library modules and the three supporting files that\n",
"power this notebook: `color_logger.py` (observability), `resilience.py`\n",
"(fault tolerance), and `mock_llm.py` (simulation mode).\n",
"\n",
"Ref: Technical Requirements (p. 118)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "7864c3dc",
"metadata": {},
"outputs": [],
"source": [
"# Standard library imports\n",
"# Ref: Technical Requirements, p. 118\n",
"import os\n",
"import sys\n",
"import time\n",
"from datetime import datetime\n",
"from dataclasses import dataclass, field\n",
"from functools import wraps\n",
"\n",
"# Ensure local modules are importable regardless of working directory\n",
"sys.path.insert(0, \".\")\n",
"\n",
"# Supporting modules (must exist alongside this notebook)\n",
"from color_logger import log_info, log_success, log_error, log_warn, log_section\n",
"from resilience import fail_gracefully\n",
"from mock_llm import MockLLM, MockVectorDB, MockResponse\n",
"\n",
"log_success(\"All imports successful\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "aacaef87",
"metadata": {},
"outputs": [],
"source": [
"# Multi-provider LLM support (OpenAI / Anthropic / Google Gemini)\n",
"# Set LLM_PROVIDER in .env to choose: openai | anthropic | google | auto\n",
"# Auto-detection uses the first available key.\n",
"# See supporting/llm_provider.py for details.\n",
"\n",
"import sys, os\n",
"sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath('.')), ''))\n",
"sys.path.insert(0, '..')\n",
"\n",
"try:\n",
" from supporting.llm_provider import detect_provider, get_llm, PROVIDER_MODELS, print_provider_banner\n",
" _PROVIDER, _PROVIDER_KEY, _PROVIDER_MODE = detect_provider()\n",
" print_provider_banner(_PROVIDER, _PROVIDER_MODE)\n",
"except ImportError:\n",
" print('[INFO] supporting/llm_provider.py not found — using default OpenAI path')\n",
" _PROVIDER, _PROVIDER_KEY, _PROVIDER_MODE = 'openai', os.getenv('OPENAI_API_KEY'), 'LIVE' if os.getenv('OPENAI_API_KEY') else 'SIMULATION'\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d62fb1d2",
"metadata": {},
"outputs": [],
"source": [
"# ── Environment Detection & Simulation Mode Toggle ─────────────\n",
"# Ref: Technical Requirements, p. 118\n",
"#\n",
"# If an API key is found in .env or via getpass, live mode activates.\n",
"# Otherwise, Simulation Mode engages the MockLLM and MockVectorDB.\n",
"\n",
"log_section(\"Environment Detection\", chapter_ref=\"Technical Requirements, p. 118\")\n",
"\n",
"# Attempt to load .env (no-op if file missing)\n",
"try:\n",
" from dotenv import load_dotenv\n",
" load_dotenv()\n",
" log_info(\"Loaded .env file\")\n",
"except ImportError:\n",
" log_warn(\"python-dotenv not installed — skipping .env loading\")\n",
"\n",
"# Check for API key\n",
"api_key = os.getenv(\"OPENAI_API_KEY\", \"\").strip()\n",
"\n",
"if not api_key:\n",
" try:\n",
" import getpass\n",
" # In non-interactive environments, this will remain empty\n",
" api_key = getpass.getpass(\n",
" \"Enter OpenAI API key (or press Enter for Simulation Mode): \"\n",
" ).strip()\n",
" except (EOFError, OSError, Exception):\n",
" api_key = \"\"\n",
"\n",
"# ── Initialize clients based on mode ───────────────────────────\n",
"SIMULATION_MODE = not bool(api_key)\n",
"\n",
"if SIMULATION_MODE:\n",
" llm_client = MockLLM()\n",
" vector_db = MockVectorDB()\n",
" log_warn(\"SIMULATION MODE active — using MockLLM and MockVectorDB\")\n",
" log_info(\"No API key required. All agent demos will run with mock data.\")\n",
"else:\n",
" # Live mode: detect provider and show banner\n",
" try:\n",
" sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath('.')), '..'))\n",
" from supporting.llm_provider import detect_provider, get_client, chat_completion, PROVIDER_MODELS, print_provider_banner\n",
" _provider = os.environ.get(\"LLM_PROVIDER\", \"auto\").strip().lower()\n",
" if _provider == \"auto\":\n",
" _provider, _, _ = detect_provider()\n",
" _model = PROVIDER_MODELS.get(_provider, {}).get(\"default\", \"gpt-4o\")\n",
" _real_client = get_client(provider=_provider, api_key=api_key)\n",
" print_provider_banner(_provider, \"LIVE\", _model)\n",
"\n",
" class LiveLLM:\n",
" \"\"\"Wraps real LLM to match MockLLM.chat() interface.\"\"\"\n",
" def __init__(self, client, provider, model):\n",
" self._client, self._provider, self._model = client, provider, model\n",
" def chat(self, messages, **kwargs):\n",
" text = chat_completion(self._client, self._provider, messages, model=self._model)\n",
" return MockResponse(content=text, model=self._model)\n",
" def __repr__(self):\n",
" return f\"LiveLLM({self._provider}, {self._model})\"\n",
"\n",
" llm_client = LiveLLM(_real_client, _provider, _model)\n",
" log_success(f\"Live LLM: {llm_client}\")\n",
" except Exception as e:\n",
" llm_client = MockLLM()\n",
" log_warn(f\"Live LLM init failed ({e}). Using MockLLM.\")\n",
" vector_db = MockVectorDB()"
]
},
{
"cell_type": "markdown",
"id": "46e94beb",
"metadata": {},
"source": [
"### How Simulation Mode Works\n",
"\n",
"The mock layer (`mock_llm.py`) provides two drop-in replacements:\n",
"\n",
"- **`MockLLM`** — Routes prompts through a keyword map to one of 6\n",
" pre-built `MockResponse` objects covering service outages, billing\n",
" issues, complex escalations, marketing campaigns, healthcare queries,\n",
" and a default fallback.\n",
"\n",
"- **`MockVectorDB`** — An in-memory store with Jaccard word-overlap\n",
" scoring, pre-seeded with 4 episodic records (3 healthcare +\n",
" 1 customer service). Supports `add()`, `search()`, and `reset()`.\n",
"\n",
"This design means **every cell in this notebook runs without any API key**.\n",
"To switch to live mode, simply add your key to `.env` and restart the kernel.\n",
"\n",
"Ref: Technical Requirements (p. 118)"
]
},
{
"cell_type": "markdown",
"id": "c9cb0d58",
"metadata": {},
"source": [
"---\n",
"## 2. Autonomous Decision-Making Agent\n",
"\n",
"**Chapter Ref:** Section 5.1 (pp. 118–132)\n",
"\n",
"The most fundamental architecture in the cognitive hierarchy. These agents perceive their environment, evaluate possible actions against goals and constraints, choose what to do, and execute — **without requiring continuous human supervision**. They learn from outcomes by updating state or memories that inform the next loop.\n",
"\n",
"The architecture implements four integrated phases: **Perception** (structuring raw inputs), **Cognition** (strategic reasoning and strategy selection), **Action** (task DAG execution), and **Learning** (feedback-driven adaptation).\n",
"\n",
"### Figure 5.1 — The Autonomous Agent Decision Loop\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "6dcd0067",
"metadata": {},
"outputs": [],
"source": [
"# ── Figure 5.1 — The Autonomous Agent Decision Loop (SVG) ──\n",
"# Ref: Section 5.1, p. 125\n",
"\n",
"from IPython.display import SVG, display\n",
"\n",
"fig_5_1 = \"\"\"\n",
"\n",
"\"\"\"\n",
"display(SVG(fig_5_1))\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a10b0100",
"metadata": {},
"outputs": [],
"source": [
"# ── Perception Module ──────────────────────────────────────────\n",
"# Ref: Section 5.1, Perception: From raw input to structured\n",
"# intelligence (pp. 119-121)\n",
"#\n",
"# The enhanced perception system goes beyond simple input processing;\n",
"# it provides the agent with situational awareness that enables\n",
"# autonomous decision-making. When a user reports an issue, the\n",
"# agent considers system load, time of day, user priority, and\n",
"# current alerts to make informed decisions.\n",
"\n",
"log_section(\"Perception Module\", chapter_ref=\"Section 5.1, pp. 119-121\")\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value={\"message\": \"\", \"error\": \"perception_failed\"},\n",
" chapter_ref=\"Section 5.1, pp. 119-121\",\n",
")\n",
"def analyze_sentiment(text: str) -> str:\n",
" \"\"\"Return sentiment label for the given text.\n",
"\n",
" Args:\n",
" text: User message to analyze.\n",
"\n",
" Returns:\n",
" Sentiment label: 'positive', 'neutral', or 'negative'.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Perception (p. 119)\n",
" \"\"\"\n",
" # Ref: p. 119 — classify_text_sentiment stub\n",
" negative_words = {\"down\", \"slow\", \"broken\", \"issue\", \"problem\",\n",
" \"can't\", \"error\", \"intermittent\", \"frustrated\"}\n",
" positive_words = {\"thanks\", \"great\", \"helped\", \"resolved\", \"good\"}\n",
" words = set(text.lower().split())\n",
" if words & negative_words:\n",
" return \"negative\"\n",
" if words & positive_words:\n",
" return \"positive\"\n",
" return \"neutral\"\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.1, p. 120\",\n",
")\n",
"def check_system_alerts() -> list:\n",
" \"\"\"Return list of active system alert IDs.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Perception — check_system_alerts() (p. 120)\n",
" \"\"\"\n",
" # Simulated system alerts\n",
" alerts = [\"NET-2026-0042: Regional connectivity degradation\"]\n",
" log_info(f\"System alerts retrieved: {len(alerts)} active\")\n",
" return alerts\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=\"standard\",\n",
" chapter_ref=\"Section 5.1, p. 120\",\n",
")\n",
"def get_user_tier(user_id: str) -> str:\n",
" \"\"\"Return account tier for the given user.\n",
"\n",
" Args:\n",
" user_id: Customer identifier.\n",
"\n",
" Returns:\n",
" Tier string: 'standard', 'premium', or 'enterprise'.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Perception — get_user_tier() (p. 120)\n",
" \"\"\"\n",
" # Simulated CRM lookup\n",
" tiers = {\n",
" \"premium_biz_123\": \"premium\",\n",
" \"enterprise_500\": \"enterprise\",\n",
" \"patient_42\": \"standard\",\n",
" }\n",
" tier = tiers.get(user_id, \"standard\")\n",
" log_info(f\"User tier for '{user_id}': {tier}\")\n",
" return tier\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=True,\n",
" chapter_ref=\"Section 5.1, p. 120\",\n",
")\n",
"def check_human_agent_status() -> bool:\n",
" \"\"\"Check whether human agents are available for escalation.\n",
"\n",
" Returns:\n",
" True if at least one human agent is online.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Perception — agent_availability (p. 120)\n",
" \"\"\"\n",
" log_info(\"Human agent availability: True (simulated)\")\n",
" return True\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value={\"message\": \"\", \"error\": \"perception_failed\"},\n",
" chapter_ref=\"Section 5.1, pp. 119-121\",\n",
")\n",
"def enhanced_perception(\n",
" user_message: str,\n",
" context: dict,\n",
" system_state: dict,\n",
") -> dict:\n",
" \"\"\"Build a rich perception dict from user message + environment.\n",
"\n",
" Combines the base perception (message, timestamp, sentiment)\n",
" with system-wide context (load, alerts, user tier, agent\n",
" availability) to enable autonomous decision-making.\n",
"\n",
" Args:\n",
" user_message: Raw user input text.\n",
" context: Session context dict with 'user_id', 'session' keys.\n",
" system_state: Dict with 'active_sessions' and other runtime data.\n",
"\n",
" Returns:\n",
" Structured perception dict with 10+ fields.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, enhanced_perception() (pp. 120-121)\n",
" \"\"\"\n",
" user_id = context.get(\"user_id\", \"unknown\")\n",
"\n",
" # Base perception (Ref: pp. 135-136)\n",
" perception = {\n",
" \"message\": user_message,\n",
" \"timestamp\": datetime.now(),\n",
" \"user_id\": user_id,\n",
" \"session_state\": context.get(\"session\", \"new\"),\n",
" \"sentiment\": analyze_sentiment(user_message),\n",
" }\n",
"\n",
" # Enhanced with environmental awareness (Ref: p. 120)\n",
" perception.update({\n",
" \"current_load\": system_state.get(\"active_sessions\", 0),\n",
" \"time_of_day\": datetime.now().hour,\n",
" \"user_tier\": get_user_tier(user_id),\n",
" \"recent_issues\": check_system_alerts(),\n",
" \"agent_availability\": check_human_agent_status(),\n",
" })\n",
"\n",
" log_success(\n",
" f\"Perception complete: intent signal='{perception['sentiment']}', \"\n",
" f\"tier='{perception['user_tier']}', \"\n",
" f\"alerts={len(perception['recent_issues'])}\"\n",
" )\n",
" return perception\n",
"\n",
"\n",
"# ── Quick test ─────────────────────────────────────────────────\n",
"test_perception = enhanced_perception(\n",
" user_message=\"My business internet has been intermittent for two days\",\n",
" context={\"user_id\": \"premium_biz_123\", \"session\": \"new\"},\n",
" system_state={\"active_sessions\": 42},\n",
")\n",
"log_info(f\"Perception keys: {list(test_perception.keys())}\")"
]
},
{
"cell_type": "markdown",
"id": "0723177b",
"metadata": {},
"source": [
"> **📝 Note — Strategy Scoring Framework (pp. 121–122)** \n",
"> The agent scores each candidate strategy across four weighted axes: **autonomy_level_score** (0–1), **urgency_score** (0–1), **complexity_score** (0–1), and **escalation_threshold** (0–1). The strategy with the highest composite score wins; ties favour guided resolution. The three strategies are: *full autonomous resolution*, *immediate escalation*, and *guided autonomous resolution* (baseline fallback at 0.5).\n",
"\n",
"> **📝 Note — Task DAGs (pp. 123–124)** \n",
"> Action plans are built as dependency-aware task DAGs (directed acyclic graphs). Each node has an `id`, `action`, and `depends_on` list. Tasks with empty `depends_on` run immediately; downstream tasks wait until all listed IDs complete. The chapter provides three DAG templates: billing issues (8 tasks), service outages (6 tasks), and a generic fallback (4 tasks).\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4a8f9386",
"metadata": {},
"outputs": [],
"source": [
"# ── Cognition Module: Strategic Reasoning ──────────────────────\n",
"# Ref: Section 5.1, Cognition: Strategic reasoning for autonomous\n",
"# operation (pp. 121-122)\n",
"#\n",
"# The reasoning component evaluates three candidate strategies\n",
"# across weighted axes (autonomy, urgency, complexity, escalation).\n",
"# The strategy with the highest composite score wins; ties favour\n",
"# guided resolution.\n",
"\n",
"log_section(\"Cognition Module\", chapter_ref=\"Section 5.1, pp. 121-122\")\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=\"general_inquiry\",\n",
" chapter_ref=\"Section 5.1, p. 121\",\n",
")\n",
"def classify_intent(message: str) -> str:\n",
" \"\"\"Classify user message intent using LLM or keyword heuristic.\n",
"\n",
" Args:\n",
" message: User message text.\n",
"\n",
" Returns:\n",
" Intent label string.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Cognition — classify_intent() (p. 121)\n",
" \"\"\"\n",
" response = llm_client.generate(message)\n",
" log_info(f\"Intent classified: '{response.intent}' \"\n",
" f\"(confidence: {response.confidence})\")\n",
" return response.intent\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=3,\n",
" chapter_ref=\"Section 5.1, p. 121\",\n",
")\n",
"def determine_priority(\n",
" intent: str,\n",
" sentiment: str,\n",
" user_history: dict | None = None,\n",
") -> int:\n",
" \"\"\"Determine issue priority on a 1-5 scale.\n",
"\n",
" Args:\n",
" intent: Classified intent label.\n",
" sentiment: Sentiment from perception ('positive'/'neutral'/'negative').\n",
" user_history: Optional user history dict.\n",
"\n",
" Returns:\n",
" Priority integer (1=lowest, 5=highest).\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Cognition — determine_priority() (p. 121)\n",
" \"\"\"\n",
" base = {\"service_outage\": 4, \"billing_issue\": 3, \"complex_technical\": 5}\n",
" priority = base.get(intent, 2)\n",
" if sentiment == \"negative\":\n",
" priority = min(priority + 1, 5)\n",
" log_info(f\"Priority determined: {priority}/5 \"\n",
" f\"(intent='{intent}', sentiment='{sentiment}')\")\n",
" return priority\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=\"guided_autonomous_resolution\",\n",
" chapter_ref=\"Section 5.1, p. 121\",\n",
")\n",
"def determine_autonomy_level(\n",
" user_tier: str,\n",
" current_load: int,\n",
" agent_availability: bool,\n",
") -> float:\n",
" \"\"\"Compute autonomy level score [0.0-1.0].\n",
"\n",
" Higher tier + higher load + lower agent availability\n",
" = higher autonomy level.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Cognition — autonomy_level (p. 121)\n",
" \"\"\"\n",
" tier_scores = {\"enterprise\": 0.9, \"premium\": 0.8, \"standard\": 0.5}\n",
" tier_score = tier_scores.get(user_tier, 0.5)\n",
" load_factor = min(current_load / 100, 1.0)\n",
" availability_factor = 0.3 if not agent_availability else 0.0\n",
" return round(min(tier_score + load_factor * 0.2 + availability_factor, 1.0), 2)\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=0.5,\n",
" chapter_ref=\"Section 5.1, p. 121\",\n",
")\n",
"def calculate_escalation_threshold(\n",
" time_of_day: int,\n",
" intent: str,\n",
" priority: int,\n",
") -> float:\n",
" \"\"\"Calculate the escalation threshold [0.0-1.0].\n",
"\n",
" Lower threshold = easier to escalate.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Cognition — escalation_threshold (p. 121)\n",
" \"\"\"\n",
" base = 0.5\n",
" if priority >= 4:\n",
" base -= 0.15\n",
" if time_of_day < 8 or time_of_day > 20: # After hours\n",
" base -= 0.1\n",
" return round(max(base, 0.1), 2)\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value={\n",
" \"intent\": \"general_inquiry\",\n",
" \"strategy\": \"guided_autonomous_resolution\",\n",
" },\n",
" chapter_ref=\"Section 5.1, pp. 121-122\",\n",
")\n",
"def autonomous_reasoning(perception_data: dict) -> dict:\n",
" \"\"\"Autonomous cognition: classify, prioritize, select strategy.\n",
"\n",
" Implements the 3-strategy scoring system from the book:\n",
" - full_autonomous_resolution\n",
" - immediate_escalation\n",
" - guided_autonomous_resolution (baseline fallback at 0.5)\n",
"\n",
" The strategy with the highest composite score wins.\n",
"\n",
" Args:\n",
" perception_data: Structured dict from enhanced_perception().\n",
"\n",
" Returns:\n",
" Decision context dict with intent, priority, strategy,\n",
" scores, and all supporting data.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, autonomous_reasoning() with strategy scoring\n",
" (pp. 121-122)\n",
" \"\"\"\n",
" # Basic intent and priority (Ref: p. 121)\n",
" intent = classify_intent(perception_data[\"message\"])\n",
" priority = determine_priority(\n",
" intent,\n",
" perception_data.get(\"sentiment\", \"neutral\"),\n",
" )\n",
"\n",
" # Autonomy and escalation parameters (Ref: p. 121)\n",
" autonomy_level = determine_autonomy_level(\n",
" perception_data.get(\"user_tier\", \"standard\"),\n",
" perception_data.get(\"current_load\", 0),\n",
" perception_data.get(\"agent_availability\", True),\n",
" )\n",
" escalation_threshold = calculate_escalation_threshold(\n",
" perception_data.get(\"time_of_day\", 12),\n",
" intent,\n",
" priority,\n",
" )\n",
"\n",
" # Build decision context\n",
" decision_context = {\n",
" \"intent\": intent,\n",
" \"priority\": priority,\n",
" \"context\": perception_data,\n",
" \"autonomy_level_score\": autonomy_level,\n",
" \"escalation_threshold\": escalation_threshold,\n",
" \"urgency_score\": priority / 5.0,\n",
" \"complexity_score\": 0.85 if intent == \"complex_technical\" else 0.3,\n",
" \"agent_availability\": perception_data.get(\"agent_availability\", True),\n",
" }\n",
"\n",
" # ── Strategy scoring (Ref: pp. 121-122) ───────────────────────\n",
" # Three weighted axes per strategy. Highest composite wins.\n",
" strategy_scores = {\n",
" \"full_autonomous_resolution\": (\n",
" decision_context[\"autonomy_level_score\"] * 0.5\n",
" + (1 - decision_context[\"escalation_threshold\"]) * 0.3\n",
" + (1 - decision_context[\"complexity_score\"]) * 0.2\n",
" ),\n",
" \"immediate_escalation\": (\n",
" decision_context[\"urgency_score\"] * 0.4\n",
" + (0.4 if not decision_context[\"agent_availability\"] else 0.0)\n",
" + decision_context[\"escalation_threshold\"] * 0.2\n",
" ),\n",
" \"guided_autonomous_resolution\": 0.5, # baseline fallback\n",
" }\n",
"\n",
" # Round scores for readability\n",
" strategy_scores = {k: round(v, 3) for k, v in strategy_scores.items()}\n",
"\n",
" # Strategy with highest score wins; ties favour guided resolution\n",
" decision_context[\"strategy\"] = max(\n",
" strategy_scores, key=strategy_scores.get\n",
" )\n",
" decision_context[\"strategy_scores\"] = strategy_scores\n",
"\n",
" log_success(\n",
" f\"Strategy selected: '{decision_context['strategy']}' \"\n",
" f\"| Scores: {strategy_scores}\"\n",
" )\n",
" return decision_context\n",
"\n",
"\n",
"# ── Quick test ─────────────────────────────────────────────────\n",
"test_decision = autonomous_reasoning(test_perception)\n",
"log_info(f\"Decision intent: {test_decision['intent']}, \"\n",
" f\"strategy: {test_decision['strategy']}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4e57582d",
"metadata": {},
"outputs": [],
"source": [
"# ── Action Module: Task DAG & Execution ────────────────────────\n",
"# Ref: Section 5.1, Action: Autonomous execution and adaptation\n",
"# (pp. 122-124)\n",
"#\n",
"# The action module translates reasoning into real-world outcomes.\n",
"# Tasks are organized as a dependency-aware DAG. Each node has an\n",
"# id, action, and depends_on list. Tasks with empty depends_on\n",
"# run immediately; downstream tasks wait until dependencies complete.\n",
"\n",
"log_section(\"Action Module\", chapter_ref=\"Section 5.1, pp. 122-124\")\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.1, pp. 123-124\",\n",
")\n",
"def create_autonomous_plan(decision_context: dict) -> list[dict]:\n",
" \"\"\"Create a dependency-aware task DAG based on intent.\n",
"\n",
" Builds a list of task dicts, each with 'id', 'action', and\n",
" 'depends_on' keys. Supports billing_issue, service_outage,\n",
" and default resolution workflows.\n",
"\n",
" Args:\n",
" decision_context: Decision dict from autonomous_reasoning().\n",
"\n",
" Returns:\n",
" List of task dicts forming a DAG.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, create_autonomous_plan() (pp. 123-124)\n",
" \"\"\"\n",
" intent = decision_context.get(\"intent\", \"general_inquiry\")\n",
"\n",
" # Ref: p. 123 — billing_issue DAG\n",
" if intent == \"billing_issue\":\n",
" plan = [\n",
" {\"id\": \"T1\", \"action\": \"fetch_account_details\", \"depends_on\": []},\n",
" {\"id\": \"T2\", \"action\": \"analyze_billing_history\", \"depends_on\": [\"T1\"]},\n",
" {\"id\": \"T3\", \"action\": \"identify_discrepancies\", \"depends_on\": [\"T2\"]},\n",
" {\"id\": \"T4\", \"action\": \"calculate_adjustments\", \"depends_on\": [\"T3\"]},\n",
" {\"id\": \"T5\", \"action\": \"generate_explanation\", \"depends_on\": [\"T3\"]},\n",
" {\"id\": \"T6\", \"action\": \"apply_account_credits\", \"depends_on\": [\"T4\"]},\n",
" {\"id\": \"T7\", \"action\": \"send_confirmation_email\", \"depends_on\": [\"T5\", \"T6\"]},\n",
" {\"id\": \"T8\", \"action\": \"schedule_follow_up\", \"depends_on\": [\"T7\"]},\n",
" ]\n",
"\n",
" # Ref: p. 124 — service_outage DAG\n",
" elif intent == \"service_outage\":\n",
" plan = [\n",
" {\"id\": \"T1\", \"action\": \"check_service_status\", \"depends_on\": []},\n",
" {\"id\": \"T2\", \"action\": \"identify_affected_areas\", \"depends_on\": [\"T1\"]},\n",
" {\"id\": \"T3\", \"action\": \"estimate_resolution_time\", \"depends_on\": [\"T1\"]},\n",
" {\"id\": \"T4\", \"action\": \"send_proactive_notifications\", \"depends_on\": [\"T2\", \"T3\"]},\n",
" {\"id\": \"T5\", \"action\": \"monitor_restoration_progress\", \"depends_on\": [\"T1\"]},\n",
" {\"id\": \"T6\", \"action\": \"update_customer_status\", \"depends_on\": [\"T4\", \"T5\"]},\n",
" ]\n",
"\n",
" # Ref: p. 124 — default resolution workflow\n",
" else:\n",
" plan = [\n",
" {\"id\": \"T1\", \"action\": \"analyze_issue\", \"depends_on\": []},\n",
" {\"id\": \"T2\", \"action\": \"research_solutions\", \"depends_on\": [\"T1\"]},\n",
" {\"id\": \"T3\", \"action\": \"implement_resolution\", \"depends_on\": [\"T2\"]},\n",
" {\"id\": \"T4\", \"action\": \"verify_success\", \"depends_on\": [\"T3\"]},\n",
" ]\n",
"\n",
" log_info(f\"Plan created for '{intent}': {len(plan)} tasks\")\n",
" for task in plan:\n",
" deps = task[\"depends_on\"] if task[\"depends_on\"] else [\"(none)\"]\n",
" log_info(f\" {task['id']}: {task['action']} → depends_on: {deps}\")\n",
" return plan\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.1, pp. 122-123\",\n",
")\n",
"def execute_task(task: dict, completed: dict) -> dict:\n",
" \"\"\"Execute a single task from the DAG.\n",
"\n",
" Simulates task execution and returns a result dict.\n",
"\n",
" Args:\n",
" task: Task dict with 'id', 'action', 'depends_on'.\n",
" completed: Dict of already-completed task results.\n",
"\n",
" Returns:\n",
" Result dict with 'task_id', 'action', 'status', 'timestamp'.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Action execution (pp. 122-123)\n",
" \"\"\"\n",
" result = {\n",
" \"task_id\": task[\"id\"],\n",
" \"action\": task[\"action\"],\n",
" \"status\": \"completed\",\n",
" \"timestamp\": datetime.now().isoformat(),\n",
" }\n",
" log_success(f\"Task {task['id']} ({task['action']}): completed\")\n",
" return result\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.1, pp. 122-124\",\n",
")\n",
"def autonomous_action_execution(decision_context: dict) -> list[dict]:\n",
" \"\"\"Execute the autonomous action plan based on selected strategy.\n",
"\n",
" Dispatches to the appropriate execution path:\n",
" - full_autonomous_resolution: build + execute full task DAG\n",
" - immediate_escalation: prepare handoff package\n",
" - guided_autonomous_resolution: execute with checkpoints\n",
"\n",
" Args:\n",
" decision_context: Decision dict from autonomous_reasoning().\n",
"\n",
" Returns:\n",
" List of task result dicts.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, autonomous_action_execution() (pp. 122-123)\n",
" \"\"\"\n",
" strategy = decision_context.get(\"strategy\", \"guided_autonomous_resolution\")\n",
" results = []\n",
"\n",
" if strategy == \"full_autonomous_resolution\":\n",
" # Execute the full resolution workflow autonomously\n",
" action_plan = create_autonomous_plan(decision_context)\n",
" completed = {}\n",
"\n",
" # Topological execution: respect depends_on ordering\n",
" remaining = list(action_plan)\n",
" while remaining:\n",
" ready = [\n",
" t for t in remaining\n",
" if all(d in completed for d in t[\"depends_on\"])\n",
" ]\n",
" if not ready:\n",
" log_error(\"Circular dependency detected in task DAG!\")\n",
" break\n",
" for task in ready:\n",
" result = execute_task(task, completed)\n",
" completed[task[\"id\"]] = result\n",
" results.append(result)\n",
" remaining.remove(task)\n",
"\n",
" elif strategy == \"immediate_escalation\":\n",
" # Prepare comprehensive handoff to human agent\n",
" log_warn(\"Escalation path: preparing handoff package\")\n",
" results.append({\n",
" \"task_id\": \"ESC-1\",\n",
" \"action\": \"prepare_escalation_package\",\n",
" \"status\": \"completed\",\n",
" \"timestamp\": datetime.now().isoformat(),\n",
" \"escalation_data\": {\n",
" \"intent\": decision_context.get(\"intent\"),\n",
" \"priority\": decision_context.get(\"priority\"),\n",
" \"complexity\": decision_context.get(\"complexity_score\"),\n",
" },\n",
" })\n",
"\n",
" else: # guided_autonomous_resolution\n",
" # Execute with checkpoints for human oversight\n",
" action_plan = create_autonomous_plan(decision_context)\n",
" completed = {}\n",
" remaining = list(action_plan)\n",
" while remaining:\n",
" ready = [\n",
" t for t in remaining\n",
" if all(d in completed for d in t[\"depends_on\"])\n",
" ]\n",
" if not ready:\n",
" break\n",
" for task in ready:\n",
" result = execute_task(task, completed)\n",
" completed[task[\"id\"]] = result\n",
" results.append(result)\n",
" remaining.remove(task)\n",
" log_info(\"Guided resolution: all checkpoints passed\")\n",
"\n",
" log_success(\n",
" f\"Action execution complete: {len(results)} tasks, \"\n",
" f\"strategy='{strategy}'\"\n",
" )\n",
" return results\n",
"\n",
"\n",
"# ── Quick test ─────────────────────────────────────────────────\n",
"test_results = autonomous_action_execution(test_decision)\n",
"log_info(f\"Execution produced {len(test_results)} task results\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e70fc9d8",
"metadata": {},
"outputs": [],
"source": [
"# ── Safety Checks & Escalation Logic ──────────────────────────\n",
"# Ref: Section 5.1, Guardrails and safety mechanisms (p. 128)\n",
"# and Escalation criteria (p. 131)\n",
"#\n",
"# Safety checks evaluate financial impact limits, data access\n",
"# permissions, and customer impact risk. Escalation uses a\n",
"# 5-factor weighted scoring system.\n",
"\n",
"log_section(\"Safety & Escalation\", chapter_ref=\"Section 5.1, pp. 128-131\")\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=(True, []),\n",
" chapter_ref=\"Section 5.1, p. 128\",\n",
")\n",
"def autonomous_safety_check(\n",
" decision_context: dict,\n",
" planned_actions: list[dict],\n",
") -> tuple[bool, list[str]]:\n",
" \"\"\"Evaluate planned actions against safety guardrails.\n",
"\n",
" Checks three categories:\n",
" 1. Financial impact limits (credit thresholds by tier)\n",
" 2. Data access permissions\n",
" 3. Customer impact risk assessment\n",
"\n",
" Args:\n",
" decision_context: Decision dict from autonomous_reasoning().\n",
" planned_actions: List of task dicts from create_autonomous_plan().\n",
"\n",
" Returns:\n",
" Tuple of (is_safe: bool, violations: list[str]).\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, autonomous_safety_check() (p. 128)\n",
" \"\"\"\n",
" safety_violations = []\n",
" action_names = [a.get(\"action\", \"\") for a in planned_actions]\n",
"\n",
" # 1. Financial impact limits (Ref: p. 128)\n",
" if \"apply_account_credits\" in action_names:\n",
" credit_limit = {\"premium\": 500, \"enterprise\": 2000, \"standard\": 100}\n",
" tier = decision_context.get(\"context\", {}).get(\"user_tier\", \"standard\")\n",
" max_credit = credit_limit.get(tier, 100)\n",
" simulated_credit = 75 # Simulated credit amount\n",
" if simulated_credit > max_credit:\n",
" safety_violations.append(\"credit_limit_exceeded\")\n",
" log_error(f\"Credit ${simulated_credit} exceeds {tier} limit ${max_credit}\")\n",
" else:\n",
" log_success(f\"Credit ${simulated_credit} within {tier} limit ${max_credit}\")\n",
"\n",
" # 2. Data access permissions (Ref: p. 128)\n",
" sensitive_actions = {\"access_medical_records\", \"view_financial_data\"}\n",
" if sensitive_actions & set(action_names):\n",
" safety_violations.append(\"insufficient_data_permissions\")\n",
" log_error(\"Sensitive data access attempted without explicit permission\")\n",
" else:\n",
" log_success(\"Data access permissions: OK\")\n",
"\n",
" # 3. Impact assessment (Ref: p. 128)\n",
" impact_risk = decision_context.get(\"complexity_score\", 0.3)\n",
" threshold = decision_context.get(\"escalation_threshold\", 0.5)\n",
" if impact_risk > threshold:\n",
" safety_violations.append(\"high_impact_risk\")\n",
" log_warn(f\"Impact risk ({impact_risk}) exceeds threshold ({threshold})\")\n",
" else:\n",
" log_success(f\"Impact risk ({impact_risk}) within threshold ({threshold})\")\n",
"\n",
" is_safe = len(safety_violations) == 0\n",
" status = \"PASSED\" if is_safe else f\"FAILED ({len(safety_violations)} violations)\"\n",
" log_info(f\"Safety check: {status}\")\n",
" return is_safe, safety_violations\n",
"\n",
"\n",
"@fail_gracefully(\n",
" fallback_value=None,\n",
" chapter_ref=\"Section 5.1, p. 131\",\n",
")\n",
"def intelligent_escalation_decision(\n",
" decision_context: dict,\n",
" safety_results: tuple[bool, list[str]],\n",
") -> dict | None:\n",
" \"\"\"Determine whether to escalate based on 5 weighted factors.\n",
"\n",
" Factors: safety violations, confidence threshold, complexity\n",
" threshold, user preference, and business impact.\n",
"\n",
" Args:\n",
" decision_context: Decision dict from autonomous_reasoning().\n",
" safety_results: Tuple from autonomous_safety_check().\n",
"\n",
" Returns:\n",
" Escalation package dict if escalation needed, else None.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, intelligent_escalation_decision() (p. 131)\n",
" \"\"\"\n",
" escalation_factors = {\n",
" \"safety_violations\": not safety_results[0],\n",
" \"confidence_threshold\": decision_context.get(\n",
" \"urgency_score\", 0\n",
" ) < 0.6,\n",
" \"complexity_threshold\": decision_context.get(\n",
" \"complexity_score\", 0\n",
" ) > 0.7,\n",
" \"user_preference\": False, # Simulated: user has no preference\n",
" \"business_impact\": decision_context.get(\"urgency_score\", 0) > 0.8,\n",
" }\n",
"\n",
" # Weighted escalation score\n",
" weights = {\n",
" \"safety_violations\": 0.30,\n",
" \"confidence_threshold\": 0.20,\n",
" \"complexity_threshold\": 0.25,\n",
" \"user_preference\": 0.10,\n",
" \"business_impact\": 0.15,\n",
" }\n",
" escalation_score = sum(\n",
" weights[k] * (1.0 if v else 0.0)\n",
" for k, v in escalation_factors.items()\n",
" )\n",
" escalation_score = round(escalation_score, 3)\n",
"\n",
" log_info(f\"Escalation factors: {escalation_factors}\")\n",
" log_info(f\"Escalation score: {escalation_score} \"\n",
" f\"(threshold: {decision_context.get('escalation_threshold', 0.5)})\")\n",
"\n",
" if escalation_score > decision_context.get(\"escalation_threshold\", 0.5):\n",
" log_warn(\"ESCALATION TRIGGERED — preparing handoff package\")\n",
" return {\n",
" \"escalation_score\": escalation_score,\n",
" \"factors\": escalation_factors,\n",
" \"decision_context_summary\": {\n",
" \"intent\": decision_context.get(\"intent\"),\n",
" \"priority\": decision_context.get(\"priority\"),\n",
" \"strategy\": decision_context.get(\"strategy\"),\n",
" },\n",
" }\n",
"\n",
" log_success(\"No escalation needed — autonomous resolution continues\")\n",
" return None\n",
"\n",
"\n",
"# ── Quick test ─────────────────────────────────────────────────\n",
"test_plan = create_autonomous_plan(test_decision)\n",
"safety_ok, violations = autonomous_safety_check(test_decision, test_plan)\n",
"escalation = intelligent_escalation_decision(test_decision, (safety_ok, violations))\n",
"log_info(f\"Safety: {safety_ok}, Escalation: {'triggered' if escalation else 'not needed'}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "42c5bbd6",
"metadata": {},
"outputs": [],
"source": [
"# ── AutonomousCustomerServiceAgent ─────────────────────────────\n",
"# Ref: Section 5.1, The continuous cognitive loop (pp. 124-126)\n",
"#\n",
"# This class integrates all components into a single cognitive loop:\n",
"# perception → reasoning → safety check → action → learning.\n",
"# It maintains session_memory, decision_history, and\n",
"# performance_metrics across interactions.\n",
"\n",
"log_section(\"AutonomousCustomerServiceAgent\", chapter_ref=\"Section 5.1, pp. 124-126\")\n",
"\n",
"\n",
"class AutonomousCustomerServiceAgent:\n",
" \"\"\"Complete autonomous agent with the cognitive loop.\n",
"\n",
" Integrates perception, cognition, action, and learning into a\n",
" continuous cycle. Maintains session memory and decision history\n",
" for adaptation across interactions.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, AutonomousCustomerServiceAgent (pp. 124-126)\n",
" \"\"\"\n",
"\n",
" def __init__(self) -> None:\n",
" self.session_memory: dict = {}\n",
" self.decision_history: list[dict] = []\n",
" self.performance_metrics: dict = {\"total_runs\": 0, \"successes\": 0}\n",
"\n",
" def get_current_system_state(self) -> dict:\n",
" \"\"\"Return simulated system state for perception.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, Perception (p. 120)\n",
" \"\"\"\n",
" return {\n",
" \"active_sessions\": 42,\n",
" \"system_load_pct\": 65,\n",
" \"known_outages\": [\"NET-2026-0042\"],\n",
" }\n",
"\n",
" @fail_gracefully(\n",
" fallback_value={\"status\": \"cognitive_loop_failed\"},\n",
" chapter_ref=\"Section 5.1, pp. 124-126\",\n",
" )\n",
" def cognitive_loop(\n",
" self, user_message: str, context: dict\n",
" ) -> list[dict]:\n",
" \"\"\"The complete autonomous decision-making cycle.\n",
"\n",
" Executes all four stages in sequence:\n",
" 1. Enhanced Perception\n",
" 2. Autonomous Reasoning\n",
" 3. Safety Check + Strategic Action Execution\n",
" 4. Learning and Adaptation\n",
"\n",
" Args:\n",
" user_message: Raw user input.\n",
" context: Session context dict.\n",
"\n",
" Returns:\n",
" List of action result dicts.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, cognitive_loop() (pp. 124-9)\n",
" \"\"\"\n",
" log_section(\"Cognitive Loop START\", chapter_ref=\"Section 5.1, pp. 124-9\")\n",
"\n",
" # 1. Enhanced Perception\n",
" system_state = self.get_current_system_state()\n",
" perception_data = enhanced_perception(\n",
" user_message, context, system_state\n",
" )\n",
"\n",
" # 2. Autonomous Reasoning\n",
" decision_context = autonomous_reasoning(perception_data)\n",
"\n",
" # 3. Safety Check + Action Execution\n",
" planned_actions = create_autonomous_plan(decision_context)\n",
" safety_ok, violations = autonomous_safety_check(\n",
" decision_context, planned_actions\n",
" )\n",
" escalation = intelligent_escalation_decision(\n",
" decision_context, (safety_ok, violations)\n",
" )\n",
"\n",
" if escalation:\n",
" decision_context[\"strategy\"] = \"immediate_escalation\"\n",
"\n",
" results = autonomous_action_execution(decision_context)\n",
"\n",
" # 4. Learning and Adaptation\n",
" self.learn_from_interaction(\n",
" perception_data, decision_context, results\n",
" )\n",
" self.update_session_memory(\n",
" context.get(\"user_id\"), decision_context, results\n",
" )\n",
"\n",
" log_section(\"Cognitive Loop END\", chapter_ref=\"Section 5.1\")\n",
" return results\n",
"\n",
" def learn_from_interaction(\n",
" self,\n",
" perception_data: dict,\n",
" decision_context: dict,\n",
" results: list[dict],\n",
" ) -> None:\n",
" \"\"\"Enhanced learning from interaction outcomes.\n",
"\n",
" Calculates success score, flags low-performing strategies,\n",
" and stores the decision pattern for future reference.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, learn_from_interaction() (pp. 125-126)\n",
" \"\"\"\n",
" # Calculate success (Ref: p. 125)\n",
" completed = sum(1 for r in results if r.get(\"status\") == \"completed\")\n",
" total = max(len(results), 1)\n",
" success_score = round(completed / total, 2)\n",
"\n",
" self.performance_metrics[\"total_runs\"] += 1\n",
" if success_score >= 0.7:\n",
" self.performance_metrics[\"successes\"] += 1\n",
" log_success(f\"Learning: success_score={success_score} (above threshold)\")\n",
" else:\n",
" log_warn(\n",
" f\"Learning: success_score={success_score} — \"\n",
" f\"flagging strategy '{decision_context.get('strategy')}' \"\n",
" f\"for improvement\"\n",
" )\n",
"\n",
" # Store decision pattern (Ref: p. 126)\n",
" self.decision_history.append({\n",
" \"perception_summary\": {\n",
" \"intent\": decision_context.get(\"intent\"),\n",
" \"sentiment\": perception_data.get(\"sentiment\"),\n",
" },\n",
" \"decision\": {\n",
" \"strategy\": decision_context.get(\"strategy\"),\n",
" \"scores\": decision_context.get(\"strategy_scores\"),\n",
" },\n",
" \"outcome\": success_score,\n",
" \"timestamp\": datetime.now(),\n",
" })\n",
" log_info(f\"Decision history: {len(self.decision_history)} entries\")\n",
"\n",
" def update_session_memory(\n",
" self,\n",
" user_id: str | None,\n",
" decision_context: dict,\n",
" results: list[dict],\n",
" ) -> None:\n",
" \"\"\"Update session memory for cross-interaction continuity.\n",
"\n",
" Chapter Reference:\n",
" Section 5.1, update_session_memory() (p. 125)\n",
" \"\"\"\n",
" if user_id:\n",
" self.session_memory[user_id] = {\n",
" \"last_intent\": decision_context.get(\"intent\"),\n",
" \"last_strategy\": decision_context.get(\"strategy\"),\n",
" \"tasks_completed\": len(results),\n",
" \"timestamp\": datetime.now().isoformat(),\n",
" }\n",
" log_info(f\"Session memory updated for '{user_id}'\")\n",
"\n",
"\n",
"log_success(\"AutonomousCustomerServiceAgent class defined\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "63c9fb70",
"metadata": {},
"outputs": [],
"source": [
"# ── Case Study: Premium Business Customer Outage ───────────────\n",
"# Ref: Section 5.1, Case Study (p. 132)\n",
"#\n",
"# Scenario: A premium customer reports:\n",
"# \"My business internet has been intermittent for two days,\n",
"# and I have a critical presentation tomorrow.\"\n",
"#\n",
"# The agent must: perceive the urgency + business context,\n",
"# reason about strategy, validate safety, execute autonomously,\n",
"# and learn from the outcome.\n",
"\n",
"log_section(\n",
" \"CASE STUDY: Business-Critical Outage\",\n",
" chapter_ref=\"Section 5.1, Case Study, p. 132\",\n",
")\n",
"\n",
"agent = AutonomousCustomerServiceAgent()\n",
"\n",
"results = agent.cognitive_loop(\n",
" user_message=(\n",
" \"My business internet has been intermittent for two days, \"\n",
" \"and I have a critical presentation tomorrow.\"\n",
" ),\n",
" context={\"user_id\": \"premium_biz_123\", \"session\": \"new\"},\n",
")\n",
"\n",
"# ── Summary ────────────────────────────────────────────────────\n",
"log_section(\"Case Study Summary\")\n",
"log_info(f\"Total tasks executed: {len(results)}\")\n",
"log_info(f\"Agent performance: {agent.performance_metrics}\")\n",
"log_info(f\"Session memory keys: {list(agent.session_memory.keys())}\")\n",
"log_info(f\"Decision history entries: {len(agent.decision_history)}\")\n",
"\n",
"if all(r.get(\"status\") == \"completed\" for r in results):\n",
" log_success(\"All tasks completed successfully — autonomous resolution achieved\")\n",
"else:\n",
" log_warn(\"Some tasks did not complete — review logs above\")"
]
},
{
"cell_type": "markdown",
"id": "f8b78378",
"metadata": {},
"source": [
"### Analysis: Autonomous Decision-Making Agent\n",
"\n",
"The case study above demonstrates the complete cognitive loop (Figure 5.1) in action:\n",
"\n",
"- **Perception** (p. 124) — Enhanced context including system alerts, user tier, and agent availability.\n",
"- **Cognition** (p. 126) — LLM-powered reasoning with strategy selection and escalation assessment.\n",
"- **Action** — Task DAG execution with dependency ordering and safety checks.\n",
"- **Learning** — Success scoring feeds back to refine future autonomy thresholds.\n",
"\n",
"> **📝 Note — Five-Factor Escalation Scoring (pp. 128–131)** \n",
"> The agent uses five escalation factors: **safety_violations**, **confidence_threshold** (< 0.8), **complexity_threshold** (> 0.7), **user_preference**, and **business_impact** (> 0.6). A weighted escalation score is compared against the decision context's escalation threshold. This ensures that the agent can gracefully defer to a human operator for ambiguous situations, ethical dilemmas, or tasks beyond its capabilities.\n",
"\n",
"> **📝 Note — Production Impact (p. 131)** \n",
"> By combining structured prompts and conditional logic, such a bot can operate autonomously for up to **80% of user queries**, freeing human agents to focus on more nuanced and complex problems.\n"
]
},
{
"cell_type": "markdown",
"id": "56728c59",
"metadata": {},
"source": [
"---\n",
"## 3. Planning Agent\n",
"\n",
"**Chapter Ref:** Section 5.2 (pp. 131–137)\n",
"\n",
"While Autonomous Decision-Making agents excel at real-time, single-step responses, many real-world challenges require breaking down complex objectives into sequences of smaller, manageable tasks executed over extended periods. The transition from reactive decision-making to proactive planning represents a significant leap in agent sophistication.\n",
"\n",
"Planning agents master several interconnected capabilities: **decomposing** high-level goals into actionable subtasks, **sequencing** tasks appropriately, **allocating** resources, **monitoring** progress, and **adapting** plans dynamically.\n",
"\n",
"> **📝 Note — Two Planning Paradigms (p. 132)** \n",
"> Modern Planning agents leverage two complementary paradigms: **Symbolic planning foundations** (STRIPS, PDDL) which provide rigorous, mathematically grounded approaches for well-defined domains, and **LLM-powered dynamic planning** (Tree-of-Thought, Self-Ask) which introduces unprecedented flexibility for novel situations. The latter enables agents to decompose tasks without rigid pre-programming.\n",
"\n",
"### Figure 5.2 — LLM-Powered Dynamic Planning\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "12677ddd",
"metadata": {},
"outputs": [],
"source": [
"# ── Figure 5.2 — Planning Agent Architecture (SVG) ──\n",
"# Ref: Section 5.2, p. 133\n",
"\n",
"from IPython.display import SVG, display\n",
"\n",
"fig_5_2 = \"\"\"\n",
"\n",
"\"\"\"\n",
"display(SVG(fig_5_2))\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "bd93f25a",
"metadata": {},
"outputs": [],
"source": [
"# ── PlanningAgent Class ────────────────────────────────────────\n",
"# Ref: Section 5.2, Planning Agent (pp. 131-137)\n",
"#\n",
"# Implements hierarchical task decomposition with dependency\n",
"# resolution via topological sort. The agent decomposes a\n",
"# high-level goal into phased subtasks, executes them in\n",
"# dependency order, monitors progress, and can revise the plan\n",
"# based on feedback.\n",
"\n",
"log_section(\"Planning Agent\", chapter_ref=\"Section 5.2, pp. 131-137\")\n",
"\n",
"\n",
"class PlanningAgent:\n",
" \"\"\"Agent for orchestrating complex, multi-step workflows.\n",
"\n",
" Decomposes high-level goals into hierarchical task plans,\n",
" executes them respecting dependency order via topological\n",
" sort, and supports adaptive plan revision.\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, The Planning Agent (pp. 131-137)\n",
" \"\"\"\n",
"\n",
" def __init__(self, llm_client_ref) -> None:\n",
" \"\"\"Initialize with an LLM client reference.\n",
"\n",
" Args:\n",
" llm_client_ref: MockLLM or real LLM client instance.\n",
" \"\"\"\n",
" self.llm = llm_client_ref\n",
" self.execution_log: list[dict] = []\n",
" self.plan_revisions: int = 0\n",
"\n",
" @fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.2, pp. 133-20\",\n",
" )\n",
" def decompose_goal(self, goal: str) -> list[dict]:\n",
" \"\"\"Decompose a high-level goal into a hierarchical task DAG.\n",
"\n",
" Uses the LLM to identify the goal type, then builds a\n",
" structured task plan matching Figure 5.2's architecture.\n",
"\n",
" Args:\n",
" goal: High-level objective string.\n",
"\n",
" Returns:\n",
" List of task dicts with 'id', 'action', 'phase',\n",
" 'depends_on', and 'tools' keys.\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, Strategic task decomposition (pp. 133-20)\n",
" \"\"\"\n",
" response = self.llm.generate(goal)\n",
" log_info(f\"Goal decomposition: '{goal}'\")\n",
" log_info(f\"LLM identified strategy: '{response.strategy}'\")\n",
"\n",
" # Ref: Figure 5.2 — marketing campaign task hierarchy\n",
" if response.intent == \"planning_request\":\n",
" plan = [\n",
" # Phase 1: Research (Ref: Figure 5.2, left branch)\n",
" {\n",
" \"id\": \"P1-T1\", \"phase\": 1,\n",
" \"action\": \"define_target_audience\",\n",
" \"depends_on\": [],\n",
" \"tools\": [\"market_research_platform\"],\n",
" },\n",
" {\n",
" \"id\": \"P1-T2\", \"phase\": 1,\n",
" \"action\": \"competitive_analysis\",\n",
" \"depends_on\": [],\n",
" \"tools\": [\"market_research_platform\"],\n",
" },\n",
" # Phase 2: Content Creation (Ref: Figure 5.2, center)\n",
" {\n",
" \"id\": \"P2-T1\", \"phase\": 2,\n",
" \"action\": \"write_blog_posts\",\n",
" \"depends_on\": [\"P1-T1\"],\n",
" \"tools\": [\"content_generation_api\"],\n",
" },\n",
" {\n",
" \"id\": \"P2-T2\", \"phase\": 2,\n",
" \"action\": \"design_banners\",\n",
" \"depends_on\": [\"P1-T1\"],\n",
" \"tools\": [\"content_generation_api\"],\n",
" },\n",
" # Phase 3: Distribution (Ref: Figure 5.2, right)\n",
" {\n",
" \"id\": \"P3-T1\", \"phase\": 3,\n",
" \"action\": \"setup_ad_campaigns\",\n",
" \"depends_on\": [\"P2-T1\", \"P2-T2\"],\n",
" \"tools\": [\"ad_platform_api\"],\n",
" },\n",
" {\n",
" \"id\": \"P3-T2\", \"phase\": 3,\n",
" \"action\": \"schedule_launch_event\",\n",
" \"depends_on\": [\"P1-T2\"],\n",
" \"tools\": [\"calendar_booking_api\"],\n",
" },\n",
" # Phase 4: Monitoring (Ref: Figure 5.2, feedback loop)\n",
" {\n",
" \"id\": \"P4-T1\", \"phase\": 4,\n",
" \"action\": \"monitor_campaign_metrics\",\n",
" \"depends_on\": [\"P3-T1\", \"P3-T2\"],\n",
" \"tools\": [\"analytics_dashboard\"],\n",
" },\n",
" {\n",
" \"id\": \"P4-T2\", \"phase\": 4,\n",
" \"action\": \"adaptive_revision_check\",\n",
" \"depends_on\": [\"P4-T1\"],\n",
" \"tools\": [],\n",
" },\n",
" ]\n",
" else:\n",
" # Generic planning decomposition\n",
" plan = [\n",
" {\"id\": \"G-T1\", \"phase\": 1, \"action\": \"analyze_requirements\",\n",
" \"depends_on\": [], \"tools\": []},\n",
" {\"id\": \"G-T2\", \"phase\": 2, \"action\": \"design_solution\",\n",
" \"depends_on\": [\"G-T1\"], \"tools\": []},\n",
" {\"id\": \"G-T3\", \"phase\": 3, \"action\": \"implement_solution\",\n",
" \"depends_on\": [\"G-T2\"], \"tools\": []},\n",
" {\"id\": \"G-T4\", \"phase\": 4, \"action\": \"validate_results\",\n",
" \"depends_on\": [\"G-T3\"], \"tools\": []},\n",
" ]\n",
"\n",
" log_success(f\"Decomposed into {len(plan)} tasks across \"\n",
" f\"{max(t['phase'] for t in plan)} phases\")\n",
" return plan\n",
"\n",
" @fail_gracefully(\n",
" fallback_value=[],\n",
" chapter_ref=\"Section 5.2, pp. 134-135\",\n",
" )\n",
" def execute_plan(self, plan: list[dict]) -> list[dict]:\n",
" \"\"\"Execute a task plan respecting dependency order.\n",
"\n",
" Uses topological sort to determine execution order.\n",
" Tasks with satisfied dependencies execute in phase order.\n",
" Each task is individually wrapped for fault tolerance.\n",
"\n",
" Args:\n",
" plan: Task DAG from decompose_goal().\n",
"\n",
" Returns:\n",
" List of execution result dicts.\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, Dynamic execution and monitoring (pp. 134-135)\n",
" \"\"\"\n",
" results = []\n",
" completed: dict[str, dict] = {}\n",
" remaining = list(plan)\n",
"\n",
" log_info(\"Beginning plan execution (topological order)...\")\n",
"\n",
" while remaining:\n",
" # Find tasks whose dependencies are all satisfied\n",
" ready = [\n",
" t for t in remaining\n",
" if all(d in completed for d in t[\"depends_on\"])\n",
" ]\n",
" if not ready:\n",
" log_error(\"Deadlock: unresolvable dependencies remain\")\n",
" break\n",
"\n",
" for task in ready:\n",
" result = self._execute_single_task(task)\n",
" completed[task[\"id\"]] = result\n",
" results.append(result)\n",
" remaining.remove(task)\n",
"\n",
" # Monitor progress (Ref: p. 134)\n",
" self.monitor_progress(plan, completed)\n",
"\n",
" log_success(f\"Plan execution complete: {len(results)}/{len(plan)} tasks\")\n",
" return results\n",
"\n",
" @fail_gracefully(\n",
" fallback_value={\"status\": \"task_failed\"},\n",
" chapter_ref=\"Section 5.2, p. 136\",\n",
" )\n",
" def _execute_single_task(self, task: dict) -> dict:\n",
" \"\"\"Execute a single task and return its result.\n",
"\n",
" Args:\n",
" task: Task dict from the plan.\n",
"\n",
" Returns:\n",
" Result dict with status, timing, and metadata.\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, Dynamic execution (p. 136)\n",
" \"\"\"\n",
" tools_str = \", \".join(task[\"tools\"]) if task[\"tools\"] else \"none\"\n",
" log_info(f\" Phase {task['phase']} | {task['id']}: \"\n",
" f\"{task['action']} (tools: {tools_str})\")\n",
"\n",
" result = {\n",
" \"task_id\": task[\"id\"],\n",
" \"phase\": task[\"phase\"],\n",
" \"action\": task[\"action\"],\n",
" \"status\": \"completed\",\n",
" \"timestamp\": datetime.now().isoformat(),\n",
" }\n",
" log_success(f\" {task['id']}: completed\")\n",
" self.execution_log.append(result)\n",
" return result\n",
"\n",
" def monitor_progress(\n",
" self, plan: list[dict], completed: dict\n",
" ) -> None:\n",
" \"\"\"Track execution progress and log phase completion.\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, Dynamic execution and monitoring (p. 134)\n",
" \"\"\"\n",
" total = len(plan)\n",
" done = len(completed)\n",
" pct = round(done / total * 100)\n",
"\n",
" # Check if a phase just completed\n",
" completed_phases = set()\n",
" for t in plan:\n",
" if t[\"id\"] in completed:\n",
" completed_phases.add(t[\"phase\"])\n",
"\n",
" # Log phase completions\n",
" for phase in sorted(completed_phases):\n",
" phase_tasks = [t for t in plan if t[\"phase\"] == phase]\n",
" if all(t[\"id\"] in completed for t in phase_tasks):\n",
" phase_actions = [t[\"action\"] for t in phase_tasks]\n",
" if done <= len(phase_tasks) or done == total:\n",
" log_info(f\" Progress: {pct}% ({done}/{total}) \"\n",
" f\"— Phase {phase} complete\")\n",
"\n",
" @fail_gracefully(\n",
" fallback_value=None,\n",
" chapter_ref=\"Section 5.2, p. 134\",\n",
" )\n",
" def revise_plan(\n",
" self, plan: list[dict], feedback: str\n",
" ) -> list[dict]:\n",
" \"\"\"Revise the plan based on execution feedback.\n",
"\n",
" Demonstrates the adaptive intelligence capability\n",
" where the agent modifies strategies based on new\n",
" information or changing conditions.\n",
"\n",
" Args:\n",
" plan: Current task plan.\n",
" feedback: Feedback string describing what changed.\n",
"\n",
" Returns:\n",
" Revised plan (or original if no changes needed).\n",
"\n",
" Chapter Reference:\n",
" Section 5.2, Adaptive intelligence and learning (p. 134)\n",
" \"\"\"\n",
" self.plan_revisions += 1\n",
" log_warn(f\"Plan revision #{self.plan_revisions}: {feedback}\")\n",
"\n",
" # Simulate adding a corrective task\n",
" max_phase = max(t[\"phase\"] for t in plan)\n",
" revision_task = {\n",
" \"id\": f\"REV-{self.plan_revisions}\",\n",
" \"phase\": max_phase + 1,\n",
" \"action\": f\"corrective_action_{self.plan_revisions}\",\n",
" \"depends_on\": [plan[-1][\"id\"]],\n",
" \"tools\": [],\n",
" }\n",
" revised = plan + [revision_task]\n",
" log_info(f\"Added corrective task: {revision_task['id']}\")\n",
" return revised\n",
"\n",
"\n",
"log_success(\"PlanningAgent class defined\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3edfe56b",
"metadata": {},
"outputs": [],
"source": [
"# ── Demo: Marketing Campaign Planning ──────────────────────────\n",
"# Ref: Section 5.2, Figure 5.2 (p. 133)\n",
"#\n",
"# Scenario: \"Launch Product Marketing Campaign\"\n",
"# The agent decomposes this into phased tasks, executes them\n",
"# in dependency order, demonstrates the feedback loop, and\n",
"# shows adaptive plan revision.\n",
"\n",
"log_section(\n",
" \"DEMO: Marketing Campaign Planning\",\n",
" chapter_ref=\"Section 5.2, Figure 5.2, p. 133\",\n",
")\n",
"\n",
"planner = PlanningAgent(llm_client)\n",
"\n",
"# Step 1: Decompose the goal\n",
"campaign_plan = planner.decompose_goal(\"Launch Product Marketing Campaign\")\n",
"\n",
"print()\n",
"log_info(\"Task DAG overview:\")\n",
"for task in campaign_plan:\n",
" deps = task[\"depends_on\"] if task[\"depends_on\"] else [\"(start)\"]\n",
" log_info(f\" Phase {task['phase']} | {task['id']}: \"\n",
" f\"{task['action']} → depends_on: {deps}\")\n",
"\n",
"# Step 2: Execute the plan\n",
"print()\n",
"log_section(\"Executing Plan\", chapter_ref=\"Section 5.2, pp. 134-135\")\n",
"campaign_results = planner.execute_plan(campaign_plan)\n",
"\n",
"# Step 3: Simulate adaptive revision (Ref: p. 134)\n",
"print()\n",
"log_section(\"Adaptive Revision\", chapter_ref=\"Section 5.2, p. 134\")\n",
"revised_plan = planner.revise_plan(\n",
" campaign_plan,\n",
" feedback=\"Competitor launched similar campaign — need differentiation\"\n",
")\n",
"\n",
"# Step 4: Execute the revision\n",
"revision_results = planner.execute_plan(\n",
" [t for t in revised_plan if t[\"id\"].startswith(\"REV-\")]\n",
")\n",
"\n",
"# Summary\n",
"print()\n",
"log_section(\"Planning Demo Summary\")\n",
"log_info(f\"Original tasks: {len(campaign_plan)}\")\n",
"log_info(f\"After revision: {len(revised_plan)} (+{len(revised_plan) - len(campaign_plan)})\")\n",
"log_info(f\"Total executions: {len(planner.execution_log)}\")\n",
"log_success(\"Marketing campaign planning demo complete\")"
]
},
{
"cell_type": "markdown",
"id": "11b924e9",
"metadata": {},
"source": [
"### Analysis: Planning Agent\n",
"\n",
"The demo above illustrates the **LLM-powered dynamic planning** paradigm\n",
"from Figure 5.2. Key observations:\n",
"\n",
"1. **Hierarchical decomposition** broke \"Launch Product Marketing Campaign\"\n",
" into 4 phases with 8 tasks, matching the figure's structure:\n",
" research → content creation → distribution → monitoring.\n",
"\n",
"2. **Topological execution** ensured that Phase 2 tasks (blog posts,\n",
" banners) waited for Phase 1 (audience definition) to complete.\n",
" Parallel tasks within the same phase executed together.\n",
"\n",
"3. **Adaptive revision** demonstrated the feedback loop — when a\n",
" competitor threat was detected, the agent appended a corrective task\n",
" to the plan without disrupting completed work.\n",
"\n",
"#### Planning vs. Decision-Making (Table 5.1)\n",
"\n",
"| Capability | Autonomous Decision-Maker | Planning Agent |\n",
"|---|---|---|\n",
"| **Focus** | Immediate, tactical responses | Strategic, long-term goal achievement |\n",
"| **Scope** | Single decisions at a time | Multi-step, coordinated processes |\n",
"| **Adaptation** | Heuristics and conditional logic | Dynamic revision and learning loops |\n",
"| **Integration** | Direct API calls for specific actions | Orchestrated tool workflows |\n",
"| **Memory Use** | Contextual within current session | Persistent, cross-project learning |\n",
"\n",
"Ref: Section 5.2 (pp. 131-137), Table 5.1 (pp. 135-136)"
]
},
{
"cell_type": "markdown",
"id": "89800fcc",
"metadata": {},
"source": [
"---\n",
"## 4. Memory-Augmented Agent\n",
"\n",
"**Chapter Ref:** Section 5.3 (pp. 135–144)\n",
"\n",
"Autonomous agents can make decisions and formulate plans, but a critical limitation remains: how can they retain context and learn from past experiences beyond the current interaction? Many foundational agents operate solely in the present. **Memory-Augmented agents** incorporate mechanisms that preserve context over time, significantly improving personalization, coherence, and long-term reasoning.\n",
"\n",
"### Three Types of Memory\n",
"\n",
"| Memory Type | Role | Trigger Condition | Failure Mode if Skipped |\n",
"|---|---|---|---|\n",
"| **Working** | Short-term session buffer (LLM prompt) | Active session, current turn | Loss of immediate conversational coherence |\n",
"| **Episodic** | Historical interactions & events | Cross-session continuity needed | Agent treats each session as new, breaking personalization |\n",
"| **Semantic** | Structured facts & world knowledge | Domain knowledge required | Hallucinated or stale answers |\n",
"\n",
"> **📝 Note — Memory Selection Guide (p. 136)** \n",
"> Use this guide to select which memory system to query: **Working memory** when the LLM context window is being populated directly during an active session. **Episodic memory** via vector similarity search over interaction history when cross-session continuity is needed. **Semantic memory** via keyword or semantic search over a knowledge base when factual or domain knowledge is required that isn't answerable from the session alone.\n",
"\n",
"> **📝 Note — Vector Database Options (pp. 137–138)** \n",
"> Practical implementations include: **Chroma** (well-suited for local development and prototyping), **Pinecone** (managed cloud service optimized for production-scale retrieval), and **Weaviate** (open source with hybrid keyword and vector search support).\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9dbbcbcc",
"metadata": {},
"outputs": [],
"source": [
"# ── MemoryAugmentedAgent Class ─────────────────────────────────\n",
"# Ref: Section 5.3, Memory-Augmented Agent (pp. 138-139)\n",
"#\n",
"# Core logic: retrieve relevant episodic memories to provide\n",
"# context for the current query, generate a response with the\n",
"# LLM, then store the new interaction for future use.\n",
"\n",
"log_section(\"Memory-Augmented Agent\", chapter_ref=\"Section 5.3, pp. 135-144\")\n",
"\n",
"\n",
"class MemoryAugmentedAgent:\n",
" \"\"\"Agent with episodic memory for context-aware interactions.\n",
"\n",
" Retrieves relevant past interactions from a vector database,\n",
" injects them into the LLM prompt as context, generates a\n",
" response, and stores the new interaction for future retrieval.\n",
"\n",
" Chapter Reference:\n",
" Section 5.3, MemoryAugmentedAgent (pp. 138-139)\n",
" \"\"\"\n",
"\n",
" def __init__(self, llm_client_ref, vector_db_ref) -> None:\n",
" \"\"\"Initialize with LLM client and vector database.\n",
"\n",
" Args:\n",
" llm_client_ref: MockLLM or real LLM client.\n",
" vector_db_ref: MockVectorDB or real vector DB.\n",
" \"\"\"\n",
" self.llm = llm_client_ref\n",
" self.episodic_memory = vector_db_ref\n",
" self.interaction_count: int = 0\n",
"\n",
" @fail_gracefully(\n",
" fallback_value=\"I apologize, but I was unable to process your request.\",\n",
" chapter_ref=\"Section 5.3, pp. 138-139\",\n",
" )\n",
" def process_interaction(\n",
" self, user_id: str, user_query: str\n",
" ) -> str:\n",
" \"\"\"Process a user query with episodic memory augmentation.\n",
"\n",
" Steps:\n",
" 1. Retrieve relevant past interactions from episodic memory\n",
" 2. Construct a prompt that includes retrieved memories\n",
" 3. Generate a response via the LLM\n",
" 4. Store the current interaction for future use\n",
"\n",
" Args:\n",
" user_id: Unique user identifier.\n",
" user_query: Current user query text.\n",
"\n",
" Returns:\n",
" Generated response string.\n",
"\n",
" Chapter Reference:\n",
" Section 5.3, process_interaction() (pp. 138-139)\n",
" \"\"\"\n",
" self.interaction_count += 1\n",
" log_info(f\"Interaction #{self.interaction_count} for user '{user_id}'\")\n",
"\n",
" # 1. Retrieve relevant past interactions (Ref: p. 141-26)\n",
" log_info(\"Step 1: Retrieving episodic memories...\")\n",
" relevant_memories = self.episodic_memory.search(\n",
" query=user_query,\n",
" user_id=user_id,\n",
" limit=3,\n",
" )\n",
"\n",
" if relevant_memories:\n",
" log_success(f\"Retrieved {len(relevant_memories)} relevant memories:\")\n",
" for i, mem in enumerate(relevant_memories):\n",
" log_info(\n",
" f\" Memory {i+1}: '{mem['query'][:50]}...' \"\n",
" f\"(relevance: {mem.get('relevance_score', 'N/A')})\"\n",
" )\n",
" else:\n",
" log_warn(\"No relevant memories found — first interaction\")\n",
"\n",
" # 2. Construct prompt with memory context (Ref: pp. 138-139)\n",
" memory_context = self._format_memories(relevant_memories)\n",
" prompt = (\n",
" f\"You are a personalized healthcare assistant.\\n\"\n",
" f\"Here is the user's current query: \\\"{user_query}\\\"\\n\"\n",
" f\"For context, here are relevant past interactions \"\n",
" f\"with this user:\\n{memory_context}\\n\"\n",
" f\"Provide a helpful and context-aware response.\"\n",
" )\n",
" log_info(\"Step 2: Prompt constructed with memory context\")\n",
"\n",
" # 3. Generate response (Ref: p. 138)\n",
" log_info(\"Step 3: Generating response...\")\n",
" response = self.llm.generate(user_query)\n",
" response_text = response.content\n",
"\n",
" # 4. Store interaction in episodic memory (Ref: p. 138)\n",
" log_info(\"Step 4: Storing interaction in episodic memory...\")\n",
" interaction_record = {\n",
" \"user_id\": user_id,\n",
" \"query\": user_query,\n",
" \"response_summary\": response_text[:100],\n",
" \"timestamp\": datetime.now().isoformat(),\n",
" }\n",
" self.episodic_memory.add(interaction_record)\n",
"\n",
" log_success(f\"Interaction #{self.interaction_count} complete\")\n",
" return response_text\n",
"\n",
" def _format_memories(self, memories: list[dict]) -> str:\n",
" \"\"\"Format retrieved memories for prompt injection.\n",
"\n",
" Args:\n",
" memories: List of memory dicts from vector DB search.\n",
"\n",
" Returns:\n",
" Formatted string for inclusion in LLM prompt.\n",
"\n",
" Chapter Reference:\n",
" Section 5.3, prompt construction (p. 139)\n",
" \"\"\"\n",
" if not memories:\n",
" return \"(No prior interactions found)\"\n",
"\n",
" lines = []\n",
" for mem in memories:\n",
" lines.append(\n",
" f\"- [{mem.get('timestamp', 'unknown')}] \"\n",
" f\"Query: '{mem.get('query', '')}' → \"\n",
" f\"Response: '{mem.get('response_summary', '')}'\"\n",
" )\n",
" return \"\\n\".join(lines)\n",
"\n",
" def get_prompt_preview(\n",
" self, user_id: str, user_query: str\n",
" ) -> str:\n",
" \"\"\"Show the full prompt that would be sent to the LLM.\n",
"\n",
" Useful for readers to understand how memory retrieval\n",
" augments generation — without actually calling the LLM.\n",
"\n",
" Args:\n",
" user_id: User identifier for memory lookup.\n",
" user_query: Current query text.\n",
"\n",
" Returns:\n",
" The complete prompt string.\n",
"\n",
" Chapter Reference:\n",
" Section 5.3, Prompt inspection (p. 139)\n",
" \"\"\"\n",
" memories = self.episodic_memory.search(\n",
" query=user_query, user_id=user_id, limit=3\n",
" )\n",
" memory_context = self._format_memories(memories)\n",
" return (\n",
" f\"You are a personalized healthcare assistant.\\n\"\n",
" f\"Here is the user's current query: \\\"{user_query}\\\"\\n\"\n",
" f\"For context, here are relevant past interactions \"\n",
" f\"with this user:\\n{memory_context}\\n\"\n",
" f\"Provide a helpful and context-aware response.\"\n",
" )\n",
"\n",
"\n",
"log_success(\"MemoryAugmentedAgent class defined\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "725ba241",
"metadata": {},
"outputs": [],
"source": [
"# ── Healthcare Case Study: 3-Turn Conversation ─────────────────\n",
"# Ref: Section 5.3, Case Study: A personalized healthcare\n",
"# assistant (pp. 139-140)\n",
"#\n",
"# Turn 1: Patient reports fatigue (new interaction)\n",
"# Turn 2: Patient follows up about diet change (retrieves Turn 1)\n",
"# Turn 3: Patient mentions iron deficiency (retrieves both priors)\n",
"\n",
"log_section(\n",
" \"CASE STUDY: Personalized Healthcare Assistant\",\n",
" chapter_ref=\"Section 5.3, pp. 139-140\",\n",
")\n",
"\n",
"# Reset vector DB to start fresh with pre-seeded records\n",
"vector_db.reset()\n",
"health_agent = MemoryAugmentedAgent(llm_client, vector_db)\n",
"\n",
"# ── Turn 1: Initial fatigue report ─────────────────────────────\n",
"log_section(\"Turn 1: Initial Fatigue Report\", chapter_ref=\"p. 139\")\n",
"response_1 = health_agent.process_interaction(\n",
" user_id=\"patient_42\",\n",
" user_query=\"I've been feeling very fatigued lately\",\n",
")\n",
"log_info(f\"Agent response: {response_1[:100]}...\")\n",
"\n",
"# ── Turn 2: Follow-up about diet change ────────────────────────\n",
"print()\n",
"log_section(\"Turn 2: Diet Change Follow-Up\", chapter_ref=\"p. 139\")\n",
"response_2 = health_agent.process_interaction(\n",
" user_id=\"patient_42\",\n",
" user_query=\"That diet change you suggested helped a bit, but I'm still tired\",\n",
")\n",
"log_info(f\"Agent response: {response_2[:100]}...\")\n",
"\n",
"# ── Turn 3: Iron deficiency update ─────────────────────────────\n",
"print()\n",
"log_section(\"Turn 3: Iron Deficiency Update\", chapter_ref=\"p. 142\")\n",
"response_3 = health_agent.process_interaction(\n",
" user_id=\"patient_42\",\n",
" user_query=\"My allergist said I'm low on iron, what should I do?\",\n",
")\n",
"log_info(f\"Agent response: {response_3[:100]}...\")\n",
"\n",
"# Summary\n",
"print()\n",
"log_section(\"Healthcare Case Study Summary\")\n",
"log_info(f\"Total interactions: {health_agent.interaction_count}\")\n",
"log_success(\"Memory accumulation demonstrated across 3 turns\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e69916ce",
"metadata": {},
"outputs": [],
"source": [
"# ── Prompt Inspection ──────────────────────────────────────────\n",
"# Ref: Section 5.3, Prompt construction (p. 139)\n",
"#\n",
"# Display the actual prompt that would be sent to the LLM for\n",
"# Turn 3, showing how retrieved episodic memories are injected\n",
"# into the context window.\n",
"\n",
"log_section(\"Prompt Inspection\", chapter_ref=\"Section 5.3, p. 139\")\n",
"\n",
"prompt_preview = health_agent.get_prompt_preview(\n",
" user_id=\"patient_42\",\n",
" user_query=\"My allergist said I'm low on iron, what should I do?\",\n",
")\n",
"\n",
"log_info(\"Full prompt sent to LLM (with injected memories):\")\n",
"print()\n",
"print(\"=\" * 60)\n",
"print(prompt_preview)\n",
"print(\"=\" * 60)\n",
"print()\n",
"log_success(\n",
" \"This demonstrates how episodic memory retrieval augments \"\n",
" \"the LLM's context, enabling personalized responses.\"\n",
")"
]
},
{
"cell_type": "markdown",
"id": "0e9474d3",
"metadata": {},
"source": [
"### Analysis: Memory-Augmented Agent\n",
"\n",
"The healthcare case study demonstrates how memory transforms agent behavior:\n",
"\n",
"**Turn 1** — The agent processes a new fatigue report. With pre-seeded\n",
"episodic memory, it retrieves relevant prior records and stores this\n",
"interaction for future reference.\n",
"\n",
"**Turn 2** — When the patient follows up, `process_interaction()` searches\n",
"episodic memory and retrieves Turn 1's record because of semantic overlap\n",
"(\"fatigue\" ↔ \"tired\", \"diet\"). The retrieved context is injected directly\n",
"into the LLM prompt, enabling a coherent response that acknowledges the\n",
"ongoing experience.\n",
"\n",
"**Turn 3** — The agent now retrieves both prior turns, demonstrating\n",
"**memory accumulation**. The prompt inspection cell reveals the exact\n",
"prompt structure described in the book (p. 139): user query + retrieved\n",
"memories formatted as context.\n",
"\n",
"#### Memory Consolidation for Scalability\n",
"\n",
"For production systems, the chapter recommends summarizing long interaction\n",
"histories into concise patient profiles (p. 142). For example:\n",
"> `Patient A, Type 2 Diabetes, responded well to diet changes (2026-01-15),\n",
"> struggled with medication adherence (2026-03-20).`\n",
"\n",
"This ensures the vector database remains efficient as interaction\n",
"history grows.\n",
"\n",
"Ref: Section 5.3 (pp. 135-144), Figure 5.3"
]
},
{
"cell_type": "markdown",
"id": "e3eac9d0",
"metadata": {},
"source": [
"---\n",
"## 5. Comparative Analysis\n",
"\n",
"**Chapter Ref:** Section 5.4 (pp. 141–142)\n",
"\n",
"Each foundational cognitive architecture excels in its specific domain. However, their true potential emerges through **strategic combinations**. These architectures are not mutually exclusive — they are complementary building blocks.\n",
"\n",
"### Table 5.2 — Comparative Analysis\n",
"\n",
"| Agent Type | Strength | Ideal Use Case | Limitations |\n",
"|---|---|---|---|\n",
"| **Autonomous Decision-Making** | Real-time response and flexibility | Customer service, triage | Risk of hallucinations; lacks deep context |\n",
"| **Planning Agent** | Structured execution and adaptability | Project orchestration, complex workflows | Slower execution; requires control logic |\n",
"| **Memory-Augmented Agent** | Long-term coherence and personalization | CRM systems, task handoff, continuity | Complex maintenance; sensitive to retrieval design |\n",
"\n",
"### Figure 5.4 — Integrated Architecture\n",
"\n",
"The demo below combines all three agent types: the **Decision-Making agent** handles the immediate query, the **Planning agent** creates a multi-step resolution plan, and the **Memory-Augmented agent** retrieves relevant history to personalize the response.\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "7fe84b9b",
"metadata": {},
"outputs": [],
"source": [
"# ── Integrated Demo: All Three Agent Types ─────────────────────\n",
"# Ref: Section 5.4, Figure 5.4 (p. 142)\n",
"#\n",
"# Scenario: A premium customer has a complex issue that requires\n",
"# all three architectures working together:\n",
"# 1. Decision agent handles initial triage\n",
"# 2. Planning agent decomposes follow-up tasks\n",
"# 3. Memory agent maintains context across the interaction\n",
"\n",
"log_section(\n",
" \"INTEGRATED DEMO: All Three Architectures\",\n",
" chapter_ref=\"Section 5.4, Figure 5.4, p. 142\",\n",
")\n",
"\n",
"# ── Step 1: Decision Agent — Initial Triage ────────────────────\n",
"log_section(\"Step 1: Decision Agent — Triage\")\n",
"\n",
"decision_agent = AutonomousCustomerServiceAgent()\n",
"triage_results = decision_agent.cognitive_loop(\n",
" user_message=\"My internet is down and I need to file a billing dispute\",\n",
" context={\"user_id\": \"premium_biz_123\", \"session\": \"integrated_demo\"},\n",
")\n",
"triage_intent = decision_agent.decision_history[-1][\"perception_summary\"][\"intent\"]\n",
"log_info(f\"Triage result: intent='{triage_intent}', \"\n",
" f\"tasks={len(triage_results)}\")\n",
"\n",
"# ── Step 2: Planning Agent — Follow-Up Decomposition ───────────\n",
"print()\n",
"log_section(\"Step 2: Planning Agent — Decomposition\")\n",
"\n",
"integrated_planner = PlanningAgent(llm_client)\n",
"followup_plan = integrated_planner.decompose_goal(\n",
" \"Plan follow-up investigation for customer billing and connectivity issues\"\n",
")\n",
"followup_results = integrated_planner.execute_plan(followup_plan)\n",
"log_info(f\"Planning result: {len(followup_results)} tasks completed\")\n",
"\n",
"# ── Step 3: Memory Agent — Context Preservation ────────────────\n",
"print()\n",
"log_section(\"Step 3: Memory Agent — Context Preservation\")\n",
"\n",
"# Reset and use fresh memory agent for this demo\n",
"vector_db.reset()\n",
"memory_agent = MemoryAugmentedAgent(llm_client, vector_db)\n",
"\n",
"# Store the triage outcome as an episodic record\n",
"memory_agent.episodic_memory.add({\n",
" \"user_id\": \"premium_biz_123\",\n",
" \"query\": \"Internet down + billing dispute (integrated triage)\",\n",
" \"response_summary\": (\n",
" f\"Triage: {triage_intent}, \"\n",
" f\"{len(triage_results)} actions executed, \"\n",
" f\"follow-up plan: {len(followup_results)} tasks\"\n",
" ),\n",
" \"timestamp\": datetime.now().isoformat(),\n",
"})\n",
"\n",
"# Later interaction — memory retrieves the triage context\n",
"memory_response = memory_agent.process_interaction(\n",
" user_id=\"premium_biz_123\",\n",
" user_query=\"What was done about my internet and billing issue?\",\n",
")\n",
"log_info(f\"Memory-augmented response: {memory_response[:80]}...\")\n",
"\n",
"# ── Integration Summary ────────────────────────────────────────\n",
"print()\n",
"log_section(\"Integration Summary\", chapter_ref=\"Figure 5.4, p. 142\")\n",
"log_success(\"Decision Agent → triaged the initial request\")\n",
"log_success(\"Planning Agent → decomposed follow-up into structured tasks\")\n",
"log_success(\"Memory Agent → preserved context for future interactions\")\n",
"log_info(\n",
" \"This demonstrates how the three architectures complement each other \"\n",
" \"when combined (Figure 5.4)\"\n",
")"
]
},
{
"cell_type": "markdown",
"id": "9de86065",
"metadata": {},
"source": [
"### When to Use Which Architecture\n",
"\n",
"- **Use Decision-Making** when you need fast, reactive responses to\n",
" individual queries — customer service chatbots, alerting systems, triage.\n",
"\n",
"- **Use Planning** when the task requires multi-step coordination with\n",
" dependencies — project orchestration, workflow automation, campaign execution.\n",
"\n",
"- **Use Memory-Augmented** when continuity across sessions matters —\n",
" personalized assistants, CRM systems, healthcare follow-ups.\n",
"\n",
"- **Combine all three** for complex scenarios where an initial triage\n",
" leads to a structured plan that must be remembered across sessions —\n",
" exactly the pattern shown in Figure 5.4.\n",
"\n",
"Ref: Section 5.4 (pp. 141-142), Table 5.2"
]
},
{
"cell_type": "markdown",
"id": "e0c5b44f",
"metadata": {},
"source": [
"---\n",
"## 6. Engineering Best Practices & Wrap-Up\n",
"\n",
"**Chapter Ref:** Section 5.5 (pp. 142–144)\n",
"\n",
"Building reliable intelligent agents requires adherence to sound engineering principles throughout the development lifecycle.\n",
"\n",
"> **📝 Note — Scalability Principles (p. 143)** \n",
"> **Start modular:** Separate capabilities into distinct modules (perception, memory, planning, action). This separation of concerns simplifies development and debugging. \n",
"> **Log everything:** Implement comprehensive logging at every stage of the agent's decision loop. Tools like LangSmith are critical for understanding *why* an agent made a particular decision.\n",
"\n",
"> **📝 Note — Context Management (p. 143)** \n",
"> **Avoid prompt bloat:** LLMs have finite context windows. Feeding them excessive or irrelevant information degrades performance, increases latency, and incurs higher costs. Optimize retrieval pipelines to provide only the most relevant context — techniques like re-ranking (Chapter 6) are crucial.\n",
"\n",
"> **📝 Note — Robustness (p. 143)** \n",
"> Design systems with clear fallbacks and escalation paths for undecidable or unsafe conditions. This might mean defaulting to a human operator, requesting clarification, or rolling back an action. Every tool call in this repository is wrapped in the `@fail_gracefully` decorator for exponential-backoff retries.\n",
"\n",
"> **What's Next:** Chapter 6 turns to *Information Retrieval and Knowledge Agents*, expanding these ideas further by exploring how agents access and leverage vast external knowledge to enrich their decision-making.\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "8935ec8b",
"metadata": {},
"outputs": [],
"source": [
"# ── Final Status Report ────────────────────────────────────────\n",
"# Simulation summary: all agent runs, success/failure counts,\n",
"# mock vs live mode status.\n",
"\n",
"log_section(\"FINAL STATUS REPORT\", chapter_ref=\"Chapter 5 Summary\")\n",
"\n",
"# Gather metrics\n",
"total_agent_runs = 0\n",
"if 'agent' in dir():\n",
" total_agent_runs += agent.performance_metrics.get(\"total_runs\", 0)\n",
"if 'decision_agent' in dir():\n",
" total_agent_runs += decision_agent.performance_metrics.get(\"total_runs\", 0)\n",
"\n",
"total_planning_tasks = 0\n",
"if 'planner' in dir():\n",
" total_planning_tasks += len(planner.execution_log)\n",
"if 'integrated_planner' in dir():\n",
" total_planning_tasks += len(integrated_planner.execution_log)\n",
"\n",
"total_memory_interactions = 0\n",
"if 'health_agent' in dir():\n",
" total_memory_interactions += health_agent.interaction_count\n",
"if 'memory_agent' in dir():\n",
" total_memory_interactions += memory_agent.interaction_count\n",
"\n",
"log_info(f\"Mode: {'SIMULATION' if SIMULATION_MODE else 'LIVE'}\")\n",
"log_info(f\"Decision Agent runs: {total_agent_runs}\")\n",
"log_info(f\"Planning Agent tasks executed: {total_planning_tasks}\")\n",
"log_info(f\"Memory Agent interactions: {total_memory_interactions}\")\n",
"\n",
"# Verify all architectures demonstrated\n",
"architectures_ok = (\n",
" total_agent_runs > 0\n",
" and total_planning_tasks > 0\n",
" and total_memory_interactions > 0\n",
")\n",
"\n",
"if architectures_ok:\n",
" print()\n",
" log_success(\"=\" * 50)\n",
" log_success(\" ALL THREE ARCHITECTURES DEMONSTRATED SUCCESSFULLY\")\n",
" log_success(\"=\" * 50)\n",
" print()\n",
" log_info(\n",
" \"This notebook implemented the three foundational cognitive \"\n",
" \"architectures from Chapter 5: Autonomous Decision-Making, \"\n",
" \"Planning, and Memory-Augmented agents.\"\n",
" )\n",
"else:\n",
" log_error(\"Some architectures were not demonstrated — check cells above\")"
]
},
{
"cell_type": "markdown",
"id": "b6322d1a",
"metadata": {},
"source": [
"### Further Reading\n",
"\n",
"- **Chapter 1** — The cognitive loop foundation (perception → reasoning → action → learning)\n",
"- **Chapter 3** — *The Art of Agent Prompting* — prompt engineering for chain-of-thought planning\n",
"- **Chapter 4** — *Agent Deployment and Responsible Development* — production safety and security\n",
"- **Chapter 6** — *Information Retrieval and Knowledge Agents* — advanced retrieval for memory systems\n",
"- **Chapter 7** — *Multi-agent systems* — combining agents into collaborative workflows\n",
"\n",
"**Book Repository:**\n",
"[PacktPublishing/30-Agents-Every-AI-Engineer-Must-Build](https://github.com/PacktPublishing/30-Agents-Every-AI-Engineer-Must-Build)\n",
"\n",
"---\n",
"\n",
"*Chapter 5: Foundational Cognitive Architectures*\n",
"*Author: Imran Ahmad | Publisher: Packt Publishing, 2026*"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"name": "python",
"version": "3.10.0"
}
},
"nbformat": 4,
"nbformat_minor": 5
}