{ "cells": [ { "cell_type": "markdown", "id": "60b96680", "metadata": {}, "source": [ "# Chapter 10: Conversational and Content Creation Agents\n", "\n", "**Book:** *30 Agents Every AI Engineer Must Build* \n", "**Author:** Imran Ahmad \n", "**Publisher:** Packt Publishing, 2026 \n", "**Chapter Pages:** pp. 281–306\n", "\n", "---\n", "\n", "> *\"We can only see a short distance ahead, but we can see plenty there that needs to be done.\"*\n", "> — Alan M. Turing\n", "\n", "## Introduction\n", "\n", "Conversational and Content Creation agents represent the creative and social interface of artificial intelligence. Unlike software development agents that prioritize syntactic correctness and logic, these agents must master the nuances of **tone, context, empathy, and aesthetic coherence**. They transform raw generative potential into structured, meaningful interactions that prioritize engaging users and producing high-value content.\n", "\n", "For the engineer, the core challenge is designing systems that **sustain coherent behavior across long interactions**, enforce safety and brand constraints deterministically, and coordinate multiple generation stages without sacrificing reliability.\n", "\n", "### What This Notebook Covers\n", "\n", "1. **The Conversational Agent** (§10.1, pp. 282–293) — A safety-aware agent with a dual-memory hierarchy (working memory via `ConversationSummaryBufferMemory` and semantic long-term recall via FAISS), a deterministic `SafetyLayer` sentinel, and a `PersonaEngine` that enforces empathetic, non-directive behavior as a constraint layer. Demonstrated through a mental health support agent case study.\n", "\n", "2. **The Content Creation Agent** (§10.2, pp. 293–305) — A multi-agent content creation pipeline built on the SMPA (Sense-Model-Plan-Act) paradigm. Specialist agents (Email, SEO, Ad Creative) produce brand-constrained drafts validated by an `EditorAgent` using a CSP (Constraint Satisfaction Problem) framework, with an `AnalyticsEngine` closing the adaptive feedback loop.\n", "\n", "### Key Architectural Principles\n", "\n", "Both agent types share a **boundary-first** design philosophy:\n", "\n", "- **Safety before generation** — Crisis detection and brand constraints are enforced *upstream* of all generative logic, never as afterthoughts\n", "- **Persona as controlled bias** — Personality is not randomness; it is a set of behavioral constraints that bias generation toward a stable behavioral region of the model's latent space\n", "- **Dual-memory hierarchy** — Working memory (recent turns, summarized progressively) + Semantic memory (FAISS vector store for long-term recall by similarity)\n", "- **Generation downstream of policy** — The system first decides whether it is safe to proceed, then constructs context, and only then invokes generative behavior under constraints\n", "\n", "### Formal Foundations\n", "\n", "A Conversational agent is characterized by five architectural properties: **persistent context** across turns and sessions, **intent awareness** that interprets goals beyond isolated utterances, **dialog management** that actively steers conversation flow, **behavioral consistency** adhering to a defined persona, and **tool and memory integration** for retrieval and recall.\n", "\n", "Content Creation agents formalize brand enforcement as a **Constraint Satisfaction Problem (CSP)**, where the consistency score is: $C = \\frac{1}{n} \\sum_{i=1}^{n} \\varphi(A_i, G)$, measuring alignment of generated artifact attributes $A_i$ against brand guidelines $G$.\n", "\n", "**Simulation Mode:** All code runs in two modes — LIVE MODE (with OpenAI API key) or SIMULATION MODE (deterministic mock responses from Chapter 10 content). No API keys required.\n", "\n", "---" ] }, { "cell_type": "markdown", "id": "ad108a05", "metadata": {}, "source": [ "---\n", "## Cell Group 1: Setup & Configuration\n", "*Ref: Chapter 10 — Technical Requirements*" ] }, { "cell_type": "code", "execution_count": null, "id": "eb89198e", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 1.1 — Imports & Dependency Check\n", "# Ref: Chapter 10 — \"Technical requirements\" — (pp. 281–282)\n", "# Author: Imran Ahmad | Chapter 10\n", "# =============================================================================\n", "\n", "import os\n", "import sys\n", "import hashlib\n", "import getpass\n", "import logging\n", "import functools\n", "import warnings\n", "from abc import ABC, abstractmethod\n", "from dataclasses import dataclass, field\n", "from typing import Any, Dict, List, Optional, Tuple\n", "\n", "warnings.filterwarnings(\"ignore\")\n", "\n", "# --- Core dependencies ---\n", "import numpy as np\n", "from dotenv import load_dotenv\n", "\n", "# --- LangChain ecosystem ---\n", "from langchain_core.messages import (\n", " AIMessage,\n", " BaseMessage,\n", " HumanMessage,\n", " SystemMessage,\n", ")\n", "from langchain.memory import ConversationSummaryBufferMemory\n", "\n", "# --- Version reporting ---\n", "import langchain\n", "import openai\n", "\n", "print(\"=\" * 65)\n", "print(\" Chapter 10: Conversational and Content Creation Agents\")\n", "print(\" Author: Imran Ahmad\")\n", "print(\"=\" * 65)\n", "print()\n", "print(f\" Python : {sys.version.split()[0]}\")\n", "print(f\" langchain : {langchain.__version__}\")\n", "print(f\" openai : {openai.__version__}\")\n", "print(f\" numpy : {np.__version__}\")\n", "\n", "try:\n", " import faiss\n", " print(f\" faiss-cpu : {faiss.__version__ if hasattr(faiss, '__version__') else 'installed'}\")\n", "except ImportError:\n", " print(\" faiss-cpu : NOT INSTALLED (will use mock embeddings)\")\n", "\n", "print()\n", "print(\" All core dependencies loaded successfully.\")\n", "print(\"=\" * 65)" ] }, { "cell_type": "code", "execution_count": null, "id": "99b52f00", "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": "6323f279", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 1.2 — Color-Coded Logger\n", "# Ref: Chapter 10 — Core Technical Requirements — Visual Logging — (p. 282)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Log Level Schema (Strategy §8):\n", "# DEBUG (10) = Blue \\033[94m — Internal tracing\n", "# INFO (20) = Blue \\033[94m — Agent lifecycle events\n", "# SUCCESS (25) = Green \\033[92m — Step completion, valid outputs\n", "# WARNING (30) = Yellow \\033[93m — Brand violations, retries\n", "# ERROR (40) = Red \\033[91m — API failures, fallback triggers\n", "# =============================================================================\n", "\n", "# Register custom SUCCESS level\n", "SUCCESS_LEVEL = 25\n", "logging.addLevelName(SUCCESS_LEVEL, \"SUCCESS\")\n", "\n", "\n", "def success(self, message, *args, **kwargs):\n", " \"\"\"Log a SUCCESS-level message.\"\"\"\n", " if self.isEnabledFor(SUCCESS_LEVEL):\n", " self._log(SUCCESS_LEVEL, message, args, **kwargs)\n", "\n", "\n", "logging.Logger.success = success\n", "\n", "# --- ANSI color codes ---\n", "BLUE = \"\\033[94m\"\n", "GREEN = \"\\033[92m\"\n", "YELLOW = \"\\033[93m\"\n", "RED = \"\\033[91m\"\n", "RESET = \"\\033[0m\"\n", "BOLD = \"\\033[1m\"\n", "\n", "\n", "class ColorFormatter(logging.Formatter):\n", " \"\"\"ANSI color formatter for Jupyter notebook output.\"\"\"\n", "\n", " LEVEL_COLORS = {\n", " logging.DEBUG: BLUE,\n", " logging.INFO: BLUE,\n", " SUCCESS_LEVEL: GREEN,\n", " logging.WARNING: YELLOW,\n", " logging.ERROR: RED,\n", " }\n", "\n", " def format(self, record):\n", " color = self.LEVEL_COLORS.get(record.levelno, RESET)\n", " timestamp = self.formatTime(record, \"%H:%M:%S\")\n", " level = record.levelname\n", " return f\"{color}[{timestamp}] [{level}] {record.getMessage()}{RESET}\"\n", "\n", "\n", "def get_logger(name: str = \"chapter10\") -> logging.Logger:\n", " \"\"\"Create a color-coded logger for agent pipeline tracing.\"\"\"\n", " logger = logging.getLogger(name)\n", " if not logger.handlers:\n", " handler = logging.StreamHandler()\n", " handler.setFormatter(ColorFormatter())\n", " logger.addHandler(handler)\n", " logger.setLevel(logging.DEBUG)\n", " return logger\n", "\n", "\n", "# --- Initialize the global logger ---\n", "logger = get_logger()\n", "logger.success(\"Color-coded logger initialized. Levels: INFO (Blue), SUCCESS (Green), WARNING (Yellow), ERROR (Red)\")" ] }, { "cell_type": "code", "execution_count": null, "id": "cba139e0", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 1.3 — API Key Detection & Mode Banner\n", "# Ref: Chapter 10 — \"Technical requirements\" — API key instruction — (pp. 281–282)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Flow: load_dotenv() → os.getenv → getpass fallback → LIVE or SIMULATION\n", "# Safety: getpass is conditional on sys.stdin.isatty() to prevent hangs\n", "# in non-interactive environments (CI/CD, Docker, nbconvert)\n", "# =============================================================================\n", "\n", "load_dotenv()\n", "\n", "\n", "def get_api_key() -> str:\n", " \"\"\"\n", " Resolve the OpenAI API key using cascading fallback:\n", " 1. Environment variable (via .env or shell export)\n", " 2. Interactive getpass prompt (only if TTY is available)\n", " 3. Empty string → triggers SIMULATION MODE\n", " \"\"\"\n", " key = os.getenv(\"OPENAI_API_KEY\", \"\")\n", "\n", " if key and \"your-key\" not in key and \"your_key\" not in key:\n", " return key\n", "\n", " # Only attempt getpass if a TTY is available\n", " if sys.stdin.isatty():\n", " try:\n", " key = getpass.getpass(\n", " \"Enter your OpenAI API key (or press Enter for Simulation Mode): \"\n", " )\n", " return key.strip()\n", " except (EOFError, OSError):\n", " pass\n", "\n", " return \"\"\n", "\n", "\n", "# --- Detect mode ---\n", "OPENAI_API_KEY = get_api_key()\n", "SIMULATION_MODE = not bool(OPENAI_API_KEY)\n", "\n", "print()\n", "if SIMULATION_MODE:\n", " logger.info(\"=\" * 60)\n", " logger.info(\"SIMULATION MODE — No API key found. Using MockChatOpenAI.\")\n", " logger.info(\"All outputs are pre-written responses from Chapter 10.\")\n", " logger.info(\"To switch to LIVE MODE: add your key to .env and restart.\")\n", " logger.info(\"=\" * 60)\n", "else:\n", " logger.success(\"=\" * 60)\n", " logger.success(\"LIVE MODE — OpenAI API key detected.\")\n", " logger.success(f\"Key: {OPENAI_API_KEY[:8]}...{OPENAI_API_KEY[-4:]}\")\n", " logger.success(\"=\" * 60)\n", "print()" ] }, { "cell_type": "code", "execution_count": null, "id": "f9657aad", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 1.4 — LLM & Embeddings Factory + @fail_gracefully Decorator\n", "# Ref: Chapter 10 — Bridges \"Technical requirements\" to implementation — (pp. 281–282)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# The @fail_gracefully decorator wraps every external call:\n", "# 1. Log BLUE on entry\n", "# 2. Try function body\n", "# ├── Success → Log GREEN, return result\n", "# └── Exception → Log RED, return fallback_value (never raises)\n", "# =============================================================================\n", "\n", "\n", "def fail_gracefully(fallback_value=None, section=\"10.x\"):\n", " \"\"\"\n", " Decorator that wraps external calls with resilient error handling.\n", "\n", " On success: logs GREEN and returns the actual result.\n", " On failure: logs RED with diagnostics and returns fallback_value.\n", " The decorated function never raises — the notebook never halts.\n", "\n", " Args:\n", " fallback_value: Value returned on exception (can be any type).\n", " section: Chapter section reference for log traceability.\n", " \"\"\"\n", " def decorator(func):\n", " @functools.wraps(func)\n", " def wrapper(*args, **kwargs):\n", " func_name = func.__name__\n", " logger.info(f\"Executing: {func_name} (Section {section})...\")\n", " try:\n", " result = func(*args, **kwargs)\n", " logger.success(f\"Step complete: {func_name} returned valid output.\")\n", " return result\n", " except Exception as e:\n", " logger.error(\n", " f\"{type(e).__name__} in {func_name}. \"\n", " f\"Falling back to mock logic for Section {section}. \"\n", " f\"Detail: {e}\"\n", " )\n", " return fallback_value\n", " return wrapper\n", " return decorator\n", "\n", "\n", "# --- Factory: LLM ---\n", "def get_llm(model: str = \"gpt-4o\", temperature: float = 0.7):\n", " \"\"\"\n", " Return the appropriate LLM instance based on current mode.\n", "\n", " LIVE MODE: Returns langchain ChatOpenAI connected to the OpenAI API.\n", " SIMULATION: Returns MockChatOpenAI with chapter-derived responses.\n", " \"\"\"\n", " if SIMULATION_MODE:\n", " from mock_llm import MockChatOpenAI\n", " logger.info(\"LLM factory: returning MockChatOpenAI (Simulation Mode)\")\n", " return MockChatOpenAI(model=model, temperature=temperature)\n", " else:\n", " try:\n", " from langchain_openai import ChatOpenAI\n", " logger.info(\"LLM factory: returning ChatOpenAI (Live Mode)\")\n", " return ChatOpenAI(\n", " model=model,\n", " temperature=temperature,\n", " openai_api_key=OPENAI_API_KEY,\n", " )\n", " except Exception as e:\n", " logger.error(f\"ChatOpenAI init failed: {e}. Falling back to MockChatOpenAI.\")\n", " from mock_llm import MockChatOpenAI\n", " return MockChatOpenAI(model=model, temperature=temperature)\n", "\n", "\n", "# --- Factory: Embeddings ---\n", "def get_embeddings():\n", " \"\"\"\n", " Return the appropriate Embeddings instance based on current mode.\n", "\n", " LIVE MODE: Returns langchain OpenAIEmbeddings.\n", " SIMULATION: Returns MockOpenAIEmbeddings with deterministic hash vectors.\n", " \"\"\"\n", " if SIMULATION_MODE:\n", " from mock_llm import MockOpenAIEmbeddings\n", " logger.info(\"Embeddings factory: returning MockOpenAIEmbeddings (Simulation Mode)\")\n", " return MockOpenAIEmbeddings()\n", " else:\n", " try:\n", " from langchain_openai import OpenAIEmbeddings\n", " logger.info(\"Embeddings factory: returning OpenAIEmbeddings (Live Mode)\")\n", " return OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY)\n", " except Exception as e:\n", " logger.error(f\"OpenAIEmbeddings init failed: {e}. Falling back to MockOpenAIEmbeddings.\")\n", " from mock_llm import MockOpenAIEmbeddings\n", " return MockOpenAIEmbeddings()\n", "\n", "\n", "# --- Instantiate LLM and Embeddings ---\n", "llm = get_llm()\n", "embeddings = get_embeddings()\n", "\n", "logger.success(\"LLM and Embeddings initialized. Ready for agent construction.\")" ] }, { "cell_type": "markdown", "id": "7c9b456d", "metadata": {}, "source": [ "---\n", "## Cell Group 2: The Conversational Agent\n", "*Ref: Chapter 10, Section 10.1 — \"The Conversational agent\"*\n", "\n", "This section implements the safety-aware conversational agent architecture from Section 10.1. The design follows a vertical pipeline where every user interaction passes through multiple layers of validation and contextualization:\n", "\n", "1. **Safety Layer (Sentinel)** — Deterministic crisis detection that bypasses the LLM entirely\n", "2. **Context Manager (Working Memory)** — `ConversationSummaryBufferMemory` with progressive summarization\n", "3. **Semantic Memory (Long-Term Store)** — FAISS vector index for similarity-based recall\n", "4. **Persona Engine (Constraint Layer)** — System prompt that reshapes output toward empathetic, non-directive behavior\n", "5. **MentalHealthAgent** — Full pipeline integrating all layers with `@fail_gracefully` resilience" ] }, { "cell_type": "markdown", "id": "1075bf65", "metadata": {}, "source": [ "#### Figure 10.1 — Safety-Aware Conversational Agent Architecture *(Book p. 288)*\n", "\n", "The architecture follows a vertical pipeline where every user interaction is subjected to multiple layers of validation and contextualization before a response is synthesized:\n", "\n", "```\n", " ┌──────────────┐\n", " │ User Input │\n", " └──────┬───────┘\n", " ▼\n", " ┌────────────────────────┐ ┌─ ─ ─ ─ ─ ─ ─ ─ ─┐\n", " │ Safety Layer │─────▶ Crisis Protocol │\n", " │ (Sentinel) │ Crisis└─ ─ ─ ─ ─ ─ ─ ─ ─┘\n", " │ Crisis Trigger │\n", " │ Detection │\n", " └────────┬───────────────┘\n", " │ Safe\n", " ▼\n", " ┌────────────────────────┐\n", " │ Cognition Core │\n", " │ (Dialog Manager) │\n", " │ │\n", " │ ┌───────────────────┐ │\n", " │ │ Working Memory │ │\n", " │ │ (RAM — Summary │ │\n", " │ │ Buffer) │──┼──▶ Archival\n", " │ └───────────────────┘ │\n", " │ ┌───────────────────┐ │\n", " │ │ Semantic Memory │ │\n", " │ │ (Disk — FAISS │ │\n", " │ │ Vector Store) │ │\n", " │ └───────────────────┘ │\n", " │ ┌───────────────────┐ │\n", " │ │ Persona Engine │ │\n", " │ │ (Behavioral │ │\n", " │ │ Constraints — │ │\n", " │ │ Empathy) │ │\n", " │ └───────────────────┘ │\n", " └────────┬───────────────┘\n", " ▼\n", " ┌────────────────────────┐\n", " │ Empathetic Response │\n", " └────────────────────────┘\n", "```\n", "\n", "**Key design choice:** Every request is processed top-down — the system first decides whether it is safe to proceed, then constructs context from short- and long-term memory, and only then invokes generative behavior under persona constraints. Generation is *downstream of policy*.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "1c3a6044", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.1 — SafetyLayer: The Crisis Sentinel\n", "# Ref: Chapter 10, \"Implementing the vertical pipeline\" — (p. 289)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Safety should be implemented as a validator that is external to the\n", "# language model, not as 'good behavior we hope the model will remember.'\n", "# The sentinel sits upstream specifically so that 'creative generation'\n", "# cannot accidentally occur when the situation demands strict escalation\n", "# behavior.\" — Chapter 10\n", "# =============================================================================\n", "\n", "\n", "class SafetyLayer:\n", " \"\"\"\n", " Deterministic circuit breaker for crisis detection.\n", "\n", " Positioned at the entry point of the vertical pipeline, this component\n", " scans incoming text for crisis triggers. If a risk is identified, the\n", " system immediately diverts to a predefined Crisis Protocol, bypassing\n", " the generative cognition core entirely.\n", "\n", " Ref: Chapter 10, \"Implementing the vertical pipeline\" — (p. 289)\n", " \"\"\"\n", "\n", " def __init__(self):\n", " self.triggers = [\"hurt myself\", \"suicide\", \"end my life\", \"harm\"]\n", " logger.success(f\"SafetyLayer initialized with {len(self.triggers)} crisis triggers.\")\n", "\n", " def is_safe(self, text: str) -> bool:\n", " \"\"\"Returns False if a crisis keyword is detected.\"\"\"\n", " return not any(trigger in text.lower() for trigger in self.triggers)\n", "\n", " def get_crisis_protocol(self) -> str:\n", " \"\"\"Return the predefined crisis escalation response.\"\"\"\n", " return (\n", " \"I'm hearing that you're in a lot of pain. I am an AI and cannot \"\n", " \"provide emergency help. Please call 988 immediately.\"\n", " )\n", "\n", "\n", "# --- Demo: SafetyLayer ---\n", "safety = SafetyLayer()\n", "\n", "# Test 1: Safe input\n", "safe_input = \"I've been feeling anxious about my exams lately.\"\n", "logger.info(f\"SafetyLayer: scanning input for crisis triggers...\")\n", "if safety.is_safe(safe_input):\n", " logger.success(f\"SafetyLayer cleared input: '{safe_input[:50]}...' → Proceeding to Cognition Core.\")\n", "else:\n", " logger.warning(\"SafetyLayer: crisis detected!\")\n", "\n", "print()\n", "\n", "# Test 2: Crisis input\n", "crisis_input = \"I want to hurt myself.\"\n", "logger.info(f\"SafetyLayer: scanning input for crisis triggers...\")\n", "if not safety.is_safe(crisis_input):\n", " logger.warning(f\"SafetyLayer INTERCEPT: crisis keyword detected in '{crisis_input}'\")\n", " protocol = safety.get_crisis_protocol()\n", " logger.warning(f\"Crisis Protocol: {protocol}\")\n", "else:\n", " logger.success(\"SafetyLayer cleared input.\")" ] }, { "cell_type": "markdown", "id": "cda889b8", "metadata": {}, "source": [ "> **📘 Implementation Insight: Efficient Context Management** *(Book p. 284)*\n", ">\n", "> To ensure the agent remains performant as dialogue grows, engineers utilize the `ConversationSummaryBufferMemory` pattern. This allows the agent to maintain situational awareness across turns by balancing the depth of context retrieval with execution speed.\n", ">\n", "> When a user provides a new query, the system performs a similarity search against the vector store to find the most relevant past entries. These recalled snippets are injected into the current context window as a \"context hint,\" allowing the agent to generate responses that are both **contextually aware of the current session** and **historically grounded in long-term user data**.\n", ">\n", "> This structural separation is what allows a mental health support agent to remember an \"exam\" mentioned weeks ago without losing the thread of the current conversation.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "551de41b", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.2 — ContextManager: Working Memory\n", "# Ref: Chapter 10, \"Memory hierarchy in practice\" — working memory — (p. 290)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Engineers often utilize the ConversationSummaryBufferMemory pattern.\n", "# This pattern allows the agent to maintain situational awareness across\n", "# turns by balancing the depth of context retrieval with execution speed.\"\n", "# — Chapter 10\n", "# =============================================================================\n", "\n", "\n", "class ContextManager:\n", " \"\"\"\n", " The Chronicler: Implements semantic continuity through a dual-memory structure.\n", "\n", " Working memory keeps recent raw text and summarizes older history using\n", " ConversationSummaryBufferMemory with max_token_limit=300. Long-term\n", " storage is handled separately by SemanticMemory (Cell 2.3).\n", "\n", " Ref: Chapter 10, \"Memory hierarchy in practice\"\n", " \"\"\"\n", "\n", " def __init__(self, llm):\n", " # working_memory keeps recent raw text and summarizes older history\n", " self.working_memory = ConversationSummaryBufferMemory(\n", " llm=llm,\n", " max_token_limit=300,\n", " return_messages=True,\n", " )\n", " self.long_term_store = [] # Simulated Vector Store for semantic retrieval\n", "\n", " def archive_meaningful_context(self, interaction: str):\n", " \"\"\"Stores key emotional data points into a high-dimensional vector space.\"\"\"\n", " self.long_term_store.append(interaction)\n", " logger.info(f\"Archived to long-term store: '{interaction[:60]}...'\")\n", "\n", "\n", "# --- Demo: ContextManager ---\n", "context_manager = ContextManager(llm=llm)\n", "logger.success(\n", " f\"ContextManager initialized: working memory (max_token_limit=300), \"\n", " f\"long-term store (empty)\"\n", ")\n", "\n", "# Demonstrate saving a turn to working memory\n", "context_manager.working_memory.save_context(\n", " {\"input\": \"I've been feeling anxious about exams.\"},\n", " {\"output\": \"It sounds like exams are creating a lot of pressure for you.\"},\n", ")\n", "logger.success(\"Working memory: saved first interaction turn.\")\n", "\n", "# Show memory state\n", "memory_vars = context_manager.working_memory.load_memory_variables({})\n", "print(f\"\\nWorking memory state ({len(memory_vars.get('history', []))} messages):\")\n", "for msg in memory_vars.get(\"history\", []):\n", " role = msg.type if hasattr(msg, \"type\") else \"unknown\"\n", " print(f\" [{role}] {msg.content[:80]}...\")" ] }, { "cell_type": "code", "execution_count": null, "id": "0033c459", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.3 — SemanticMemory: FAISS Long-Term Store\n", "# Ref: Chapter 10, \"Memory hierarchy in practice\" — semantic memory — (pp. 290–291)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Semantic recall is not keyword search. The system embeds both stored\n", "# memories and the current query into a high-dimensional space and\n", "# retrieves by similarity.\" — Chapter 10\n", "# =============================================================================\n", "\n", "from langchain_community.vectorstores import FAISS\n", "\n", "\n", "class SemanticMemory:\n", " \"\"\"\n", " The Long-Term Store: Enables retrieval of past associations using\n", " semantic similarity via FAISS vector index.\n", "\n", " In SIMULATION MODE, uses MockOpenAIEmbeddings (deterministic 256-dim\n", " hash vectors). In LIVE MODE, uses OpenAIEmbeddings (text-embedding-3-small).\n", "\n", " Ref: Chapter 10, \"Memory hierarchy in practice\"\n", " \"\"\"\n", "\n", " def __init__(self, embeddings_model=None):\n", " self.embeddings = embeddings_model or embeddings\n", " self.vector_db = None\n", "\n", " @fail_gracefully(fallback_value=None, section=\"10.1\")\n", " def archive_event(self, text: str):\n", " \"\"\"Converts a conversation snippet into a vector and stores it.\"\"\"\n", " if self.vector_db is None:\n", " self.vector_db = FAISS.from_texts([text], self.embeddings)\n", " else:\n", " self.vector_db.add_texts([text])\n", " logger.info(f\"SemanticMemory: archived '{text[:50]}...'\")\n", "\n", " @fail_gracefully(fallback_value=\"\", section=\"10.1\")\n", " def retrieve_relevant_context(self, current_query: str) -> str:\n", " \"\"\"\n", " Given the current query, fetch the most semantically relevant\n", " prior snippet and inject it into the live context window.\n", " \"\"\"\n", " if self.vector_db is None:\n", " return \"\"\n", " docs = self.vector_db.similarity_search(current_query, k=1)\n", " return docs[0].page_content if docs else \"\"\n", "\n", "\n", "# --- Demo: SemanticMemory ---\n", "semantic_memory = SemanticMemory()\n", "logger.success(\"SemanticMemory initialized (FAISS vector store, empty).\")\n", "\n", "# Archive a text snippet\n", "semantic_memory.archive_event(\n", " \"User discussed feeling anxious about upcoming math exam. \"\n", " \"Expressed concern about time pressure and preparation adequacy.\"\n", ")\n", "\n", "# Retrieve by semantic query\n", "retrieved = semantic_memory.retrieve_relevant_context(\"How are you feeling about the exam?\")\n", "print(f\"\\nSemantic retrieval result:\")\n", "print(f\" Query: 'How are you feeling about the exam?'\")\n", "print(f\" Retrieved: '{retrieved[:80]}...'\")\n", "print(f\" Match: {'Yes' if retrieved else 'No'} (vector similarity)\")" ] }, { "cell_type": "markdown", "id": "c17f62dc", "metadata": {}, "source": [ "> **⚠️ Security Note** *(Book p. 291)*\n", ">\n", "> Because the semantic memory store may contain highly sensitive personal data, the architectural security requirement is **not optional**. At minimum, encryption at rest and strong access controls should be treated as baseline requirements for the semantic layer in this domain.\n" ] }, { "cell_type": "markdown", "id": "14ff2209", "metadata": {}, "source": [ "### Controlled Bias: Persona as an Architectural Choice\n", "\n", "As discussed in Chapter 10, a *persona* can be understood as a set of behavioral constraints applied to the language model's output distribution. These constraints do not determine exact responses, but they **bias generation toward a consistent behavioral region** of the model's latent space.\n", "\n", "For example, defining a persona as \"empathetic\" increases the likelihood of validation-oriented language — acknowledgment, reflective phrasing, emotional mirroring — while reducing the probability of terse, prescriptive, or emotionally neutral responses.\n", "\n", "> *\"Personality is not randomness; it is controlled bias.\"* — Chapter 10\n", "\n", "The `system_prompt` defined above enforces three concrete constraints:\n", "1. **Reflective questioning** — responses begin with acknowledgment (\"It sounds like...\")\n", "2. **Non-directive stance** — the agent avoids prescriptive advice\n", "3. **Validation-first style** — empathy takes priority over problem-solving" ] }, { "cell_type": "code", "execution_count": null, "id": "c987b6f6", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.4 — PersonaEngine: The Constraint Layer\n", "# Ref: Chapter 10, \"The persona engine as a constraint layer\" — (p. 293)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Personality is not randomness; it is controlled bias.\" — Chapter 10\n", "#\n", "# The persona engine ensures that \"empathetic\" is not an emergent property\n", "# that appears sometimes and disappears under stress. Persona constraints\n", "# are encoded as stable system-level configuration, reused every turn.\n", "# =============================================================================\n", "\n", "system_prompt = SystemMessage(content=(\n", " \"You are a supportive peer. Use reflective questioning \"\n", " \"(e.g., 'It sounds like...'). Avoid giving directive advice. \"\n", " \"Weight your responses toward validation and empathy.\"\n", "))\n", "\n", "logger.success(f\"PersonaEngine: loaded 'supportive peer' persona.\")\n", "print(f\"\\nSystem prompt ({len(system_prompt.content)} chars):\")\n", "print(f\" {system_prompt.content}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "4c94aa2b", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.5 — MentalHealthAgent: The Full Vertical Pipeline\n", "# Ref: Chapter 10, \"Case study: Implementing an empathetic mental health — (pp. 287–292)\n", "# support agent\"\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Pipeline ordering (from Figure 10.1):\n", "# 1. Safety Check (Immediate) — deterministic sentinel\n", "# 2. Context Retrieval (Semantic Continuity) — working + long-term memory\n", "# 3. Empathetic Generation — LLM under persona constraints\n", "# 4. Memory Update — save context for future turns\n", "# =============================================================================\n", "\n", "\n", "class MentalHealthAgent:\n", " \"\"\"\n", " Complete empathetic mental health support agent combining:\n", " - SafetyLayer (crisis sentinel)\n", " - ContextManager (working memory with summarization)\n", " - SemanticMemory (FAISS long-term store)\n", " - PersonaEngine (system prompt constraint layer)\n", "\n", " The handle_query() method enforces the vertical pipeline ordering\n", " from Figure 10.1: safety → context → generation → memory update.\n", "\n", " Ref: Chapter 10, \"Case study: Implementing an empathetic mental — (pp. 287–292)\n", " health support agent\"\n", " \"\"\"\n", "\n", " def __init__(self, llm_instance, embeddings_instance, persona):\n", " self.llm = llm_instance\n", " self.safety = SafetyLayer()\n", " self.context = ContextManager(llm=self.llm)\n", " self.semantic = SemanticMemory(embeddings_model=embeddings_instance)\n", " self.system_prompt = persona\n", " logger.success(\"MentalHealthAgent initialized — all pipeline layers active.\")\n", "\n", " @fail_gracefully(\n", " fallback_value=\"[Simulated empathetic response — see Section 10.1]\",\n", " section=\"10.1\",\n", " )\n", " def handle_query(self, user_input: str) -> str:\n", " \"\"\"\n", " Process a user query through the full vertical pipeline.\n", "\n", " 1. Safety Check (Immediate)\n", " 2. Context Retrieval (Semantic Continuity)\n", " 3. Empathetic Generation\n", " 4. Memory Update\n", " \"\"\"\n", " # 1. Safety Check (Immediate)\n", " logger.info(\"SafetyLayer: scanning input for crisis triggers...\")\n", " if not self.safety.is_safe(user_input):\n", " protocol = self.safety.get_crisis_protocol()\n", " logger.warning(f\"SafetyLayer INTERCEPT — crisis protocol activated.\")\n", " return protocol\n", "\n", " logger.success(\"SafetyLayer cleared input. Proceeding to Cognition Core.\")\n", "\n", " # 2. Context Retrieval (Semantic Continuity)\n", " logger.info(\"Retrieving working memory context...\")\n", " history = self.context.working_memory.load_memory_variables({}).get(\n", " \"history\", []\n", " )\n", "\n", " # Also check semantic memory for long-term recall\n", " semantic_context = self.semantic.retrieve_relevant_context(user_input)\n", " if semantic_context:\n", " logger.info(f\"Semantic recall: '{semantic_context[:50]}...'\")\n", "\n", " # 3. Empathetic Generation\n", " full_context = [self.system_prompt] + list(history)\n", " if semantic_context:\n", " full_context.append(\n", " SystemMessage(content=f\"Relevant past context: {semantic_context}\")\n", " )\n", " full_context.append(HumanMessage(content=user_input))\n", "\n", " response = self.llm.invoke(full_context)\n", "\n", " # 4. Memory Update\n", " self.context.working_memory.save_context(\n", " {\"input\": user_input}, {\"output\": response.content}\n", " )\n", "\n", " return response.content\n", "\n", "\n", "# --- Instantiate the agent ---\n", "agent = MentalHealthAgent(\n", " llm_instance=llm,\n", " embeddings_instance=embeddings,\n", " persona=system_prompt,\n", ")" ] }, { "cell_type": "code", "execution_count": null, "id": "ef3afee5", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 2.6 — Demo: Multi-Turn Dialogue with Memory Recall & Crisis Bypass\n", "# Ref: Chapter 10, Full Section 10.1 integration — (pp. 282–293)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Scripted conversation demonstrating:\n", "# Turn 1: \"I've been feeling anxious about my exams\" → empathetic response\n", "# Turn 2: \"The math exam is tomorrow\" → follow-up with context\n", "# Turn 3: Archive to semantic memory, then retrieve\n", "# Turn 4: Crisis input → deterministic safety protocol bypass\n", "# =============================================================================\n", "\n", "print(\"=\" * 65)\n", "print(\" DEMO: Multi-Turn Dialogue — Mental Health Support Agent\")\n", "print(\" Ref: Chapter 10, Section 10.1\")\n", "print(\"=\" * 65)\n", "print()\n", "\n", "# --- Turn 1: Initial anxiety disclosure ---\n", "print(f\"{BOLD}--- Turn 1 ---{RESET}\")\n", "turn1_input = \"I've been feeling really anxious about my exams lately.\"\n", "print(f\" User: {turn1_input}\")\n", "turn1_response = agent.handle_query(turn1_input)\n", "print(f\" Agent: {turn1_response}\")\n", "print()\n", "\n", "# --- Turn 2: Follow-up with specific context ---\n", "print(f\"{BOLD}--- Turn 2 ---{RESET}\")\n", "turn2_input = \"The math exam is tomorrow and I don't feel prepared at all.\"\n", "print(f\" User: {turn2_input}\")\n", "turn2_response = agent.handle_query(turn2_input)\n", "print(f\" Agent: {turn2_response}\")\n", "print()\n", "\n", "# --- Turn 3: Archive to semantic memory and demonstrate retrieval ---\n", "print(f\"{BOLD}--- Turn 3: Semantic Memory Archive & Retrieval ---{RESET}\")\n", "archive_text = (\n", " \"User expressed significant exam anxiety, particularly about a math exam. \"\n", " \"Discussed feelings of being unprepared and time pressure.\"\n", ")\n", "agent.semantic.archive_event(archive_text)\n", "logger.success(\"Archived Turn 1-2 summary to semantic memory.\")\n", "\n", "# Retrieve from semantic memory with a new query\n", "retrieval_query = \"Do you remember what we discussed about my studying?\"\n", "semantic_result = agent.semantic.retrieve_relevant_context(retrieval_query)\n", "print(f\" Retrieval query: '{retrieval_query}'\")\n", "print(f\" Semantic recall: '{semantic_result[:80]}...'\")\n", "print()\n", "\n", "# Continue conversation with memory-aware response\n", "turn3_input = \"Do you remember what I said earlier about my exams?\"\n", "print(f\" User: {turn3_input}\")\n", "turn3_response = agent.handle_query(turn3_input)\n", "print(f\" Agent: {turn3_response}\")\n", "print()\n", "\n", "# --- Turn 4: Crisis input — deterministic bypass ---\n", "print(f\"{BOLD}--- Turn 4: Crisis Detection ---{RESET}\")\n", "turn4_input = \"I feel so overwhelmed I want to hurt myself.\"\n", "print(f\" User: {turn4_input}\")\n", "turn4_response = agent.handle_query(turn4_input)\n", "print(f\" Agent: {turn4_response}\")\n", "print()\n", "\n", "# Verify crisis protocol was triggered\n", "assert \"988\" in turn4_response, \"Crisis protocol should reference 988 hotline\"\n", "logger.success(\"Crisis protocol verified — 988 hotline reference present.\")\n", "\n", "print()\n", "print(\"=\" * 65)\n", "print(\" Section 10.1 Demo Complete\")\n", "print(\"=\" * 65)" ] }, { "cell_type": "markdown", "id": "febbd732", "metadata": {}, "source": [ "---\n", "## Cell Group 3: The Content Creation Agent\n", "*Ref: Chapter 10, Section 10.2 — \"The Content Creation agent\"*\n", "\n", "This section implements the multi-agent content creation pipeline from Section 10.2. The architecture is built on two interlocking frameworks:\n", "\n", "1. **SMPA (Sense-Model-Plan-Act)** — A structured cognitive loop that decomposes generation into discrete stages, preventing monolithic prompt drift.\n", "\n", "2. **CSP (Constraint Satisfaction Problem)** — Brand guidelines modeled as hard constraints. The `EditorAgent` evaluates drafts against the Brand Style Guide using a consistency score, triggering revision loops until all constraints are satisfied.\n", "\n", "The pipeline coordinates three specialist agents (Email, SEO, Ad Creative) under a Campaign Planner, with an `AnalyticsEngine` closing the adaptive feedback loop." ] }, { "cell_type": "markdown", "id": "c749ab16", "metadata": {}, "source": [ "#### Figure 10.2 — Agentic Content Pipeline *(Book p. 295)*\n", "\n", "```\n", " ┌─────────────┐\n", " │ Style Guide │\n", " │ Brand │\n", " │ Constraints │\n", " └──────┬──────┘\n", " │\n", "┌──────────────┐ ┌──────────────┐ ┌─────▼────────┐ ┌──────────────┐\n", "│ Requirements │─▶│ Researcher │─▶│ Writer │─▶│ Editor │\n", "│ Prompt & │ │ Data & RAG │ │ Drafting │ │ QC & │\n", "│ Goals │ │ │ │ (SMPA) │ │ Critique │\n", "└──────────────┘ └──────────────┘ └──────────────┘ └──────┬──────┘\n", " ▲ │\n", " │ Feedback & │\n", " └── Revisions ─────┘\n", " │\n", " ▼\n", " ┌──────────────────┐\n", " │ Final Artifact │\n", " └──────────────────┘\n", "```\n", "\n", "**Without the Editor agent**, drafts containing forbidden terminology or incorrect tone would be published unchanged. The Feedback & Revisions loop triggers iterative refinement until the consistency score $C = \\frac{1}{n} \\sum \\varphi(A_i, G)$ meets the threshold.\n" ] }, { "cell_type": "markdown", "id": "c1cbff76", "metadata": {}, "source": [ "### The SMPA Cognitive Loop\n", "\n", "The SMPA paradigm (from Chapter 1) decomposes the generation process into four discrete stages:\n", "\n", "| Phase | Purpose | Example |\n", "|:------|:--------|:--------|\n", "| **Sense** | Capture prompt requirements, audience constraints, target objectives | Parse campaign brief, identify product and audience |\n", "| **Model** | Build a dynamically updated representation of desired output structure | Define header hierarchies, channel templates |\n", "| **Plan** | Formulate a narrative arc, identify data citations and visual placeholders | Create content calendar, allocate channels |\n", "| **Act** | Execute generation of prose and assets aligned with the plan | Draft email, SEO post, ad copy |\n", "\n", "This separation prevents the model from introducing factual inaccuracies mid-sentence and facilitates human-in-the-loop checkpoints." ] }, { "cell_type": "code", "execution_count": null, "id": "60166daa", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.1 — Agent(ABC): The SMPA Foundation\n", "# Ref: Chapter 10, \"The SMPA foundation\" — (p. 299)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Rather than treating generation as a single event, we define an abstract\n", "# base class that forces every agent to follow a structured cognitive loop.\"\n", "# — Chapter 10\n", "# =============================================================================\n", "\n", "\n", "class Agent(ABC):\n", " \"\"\"\n", " Base class for all agents implementing the SMPA cycle\n", " (Sense-Model-Plan-Act).\n", "\n", " By enforcing this separation of concerns, we prevent the model from\n", " drifting or introducing inaccuracies mid-generation.\n", "\n", " Ref: Chapter 10, \"The SMPA foundation\"; references Chapter 1 — (p. 299)\n", " for the original SMPA paradigm.\n", " \"\"\"\n", "\n", " @abstractmethod\n", " def sense(self, input_data: Dict[str, Any]) -> Any:\n", " \"\"\"Capture prompt requirements and audience constraints.\"\"\"\n", " ...\n", "\n", " @abstractmethod\n", " def model(self, sensed: Any) -> Any:\n", " \"\"\"Build a representation of the desired output structure.\"\"\"\n", " ...\n", "\n", " @abstractmethod\n", " def plan(self, modeled: Any) -> Any:\n", " \"\"\"Formulate a narrative arc and identify asset placeholders.\"\"\"\n", " ...\n", "\n", " @abstractmethod\n", " def act(self, planned: Any) -> Any:\n", " \"\"\"Execute the generation of prose and assets.\"\"\"\n", " ...\n", "\n", " def execute(self, input_data: Dict[str, Any]) -> Any:\n", " \"\"\"\n", " Run the full SMPA cognitive loop in sequence.\n", "\n", " Sense → Model → Plan → Act\n", " \"\"\"\n", " # Sense: Capture prompt requirements and audience constraints\n", " sensed = self.sense(input_data)\n", " # Model: Build a representation of the desired output structure\n", " modeled = self.model(sensed)\n", " # Plan: Formulate a narrative arc and identify asset placeholders\n", " planned = self.plan(modeled)\n", " # Act: Execute the generation of prose and assets\n", " result = self.act(planned)\n", " return result\n", "\n", "\n", "logger.success(\"Agent(ABC) base class defined — SMPA cycle enforced.\")" ] }, { "cell_type": "code", "execution_count": null, "id": "705e3076", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.2 — BrandGuidelines: Brand Constraints as a CSP\n", "# Ref: Chapter 10, \"Implementing brand constraints as a CSP\" — (p. 300)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"Enterprise agents must treat brand guidelines as hard constraints.\n", "# We model this as a logic-based validation step within our BrandGuidelines\n", "# class, to ensure that forbidden terminology (like 'cheap' in a luxury\n", "# context) is never used.\" — Chapter 10\n", "# =============================================================================\n", "\n", "\n", "@dataclass\n", "class BrandGuidelines:\n", " \"\"\"Brand consistency constraints (CSP hard constraints).\"\"\"\n", " tone: str = \"authoritative but friendly\"\n", " forbidden_words: List[str] = field(default_factory=lambda: [\"cheap\", \"free trial\"])\n", "\n", " def validate_content(self, content: str) -> Tuple[bool, List[str]]:\n", " \"\"\"Validate content against brand guidelines.\"\"\"\n", " violations = []\n", " content_lower = content.lower()\n", " for word in self.forbidden_words:\n", " if word.lower() in content_lower:\n", " violations.append(f\"Forbidden word found: '{word}'\")\n", " return len(violations) == 0, violations\n", "\n", "\n", "# --- Demo: BrandGuidelines ---\n", "brand = BrandGuidelines()\n", "logger.success(f\"BrandGuidelines initialized: tone='{brand.tone}', \"\n", " f\"forbidden_words={brand.forbidden_words}\")\n", "\n", "# Test 1: Clean content\n", "clean_text = \"DataVault Pro delivers enterprise-grade data governance.\"\n", "passed, violations = brand.validate_content(clean_text)\n", "logger.success(f\"Clean text: passed={passed}, violations={violations}\")\n", "\n", "# Test 2: Violating content\n", "dirty_text = \"DataVault Pro is the cheap alternative with a free trial offer.\"\n", "passed, violations = brand.validate_content(dirty_text)\n", "logger.warning(f\"Violating text: passed={passed}, violations={violations}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "4083b91b", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.3 — validate_against_brand(): Editor Pre-Check\n", "# Ref: Chapter 10, \"Brand consistency as a constraint satisfaction problem\" — (pp. 294–296)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# This standalone function demonstrates the Editor Agent's constraint\n", "# checking logic before it is integrated into the full pipeline.\n", "# The chapter's example draft deliberately violates three brand terms.\n", "# =============================================================================\n", "\n", "BRAND_GUIDELINES = {\n", " \"forbidden_terms\": [\"cheaper\", \"best-in-class\", \"industry-leading\"],\n", " \"required_tone\": \"authoritative-but-approachable\",\n", "}\n", "\n", "\n", "def validate_against_brand(draft: str, guidelines: dict) -> dict:\n", " \"\"\"Editor Agent: check draft against brand constraint set.\"\"\"\n", " violations = [\n", " term for term in guidelines[\"forbidden_terms\"]\n", " if term.lower() in draft.lower()\n", " ]\n", " return {\n", " \"passed\": len(violations) == 0,\n", " \"violations\": violations,\n", " \"revision_instruction\": (\n", " f\"Remove or replace: {violations}. \"\n", " f\"Tone must be: {guidelines['required_tone']}.\"\n", " ) if violations else \"\",\n", " }\n", "\n", "\n", "# --- Demo: Chapter's example violating draft ---\n", "raw_draft = (\n", " \"Our platform is the industry-leading, cheaper alternative to \"\n", " \"legacy solutions. Best-in-class performance guaranteed.\"\n", ")\n", "\n", "result = validate_against_brand(raw_draft, BRAND_GUIDELINES)\n", "\n", "print(f\"Draft: '{raw_draft}'\")\n", "print()\n", "logger.warning(f\"Passed: {result['passed']}\")\n", "logger.warning(f\"Violations: {result['violations']}\")\n", "logger.warning(f\"Revision instruction: {result['revision_instruction']}\")\n", "\n", "# Verify: chapter states 3 violations\n", "assert result[\"passed\"] == False, \"Should have failed validation\"\n", "assert len(result[\"violations\"]) == 3, (\n", " f\"Expected 3 violations (cheaper, industry-leading, Best-in-class), \"\n", " f\"got {len(result['violations'])}: {result['violations']}\"\n", ")\n", "logger.success(\"validate_against_brand() correctly detected all 3 violations.\")" ] }, { "cell_type": "code", "execution_count": null, "id": "719d7d56", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.4 — EditorAgent: The Quality Control Layer\n", "# Ref: Chapter 10, \"The editor agent: The quality control layer\" — (p. 300)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Consistency score formula from the chapter:\n", "# C = 1/n × Σ φ(Aᵢ, G)\n", "# where:\n", "# n = number of brand parameters\n", "# Aᵢ = attributes of the generated artifact\n", "# G = definitive brand guidelines\n", "# φ = alignment function measuring degree of adherence\n", "# =============================================================================\n", "\n", "\n", "@dataclass\n", "class ContentArtifact:\n", " \"\"\"Container for a generated content draft with metadata.\"\"\"\n", " channel: str = \"\"\n", " body: str = \"\"\n", " tone_score: float = 0.8\n", "\n", "\n", "class EditorAgent:\n", " \"\"\"\n", " Implements the consistency score: C = 1/n × Σ φ(Aᵢ, G)\n", "\n", " The editor does not merely flag errors — it calculates a multi-dimensional\n", " consistency score based on alignment of the generated artifact with the\n", " brand guidelines.\n", "\n", " Ref: Chapter 10, \"The editor agent: The quality control layer\" — (p. 300)\n", " \"\"\"\n", "\n", " def _calculate_consistency_score(\n", " self,\n", " artifact: ContentArtifact,\n", " violations: List[str],\n", " tone_score: float,\n", " ) -> float:\n", " \"\"\"\n", " Calculate the brand consistency score.\n", "\n", " C = 1/n × Σ φ(Aᵢ, G)\n", " where n=3 brand parameters:\n", " - forbidden_score: 1.0 if no violations, 0.0 otherwise\n", " - tone_score: provided externally (0.0-1.0)\n", " - structure_score: 1.0 if body > 100 chars, 0.5 otherwise\n", " \"\"\"\n", " n = 3 # number of brand parameters\n", " forbidden_score = 1.0 if len(violations) == 0 else 0.0\n", " structure_score = 1.0 if len(artifact.body) > 100 else 0.5\n", " # φ is our alignment function measuring adherence\n", " consistency_score = (forbidden_score + tone_score + structure_score) / n\n", " return consistency_score\n", "\n", " def evaluate(\n", " self,\n", " artifact: ContentArtifact,\n", " guidelines: dict,\n", " ) -> Dict[str, Any]:\n", " \"\"\"Full editor evaluation: constraint check + consistency score.\"\"\"\n", " violations = [\n", " t for t in guidelines.get(\"forbidden_terms\", [])\n", " if t.lower() in artifact.body.lower()\n", " ]\n", " score = self._calculate_consistency_score(\n", " artifact, violations, artifact.tone_score\n", " )\n", " return {\n", " \"passed\": len(violations) == 0,\n", " \"violations\": violations,\n", " \"consistency_score\": round(score, 3),\n", " }\n", "\n", "\n", "# --- Demo: EditorAgent ---\n", "editor = EditorAgent()\n", "\n", "# Test 1: Clean artifact (long body, no violations, good tone)\n", "clean_artifact = ContentArtifact(\n", " channel=\"email\",\n", " body=\"DataVault Pro delivers enterprise-grade data governance with automated \"\n", " \"lineage tracking, role-based access controls, and real-time compliance \"\n", " \"monitoring for modern data teams.\",\n", " tone_score=0.9,\n", ")\n", "eval_clean = editor.evaluate(clean_artifact, BRAND_GUIDELINES)\n", "logger.success(f\"Clean artifact: score={eval_clean['consistency_score']}, \"\n", " f\"passed={eval_clean['passed']}\")\n", "\n", "# Test 2: Violating artifact\n", "dirty_artifact = ContentArtifact(\n", " channel=\"ad\",\n", " body=\"Our industry-leading platform is the cheaper alternative.\",\n", " tone_score=0.4,\n", ")\n", "eval_dirty = editor.evaluate(dirty_artifact, BRAND_GUIDELINES)\n", "logger.warning(f\"Violating artifact: score={eval_dirty['consistency_score']}, \"\n", " f\"violations={eval_dirty['violations']}\")\n", "\n", "print()\n", "print(\"Consistency Score Formula: C = 1/n × Σ φ(Aᵢ, G)\")\n", "print(f\" Clean artifact: C = {eval_clean['consistency_score']} \"\n", " f\"(forbidden=1.0, tone=0.9, structure=1.0) / 3\")\n", "print(f\" Violating artifact: C = {eval_dirty['consistency_score']} \"\n", " f\"(forbidden=0.0, tone=0.4, structure=0.5) / 3\")" ] }, { "cell_type": "code", "execution_count": null, "id": "d206af3d", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.5 — AssetRequest & dispatch_asset_request()\n", "# Ref: Chapter 10, \"Multimodal orchestration via function calling\" — (p. 301)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# \"The text-generating agent never calls the image API directly; it emits\n", "# a structured AssetRequest, and the dispatcher handles routing. This\n", "# separation keeps the generation agents testable and the multimodal\n", "# backend swappable.\" — Chapter 10\n", "# =============================================================================\n", "\n", "\n", "@dataclass\n", "class AssetRequest:\n", " \"\"\"Structured request for multimodal asset generation.\"\"\"\n", " asset_type: str # e.g., 'image' or 'chart'\n", " asset_id: str\n", " prompt: str\n", " constraints: Dict[str, Any] # e.g., 'aspect_ratio': '16:9'\n", "\n", "\n", "@fail_gracefully(\n", " fallback_value={\"asset_id\": \"mock\", \"url\": \"https://placehold.co/1024x1024\"},\n", " section=\"10.2\",\n", ")\n", "def dispatch_asset_request(req: AssetRequest) -> Dict[str, Any]:\n", " \"\"\"\n", " Route an AssetRequest to the appropriate generation backend.\n", "\n", " In LIVE MODE: calls OpenAI Images API (DALL-E 3).\n", " In SIMULATION MODE: @fail_gracefully returns the mock fallback URL.\n", "\n", " The pattern is deliberately minimal: one function, one API call, one\n", " return contract. In production, teams wrap this dispatcher with retry\n", " logic, cost tracking, and content-policy filtering.\n", " \"\"\"\n", " if SIMULATION_MODE:\n", " # Return mock directly in simulation — no API call needed\n", " logger.info(f\"[Simulation] dispatch_asset_request: '{req.asset_id}' → placeholder\")\n", " return {\"asset_id\": req.asset_id, \"url\": \"https://placehold.co/1024x1024\"}\n", "\n", " import openai\n", " client = openai.OpenAI() # uses OPENAI_API_KEY from env\n", " size = req.constraints.get(\"size\", \"1024x1024\")\n", " response = client.images.generate(\n", " model=\"dall-e-3\",\n", " prompt=req.prompt,\n", " size=size,\n", " n=1,\n", " )\n", " return {\"asset_id\": req.asset_id, \"url\": response.data[0].url}\n", "\n", "\n", "# --- Demo: dispatch_asset_request ---\n", "test_request = AssetRequest(\n", " asset_type=\"image\",\n", " asset_id=\"hero_banner_001\",\n", " prompt=\"Modern data governance dashboard with blue accent lighting\",\n", " constraints={\"size\": \"1024x1024\", \"aspect_ratio\": \"1:1\"},\n", ")\n", "\n", "result = dispatch_asset_request(test_request)\n", "print(f\"\\nAsset dispatch result:\")\n", "print(f\" asset_id: {result['asset_id']}\")\n", "print(f\" url: {result['url']}\")\n", "logger.success(f\"dispatch_asset_request: returned URL for '{result['asset_id']}'\")" ] }, { "cell_type": "markdown", "id": "63628142", "metadata": {}, "source": [ "### The Reinforcement Learning Formulation\n", "\n", "The adaptive optimization cycle can be modeled as an RL problem. The agent seeks to maximize the expected reward over a sequence of content iterations:\n", "\n", "$$J(\\theta) = E_{\\pi_\\theta} \\left[ \\sum_{t=0}^{T} \\gamma^t R_t \\right]$$\n", "\n", "Where:\n", "- **$R_t$** = engagement metrics (e.g., CTR, conversion rate) at iteration $t$\n", "- **$\\gamma$** = discount factor for long-term brand equity\n", "- **$\\pi_\\theta$** = agent's generation policy parameterized by $\\theta$\n", "\n", "When an experiment reveals that a specific tonal variant or visual treatment consistently outperforms others, the agent adjusts its internal preferences. Over time, the system transitions from a static drafting capability into an **adaptive optimization cycle**." ] }, { "cell_type": "markdown", "id": "f02e112c", "metadata": {}, "source": [ "#### Figure 10.3 — Marketing Content Assistant Architecture *(Book p. 298)*\n", "\n", "```\n", " ┌──────────────────┐\n", " │ Email Agent │\n", " │ Subject Line Opt.│\n", " └────────┬─────────┘\n", " │\n", "┌─────────────────┐ ┌────────────────┐│ ┌──────────────┐ ┌──────────────────┐\n", "│ Strategic Intent │─▶│Campaign Planner│├▶│ SEO Agent │─▶│ Analytics Engine │\n", "│ Constraints & │ │(SMPA ││ │Keyword Context│ │ Engagement KPIs │\n", "│ Persona │ │ Orchestrator) │┘ └──────────────┘ └──────────────────┘\n", "└─────────────────┘ └───────┬────────┘ │\n", " │ │\n", " ▼ │\n", " ┌──────────────────┐ │\n", " │ Ad Creative Agent │ │\n", " │Multimodal Synth. │ │\n", " └──────────────────┘ │\n", " │ │\n", " └──── Adaptive Feedback Loop ─────────┘\n", " (Performance Signals)\n", "```\n", "\n", "**Adaptive optimization:** The agent treats performance signals (CTR, conversion depth) as feedback that updates future generative parameters, modeled as: $J(\\theta) = E_{\\pi_\\theta}\\left[\\sum_{t=0}^{T} \\gamma^t R_t\\right]$\n" ] }, { "cell_type": "code", "execution_count": null, "id": "5e0e486a", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.6 — AnalyticsEngine: The Adaptive Optimization Cycle\n", "# Ref: Chapter 10, \"The adaptive optimization cycle\" — (pp. 298, 302)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# The RL optimization objective from the chapter:\n", "# J(θ) = E_πθ [ Σ γᵗ Rₜ ]\n", "# where:\n", "# Rₜ = engagement metrics (e.g., CTR) at time t\n", "# γ = discount factor for long-term brand equity\n", "# πθ = agent's generation policy parameterized by θ\n", "# =============================================================================\n", "\n", "\n", "class AnalyticsEngine:\n", " \"\"\"\n", " Generates feedback for Campaign Planner based on performance.\n", "\n", " Transforms the content agent from a static writer into a self-optimizing\n", " engine that learns which creative patterns resonate with its audience.\n", "\n", " Ref: Chapter 10, \"The adaptive optimization cycle\" — (pp. 298, 302)\n", " \"\"\"\n", "\n", " def generate_adaptive_feedback(\n", " self, campaign_analysis: Dict[str, Any]\n", " ) -> Dict[str, Any]:\n", " \"\"\"\n", " Generate strategic recommendations based on CTR threshold.\n", "\n", " If overall CTR > 0.05: maintain current strategy.\n", " If overall CTR ≤ 0.05: trigger adaptive feedback loop.\n", " \"\"\"\n", " ctr = campaign_analysis[\"overall_ctr\"]\n", " if ctr > 0.05:\n", " return {\"recommendations\": [\"Maintain current creative strategy\"]}\n", " else:\n", " # Triggering the adaptive feedback loop\n", " return {\"recommendations\": [\"Revise value proposition messaging\"]}\n", "\n", "\n", "# --- Demo: AnalyticsEngine ---\n", "analytics = AnalyticsEngine()\n", "\n", "# Scenario 1: High CTR\n", "high_ctr = analytics.generate_adaptive_feedback({\"overall_ctr\": 0.08})\n", "logger.success(f\"High CTR (0.08): {high_ctr['recommendations']}\")\n", "\n", "# Scenario 2: Low CTR\n", "low_ctr = analytics.generate_adaptive_feedback({\"overall_ctr\": 0.03})\n", "logger.warning(f\"Low CTR (0.03): {low_ctr['recommendations']}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "14e45c85", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.7 — Campaign Infrastructure: CampaignBrief, CampaignAssets,\n", "# _draft(), _validate(), _validated_draft()\n", "# Ref: Chapter 10, \"End-to-end campaign walkthrough: execute_campaign()\" — (pp. 302–305)\n", "# Author: Imran Ahmad | Chapter 10\n", "# =============================================================================\n", "\n", "\n", "@dataclass\n", "class CampaignBrief:\n", " \"\"\"Strategic input for the Campaign Planner.\"\"\"\n", " product: str\n", " audience: str\n", " tone: str = \"authoritative-but-approachable\"\n", " forbidden_terms: List[str] = field(\n", " default_factory=lambda: [\"cheaper\", \"best-in-class\", \"industry-leading\"]\n", " )\n", "\n", "\n", "@dataclass\n", "class CampaignAssets:\n", " \"\"\"Output container for all campaign deliverables.\"\"\"\n", " email: str = \"\"\n", " seo_post: str = \"\"\n", " ad_copy: str = \"\"\n", " analytics: Dict[str, float] = field(default_factory=dict)\n", "\n", "\n", "def _draft(role: str, brief: CampaignBrief, channel: str) -> str:\n", " \"\"\"\n", " Single-channel Writer Agent using SMPA Act phase.\n", "\n", " Constructs a system prompt encoding the role, tone, and forbidden terms,\n", " then invokes the LLM to generate channel-specific content.\n", " \"\"\"\n", " messages = [\n", " SystemMessage(content=(\n", " f\"You are a {role} writing {channel} copy. \"\n", " f\"Tone: {brief.tone}. \"\n", " f\"Never use: {brief.forbidden_terms}.\"\n", " )),\n", " HumanMessage(content=(\n", " f\"Write {channel} content for: {brief.product}. \"\n", " f\"Audience: {brief.audience}.\"\n", " )),\n", " ]\n", " return llm.invoke(messages).content\n", "\n", "\n", "def _validate(draft: str, brief: CampaignBrief) -> Dict[str, Any]:\n", " \"\"\"Editor Agent: enforce brand constraints (CSP hard check).\"\"\"\n", " violations = [\n", " t for t in brief.forbidden_terms if t.lower() in draft.lower()\n", " ]\n", " return {\n", " \"passed\": len(violations) == 0,\n", " \"violations\": violations,\n", " \"instruction\": (\n", " f\"Revise to remove: {violations}. Tone must be: {brief.tone}.\"\n", " if violations else \"\"\n", " ),\n", " }\n", "\n", "\n", "def _validated_draft(\n", " role: str, brief: CampaignBrief, channel: str, max_retries: int = 2\n", ") -> str:\n", " \"\"\"\n", " Writer + Editor feedback loop: retry until brand constraints pass.\n", "\n", " The Editor returns a revision_instruction; the Writer retries with\n", " that instruction appended. This process continues until the draft\n", " satisfies the brand constraints or max_retries is exhausted.\n", " \"\"\"\n", " draft = _draft(role, brief, channel)\n", "\n", " for attempt in range(max_retries):\n", " result = _validate(draft, brief)\n", " if result[\"passed\"]:\n", " logger.success(f\"{channel}: passed brand validation.\")\n", " return draft\n", "\n", " logger.warning(\n", " f\"Brand violation in {channel} draft (attempt {attempt + 1}): \"\n", " f\"{result['violations']}. Triggering Editor retry.\"\n", " )\n", " # Editor returns revision_instruction; Writer retries\n", " messages = [\n", " SystemMessage(content=result[\"instruction\"]),\n", " HumanMessage(content=f\"Revise this {channel} draft:\\n{draft}\"),\n", " ]\n", " draft = llm.invoke(messages).content\n", "\n", " logger.info(f\"{channel}: returning best effort after {max_retries} retries.\")\n", " return draft # return best effort after max_retries\n", "\n", "\n", "logger.success(\"Campaign infrastructure defined: CampaignBrief, CampaignAssets, \"\n", " \"_draft, _validate, _validated_draft.\")" ] }, { "cell_type": "markdown", "id": "373e3c24", "metadata": {}, "source": [ "> **📘 Implementation Insight: When to Use a Full Agent Chain** *(Book p. 302)*\n", ">\n", "> Before committing to a multi-agent pipeline, engineers should evaluate whether a single LLM call would suffice. Use a full agent chain when the task requires:\n", "> - **Brand-constraint enforcement** across more than one output type\n", "> - **Coordination of three or more** distinct specialized roles (researcher, writer, editor)\n", "> - **Analytics-driven feedback** that must update future generation\n", ">\n", "> Use a simple chatbot when the output is single-pass, the quality bar is conversational rather than publication-grade, and no external data retrieval is required.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "3b112881", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.8 — execute_campaign(): The Campaign Planner Coordinator\n", "# Ref: Chapter 10, \"End-to-end campaign walkthrough: execute_campaign()\" — (pp. 302–305)\n", "# Author: Imran Ahmad | Chapter 10\n", "#\n", "# Three-phase structure:\n", "# Phase 1 — Dispatch: Email, SEO, Ad agents produce validated drafts\n", "# Phase 2 — Analytics: record mock engagement signals\n", "# Phase 3 — Feedback: update Planner preferences for next run\n", "# =============================================================================\n", "\n", "\n", "@fail_gracefully(\n", " fallback_value=CampaignAssets(\n", " email=\"[Mock email — see Section 10.2]\",\n", " seo_post=\"[Mock SEO post — see Section 10.2]\",\n", " ad_copy=\"[Mock ad copy — see Section 10.2]\",\n", " analytics={\"email_open_rate\": 0.28, \"seo_click_rate\": 0.14, \"ad_conversion\": 0.04},\n", " ),\n", " section=\"10.2\",\n", ")\n", "def execute_campaign(brief: CampaignBrief) -> CampaignAssets:\n", " \"\"\"\n", " Campaign Planner: coordinate specialist agents for a product launch.\n", "\n", " Phase 1 — Dispatch: Email, SEO, Ad agents produce validated drafts.\n", " Phase 2 — Analytics: record mock engagement signals.\n", " Phase 3 — Feedback: update Planner preferences for next run.\n", " \"\"\"\n", " assets = CampaignAssets()\n", "\n", " # Phase 1 — Dispatch specialist agents with Editor validation\n", " logger.info(\"Phase 1: Dispatching specialist agents...\")\n", "\n", " logger.info(\"Dispatching Email Agent...\")\n", " assets.email = _validated_draft(\"email specialist\", brief, \"email newsletter\")\n", "\n", " logger.info(\"Dispatching SEO Agent...\")\n", " assets.seo_post = _validated_draft(\"SEO writer\", brief, \"long-form blog post\")\n", "\n", " logger.info(\"Dispatching Ad Creative Agent...\")\n", " assets.ad_copy = _validated_draft(\"ad creative\", brief, \"display ad copy\")\n", "\n", " # Phase 2 — Analytics Engine: record performance signals (mock)\n", " logger.info(\"Phase 2: Recording analytics signals...\")\n", " assets.analytics = {\n", " \"email_open_rate\": 0.28, # 28% open rate\n", " \"seo_click_rate\": 0.14, # 14% CTR on search impression\n", " \"ad_conversion\": 0.04, # 4% conversion from ad click\n", " }\n", "\n", " # Phase 3 — Adaptive feedback: surfaces low-performer for next iteration\n", " logger.info(\"Phase 3: Generating adaptive feedback...\")\n", " lowest = min(assets.analytics, key=assets.analytics.get)\n", " logger.info(f\"[Planner] Weakest channel: {lowest} — schedule A/B revision.\")\n", "\n", " return assets\n", "\n", "\n", "logger.success(\"execute_campaign() defined — ready for full pipeline execution.\")" ] }, { "cell_type": "markdown", "id": "1b1776b1", "metadata": {}, "source": [ "> **📘 Implementation Insight: When to Use a Full Agent Chain** *(Book p. 302)*\n", ">\n", "> Before committing to a multi-agent pipeline, engineers should evaluate whether a single LLM call would suffice. Use a full agent chain when the task requires:\n", "> - **Brand-constraint enforcement** across more than one output type\n", "> - **Coordination of three or more** distinct specialized roles (researcher, writer, editor)\n", "> - **Analytics-driven feedback** that must update future generation\n", ">\n", "> Use a simple chatbot when the output is single-pass, the quality bar is conversational rather than publication-grade, and no external data retrieval is required.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "49d73782", "metadata": {}, "outputs": [], "source": [ "# =============================================================================\n", "# Cell 3.9 — Demo: Full Campaign Execution with DataVault Pro\n", "# Ref: Chapter 10, Full Section 10.2 integration — (pp. 293–305)\n", "# Author: Imran Ahmad | Chapter 10\n", "# =============================================================================\n", "\n", "print(\"=\" * 65)\n", "print(\" DEMO: Full Campaign Execution — Marketing Content Assistant\")\n", "print(\" Ref: Chapter 10, Section 10.2\")\n", "print(\"=\" * 65)\n", "print()\n", "\n", "# --- Create campaign brief (from chapter) ---\n", "brief = CampaignBrief(\n", " product=\"DataVault Pro — enterprise data-governance platform\",\n", " audience=\"CTOs and data engineering leads at mid-market companies\",\n", ")\n", "\n", "logger.info(f\"Campaign brief: product='{brief.product}'\")\n", "logger.info(f\" audience='{brief.audience}'\")\n", "logger.info(f\" tone='{brief.tone}'\")\n", "logger.info(f\" forbidden_terms={brief.forbidden_terms}\")\n", "print()\n", "\n", "# --- Execute the full campaign pipeline ---\n", "campaign = execute_campaign(brief)\n", "\n", "# --- Display results ---\n", "print()\n", "print(f\"{BOLD}--- Email Draft (first 120 chars) ---{RESET}\")\n", "print(f\" {campaign.email[:120]}...\")\n", "print()\n", "\n", "print(f\"{BOLD}--- SEO Post (first 120 chars) ---{RESET}\")\n", "print(f\" {campaign.seo_post[:120]}...\")\n", "print()\n", "\n", "print(f\"{BOLD}--- Ad Copy ---{RESET}\")\n", "print(f\" {campaign.ad_copy[:120]}...\")\n", "print()\n", "\n", "print(f\"{BOLD}--- Analytics ---{RESET}\")\n", "for metric, value in campaign.analytics.items():\n", " print(f\" {metric}: {value}\")\n", "print()\n", "\n", "# --- Adaptive feedback ---\n", "analytics_engine = AnalyticsEngine()\n", "overall_ctr = sum(campaign.analytics.values()) / len(campaign.analytics)\n", "feedback = analytics_engine.generate_adaptive_feedback({\"overall_ctr\": overall_ctr})\n", "logger.info(f\"Overall CTR: {overall_ctr:.3f}\")\n", "logger.success(f\"Adaptive feedback: {feedback['recommendations']}\")\n", "\n", "# --- Weakest channel identification ---\n", "weakest = min(campaign.analytics, key=campaign.analytics.get)\n", "logger.info(f\"[Planner] Weakest channel: {weakest} — schedule A/B revision.\")\n", "\n", "print()\n", "print(\"=\" * 65)\n", "print(\" Section 10.2 Demo Complete\")\n", "print(\"=\" * 65)" ] }, { "cell_type": "markdown", "id": "430e300a", "metadata": {}, "source": [ "---\n", "## Cell Group 4: Summary & Reflection\n", "*Ref: Chapter 10 — \"Summary\"*" ] }, { "cell_type": "markdown", "id": "3c5ccdab", "metadata": {}, "source": [ "### Key Architectural Takeaways\n", "\n", "This chapter examined the class of agents that operate at the boundary between algorithmic capability and human experience. The core engineering challenges — sustaining coherent behavior across long interactions, enforcing safety and brand constraints deterministically, and coordinating multiple generation stages — require architectures that go far beyond simple request-response loops.\n", "\n", "**Section 10.1 — The Conversational Agent:**\n", "\n", "- **Dual-Memory Hierarchy:** Working memory (`ConversationSummaryBufferMemory`, max 300 tokens) provides low-latency recency-based context, while semantic memory (FAISS vector store) enables long-term recall by similarity. This separation mirrors the distinction between RAM and disk in traditional systems — different latency, retention, and access patterns for different purposes.\n", "\n", "- **Safety-First Pipeline:** The `SafetyLayer` sentinel sits upstream of all generative logic. Crisis detection is deterministic (keyword-based), not probabilistic — generation is downstream of policy. This ensures that creative text generation can never accidentally occur when the situation demands strict escalation behavior.\n", "\n", "- **Persona as Constraint, Not Randomness:** The `PersonaEngine` reshapes the model's output distribution toward a stable behavioral region. \"Personality is not randomness; it is controlled bias.\" The system prompt encodes immutable behavioral rules — what the agent is, how uncertainty is handled, and what boundaries must not be crossed.\n", "\n", "- **Vertical Pipeline Ordering:** Every request follows a strict sequence: Safety → Context → Generation → Memory Update. This ordering is not a suggestion; it is an architectural invariant that prevents policy violations.\n", "\n", "**Section 10.2 — The Content Creation Agent:**\n", "\n", "- **SMPA Decomposition:** The Sense-Model-Plan-Act cycle (from Chapter 1) decomposes generation into discrete stages. This prevents monolithic prompt drift and enables human-in-the-loop checkpoints at each stage boundary.\n", "\n", "- **CSP Brand Enforcement:** Brand guidelines are modeled as hard constraints in a Constraint Satisfaction Problem. The `EditorAgent` calculates a multi-dimensional consistency score (C = 1/n × Σ φ(Aᵢ, G)) and triggers Writer retry loops until all constraints are satisfied. This is not optional review — it is a pipeline gate.\n", "\n", "- **Multi-Agent Orchestration:** Specialist agents (Email, SEO, Ad Creative) operate independently but under shared brand contracts. The Campaign Planner coordinates dispatch, validation, and feedback aggregation.\n", "\n", "- **Adaptive Feedback Loop:** The `AnalyticsEngine` closes the optimization cycle by treating real-world engagement metrics (CTR, conversion) as feedback signals. This transforms static content generation into a self-optimizing process modeled as J(θ) = E_πθ[Σ γᵗRₜ].\n", "\n", "**Cross-Cutting Patterns:**\n", "\n", "- **`@fail_gracefully` Decorator:** Every external call is wrapped with resilient error handling. Failures produce diagnostic logs and return mock fallbacks — the notebook never halts. This is essential for both educational stability and production resilience.\n", "\n", "- **Simulation Mode:** The `MockChatOpenAI` and `MockOpenAIEmbeddings` classes enable full notebook execution without API keys. Context-aware dispatch ensures mock responses are educationally meaningful, not generic placeholders." ] }, { "cell_type": "markdown", "id": "7ad7af1e", "metadata": {}, "source": [ "### Next Steps\n", "\n", "In the next chapter, we move from agents that engage humans through dialogue and published content to agents that operate within more specialized professional domains. The architectural patterns introduced here — persona modeling, constraint-layer enforcement, and analytics-driven feedback loops — transfer directly to settings where the success criteria and risk profiles demand different design trade-offs.\n", "\n", "**Suggested Extension Exercises:**\n", "\n", "1. **Extend the Memory Hierarchy:** Add a third memory tier (e.g., Redis-backed persistent store) and implement a memory consolidation policy that migrates high-salience entries from working memory to long-term storage based on emotional weight.\n", "\n", "2. **Add A/B Testing to the Campaign Pipeline:** Modify `execute_campaign()` to generate two variants per channel and select the winner based on simulated CTR scores from the `AnalyticsEngine`.\n", "\n", "3. **Implement Dynamic Persona Modulation:** Extend the `PersonaEngine` to support context-dependent persona shifts (e.g., warmer tone in exploratory phases, more concise tone during task execution) governed by a policy layer rather than ad hoc prompt changes.\n", "\n", "4. **Build a Compliance Audit Trail:** Add structured logging that captures every Editor evaluation, brand violation, and retry decision in a format suitable for regulatory review.\n", "\n", "---\n", "\n", "*Chapter 10 — Conversational and Content Creation Agents*\n", "*Book: 30 Agents Every AI Engineer Must Build*\n", "*Author: Imran Ahmad | 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 }