diff --git a/patterns/workflows/2-workflow-patterns/1a-prompt-chaining-full-workflow.py b/patterns/workflows/2-workflow-patterns/1a-prompt-chaining-full-workflow.py new file mode 100644 index 0000000..d9d6315 --- /dev/null +++ b/patterns/workflows/2-workflow-patterns/1a-prompt-chaining-full-workflow.py @@ -0,0 +1,284 @@ +''' +This is an enhanced version of example 1-prompt-chaining.py + +It demonstrates end-to-end agentic worflow - setting up meeting in Google Calendar based on user's voice command. + +Detailed flow: +- get voice command from user and transcribe it to text (using OpenAI whisper) +- analyze the text whether it contains request to create a meeting (using OpenAI gpt-4o-mini) +- based on the text details, set up a new meeting in the Google Calendar (using Google Calendar API) +- generate confirmation message about the new meeting (using OpenAI gpt-4o-mini) + + IMPORTANT! Before running this example: + 1. Install required libraries: pip install -r rquirements_for_1a.txt + 2. Add .env file with your OpenAI API key + 3. Complete authenticatication setup in Google Cloud. See calendar_tools.py file for details. + +''' + + +import calendar_tools + +from typing import Optional +from datetime import datetime +from pydantic import BaseModel, Field +from openai import OpenAI +import os +import logging +import speech_recognition as sr +from dotenv import load_dotenv + +# -------------------------------------------------------------- +# Step 0: Setup and helper functions +# -------------------------------------------------------------- + +# method for audio recording and transcription +# returns full text transcription +def live_transcription() -> str: + recognizer = sr.Recognizer() + mic = sr.Microphone() + + print("🎤 Listening... (Press Ctrl+C to stop)") + + with mic as source: + recognizer.adjust_for_ambient_noise(source) + + while True: + try: + print("\nListening...") + audio = recognizer.listen(source, timeout=15, phrase_time_limit=30) + + # Save audio as a temporary file + with open("temp_audio.wav", "wb") as f: + f.write(audio.get_wav_data()) + + # Transcribe audio using OpenAI Whisper API + with open("temp_audio.wav", "rb") as audio_file: + transcript = audio_client.audio.transcriptions.create( + model=audio_model, + file=audio_file + ) + print(f"Transcript object: {transcript}") + return transcript.text + + + except sr.WaitTimeoutError: + print("⏳ No speech detected, waiting...") + except Exception as e: + print("⚠️ Error:", e) + break + return None + +# Set up logging configuration +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger(__name__) + +# setting up OpenAI clients +load_dotenv() +client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) +model = "gpt-4o-mini" + +audio_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) +audio_model = "whisper-1" + + +# -------------------------------------------------------------- +# Step 1: Define the data models for each stage +# -------------------------------------------------------------- + +class EventExtraction(BaseModel): + """First LLM call: Extract basic event information""" + + description: str = Field(description="Raw description of the event") + is_calendar_event: bool = Field( + description="Whether this text describes a calendar event" + ) + confidence_score: float = Field(description="Confidence score between 0 and 1") + + +class EventDetails(BaseModel): + """Second LLM call: Parse specific event details""" + + name: str = Field(description="Name of the event") + date: str = Field( + description="Date and time of the event. Use '%Y-%m-%d %H:%M' syntax to format this value." + ) + duration_minutes: int = Field(description="Expected duration in minutes") + participants: list[str] = Field(description="List of participants e-mails") + + +class EventConfirmation(BaseModel): + """Third LLM call: Generate confirmation message""" + + confirmation_message: str = Field( + description="Natural language confirmation message" + ) + calendar_link: Optional[str] = Field( + description="Generated calendar link if applicable" + ) + + +# -------------------------------------------------------------- +# Step 2: Define the functions +# -------------------------------------------------------------- + + +def extract_event_info(user_input: str) -> EventExtraction: + """First LLM call to determine if input is a calendar event""" + logger.info("Starting event extraction analysis") + logger.debug(f"Input text: {user_input}") + + today = datetime.now() + date_context = f"Today is {today.strftime('%A, %B %d, %Y')}." + + completion = client.beta.chat.completions.parse( + model=model, + messages=[ + { + "role": "system", + "content": f"{date_context} Analyze if the text describes a calendar event.", + }, + {"role": "user", "content": user_input}, + ], + response_format=EventExtraction, + ) + result = completion.choices[0].message.parsed + logger.info( + f"Extraction complete - Is calendar event: {result.is_calendar_event}, Confidence: {result.confidence_score:.2f}" + ) + return result + + +def parse_event_details(description: str) -> EventDetails: + """Second LLM call to extract specific event details""" + logger.info("Starting event details parsing") + + today = datetime.now() + date_context = f"Today is {today.strftime('%A, %B %d, %Y')}." + + completion = client.beta.chat.completions.parse( + model=model, + messages=[ + { + "role": "system", + "content": f"{date_context} Extract detailed event information. When dates reference 'next Tuesday' or similar relative dates, use this current date as reference. Assume Europe West timezone. If extracted participand e-mail info doesn't contain @ sign, then add @ sign to create valid e-mail address. ", + }, + {"role": "user", "content": description}, + ], + response_format=EventDetails, + ) + result = completion.choices[0].message.parsed + logger.info( + f"Parsed event details - Name: {result.name}, Date: {result.date}, Duration: {result.duration_minutes}min" + ) + logger.debug(f"Participants: {', '.join(result.participants)}") + return result + + +def generate_confirmation(event_details: EventDetails) -> EventConfirmation: + """Third LLM call to generate a confirmation message""" + logger.info("Generating confirmation message") + + completion = client.beta.chat.completions.parse( + model=model, + messages=[ + { + "role": "system", + "content": "Generate a natural confirmation message for the event. Sign off with your name; Susie", + }, + {"role": "user", "content": str(event_details.model_dump())}, + ], + response_format=EventConfirmation, + ) + result = completion.choices[0].message.parsed + logger.info("Confirmation message generated successfully") + return result + + +# -------------------------------------------------------------- +# Step 3: Chain the functions together +# -------------------------------------------------------------- + + +def process_calendar_request(user_input: str) -> Optional[EventConfirmation]: + """Main function implementing the prompt chain with gate check""" + logger.info("Processing calendar request") + logger.debug(f"Raw input: {user_input}") + + # First LLM call: Extract basic info + initial_extraction = extract_event_info(user_input) + + # Gate check: Verify if it's a calendar event with sufficient confidence + if ( + not initial_extraction.is_calendar_event + or initial_extraction.confidence_score < 0.7 + ): + logger.warning( + f"Gate check failed - is_calendar_event: {initial_extraction.is_calendar_event}, confidence: {initial_extraction.confidence_score:.2f}" + ) + return None + + logger.info("Gate check passed, proceeding with event processing") + + # Second LLM call: Get detailed event information + event_details = parse_event_details(initial_extraction.description) + + # The following lines create new meeting in user's Google Calendar. + # We utilize Google Calendar API (see detailed explanation and source code in calendar_tools.py) + service = calendar_tools.authenticate_and_connect_to_GoogleCalendarAPI() + + patricipants_emails: list[str] = [] + for participant_email in event_details.participants: + patricipants_emails.append(participant_email) + + if service: + meeting_created = calendar_tools.create_event(service, + title=event_details.name, + start_datetime=event_details.date, + duration_minutes=event_details.duration_minutes, + attendees_emails=patricipants_emails, + description=None, + location=None, + reminders=True, + visibility='default', + color_id='1', + add_meet_link=True) + else: + logger.warning("Could not connect to the calendar using Calendar API!") + return "The meeting has not been created - there were technical problems." + + if meeting_created: + # Third LLM call: Generate confirmation + confirmation = generate_confirmation(event_details) + logger.info("Calendar request processing completed successfully") + else: + logger.warning("Could not create a meeting in the calendar!") + return "The meeting has not been created - there were technical problems." + + return confirmation + + +# -------------------------------------------------------------- +# Step 4: Test the chain with a valid input +# -------------------------------------------------------------- + +if __name__ == "__main__": + + user_input = live_transcription() + + # You can say one of these example voice inputs or say uncomment one of these lines just to run the workflow for plain text + # user_input = "Schedule 1h team meeting next Tuesday at 1pm with Alice@xyz.ai and Bob@xyz.ai to discuss the project roadmap." + # user_input = "Ustaw 30 minutowe spotkanie w przyszły czwartek o 14tej z Jacek@ai.pl oraz Agatka@ai.pl żeby zademostrować możliwości agentów sztucznej inteligencji." + # user_input = "And I think to myself what a wonderful world!" # this is not going to create a calendar event + result = process_calendar_request(user_input) + if result: + print(f"Confirmation: {result.confirmation_message}") + if result.calendar_link: + print(f"Calendar Link: {result.calendar_link}") + else: + print("This doesn't appear to be a calendar event request.") + diff --git a/patterns/workflows/2-workflow-patterns/calendar_tools.py b/patterns/workflows/2-workflow-patterns/calendar_tools.py new file mode 100644 index 0000000..d88a0db --- /dev/null +++ b/patterns/workflows/2-workflow-patterns/calendar_tools.py @@ -0,0 +1,231 @@ +''' +Krzysztof Kućmierz, krzysztof.kucmierz@artificiuminformatica.pl +'LICENSE' file in the ai-cookbook project contains license details. + +IMPORTANT! Before authenticating to Google Calendar you need to enable Calendar API and create OAuth credentials in Google Cloud. +Follow detailed steps -> https://developers.google.com/calendar/api/quickstart/python + +This file contains helper methods to access Google Calendar using Google Calendar API: +- authentication +- create a meeting +- list upcoming meetings +- show details of a meeting +''' + +import os +import sys +from datetime import datetime, timezone, timedelta +from google.oauth2.credentials import Credentials +from google_auth_oauthlib.flow import InstalledAppFlow +from google.auth.transport.requests import Request +from googleapiclient.discovery import build +from googleapiclient.errors import HttpError + +# Define the scopes required +SCOPES = ['https://www.googleapis.com/auth/calendar'] + + +''' +Connect to user's calendar using open authentication (OAuth) +Returns: service object or None in case of HttpError +''' +def authenticate_and_connect_to_GoogleCalendarAPI(): + creds = None + + # Load credentials if previously saved + if os.path.exists('token.json'): + creds = Credentials.from_authorized_user_file('token.json', SCOPES) + + # If credentials are not available or invalid, request login + if not creds or not creds.valid: + if creds and creds.expired and creds.refresh_token: + creds.refresh(Request()) + else: + flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES) + creds = flow.run_local_server(port=0) + + # Save the credentials for future use + with open('token.json', 'w') as token: + token.write(creds.to_json()) + try: + # Connect to Google Calendar API + service = build('calendar', 'v3', credentials=creds) + + except HttpError as error: + print(f"An error occurred: {error}.") + service = None + + return service + +''' +Shows event details on console output +Input parameters: +event calndar event object +Returns: None +''' +def show_event_details(event): + # Extract event URI + event_uri = event['htmlLink'] + # Extract Google Meet link (if added) + meet_link = event.get('conferenceData', {}).get('entryPoints', [{}])[0].get('uri', 'No Meet link') + print(f"✅ Event Created Successfully: {event['summary']}") + print(f"🔗 Event URI: {event_uri}") + print(f"📍 Location: {event.get('location', 'Not provided')}") + print(f"🕒 Start: {event['start']['dateTime']}") + print(f"🕒 End: {event['end']['dateTime']}") + print(f"📧 Attendees: {[attendee['email'] for attendee in event.get('attendees', [])]}") + print(f"🎨 Color ID: {event.get('colorId', 'Default')}") + default_reminders = event.get('reminders', {}).get('useDefault', True) + print(f"🔔 Reminders: {'Default' if default_reminders else 'Non default or disabled'}") + print(f"📹 Google Meet Link: {meet_link}") + + +''' +Creates a calendar event (meeting) +Input parameters: +service resource object (client allowing to call Google Calendar API methods) +title meeting title +start_datetime day and time of the meeting +duration_minutes meeting duration +attendees_emails list of e-mails +description (optional) short description of the meeting +location (optional) meeting location +reminders (optional) if True meeting reminders will be sent +visibility (optional) visibility of the meeting. Options are: 'default', 'public', or 'private' +color_id (optional) color assigned to the meeting +add_meet_link (optional) if True add Google Meet link to a meeting + +Returns: True if event created, False if not created +''' +def create_event(service, + title: str, + start_datetime: str, + duration_minutes: int, + attendees_emails: list[str], + description: str = None, + location: str = None, + reminders: bool = True, + visibility: str = 'default', + color_id: str = '1', + add_meet_link: bool = False): + + # Convert start time to ISO 8601 format (RFC3339) + start_time = datetime.strptime(start_datetime, "%Y-%m-%d %H:%M").replace(tzinfo=timezone.utc) + end_time = start_time + timedelta(minutes=duration_minutes) + + event = { + 'summary': title, + 'description': description, + 'start': { + 'dateTime': start_time.isoformat(), + 'timeZone': 'UTC', # Adjust to your timezone if needed + }, + 'end': { + 'dateTime': end_time.isoformat(), + 'timeZone': 'UTC', + }, + 'location': location if location else "", # Optional location + 'attendees': [{'email': email} for email in attendees_emails], + 'visibility': visibility, + 'colorId': color_id, # Color-coded event (1-11) + 'reminders': { + 'useDefault': False if reminders else True, + 'overrides': [ + {'method': 'email', 'minutes': 30}, # Email reminder 30 mins before + {'method': 'popup', 'minutes': 10} # Pop-up reminder 10 mins before + ] if reminders else [], + }, + + # Add Google Meet link if requested + 'conferenceData': { + 'createRequest': { + 'requestId': f"meet-{start_time.timestamp()}", + 'conferenceSolutionKey': {'type': 'hangoutsMeet'} + } + } if add_meet_link else None + } + + # Insert event with conference data enabled + try: + event = service.events().insert( + calendarId='primary', + body=event, + conferenceDataVersion=1 # Required for Google Meet links + ).execute() + + except HttpError as error: + print(f"An error occurred: {error}.") + + # Verify event creation + if 'id' in event and 'htmlLink' in event: + print(f"✅ Event Created Successfully.") + show_event_details(event) + return True + else: + print("❌ Event creation failed: No valid response received.") + return False + + +''' +Shows up to 10 future events in the calendar +Input parameters: +service resource object (client allowing to call Google Calendar API methods) +Returns: None +''' +# +def show_upcoming_events(service): + # Fetch upcoming events + print('\nFetching upcoming events...') + now = datetime.now(timezone.utc).isoformat() + + try: + events_result = service.events().list(calendarId='primary', maxResults=10, singleEvents=True, + orderBy='startTime', timeMin=now).execute() + events = events_result.get('items', []) + if not events: + print('No upcoming events found.') + else: + print("\n📅 Upcoming Events:\n" + "="*30) + for event in events: + start = event['start'].get('dateTime', event['start'].get('date')) # Handles all-day events + start_dt = datetime.fromisoformat(start) if 'T' in start else datetime.strptime(start, "%Y-%m-%d") + formatted_start = start_dt.strftime("%A, %d %B %Y %I:%M %p") if 'T' in start else start_dt.strftime("%A, %d %B %Y") + + print(f"🕒 {formatted_start} - {event.get('summary', 'No Title')}") + print("="*30) + + except HttpError as error: + print(f"An error occurred: {error}.") + + +''' +A flow to test above methods +''' +def test_calendar_tools(): + + service = authenticate_and_connect_to_GoogleCalendarAPI() + + if service is None: + print("Couldn't connect to Calendar API. Exiting!") + sys.exit(1) + + # if connected to service, add a meeting + create_event( + service=service, + title="Let's talk about AI agents and workflows.", + description="Meeting generated using Calendar API.", + start_datetime="2025-03-28 11:00", # Format: YYYY-MM-DD HH:MM (24-hour) + duration_minutes=45, + attendees_emails=["architect@xyz.ai", "program_manager@xyz.ai"], + location="Google Meet", + reminders=True, + visibility="public", # Can be 'default', 'public', or 'private' + color_id="4", # Google Calendar color ID (1-11) + add_meet_link=True + ) + + show_upcoming_events(service) + + +if __name__ == '__main__': + test_calendar_tools() \ No newline at end of file diff --git a/patterns/workflows/2-workflow-patterns/requirements_for_1a.txt b/patterns/workflows/2-workflow-patterns/requirements_for_1a.txt new file mode 100644 index 0000000..cd93d97 --- /dev/null +++ b/patterns/workflows/2-workflow-patterns/requirements_for_1a.txt @@ -0,0 +1,9 @@ +google_api_python_client==2.165.0 +google_auth_oauthlib==1.2.1 +nest_asyncio==1.6.0 +openai==1.68.2 +protobuf==6.30.1 +pydantic==2.10.6 +speechrecognition==3.14.2 +pyaudio +python-dotenv \ No newline at end of file