Merge pull request #153 from Sakalya100/main

Added new demo: Self-Guided AI Tour Agent using OpenAI Agents SDK
This commit is contained in:
Shubham Saboo 2025-03-25 15:49:55 -05:00 committed by GitHub
commit 471ab9150d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 683 additions and 0 deletions

View file

@ -0,0 +1,73 @@
# 🗺️ Self-Guided Audio Tour Agent
A conversational voice agent system that generates immersive, self-guided audio tours based on the users **location**, **areas of interest**, and **tour duration**. Built on a multi-agent architecture using OpenAI Agents SDK, real-time information retrieval, and expressive TTS for natural speech output.
---
## 🚀 Features
### 🎙️ Multi-Agent Architecture
- **Orchestrator Agent**
Coordinates the overall tour flow, manages transitions, and assembles content from all expert agents.
- **History Agent**
Delivers insightful historical narratives with an authoritative voice.
- **Architecture Agent**
Highlights architectural details, styles, and design elements using a descriptive and technical tone.
- **Culture Agent**
Explores local customs, traditions, and artistic heritage with an enthusiastic voice.
- **Culinary Agent**
Describes iconic dishes and food culture in a passionate and engaging tone.
---
### 📍 Location-Aware Content Generation
- Dynamic content generation based on user-input **location**
- Real-time **web search integration** to fetch relevant, up-to-date details
- Personalized content delivery filtered by user **interest categories**
---
### ⏱️ Customizable Tour Duration
- Selectable tour length: **15, 30, or 60 minutes**
- Time allocations adapt to user interest weights and location relevance
- Ensures well-paced and proportioned narratives across sections
---
### 🔊 Expressive Speech Output
- High-quality audio generated using **Text-to-Speech (TTS)**
### How to get Started?
1. Clone the GitHub repository
```bash
git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git
cd ai_agent_tutorials/ai_audio_tour_agent
```
2. Install the required dependencies:
```bash
pip install -r requirements.txt
```
3. Get your OpenAI API Key
- Sign up for an [OpenAI account](https://platform.openai.com/) (or the LLM provider of your choice) and obtain your API key.
4. Run the Streamlit App
```bash
streamlit run ai_audio_tour_agent.py
```

View file

@ -0,0 +1,306 @@
from pydantic import BaseModel
from agents import Agent, WebSearchTool
from agents.model_settings import ModelSettings
ARCHITECTURE_AGENT_INSTRUCTIONS = ("""
You are the Architecture agent for a self-guided audio tour system. Given a location and the areas of interest of user, your role is to:
1. Describe architectural styles, notable buildings, urban planning, and design elements
2. Provide technical insights balanced with accessible explanations
3. Highlight the most visually striking or historically significant structures
4. Adopt a detailed, descriptive voice style when delivering architectural content
5. Make sure not to add any headings like ## Architecture. Just provide the content
6. Make sure the details are conversational and don't include any formatting or headings. It will be directly used in a audio model for converting to speech and the entire content should feel like natural speech.
7. Make sure the content is strictly between the upper and lower Word Limit as specified. For example, If the word limit is 100 to 120, it should be within that, not less than 100 or greater than 120
NOTE: Given a location, use web search to retrieve uptodate context and architectural information about the location
NOTE: Do not add any Links or Hyperlinks in your answer or never cite any source
Help users see and appreciate architectural details they might otherwise miss. Make it as detailed and elaborative as possible
""")
class Architecture(BaseModel):
output: str
architecture_agent = Agent(
name="ArchitectureAgent",
instructions=ARCHITECTURE_AGENT_INSTRUCTIONS,
model="gpt-4o",
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
output_type=Architecture
)
CULINARY_AGENT_INSTRUCTIONS = ("""
You are the Culinary agent for a self-guided audio tour system. Given a location and the areas of interest of user, your role is to:
1. Highlight local food specialties, restaurants, markets, and culinary traditions in the user's location
2. Explain the historical and cultural significance of local dishes and ingredients
3. Suggest food stops suitable for the tour duration
4. Adopt an enthusiastic, passionate voice style when delivering culinary content
5. Make sure not to add any headings like ## Culinary. Just provide the content
6. Make sure the details are conversational and don't include any formatting or headings. It will be directly used in a audio model for converting to speech and the entire content should feel like natural speech.
7. Make sure the content is strictly between the upper and lower Word Limit as specified. For example, If the word limit is 100 to 120, it should be within that, not less than 100 or greater than 120
NOTE: Given a location, use web search to retrieve uptodate context and culinary information about the location
NOTE: Do not add any Links or Hyperlinks in your answer or never cite any source
Make your descriptions vivid and appetizing. Include practical information like operating hours when relevant. Make it as detailed and elaborative as possible
""")
class Culinary(BaseModel):
output: str
culinary_agent = Agent(
name="CulinaryAgent",
instructions=CULINARY_AGENT_INSTRUCTIONS,
model="gpt-4o",
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
output_type=Culinary
)
CULTURE_AGENT_INSTRUCTIONS = ("""
You are the Culture agent for a self-guided audio tour system. Given a location and the areas of interest of user, your role is to:
1. Provide information about local traditions, customs, arts, music, and cultural practices
2. Highlight cultural venues and events relevant to the user's interests
3. Explain cultural nuances and significance that enhance the visitor's understanding
4. Adopt a warm, respectful voice style when delivering cultural content
5. Make sure not to add any headings like ## Culture. Just provide the content
6. Make sure the details are conversational and don't include any formatting or headings. It will be directly used in a audio model for converting to speech and the entire content should feel like natural speech.
7. Make sure the content is strictly between the upper and lower Word Limit as specified. For example, If the word limit is 100 to 120, it should be within that, not less than 100 or greater than 120
NOTE: Given a location, use web search to retrieve uptodate context and all the cultural information about the location
NOTE: Do not add any Links or Hyperlinks in your answer or never cite any source
Focus on authentic cultural insights that help users appreciate local ways of life. Make it as detailed and elaborative as possible
""")
class Culture(BaseModel):
output: str
culture_agent = Agent(
name="CulturalAgent",
instructions=CULTURE_AGENT_INSTRUCTIONS,
model="gpt-4o",
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
output_type=Culture
)
HISTORY_AGENT_INSTRUCTIONS = ("""
You are the History agent for a self-guided audio tour system. Given a location and the areas of interest of user, your role is to:
1. Provide historically accurate information about landmarks, events, and people related to the user's location
2. Prioritize the most significant historical aspects based on the user's time constraints
3. Include interesting historical facts and stories that aren't commonly known
4. Adopt an authoritative, professorial voice style when delivering historical content
5. Make sure not to add any headings like ## History. Just provide the content
6. Make sure the details are conversational and don't include any formatting or headings. It will be directly used in a audio model for converting to speech and the entire content should feel like natural speech.
7. Make sure the content is strictly between the upper and lower Word Limit as specified. For example, If the word limit is 100 to 120, it should be within that, not less than 100 or greater than 120
NOTE: Given a location, use web search to retrieve uptodate context and historical information about the location
NOTE: Do not add any Links or Hyperlinks in your answer or never cite any source
Focus on making history come alive through engaging narratives. Keep descriptions concise but informative. Make it as detailed and elaborative as possible
""")
class History(BaseModel):
output: str
historical_agent = Agent(
name="HistoricalAgent",
instructions=HISTORY_AGENT_INSTRUCTIONS,
model="gpt-4o",
output_type=History,
tools=[WebSearchTool()],
model_settings=ModelSettings(tool_choice="required"),
)
ORCHESTRATOR_INSTRUCTIONS = ("""
Your Role
You are the Orchestrator Agent for a self-guided audio tour system. Your task is to assemble a comprehensive and engaging tour for a single location by integrating pre-timed content from four specialist agents (Architecture, History, Culinary, and Culture), while adding introduction and conclusion elements.
Input Parameters
- User Location: The specific location for the tour (e.g., a landmark, neighborhood, or district)
- User Interests: User's preference across categories (Architecture, History, Culinary, Culture)
- Specialist Agent Outputs: Pre-sized content from each domain expert (Architecture, History, Culinary, Culture)
- Specialist Agent Word Limit: Word Limit from each domain expert (Architecture, History, Culinary, Culture)
Your Tasks
1. Introduction Creation (1-2 minutes)
Create an engaging and warm introduction that:
- Welcomes the user to the specific location
- Briefly outlines what the tour will cover
- Highlights which categories are emphasized based on user interests
- Sets the tone for the experience (conversational and immersive)
2. Content Integration with Deduplication
Integrate the content from all four agents in the correct order:
- Architecture History Culture Culinary
- Maintain each agent's voice and expertise
- Ensure all content fits within its allocated time budget
- Don't edit anything from your end and just accumulate the content from the specialised agents
3. Transition Development
Develop smooth transitions between the sections:
- Use natural language to move from one domain to another
- Connect themes when possible (e.g., how architecture influenced culture, or how history shaped food)
4. Conclusion Creation
Write a thoughtful concise and short conclusion that:
- Summarizes key highlights from the tour
- Reinforces the uniqueness of the location
- Connects the explored themes holistically
- Encourages the listener to explore further based on their interests
5. Final Assembly
Assemble the complete tour in the following order:
- Introduction
- Architecture
- History
- Culture
- Culinary
- Conclusion
Ensure:
- Transitions are smooth
- Content is free from redundancy
- Total duration respects the time allocation plan
- The entire output sounds like one cohesive guided experience
""")
class FinalTour(BaseModel):
introduction: str
"""A short introduction of the Tour."""
architecture: str
"""The Architectural Content"""
history: str
"""The Historical Content"""
culture: str
"""The Culture Content"""
culinary: str
"""The Culinary Content"""
conclusion: str
"""A short conclusion of the Tour."""
orchestrator_agent = Agent(
name="OrchestratorAgent",
instructions=ORCHESTRATOR_INSTRUCTIONS,
model="gpt-4o",
output_type=FinalTour,
)
PLANNER_INSTRUCTIONS = ("""
Your Role
You are the Planner Agent for a self-guided tour system. Your primary responsibility is to analyze the user's location, interests, and requested tour duration to create an optimal time allocation plan for content generation by specialist agents (Architecture, History, Culture, and Culinary).
Input Parameters
User Location: The specific location for the tour
User Interests: User's ranked preferences across categories (Architecture, History, Culture, Culinary)
Tour Duration: User's selected time (15, 30, or 60 minutes)
Your Tasks
1. Interest Analysis
Evaluate the user's interest preferences
Assign weight to each category based on expressed interest level
If no specific preferences are provided, assume equal interest in all categories
2. Location Assessment
Analyze the significance of the specified location for each category
Determine if the location has stronger relevance in particular categories
Example: A cathedral might warrant more time for Architecture and History than Culinary
3. Time Allocation Calculation
Calculate the total content time (excluding introduction and conclusion)
Reserve 1-2 minutes for introduction and 1 minute for conclusion
Distribute the remaining time among the four categories based on:
User interest weights (primary factor)
Location relevance to each category (secondary factor)
Ensure minimum time thresholds for each category (even low-interest categories get some coverage)
4. Scaling for Different Durations
15-minute tour:
Introduction: ~1 minute
Content sections: ~12-13 minutes total (divided among categories)
Conclusion: ~1 minute
Each category gets at least 1 minute, with preferred categories getting more
30-minute tour:
Introduction: ~1.5 minutes
Content sections: ~27 minutes total (divided among categories)
Conclusion: ~1.5 minutes
Each category gets at least 3 minutes, with preferred categories getting more
60-minute tour:
Introduction: ~2 minutes
Content sections: ~56 minutes total (divided among categories)
Conclusion: ~2 minutes
Each category gets at least 5 minutes, with preferred categories getting more
Your output must be a JSON object with numeric time allocations (in minutes) for each section:
- introduction
- architecture
- history
- culture
- culinary
- conclusion
Only return the number of minutes allocated to each section. Do not include explanations or text descriptions.
Example:
{
"introduction": 2,
"architecture": 15,
"history": 20,
"culture": 10,
"culinary": 9,
"conclusion": 2
}
Make sure the time allocation adheres to the interests, and the interested section is allocated more time than others.
""")
class Planner(BaseModel):
introduction: float
architecture: float
history: float
culture: float
culinary: float
conclusion: float
planner_agent = Agent(
name="PlannerAgent",
instructions=PLANNER_INSTRUCTIONS,
model="gpt-4o",
output_type=Planner,
)

View file

@ -0,0 +1,83 @@
import streamlit as st
import asyncio
from manager import TourManager
from agents import set_default_openai_key
def tts(text):
from pathlib import Path
from openai import OpenAI
client = OpenAI()
speech_file_path = Path(__file__).parent / f"speech_tour.mp3"
response = client.audio.speech.create(
model="gpt-4o-mini-tts",
voice="coral",
input=text,
instructions="""Speak with different tones based on the content sections:
- Use an inviting and warm tone for the Introduction.
- Speak in an authoritative tone for History sections.
- Use a positive and curious tone when discussing Architecture.
- Use an enthusiastic and lively tone when covering Culture.
- Use a joyful and passionate tone when talking about Culinary topics.
Keep the transitions smooth and natural between sections."""
)
response.stream_to_file(speech_file_path)
return speech_file_path
def run_async(func, *args, **kwargs):
try:
return asyncio.run(func(*args, **kwargs))
except RuntimeError:
loop = asyncio.get_event_loop()
return loop.run_until_complete(func(*args, **kwargs))
st.sidebar.title("🔑 API Settings")
api_key = st.sidebar.text_input("Enter your OpenAI API key:", type="password")
if api_key:
st.session_state["OPENAI_API_KEY"] = api_key
st.sidebar.success("API key saved!")
set_default_openai_key(api_key)
st.title("🏛️ Self-Guided Audio Tour Generator")
location = st.text_input("📍 Enter the location for your tour:")
interests = st.multiselect(
"🎯 Select your interests:",
options=["History", "Architecture", "Culinary", "Culture"]
)
duration = st.slider(
"⏱️ Select the duration of the tour (in minutes):",
min_value=5,
max_value=60,
value=30,
step=5
)
if st.button("🎧 Generate Tour"):
if "OPENAI_API_KEY" not in st.session_state:
st.error("Please enter your OpenAI API key in the sidebar.")
elif not location:
st.error("Please enter a location.")
elif not interests:
st.error("Please select at least one interest.")
else:
with st.spinner(f"Generating a {duration}-minute tour in {location} focused on {', '.join(interests)}."):
mgr = TourManager()
final_tour = run_async(
mgr.run, location, interests, duration
)
with st.expander("🎬 Tour"):
st.markdown(final_tour)
with st.spinner("Generating Audio"):
tour_audio = tts(final_tour)
st.audio(tour_audio, format="audio/mp3")

View file

@ -0,0 +1,168 @@
from __future__ import annotations
import asyncio
import time
import json
from collections.abc import Sequence
from rich.console import Console
from agents import Runner, RunResult, custom_span, gen_trace_id, trace
from agent import History, historical_agent
from agent import Culinary,culinary_agent
from agent import Culture,culture_agent
from agent import Architecture,architecture_agent
from agent import Planner, planner_agent
from agent import FinalTour, orchestrator_agent
from printer import Printer
class TourManager:
"""
Orchestrates the full flow
"""
def __init__(self) -> None:
self.console = Console()
self.printer = Printer(self.console)
async def run(self, query: str, interest: str, duration: str) -> None:
trace_id = gen_trace_id()
with trace("Tour Research trace", trace_id=trace_id):
self.printer.update_item(
"trace_id",
f"View trace: https://platform.openai.com/traces/{trace_id}",
is_done=True,
hide_checkmark=True,
)
self.printer.update_item("start", "Starting tour research...", is_done=True)
planner = await self._get_plan(query, interest, duration)
print(planner)
architecture_research = await self._get_architecture(query, interest, planner.architecture)
historical_research = await self._get_history(query, interest, planner.history)
culinary_research = await self._get_culinary(query, interest, planner.culinary)
culture_research = await self._get_culture(query, interest, planner.culture)
final_tour = await self._get_final_tour(query, interest, duration, architecture_research.output, planner.architecture, historical_research.output, planner.history, culinary_research.output, planner.culinary, culture_research.output, planner.culture)
self.printer.update_item("final_report", "", is_done=True)
self.printer.end()
tour_intro = final_tour.introduction
architecture = final_tour.architecture
history = final_tour.history
culinary = final_tour.culinary
culture = final_tour.culture
conclusion = final_tour.conclusion
final = (
"INTRODUCTION\n\n"
+ tour_intro + "\n\n"
+ "ARCHITECTURE\n\n"
+ architecture + "\n\n"
+ "HISTORY\n\n"
+ history + "\n\n"
+ "CULTURE\n\n"
+ culture + "\n\n"
+ "CULINARY\n\n"
+ culinary + "\n\n"
+ "CONCLUSION\n\n"
+ conclusion
)
return final
async def _get_plan(self, query: str, interest: str, duration: str) -> Planner:
self.printer.update_item("Planner", "Getting Time allocation for each section")
result = await Runner.run(planner_agent, f"Query: {query} Interest: {interest} Duration: {duration}")
self.printer.update_item(
"Planner",
f"Completed planning",
is_done=True,
)
return result.final_output_as(Planner)
async def _get_history(self, query: str, interest: str, duration: float) -> History:
self.printer.update_item("History", "Getting Historical Data for Location")
word_limit = int(duration) * 120
result = await Runner.run(historical_agent, f"Query: {query} Interest: {interest} Word Limit: {word_limit} - {word_limit + 20}")
self.printer.update_item(
"History",
f"Completed history research",
is_done=True,
)
return result.final_output_as(History)
# return result.final_output_as(FinancialSearchPlan)
async def _get_architecture(self, query: str, interest: str, duration: float):
self.printer.update_item("Architecture", "Getting Architectural Data for Location")
word_limit = int(duration) * 120
result = await Runner.run(architecture_agent, f"Query: {query} Interest: {interest} Word Limit: {word_limit} - {word_limit + 20}")
self.printer.update_item(
"Architecture",
f"Completed architecture research",
is_done=True,
)
return result.final_output_as(Architecture)
async def _get_culinary(self, query: str, interest: str, duration: float):
self.printer.update_item("Culinary", "Getting Culinary Data for Location")
word_limit = int(duration) * 120
result = await Runner.run(culinary_agent, f"Query: {query} Interest: {interest} Word Limit: {word_limit} - {word_limit + 20}")
self.printer.update_item(
"Culinary",
f"Completed culinary research",
is_done=True,
)
return result.final_output_as(Culinary)
async def _get_culture(self, query: str, interest: str, duration: float):
self.printer.update_item("Culture", "Getting Cultural Data for Location")
word_limit = int(duration) * 120
result = await Runner.run(culture_agent, f"Query: {query} Interest: {interest} Word Limit: {word_limit} - {word_limit + 20}")
self.printer.update_item(
"Culture",
f"Completed culture research",
is_done=True,
)
return result.final_output_as(Culture)
async def _get_final_tour(self, query: str, interest: str, duration: float, architecture: str, architecture_dur: float, history: str, history_dur: float, culinary: str, culinary_dur: float, culture: str, culture_dur:float):
self.printer.update_item("Final Tour", "Getting Final Tour")
result = await Runner.run(
orchestrator_agent,
f"""Query: {query}
Interest: {interest}
Total Tour Duration (in minutes): {duration}
Word Limit Allocation:
- Architecture: {architecture_dur*100}
- History: {history_dur*100}
- Culture: {culture_dur*100}
- Culinary: {culinary_dur*100}
Content Sections:
Architecture:
{architecture}
History:
{history}
Culture:
{culture}
Culinary:
{culinary}
"""
)
self.printer.update_item(
"Final Tour",
f"Completed Final Tour Guide Creation",
is_done=True,
)
return result.final_output_as(FinalTour)

View file

@ -0,0 +1,46 @@
from typing import Any
from rich.console import Console, Group
from rich.live import Live
from rich.spinner import Spinner
class Printer:
"""
Simple wrapper to stream status updates. Used by the financial bot
manager as it orchestrates planning, search and writing.
"""
def __init__(self, console: Console) -> None:
self.live = Live(console=console)
self.items: dict[str, tuple[str, bool]] = {}
self.hide_done_ids: set[str] = set()
self.live.start()
def end(self) -> None:
self.live.stop()
def hide_done_checkmark(self, item_id: str) -> None:
self.hide_done_ids.add(item_id)
def update_item(
self, item_id: str, content: str, is_done: bool = False, hide_checkmark: bool = False
) -> None:
self.items[item_id] = (content, is_done)
if hide_checkmark:
self.hide_done_ids.add(item_id)
self.flush()
def mark_item_done(self, item_id: str) -> None:
self.items[item_id] = (self.items[item_id][0], True)
self.flush()
def flush(self) -> None:
renderables: list[Any] = []
for item_id, (content, is_done) in self.items.items():
if is_done:
prefix = "" if item_id not in self.hide_done_ids else ""
renderables.append(prefix + content)
else:
renderables.append(Spinner("dots", text=content))
self.live.update(Group(*renderables))

View file

@ -0,0 +1,7 @@
openai==1.68.2
openai-agents==0.0.6
pydantic==2.10.6
pydantic_core==2.27.2
python-dotenv==1.0.1
rich==13.9.4
streamlit==1.43.2