> ## Documentation Index
> Fetch the complete documentation index at: https://notes.kodekloud.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Lab Walkthrough Apply Agentic Patterns
> Lab showing how to extend a ReAct agent with structured output, local input guardrails, and human confirmation for safer, auditable scheduling actions.
Time to get hands-on.
In this lesson you'll extend a minimal ReAct-style agent with three practical production patterns: structured output, input guardrails, and a human-in-the-loop confirmation step. These are additive patterns — you do not need to rewrite the agent core; you augment it.
This lab assumes a ready environment:
* Python 3.11 and a virtual environment
* The [OpenAI Python SDK](https://platform.openai.com/docs/libraries/python) installed
* Your working directory: `/root/code`
Create a file named `patterns_agent.py`. Start from the minimal ReAct structure: import modules, create an OpenAI client, set a system prompt that describes the Thought / Action / Observation markers, add a `check_calendar` tool and handler, and wire up the loop that runs until the agent produces a final answer.
Minimal starting template (ReAct structure):
```python theme={null}
# patterns_agent.py (start)
import os, json, re
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_API_BASE"),
)
system_prompt = """You are a scheduling assistant. Use the ReAct pattern.
Thought: reason about what to do next.
Action: call a tool if needed.
Observation: results of the action.
Repeat until you can give a final answer.
"""
def check_calendar(date: str) -> str:
return f"Standup 9am, Review 2pm on {date}"
```
Next, add the three patterns one-by-one and then run the agent. Below is a consolidated, runnable example that demonstrates all three patterns together: structured output, input guardrails, and a human-in-the-loop confirmation for high-stakes actions such as sending email. The script is intentionally simple and synchronous for clarity — its goal is to illustrate how these patterns integrate into a ReAct-style loop.
The `check_input` guard runs locally before any API call, so disallowed queries never reach the model and cost no tokens.
Final consolidated example (`patterns_agent.py`):
```python theme={null}
# patterns_agent.py
import os
import json
import re
from typing import Any, Dict, Optional
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_API_BASE"),
)
# 1) Structured output: instruct the model to always end the final reply
# with a machine-readable JSON summary block that contains "summary"
# and "actions_taken".
system_prompt = """You are a scheduling assistant. Use the ReAct pattern.
Thought: reason about what to do next.
Action: call a tool if needed (use the exact JSON action format described below).
Observation: results of the action.
Repeat until you can give a final answer.
When you wish to call a tool, emit a single line Action containing JSON, for example:
Action: {"name": "check_calendar", "args": {"date": "2023-08-01"}}
Always end your final response with a JSON summary block, for example:
{"summary": "I checked the calendar and emailed Sarah.", "actions_taken": ["check_calendar", "send_email"]}
"""
# Tools
def check_calendar(date: str) -> str:
# Replace with real calendar lookup in production
return f"Standup 9am, Review 2pm on {date}"
def send_email(to: str, subject: str, body: str) -> str:
# Replace with real email sending logic in production
return f"Email sent to {to} with subject '{subject}'."
# 2) Input guardrails: check input locally for disallowed request types.
def check_input(message: str) -> Optional[str]:
blocked = ["medical", "legal", "financial advice"]
message_lower = message.lower()
for term in blocked:
if term in message_lower:
return "I can only help with scheduling and contacts."
return None
# Helpers to parse action JSON from the assistant's text
ACTION_RE = re.compile(r'Action:\s*(\{.*\})', re.DOTALL)
SUMMARY_RE = re.compile(r'(\{[\s\S]*?"summary"[\s\S]*?\})', re.DOTALL)
def extract_action(text: str) -> Optional[Dict[str, Any]]:
m = ACTION_RE.search(text)
if not m:
return None
try:
action = json.loads(m.group(1))
return action
except json.JSONDecodeError:
return None
def extract_summary(text: str) -> Optional[Dict[str, Any]]:
m = SUMMARY_RE.search(text)
if not m:
return None
try:
summary = json.loads(m.group(1))
return summary
except json.JSONDecodeError:
return None
# Main agent loop
def run_agent(user_message: str):
# Guardrails
guard = check_input(user_message)
if guard:
print(guard)
return
# Conversation history (system + chat messages)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
actions_taken = []
while True:
# Call the model
resp = client.chat.completions.create(
model="gpt-4o-mini", # choose a model available to you
messages=messages,
temperature=0.0,
max_tokens=800,
)
assistant_msg = resp.choices[0].message.content
print("Assistant:")
print(assistant_msg.strip())
messages.append({"role": "assistant", "content": assistant_msg})
# Check for action invocation
action = extract_action(assistant_msg)
if action:
name = action.get("name")
args = action.get("args", {})
actions_taken.append(name)
# Tool dispatch with human-in-the-loop for high-stakes actions
if name == "send_email":
# Human confirmation step
print(f"\nProposed email arguments: {json.dumps(args, indent=2)}")
confirm = input("Send this email? (y/n): ").strip().lower()
if confirm != "y":
result = "Email cancelled by user."
else:
# call the tool (in production, handle exceptions)
result = send_email(**args)
elif name == "check_calendar":
# simple tool call
date = args.get("date", "")
result = check_calendar(date)
else:
result = f"Unknown tool: {name}"
# Append Observation and continue loop
observation_text = f"Observation: {result}"
print(observation_text)
messages.append({"role": "user", "content": observation_text})
continue
# Check for final JSON summary block in assistant output to finish
summary = extract_summary(assistant_msg)
if summary:
# Ensure actions_taken is included or merge if model provided its own
if "actions_taken" not in summary:
summary["actions_taken"] = actions_taken
print("\nFinal JSON summary:")
print(json.dumps(summary, indent=2))
break
# If neither action nor summary found, the model may be expecting more context.
# Optionally, enforce a safety max loop count (not shown) to avoid infinite loops.
if __name__ == "__main__":
# Example: Set the user message to ask for an email with a calendar summary
user_message = "Email Sarah my calendar summary for today."
run_agent(user_message)
```
How the script demonstrates each pattern
* Structured output
* The system prompt instructs the model to emit actions in a single-line JSON `Action` form and to conclude with a JSON summary block containing `summary` and `actions_taken`. This makes downstream parsing deterministic and machine-readable.
* Input guardrails
* `check_input` runs entirely locally before any API call. If it detects blocked terms (for example: `medical`, `legal`, or `financial advice`), the script rejects the user request and never calls the model — saving tokens and preventing the model from handling sensitive requests.
* Human-in-the-loop
* When the agent emits an action whose `name` is `send_email`, the driver pauses and prompts a human operator to confirm. If the user denies, the tool returns an "Email cancelled by user." observation and the agent re-plans. Lower-risk tools such as `check_calendar` run automatically.
Summary table — patterns at a glance:
| Pattern | Purpose | Example / Notes |
| - | - | - |
| Structured output | Make agent responses deterministic and parseable | System prompt requires `Action: {"name": "...", "args": {...}}` and final JSON summary `{"summary": "...", "actions_taken": [...]}` |
| Input guardrails | Block disallowed topics before calling API | `check_input(message)` returns an error string for blocked terms like `medical` |
| Human-in-the-loop | Add a confirmation step for sensitive actions | Pause on `send_email` and prompt: `Send this email? (y/n):` |
Run the full stack
1. Ensure the environment variables `OPENAI_API_KEY` (and `OPENAI_API_BASE` if using a non-default base) are set.
2. Run:
* `python patterns_agent.py`
3. Example behavior with `user_message = "Email Sarah my calendar summary for today."`:
* The guardrail runs and passes.
* The agent may call `check_calendar` automatically.
* When the agent requests `send_email`, the script prompts: `Send this email? (y/n):`
* Type `y` to simulate sending (the `send_email` implementation is a stub).
* The agent finishes and prints a final machine-readable JSON summary.
Best practices and next steps
* Replace the stubbed `check_calendar` and `send_email` with real integrations and robust error handling in production.
* Consider adding a loop counter or timeout to protect against infinite planning loops.
* Expand `check_input` to use more advanced safety checks (regular expressions, allowlists, or a dedicated moderation service).
* Log `actions_taken` and Observations to a secure audit trail for observability and compliance.
Links and references
* [OpenAI Python SDK docs](https://platform.openai.com/docs/libraries/python)
* ReAct and agent design patterns: search for ReAct agent papers and blog posts for design inspiration.
This pattern preserves the ReAct flow while adding small, composable production controls that improve safety, observability, and human oversight without rewriting the agent logic.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.