Lab Walkthrough Apply Agentic Patterns

# 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}"

# 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)