Module 26: Callbacks and Guardrails - Building a Content Moderator

Lab 26: Building a Content Moderation Assistant with Caching

Goal

In this lab, you will implement a suite of callbacks to create a Content Moderation Assistant. You will learn to build safety guardrails, validate tool arguments, filter responses, and, crucially, implement a Caching mechanism using before_agent_callback to save tokens and time.

Step 1: Create the Project Structure

Environment setup:

• Local machine: If you're continuing the course in sequence, cd into the adk-training project you created in Module 2 and skip the block below -- your environment and dependencies already carry forward. If you're starting fresh at this module, create that project first:

  uv init adk-training --python 3.10 && cd adk-training
  uv add "google-adk>=2.1.0" python-dotenv

• DevContainer / Codespaces (Alternative): If you opened this repository in a DevContainer or GitHub Codespaces, Python, uv, and the core ADK dependencies are already pre-installed in your container environment.

1. Create the agent project:

   uv run adk create content_moderator

2. Navigate into the new directory:

   cd content_moderator

Step 2: Implement the Callbacks

Exercise: Open agent.py. Your task is to implement the logic for the six core callbacks. Use the # TODO comments as your guide.

import hashlib
import os
import re
import logging
from typing import Dict, Any, Optional
from google.adk import Agent
from google.adk.agents.callback_context import CallbackContext
from google.adk.tools import ToolContext
from google.adk.tools.base_tool import BaseTool
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types
from dotenv import load_dotenv

load_dotenv()

# --- Configuration ---
BLOCKED_WORDS = ['unsafe', 'offensive']

# ============================================================================
# CALLBACK FUNCTIONS
# ============================================================================

def _cache_key(callback_context: CallbackContext) -> str:
    """
    TODO: Helper.
    A cache keyed by a single global name (e.g. 'cached_response') would
    return the SAME cached answer no matter what the user asks next — you
    must derive the key from the CURRENT turn's user input instead.
    Use callback_context.get_invocation_context().user_content to get the
    current user message (a types.Content), concatenate its parts' text,
    and hash it (e.g. hashlib.md5(...).hexdigest()) into a key such as
    f"cache:{digest}".
    """
    pass

def before_agent_callback(callback_context: CallbackContext) -> Optional[types.Content]:
    """
    TODO: Caching (Check).
    Compute this turn's cache key with _cache_key() and read it from
    callback_context.state. If present, print a cache-hit message and
    return a types.Content wrapping it (role="model") to skip the LLM
    entirely. Otherwise return None.
    """
    pass

def after_agent_callback(callback_context: CallbackContext) -> None:
    """
    TODO: Caching (Save).
    Walk callback_context.session.events in reverse to find the last
    non-user event with content, and save its text into
    callback_context.state[_cache_key(callback_context)] for future reuse.
    Using the same per-input key as before_agent_callback is essential —
    otherwise a later, unrelated question would incorrectly hit the cache
    entry saved for a previous, different question.
    Guard against event.content.parts being empty before indexing
    parts[0] — a content object with no parts would otherwise raise an
    IndexError.
    """
    pass

def before_model_callback(
    callback_context: CallbackContext,
    llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """
    TODO: Input Guardrail.
    Check ONLY the current turn's user input against BLOCKED_WORDS — do
    NOT scan the full llm_request.contents history. llm_request.contents
    holds the ENTIRE conversation so far, and once a blocked word appears
    anywhere in it, every later turn would keep matching and get refused
    forever, even completely unrelated questions. Mirror the pattern
    _cache_key() uses: pull the current message via
    callback_context.get_invocation_context().user_content, concatenate
    its parts' text, and check only that against BLOCKED_WORDS. If a
    blocked word is found, print a warning and return an LlmResponse with
    a refusal message instead of calling the model. Otherwise return None.
    """
    pass

def after_model_callback(
    callback_context: CallbackContext,
    llm_response: LlmResponse
) -> Optional[LlmResponse]:
    """
    TODO: Output Filtering.
    Use re.sub to redact email addresses from the LLM response text.
    If redacted, return a new LlmResponse; otherwise return None.
    Note: llm_response.content.parts[0].text can be None — the model's
    response may be a pure function call (e.g. deciding to invoke
    generate_text) with no text to filter. Guard against that first.
    """
    pass

def before_tool_callback(
    tool: BaseTool,
    args: Dict[str, Any],
    tool_context: ToolContext
) -> Optional[Dict[str, Any]]:
    """
    TODO: Tool Validation.
    If tool.name == 'generate_text' and args['word_count'] exceeds 5000,
    print a warning and return an error dict (e.g. {'status': 'error',
    'message': '...'}) to block execution. Otherwise return None.
    """
    pass

def after_tool_callback(
    tool: BaseTool,
    args: Dict[str, Any],
    tool_context: ToolContext,
    tool_response: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
    """
    TODO: Output Audit.
    If tool.name == 'generate_text', check tool_response['text'] against
    BLOCKED_WORDS (defense-in-depth, in case the model injects one via
    arguments). If found, print a warning and return a modified copy of
    tool_response with the blocked words replaced by '***'. Otherwise,
    print an audit log line (e.g. tool name + status) and return None.
    """
    pass

# --- Tools ---
def generate_text(topic: str, word_count: int) -> dict:
    """Generates text on a topic."""
    return {"status": "success", "text": f"A {word_count}-word essay on {topic}..."}

# ============================================================================
# AGENT DEFINITION
# ============================================================================

# TODO: Define root_agent and register ALL six callbacks defined above.
root_agent = Agent(
    name="secure_moderator",
    model="gemini-3.5-flash",
    # instruction=...,
    # tools=[...],
    # before_agent_callback=...,
    # after_agent_callback=...,
    # before_model_callback=...,
    # after_model_callback=...,
    # before_tool_callback=...,
    # after_tool_callback=...,
)

Step 3: Run and Test

Start the agent and try both paths: a normal prompt (should reach the model), and a prompt containing a blocked word like "unsafe" (should be refused by before_model_callback without ever calling the model). Then repeat the same prompt in the same session — before_agent_callback should return the cached response instantly. Finally, ask a different question in the same session — this should NOT return the previous answer; it should call the LLM again and get a fresh, correct response. This third check is what actually proves the cache is keyed per-question rather than one global slot for the whole session.

uv run adk run content_moderator

Self-Reflection Questions

• What is the key difference between a callback and a plugin in the ADK? When would you choose one over the other?
• Why does returning a types.Content object from before_agent_callback cause the agent to skip the LLM call entirely?
• How does using callbacks for guardrails and validation make an agent more reliable and safer to deploy in a production environment?

---

🔐 Reference Solution Vault

🔐

Module 26 Reference Solution Vault

Active Comprehension Gate • Zero Guessable URLs • Guided Verification

Step 1/3: Question 1 of 3

In Module 26 (Callbacks), what is the primary role of before/after turn hooks in ADK architectures?