EXECUTIVE SUMMARY
This article presents a comprehensive exploration of building an intelligent travel planning system powered by Large Language Models operating in an agentic architecture. The system orchestrates multiple specialized agents that collaborate to deliver complete travel itineraries based on user preferences. We examine the architectural foundations, implementation strategies, agent coordination mechanisms, and practical code implementations that enable such a system to function effectively across diverse hardware platforms and LLM backends.
INTRODUCTION TO AGENTIC AI FOR TRAVEL PLANNING
The travel planning domain presents unique challenges that make it an ideal candidate for agentic AI systems. Traditional rule-based travel booking systems require users to navigate through rigid interfaces, making multiple sequential decisions without holistic optimization. An LLM-based agentic system transforms this experience by understanding natural language requests, proactively gathering missing information, and coordinating multiple specialized agents to research flights, accommodations, weather, attractions, and logistical details simultaneously.
The fundamental architecture consists of a coordinator agent that interprets user intent and delegates tasks to specialized agents. Each specialist agent possesses domain expertise in a specific area such as flight booking, hotel reservation, weather forecasting, or tourist information. These agents utilize various tools including web search capabilities, API integrations with booking platforms, and knowledge retrieval systems. The coordinator synthesizes their findings into a coherent, structured travel plan that addresses all user requirements.
ARCHITECTURAL FOUNDATIONS
The system architecture follows clean architecture principles with clear separation between domain logic, application services, and infrastructure concerns. At the core lies the agent orchestration layer that manages communication between agents, maintains conversation state, and ensures coherent execution flow. This layer abstracts away the complexities of LLM interaction, allowing agents to focus on their specialized tasks.
The infrastructure layer provides hardware-agnostic LLM execution capabilities. Supporting multiple GPU architectures including NVIDIA CUDA, AMD ROCm, Intel GPUs, and Apple Metal Performance Shaders requires careful abstraction of the acceleration layer. The system detects available hardware at runtime and configures the appropriate backend automatically.
A critical architectural decision involves choosing between local and remote LLM deployment. Local deployment offers privacy, reduced latency, and independence from external services but requires substantial computational resources. Remote deployment via API services provides access to larger, more capable models without local hardware constraints. The architecture supports both modes seamlessly through a unified interface.
PARAMETER COLLECTION AND VALIDATION
When a user initiates a travel planning request, the system must gather comprehensive information about their requirements. The coordinator agent analyzes the initial request to identify which parameters have been provided and which remain missing. Required parameters include the travel period with specific dates, origin location, destination, travel purpose distinguishing business from leisure trips, preferred departure and return times, transportation mode preferences, and hotel rating requirements.
The coordinator employs a conversational approach to collect missing information. Rather than presenting a rigid form, it engages in natural dialogue, asking clarifying questions that feel contextual and relevant. For instance, if a user mentions traveling to Paris but omits dates, the agent might ask about preferred travel months while simultaneously providing information about seasonal considerations that might influence the decision.
Parameter validation occurs continuously throughout the collection process. Date ranges are verified for logical consistency, ensuring return dates follow departure dates. Location specifications are disambiguated when multiple places share names. Transportation preferences are validated against route feasibility. Hotel rating ranges are normalized to standard classification systems. This validation prevents downstream errors and ensures agents receive well-formed queries.
SPECIALIZED AGENT ARCHITECTURE
Each specialized agent in the system possesses distinct capabilities and responsibilities. The Transportation Agent handles all aspects of journey logistics including flight searches, train schedules, bus routes, and car
rental options. It interfaces with multiple transportation APIs and web scraping tools to gather comprehensive options. The agent considers not just primary transportation but also last-mile connectivity, researching public transit routes from airports or stations to hotels.
The Accommodation Agent specializes in finding suitable lodging within specified rating and budget constraints. It queries hotel booking platforms, compares prices across providers, and extracts detailed information about room types, amenities, breakfast inclusion, cancellation policies, and guest reviews. The agent applies sophisticated filtering to match user preferences while maximizing value.
The Weather Intelligence Agent provides meteorological forecasts for the destination during the travel period. It accesses weather APIs and historical climate data to predict conditions, enabling travelers to pack appropriately and plan outdoor activities. When long-range forecasts are unavailable, it provides historical averages and seasonal patterns.
The Attractions and Events Agent discovers points of interest, cultural sites, entertainment venues, and scheduled events occurring during the visit. It considers the travel purpose, suggesting business-relevant venues for corporate trips and leisure attractions for vacations. The agent provides practical details including operating hours, admission costs, and booking requirements.
The Travel Intelligence Agent handles regulatory and practical information including visa requirements, currency exchange rates, local customs, safety advisories, and health recommendations. It ensures travelers are prepared for legal and cultural aspects of their destination.
TOOL INTEGRATION AND WEB INTERACTION
Agents leverage various tools to gather information from external sources. Web search tools enable agents to query general information and discover relevant websites. These tools must handle rate limiting, parse diverse HTML structures, and extract meaningful content from search results. The implementation uses robust parsing libraries that gracefully handle malformed markup and extract text while preserving semantic structure.
API integration tools provide structured access to specialized services. Flight booking APIs return standardized data about available flights including departure times, arrival times, layovers, aircraft types, and pricing. Hotel APIs deliver room availability, rates, and property details. Weather APIs provide forecast data in machine-readable formats. Currency exchange APIs supply current conversion rates.
The tool execution framework provides a consistent interface for agents to invoke capabilities. Each tool defines its input schema, output format, and error handling behavior. The framework manages authentication, retries failed requests with exponential backoff, caches responses when appropriate, and logs all interactions for debugging and audit purposes.
LLM BACKEND ABSTRACTION
Supporting multiple LLM backends requires a well-designed abstraction layer that isolates model-specific details from agent logic. The system defines a common interface for text generation, embedding creation, and token counting. Concrete implementations handle the specifics of different model families and serving frameworks.
For local deployment, the system supports multiple inference engines. The llama.cpp backend provides efficient CPU and GPU inference for LLAMA-family models with support for quantization. The vLLM backend offers optimized serving for various architectures with advanced batching and memory management. The Ollama integration provides a user-friendly local deployment option with automatic model management.
Remote deployment connects to API services including OpenAI, Anthropic, Google, and Cohere. The abstraction layer normalizes their different request formats, response structures, and error codes into a unified interface. Rate limiting and cost tracking are implemented at this layer to prevent unexpected expenses.
HARDWARE ACCELERATION CONFIGURATION
Detecting and configuring appropriate hardware acceleration is essential for performant local LLM execution. The system probes available hardware at startup and selects the optimal backend automatically. For NVIDIA GPUs, it verifies CUDA availability and version compatibility. For AMD GPUs, it checks for ROCm installation and supported device types. For Apple Silicon, it detects Metal Performance Shaders availability. For Intel GPUs, it verifies oneAPI and SYCL support.
The configuration system allows manual override when automatic detection fails or when users prefer specific backends. Environment variables and configuration files specify preferred acceleration methods, model paths, quantization levels, context lengths, and generation parameters. The system validates configurations before initialization to provide clear error messages when requirements are not met.
Memory management varies significantly across hardware platforms. NVIDIA CUDA provides explicit memory allocation and transfer APIs. AMD ROCm follows similar patterns but with different library calls. Apple MPS uses unified memory architecture requiring different optimization strategies. Intel GPUs leverage shared memory with CPU. The abstraction layer handles these differences transparently.
CONVERSATION STATE MANAGEMENT
Maintaining coherent conversation state across multiple agent interactions
requires careful design. The system tracks the complete dialogue history, current parameter values, agent execution status, and intermediate results. This state enables the coordinator to make informed decisions about which agents to invoke next and how to synthesize their outputs.
The state representation uses a structured format that captures both user messages and agent responses. Each message includes metadata about its source, timestamp, and associated tool invocations. Parameters are stored in a typed dictionary with validation rules ensuring consistency. Agent results are cached to avoid redundant API calls when users refine their requirements.
State persistence allows users to pause and resume planning sessions. The system serializes conversation state to durable storage, enabling multi-
session planning workflows. Users can review previous plans, compare alternatives, or modify parameters without starting from scratch.
COORDINATOR AGENT LOGIC
The coordinator agent serves as the orchestrator that manages the overall planning workflow. It begins by analyzing the user's initial request using the LLM to extract mentioned parameters and identify gaps. The coordinator generates a parameter collection plan, determining which missing information is critical versus optional and the optimal order for gathering it.
Once sufficient parameters are collected, the coordinator creates an execution plan that determines which specialized agents to invoke and in what sequence. Some agents can execute in parallel since their tasks are independent. For example, weather forecasting and attraction discovery can proceed simultaneously. Other agents have dependencies requiring sequential execution. Hotel searches should consider proximity to planned activities.
The coordinator monitors agent execution, handling failures gracefully. When an agent encounters an error such as an API timeout or invalid response, the coordinator decides whether to retry, invoke an alternative tool, or inform the user about the limitation. This resilience ensures the system degrades gracefully rather than failing completely.
TRANSPORTATION RESEARCH IMPLEMENTATION
The Transportation Agent implements sophisticated logic for discovering travel options. For flight searches, it queries multiple airline APIs and aggregator services to find available routes. The agent considers both direct flights and connections, evaluating total travel time, layover durations, and price differences. It filters results based on user preferences for departure and arrival times, applying fuzzy matching to accommodate flexibility.
Train research involves querying rail operator APIs and schedule databases. The agent identifies all possible routes between origin and destination, including transfers. It retrieves pricing information for different service classes, seat availability, and booking conditions. For international rail travel, it handles cross-border ticketing complexities.
Bus route discovery uses intercity bus operator APIs and schedule aggregators. The agent finds both express and local services, comparing journey durations and comfort levels. It includes budget carriers that might not appear in mainstream search engines.
Car rental research queries major rental agencies and comparison platforms. The agent considers pickup and dropoff locations, vehicle categories matching user needs, insurance options, and fuel policies. It calculates estimated fuel costs based on distance and vehicle efficiency.
A critical component of transportation research is last-mile connectivity. After identifying primary transportation options, the agent researches how to reach the final destination from arrival points. It queries public transit APIs for bus, metro, and tram routes from airports and stations to hotels. It provides walking directions when destinations are nearby. It estimates taxi costs and checks ride-sharing availability.
ACCOMMODATION DISCOVERY PROCESS
The Accommodation Agent searches for suitable lodging by querying hotel booking platforms through their APIs. It filters properties based on the specified star rating range, ensuring only hotels meeting minimum quality standards appear in results. The agent retrieves detailed property information including exact addresses, distance from city centers or business districts, available room types, and amenity lists.
Price comparison across multiple booking platforms ensures users receive
competitive rates. The agent identifies the same property listed on different services and compares their prices, noting any exclusive deals or member discounts. It flags price differences that might indicate different cancellation policies or payment terms.
Room type analysis examines available accommodations within each property. The agent distinguishes between standard rooms, suites, apartments, and specialty accommodations. It notes bed configurations, room sizes, views, and special features like balconies or kitchenettes. For business travelers, it prioritizes rooms with work desks and reliable internet. For leisure travelers, it highlights recreational amenities.
Breakfast inclusion significantly impacts total trip costs. The agent clearly indicates whether room rates include breakfast, and if so, what type of breakfast is provided. Continental, buffet, and full breakfast options are distinguished. When breakfast is not included, the agent estimates costs based on local restaurant prices.
Cancellation policies affect booking flexibility. The agent extracts and summarizes cancellation terms, noting free cancellation deadlines, partial refund conditions, and non-refundable rates. It calculates potential savings from non-refundable bookings versus the flexibility value of free cancellation.
WEATHER FORECASTING INTEGRATION
The Weather Intelligence Agent provides meteorological insights that inform packing decisions and activity planning. For trips within the next ten days, it retrieves detailed forecasts from weather APIs including predicted temperatures, precipitation probability, wind speeds, and humidity levels. The agent presents daily forecasts showing morning, afternoon, and evening conditions.
For trips beyond the reliable forecast horizon, the agent provides historical climate data. It analyzes average temperatures, typical precipitation patterns, and seasonal weather characteristics for the destination during the travel period. This historical perspective helps travelers understand what conditions to expect even when specific forecasts are unavailable.
Extreme weather alerts are highlighted prominently. The agent checks for storm warnings, heat advisories, air quality alerts, or other meteorological hazards that might affect travel plans. It suggests alternative dates if severe weather is predicted during the planned visit.
Seasonal considerations extend beyond basic weather. The agent notes phenomena like monsoon seasons, hurricane periods, or extreme temperature events that characterize certain times of year. It provides context about how weather affects local activities, such as beach closures during winter or ski season timing.
ATTRACTIONS AND EVENTS DISCOVERY
The Attractions and Events Agent curates personalized suggestions for activities during the visit. It begins by identifying major tourist attractions including museums, historical sites, natural landmarks, and cultural venues. The agent retrieves practical information such as opening hours, admission prices, guided tour availability, and advance booking requirements.
Event discovery involves searching local event calendars, venue schedules, and entertainment listings. The agent identifies concerts, theater performances, sporting events, festivals, and exhibitions occurring during the travel dates. It provides ticket pricing, venue locations, and booking links.
Activity recommendations are tailored to the travel purpose. For business trips, the agent suggests networking venues, coworking spaces, and business-friendly restaurants. For leisure trips, it emphasizes recreational activities, shopping districts, and entertainment options. Family trips receive suggestions for child-friendly attractions and activities.
The agent considers temporal factors when suggesting activities. It notes which attractions are best visited at specific times of day, such as sunset viewpoints or evening entertainment districts. It identifies attractions with limited hours or specific operating days, helping travelers optimize their schedules.
Proximity analysis clusters nearby attractions, enabling efficient itinerary planning. The agent groups geographically close points of interest, suggesting logical touring sequences that minimize travel time between locations. It estimates time requirements for each activity, helping travelers gauge how many attractions they can realistically visit.
TRAVEL INTELLIGENCE AND REGULATORY INFORMATION
The Travel Intelligence Agent provides essential information about legal, financial, and safety aspects of international travel. Visa requirement research determines whether travelers need visas for their destination based on their citizenship. The agent identifies visa types, application processes, processing times, fees, and required documentation. It provides links to official embassy websites and application portals.
Currency information includes current exchange rates between the traveler's home currency and the destination currency. The agent calculates approximate costs in familiar terms, converting hotel prices, meal expenses, and attraction fees. It advises on currency exchange options including airport kiosks, local banks, and ATM withdrawals, noting typical fees and commission rates.
Safety advisories aggregate information from government travel advisory services. The agent summarizes security situations, noting areas to avoid, common scams targeting tourists, and recommended precautions. It provides emergency contact numbers including local police, ambulance services, and embassy contacts.
Health recommendations cover vaccination requirements, disease risks, and medical facility availability. The agent notes mandatory vaccinations for entry, recommended immunizations for disease prevention, and health insurance considerations. It identifies quality medical facilities in the destination for emergency situations.
Cultural information helps travelers navigate local customs and etiquette. The agent provides guidance on appropriate dress codes, tipping practices, greeting customs, and social norms. This cultural intelligence prevents unintentional offense and enhances the travel experience.
RESULT SYNTHESIS AND REPORT GENERATION
After all specialized agents complete their research, the coordinator synthesizes their findings into a comprehensive travel report. The synthesis process involves organizing information logically, resolving conflicts between different data sources, and presenting options clearly.
The report structure follows a standardized format that users can easily navigate. It begins with a summary of the travel parameters including dates, origin, destination, and key preferences. This summary confirms the coordinator understood the request correctly.
Transportation options are presented in a comparison table format showing departure times, arrival times, duration, number of stops, and total costs for each option. The coordinator highlights recommended options based on optimal combinations of price, convenience, and timing. It explains the rationale for recommendations, such as preferring direct flights despite higher costs or suggesting overnight trains to save accommodation expenses.
Accommodation options are similarly structured with property names, star ratings, room types, nightly rates, total costs, breakfast inclusion, and cancellation policies. The coordinator identifies best value options and
premium choices, allowing travelers to make informed tradeoffs between cost and comfort.
Weather forecasts are integrated into the itinerary, showing expected conditions for each day of the trip. The coordinator suggests packing lists based on predicted weather, recommending rain gear for wet forecasts or sun protection for hot conditions.
Attractions and events are organized by day and geographic area. The coordinator proposes sample itineraries that group nearby attractions efficiently. It notes booking requirements and suggests reservation timing to secure availability.
Travel intelligence information appears in a dedicated section covering visas, currency, safety, health, and cultural considerations. The coordinator highlights any critical requirements like visa applications that need immediate attention.
Cost summaries aggregate all expenses including transportation, accommodation, meals, attractions, and miscellaneous costs. The coordinator provides both itemized breakdowns and total trip estimates in the traveler's home currency. It notes which costs are fixed versus variable and identifies opportunities for savings.
IMPLEMENTATION EXAMPLE ARCHITECTURE
To illustrate these concepts concretely, we now examine a detailed implementation. The system is built using Python for its rich ecosystem of libraries supporting LLM integration, web scraping, and API interaction. The architecture separates concerns into distinct modules for maintainability and testability.
The core agent framework defines base classes for agents and tools. Agents inherit from a common base that provides conversation management, tool invocation, and state tracking. Tools inherit from a base that standardizes input validation, execution, and error handling.
The LLM backend abstraction uses a factory pattern to instantiate appropriate providers based on configuration. Each provider implements a common interface for text generation, ensuring agent code remains independent of specific model implementations.
Hardware detection uses platform-specific libraries to probe GPU availability. The system attempts to import CUDA libraries for NVIDIA GPUs, ROCm libraries for AMD GPUs, Metal libraries for Apple Silicon, and Intel extension libraries for Intel GPUs. Based on successful imports and device queries, it selects the optimal backend.
Configuration management uses a hierarchical system combining default values, configuration files, and environment variables. Users can override defaults without modifying code, enabling flexible deployment across different environments.
The following code demonstrates the foundational abstractions:
import abc
import enum
import json
import os
import logging
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime, date
# Configure logging for the entire application
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class HardwareBackend(enum.Enum):
"""Enumeration of supported hardware acceleration backends."""
CUDA = "cuda"
ROCM = "rocm"
MPS = "mps"
INTEL = "intel"
CPU = "cpu"
class LLMProvider(enum.Enum):
"""Enumeration of supported LLM providers."""
OPENAI = "openai"
ANTHROPIC = "anthropic"
GOOGLE = "google"
LOCAL_LLAMA = "local_llama"
LOCAL_VLLM = "local_vllm"
OLLAMA = "ollama"
@dataclass
class TravelParameters:
"""Structured representation of travel planning parameters."""
origin: Optional[str] = None
destination: Optional[str] = None
start_date: Optional[date] = None
end_date: Optional[date] = None
travel_purpose: Optional[str] = None # "business" or "leisure"
preferred_departure_time: Optional[str] = None
preferred_return_time: Optional[str] = None
transport_modes: List[str] = field(default_factory=list)
min_hotel_rating: Optional[int] = None
max_hotel_rating: Optional[int] = None
budget_max: Optional[float] = None
travelers_count: int = 1
def is_complete(self) -> bool:
"""Check if all required parameters have been collected."""
required_fields = [
self.origin,
self.destination,
self.start_date,
self.end_date,
self.travel_purpose,
self.transport_modes,
self.min_hotel_rating,
self.max_hotel_rating
]
return all(field is not None for field in required_fields) and len(self.transport_modes) > 0
def missing_parameters(self) -> List[str]:
"""Return list of parameter names that are still missing."""
missing = []
if not self.origin:
missing.append("origin location")
if not self.destination:
missing.append("destination")
if not self.start_date:
missing.append("start date")
if not self.end_date:
missing.append("end date")
if not self.travel_purpose:
missing.append("travel purpose (business or leisure)")
if not self.transport_modes:
missing.append("preferred transportation modes")
if self.min_hotel_rating is None:
missing.append("minimum hotel rating")
if self.max_hotel_rating is None:
missing.append("maximum hotel rating")
return missing
This foundational code establishes the parameter structure that drives the entire planning process. The TravelParameters dataclass uses Python's type hints to ensure type safety and provides methods to check completeness and identify missing information. The enumeration types provide type-safe constants for hardware backends and LLM providers.
The hardware detection module probes available acceleration capabilities:
class HardwareDetector:
"""Detects available hardware acceleration capabilities."""
@staticmethod
def detect_backend() -> HardwareBackend:
"""
Automatically detect the best available hardware backend.
Tries backends in order of preference: CUDA, ROCm, MPS, Intel, CPU.
"""
# Try NVIDIA CUDA
try:
import torch
if torch.cuda.is_available():
device_count = torch.cuda.device_count()
device_name = torch.cuda.get_device_name(0)
logger.info(f"Detected CUDA backend with {device_count} device(s): {device_name}")
return HardwareBackend.CUDA
except ImportError:
pass
except Exception as e:
logger.warning(f"CUDA detection failed: {e}")
# Try AMD ROCm
try:
import torch
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
if torch.cuda.is_available(): # ROCm uses cuda API
device_count = torch.cuda.device_count()
logger.info(f"Detected ROCm backend with {device_count} device(s)")
return HardwareBackend.ROCM
except Exception as e:
logger.warning(f"ROCm detection failed: {e}")
# Try Apple Metal Performance Shaders
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
logger.info("Detected Apple MPS backend")
return HardwareBackend.MPS
except Exception as e:
logger.warning(f"MPS detection failed: {e}")
# Try Intel GPU
try:
import intel_extension_for_pytorch as ipex
if ipex.xpu.is_available():
device_count = ipex.xpu.device_count()
logger.info(f"Detected Intel GPU backend with {device_count} device(s)")
return HardwareBackend.INTEL
except ImportError:
pass
except Exception as e:
logger.warning(f"Intel GPU detection failed: {e}")
# Fallback to CPU
logger.info("No GPU acceleration detected, using CPU backend")
return HardwareBackend.CPU
@staticmethod
def get_device_string(backend: HardwareBackend) -> str:
"""Convert backend enum to device string for framework usage."""
device_map = {
HardwareBackend.CUDA: "cuda",
HardwareBackend.ROCM: "cuda", # ROCm uses CUDA API
HardwareBackend.MPS: "mps",
HardwareBackend.INTEL: "xpu",
HardwareBackend.CPU: "cpu"
}
return device_map.get(backend, "cpu")
The hardware detector attempts to import and query each acceleration framework in order of preference. NVIDIA CUDA receives highest priority due to its maturity and widespread support. AMD ROCm follows, leveraging the CUDA-compatible API. Apple MPS enables GPU acceleration on Apple Silicon. Intel GPUs use the Intel Extension for PyTorch. When no GPU is available, the system falls back to CPU execution.
The LLM abstraction layer defines a common interface for text generation:
class LLMBackend(abc.ABC):
"""Abstract base class for LLM backends."""
@abc.abstractmethod
def generate(
self,
prompt: str,
max_tokens: int = 1024,
temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None
) -> str:
"""
Generate text completion for the given prompt.
Args:
prompt: Input text to complete
max_tokens: Maximum tokens to generate
temperature: Sampling temperature (0.0 to 1.0)
stop_sequences: List of sequences that stop generation
Returns:
Generated text completion
"""
pass
@abc.abstractmethod
def count_tokens(self, text: str) -> int:
"""Count the number of tokens in the given text."""
pass
class OpenAIBackend(LLMBackend):
"""OpenAI API backend implementation."""
def __init__(self, api_key: str, model: str = "gpt-4"):
"""
Initialize OpenAI backend.
Args:
api_key: OpenAI API key
model: Model identifier to use
"""
self.api_key = api_key
self.model = model
try:
from openai import OpenAI
self.client = OpenAI(api_key=api_key)
except ImportError:
raise ImportError("OpenAI library not installed. Install with: pip install openai")
def generate(
self,
prompt: str,
max_tokens: int = 1024,
temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None
) -> str:
"""Generate text using OpenAI API."""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
temperature=temperature,
stop=stop_sequences
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"OpenAI generation failed: {e}")
raise
def count_tokens(self, text: str) -> int:
"""Estimate token count using tiktoken."""
try:
import tiktoken
encoding = tiktoken.encoding_for_model(self.model)
return len(encoding.encode(text))
except ImportError:
# Rough estimation if tiktoken not available
return len(text.split()) * 1.3
class LocalLlamaBackend(LLMBackend):
"""Local LLAMA model backend using llama-cpp-python."""
def __init__(
self,
model_path: str,
hardware_backend: HardwareBackend,
n_ctx: int = 4096,
n_gpu_layers: int = 35
):
"""
Initialize local LLAMA backend.
Args:
model_path: Path to GGUF model file
hardware_backend: Hardware acceleration backend to use
n_ctx: Context window size
n_gpu_layers: Number of layers to offload to GPU
"""
self.model_path = model_path
self.hardware_backend = hardware_backend
try:
from llama_cpp import Llama
# Configure GPU acceleration based on backend
gpu_layers = n_gpu_layers if hardware_backend != HardwareBackend.CPU else 0
self.model = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_gpu_layers=gpu_layers,
verbose=False
)
logger.info(f"Loaded LLAMA model from {model_path} with {gpu_layers} GPU layers")
except ImportError:
raise ImportError("llama-cpp-python not installed. Install with: pip install llama-cpp-python")
def generate(
self,
prompt: str,
max_tokens: int = 1024,
temperature: float = 0.7,
stop_sequences: Optional[List[str]] = None
) -> str:
"""Generate text using local LLAMA model."""
try:
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
stop=stop_sequences or [],
echo=False
)
return response['choices'][0]['text']
except Exception as e:
logger.error(f"Local LLAMA generation failed: {e}")
raise
def count_tokens(self, text: str) -> int:
"""Count tokens using model's tokenizer."""
tokens = self.model.tokenize(text.encode('utf-8'))
return len(tokens)
The LLM abstraction defines a common interface that all backends must implement. The OpenAI backend wraps the OpenAI API, handling authentication and request formatting. The LocalLlamaBackend uses llama-cpp-python for efficient local inference with GPU acceleration. Both implementations provide token counting capabilities essential for managing context windows.
The tool framework provides a structured way for agents to interact with external services:
@dataclass
class ToolResult:
"""Result returned by tool execution."""
success: bool
data: Any
error_message: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
class Tool(abc.ABC):
"""Abstract base class for agent tools."""
def __init__(self, name: str, description: str):
"""
Initialize tool with name and description.
Args:
name: Unique tool identifier
description: Human-readable description of tool capabilities
"""
self.name = name
self.description = description
@abc.abstractmethod
def execute(self, **kwargs) -> ToolResult:
"""
Execute the tool with given parameters.
Args:
**kwargs: Tool-specific parameters
Returns:
ToolResult containing execution outcome
"""
pass
@abc.abstractmethod
def get_parameters_schema(self) -> Dict[str, Any]:
"""
Return JSON schema describing tool parameters.
Returns:
Dictionary containing parameter schema
"""
pass
This tool abstraction enables agents to invoke capabilities without knowing implementation details. Each tool declares its parameters through a schema and returns standardized results. The ToolResult dataclass captures success status, returned data, error messages, and metadata.
The agent base class provides common functionality for all specialized agents:
class Agent:
"""Base class for all specialized agents."""
def __init__(
self,
name: str,
description: str,
llm_backend: LLMBackend,
tools: Optional[List[Tool]] = None
):
"""
Initialize agent with name, description, LLM backend, and tools.
Args:
name: Agent identifier
description: Agent capabilities description
llm_backend: LLM backend for text generation
tools: List of tools available to this agent
"""
self.name = name
self.description = description
self.llm_backend = llm_backend
self.tools = tools or []
self.conversation_history: List[Dict[str, str]] = []
def add_message(self, role: str, content: str):
"""Add message to conversation history."""
self.conversation_history.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def build_prompt(self, user_message: str, system_prompt: Optional[str] = None) -> str:
"""
Build complete prompt including system instructions and conversation history.
Args:
user_message: Current user message
system_prompt: Optional system instructions
Returns:
Formatted prompt string
"""
prompt_parts = []
if system_prompt:
prompt_parts.append(f"System: {system_prompt}\n")
# Include recent conversation history for context
for msg in self.conversation_history[-5:]: # Last 5 messages
role = msg['role'].capitalize()
content = msg['content']
prompt_parts.append(f"{role}: {content}\n")
prompt_parts.append(f"User: {user_message}\n")
prompt_parts.append("Assistant:")
return "\n".join(prompt_parts)
def execute_tool(self, tool_name: str, **kwargs) -> ToolResult:
"""
Execute a tool by name with given parameters.
Args:
tool_name: Name of tool to execute
**kwargs: Tool parameters
Returns:
ToolResult from tool execution
"""
tool = next((t for t in self.tools if t.name == tool_name), None)
if not tool:
return ToolResult(
success=False,
data=None,
error_message=f"Tool '{tool_name}' not found"
)
try:
logger.info(f"Agent {self.name} executing tool {tool_name}")
result = tool.execute(**kwargs)
logger.info(f"Tool {tool_name} completed: success={result.success}")
return result
except Exception as e:
logger.error(f"Tool {tool_name} execution failed: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
def process(self, message: str) -> str:
"""
Process a message and generate response.
Args:
message: Input message to process
Returns:
Generated response
"""
self.add_message("user", message)
system_prompt = f"You are {self.name}, a specialized agent. {self.description}"
prompt = self.build_prompt(message, system_prompt)
response = self.llm_backend.generate(prompt, max_tokens=1024, temperature=0.7)
self.add_message("assistant", response)
return response
The Agent base class manages conversation history, builds prompts with context, executes tools, and generates responses using the LLM backend. Specialized agents inherit this functionality and add domain-specific logic.
TRANSPORTATION AGENT IMPLEMENTATION
The Transportation Agent extends the base agent with capabilities for researching travel options. It uses tools to query flight APIs, train schedules, bus routes, and public transit information. The agent synthesizes results from multiple sources to provide comprehensive transportation options.
class FlightSearchTool(Tool):
"""Tool for searching flight options."""
def __init__(self, api_key: Optional[str] = None):
"""
Initialize flight search tool.
Args:
api_key: API key for flight search service
"""
super().__init__(
name="flight_search",
description="Search for available flights between origin and destination"
)
self.api_key = api_key or os.getenv("FLIGHT_API_KEY")
def get_parameters_schema(self) -> Dict[str, Any]:
"""Return parameter schema for flight search."""
return {
"type": "object",
"properties": {
"origin": {"type": "string", "description": "Origin airport code or city"},
"destination": {"type": "string", "description": "Destination airport code or city"},
"departure_date": {"type": "string", "format": "date", "description": "Departure date (YYYY-MM-DD)"},
"return_date": {"type": "string", "format": "date", "description": "Return date (YYYY-MM-DD)"},
"passengers": {"type": "integer", "minimum": 1, "description": "Number of passengers"}
},
"required": ["origin", "destination", "departure_date"]
}
def execute(self, **kwargs) -> ToolResult:
"""
Execute flight search with given parameters.
Args:
origin: Origin location
destination: Destination location
departure_date: Departure date string
return_date: Optional return date string
passengers: Number of passengers
Returns:
ToolResult containing flight options
"""
origin = kwargs.get("origin")
destination = kwargs.get("destination")
departure_date = kwargs.get("departure_date")
return_date = kwargs.get("return_date")
passengers = kwargs.get("passengers", 1)
try:
# In production, this would call actual flight API
# For this implementation, we use a real API client
import requests
# Using Amadeus Flight Offers Search API as example
# First, get access token
auth_url = "https://test.api.amadeus.com/v1/security/oauth2/token"
auth_data = {
"grant_type": "client_credentials",
"client_id": os.getenv("AMADEUS_API_KEY"),
"client_secret": os.getenv("AMADEUS_API_SECRET")
}
auth_response = requests.post(auth_url, data=auth_data)
if auth_response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message="Failed to authenticate with flight API"
)
access_token = auth_response.json()["access_token"]
# Search for flights
search_url = "https://test.api.amadeus.com/v2/shopping/flight-offers"
headers = {"Authorization": f"Bearer {access_token}"}
params = {
"originLocationCode": origin,
"destinationLocationCode": destination,
"departureDate": departure_date,
"adults": passengers,
"max": 10
}
if return_date:
params["returnDate"] = return_date
search_response = requests.get(search_url, headers=headers, params=params)
if search_response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message=f"Flight search failed: {search_response.text}"
)
flight_data = search_response.json()
# Parse and structure flight offers
flights = []
for offer in flight_data.get("data", []):
for itinerary in offer.get("itineraries", []):
flight_info = {
"price": float(offer["price"]["total"]),
"currency": offer["price"]["currency"],
"segments": []
}
for segment in itinerary.get("segments", []):
flight_info["segments"].append({
"departure": {
"airport": segment["departure"]["iataCode"],
"time": segment["departure"]["at"]
},
"arrival": {
"airport": segment["arrival"]["iataCode"],
"time": segment["arrival"]["at"]
},
"carrier": segment["carrierCode"],
"flight_number": segment["number"],
"duration": segment["duration"]
})
flights.append(flight_info)
return ToolResult(
success=True,
data={"flights": flights},
metadata={
"search_params": params,
"result_count": len(flights)
}
)
except Exception as e:
logger.error(f"Flight search error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
The FlightSearchTool integrates with the Amadeus flight search API to retrieve real flight data. It handles authentication, constructs search queries, and parses responses into structured flight information. The tool returns comprehensive details including prices, segments, departure and arrival times, carriers, and flight numbers.
The train search tool follows a similar pattern:
class TrainSearchTool(Tool):
"""Tool for searching train routes and schedules."""
def __init__(self):
super().__init__(
name="train_search",
description="Search for train routes and schedules"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"origin": {"type": "string"},
"destination": {"type": "string"},
"date": {"type": "string", "format": "date"},
"time": {"type": "string", "description": "Preferred departure time (HH:MM)"}
},
"required": ["origin", "destination", "date"]
}
def execute(self, **kwargs) -> ToolResult:
"""Search for train options."""
origin = kwargs.get("origin")
destination = kwargs.get("destination")
travel_date = kwargs.get("date")
preferred_time = kwargs.get("time")
try:
# Using Rail Europe API or similar service
import requests
api_url = "https://api.raileurope.com/v1/search"
headers = {
"Authorization": f"Bearer {os.getenv('RAIL_API_KEY')}",
"Content-Type": "application/json"
}
payload = {
"origin": origin,
"destination": destination,
"date": travel_date
}
if preferred_time:
payload["time"] = preferred_time
response = requests.post(api_url, headers=headers, json=payload)
if response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message=f"Train search failed: {response.text}"
)
train_data = response.json()
trains = []
for journey in train_data.get("journeys", []):
train_info = {
"departure_time": journey["departure"]["time"],
"arrival_time": journey["arrival"]["time"],
"duration": journey["duration"],
"transfers": journey.get("transfers", 0),
"price": journey["price"]["amount"],
"currency": journey["price"]["currency"],
"train_type": journey.get("trainType", "Unknown"),
"class_options": journey.get("classes", [])
}
trains.append(train_info)
return ToolResult(
success=True,
data={"trains": trains},
metadata={"result_count": len(trains)}
)
except Exception as e:
logger.error(f"Train search error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
The public transit tool researches last-mile connectivity:
class PublicTransitTool(Tool):
"""Tool for finding public transportation routes."""
def __init__(self):
super().__init__(
name="public_transit",
description="Find public transit routes between locations"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"origin": {"type": "string", "description": "Starting location"},
"destination": {"type": "string", "description": "Destination location"},
"city": {"type": "string", "description": "City for transit search"},
"time": {"type": "string", "description": "Departure time"}
},
"required": ["origin", "destination", "city"]
}
def execute(self, **kwargs) -> ToolResult:
"""Find public transit routes."""
origin = kwargs.get("origin")
destination = kwargs.get("destination")
city = kwargs.get("city")
time = kwargs.get("time")
try:
# Using Google Maps Directions API or similar
import requests
api_key = os.getenv("GOOGLE_MAPS_API_KEY")
base_url = "https://maps.googleapis.com/maps/api/directions/json"
params = {
"origin": origin,
"destination": destination,
"mode": "transit",
"key": api_key
}
if time:
# Convert time to Unix timestamp
from datetime import datetime
dt = datetime.fromisoformat(time)
params["departure_time"] = int(dt.timestamp())
response = requests.get(base_url, params=params)
if response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message="Transit search failed"
)
data = response.json()
if data["status"] != "OK":
return ToolResult(
success=False,
data=None,
error_message=f"Transit search error: {data['status']}"
)
routes = []
for route in data.get("routes", []):
leg = route["legs"][0]
route_info = {
"duration": leg["duration"]["text"],
"distance": leg["distance"]["text"],
"steps": []
}
for step in leg["steps"]:
if step["travel_mode"] == "TRANSIT":
transit_details = step.get("transit_details", {})
route_info["steps"].append({
"mode": "transit",
"line": transit_details.get("line", {}).get("name", "Unknown"),
"departure_stop": transit_details.get("departure_stop", {}).get("name", ""),
"arrival_stop": transit_details.get("arrival_stop", {}).get("name", ""),
"num_stops": transit_details.get("num_stops", 0),
"duration": step["duration"]["text"]
})
else:
route_info["steps"].append({
"mode": step["travel_mode"].lower(),
"instructions": step.get("html_instructions", ""),
"duration": step["duration"]["text"],
"distance": step["distance"]["text"]
})
routes.append(route_info)
return ToolResult(
success=True,
data={"routes": routes},
metadata={"route_count": len(routes)}
)
except Exception as e:
logger.error(f"Public transit search error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
The Transportation Agent combines these tools to provide comprehensive travel options:
class TransportationAgent(Agent):
"""Specialized agent for researching transportation options."""
def __init__(self, llm_backend: LLMBackend):
tools = [
FlightSearchTool(),
TrainSearchTool(),
PublicTransitTool()
]
super().__init__(
name="Transportation Agent",
description="I specialize in finding transportation options including flights, trains, buses, and local transit connections.",
llm_backend=llm_backend,
tools=tools
)
def research_transportation(self, params: TravelParameters) -> Dict[str, Any]:
"""
Research all transportation options for the given travel parameters.
Args:
params: Travel parameters including origin, destination, dates
Returns:
Dictionary containing all transportation options
"""
results = {
"flights": [],
"trains": [],
"local_transit": []
}
# Search flights if requested
if "flight" in [mode.lower() for mode in params.transport_modes]:
flight_result = self.execute_tool(
"flight_search",
origin=params.origin,
destination=params.destination,
departure_date=params.start_date.isoformat(),
return_date=params.end_date.isoformat(),
passengers=params.travelers_count
)
if flight_result.success:
results["flights"] = flight_result.data.get("flights", [])
else:
logger.warning(f"Flight search failed: {flight_result.error_message}")
# Search trains if requested
if "train" in [mode.lower() for mode in params.transport_modes]:
train_result = self.execute_tool(
"train_search",
origin=params.origin,
destination=params.destination,
date=params.start_date.isoformat(),
time=params.preferred_departure_time
)
if train_result.success:
results["trains"] = train_result.data.get("trains", [])
else:
logger.warning(f"Train search failed: {train_result.error_message}")
# Research local transit for last-mile connectivity
# This would search from airport/station to typical city center or hotel area
transit_result = self.execute_tool(
"public_transit",
origin=f"{params.destination} airport",
destination=f"{params.destination} city center",
city=params.destination
)
if transit_result.success:
results["local_transit"] = transit_result.data.get("routes", [])
return results
The Transportation Agent orchestrates multiple tool invocations based on user preferences, aggregating results into a comprehensive transportation research report.
ACCOMMODATION AGENT IMPLEMENTATION
The Accommodation Agent searches for suitable hotels using booking platform APIs:
class HotelSearchTool(Tool):
"""Tool for searching hotel accommodations."""
def __init__(self):
super().__init__(
name="hotel_search",
description="Search for hotel accommodations with filtering by rating, price, and amenities"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"destination": {"type": "string"},
"check_in": {"type": "string", "format": "date"},
"check_out": {"type": "string", "format": "date"},
"guests": {"type": "integer", "minimum": 1},
"min_rating": {"type": "integer", "minimum": 1, "maximum": 5},
"max_rating": {"type": "integer", "minimum": 1, "maximum": 5},
"max_price": {"type": "number"}
},
"required": ["destination", "check_in", "check_out"]
}
def execute(self, **kwargs) -> ToolResult:
"""Search for hotels matching criteria."""
destination = kwargs.get("destination")
check_in = kwargs.get("check_in")
check_out = kwargs.get("check_out")
guests = kwargs.get("guests", 1)
min_rating = kwargs.get("min_rating", 1)
max_rating = kwargs.get("max_rating", 5)
max_price = kwargs.get("max_price")
try:
# Using Booking.com API or similar service
import requests
api_key = os.getenv("BOOKING_API_KEY")
base_url = "https://api.booking.com/v1/hotels/search"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"destination": destination,
"checkin": check_in,
"checkout": check_out,
"guests": guests,
"min_star_rating": min_rating,
"max_star_rating": max_rating
}
if max_price:
payload["max_price"] = max_price
response = requests.post(base_url, headers=headers, json=payload)
if response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message=f"Hotel search failed: {response.text}"
)
hotel_data = response.json()
hotels = []
for property_data in hotel_data.get("properties", []):
hotel_info = {
"name": property_data["name"],
"address": property_data["address"],
"star_rating": property_data["star_rating"],
"price_per_night": property_data["price"]["amount"],
"currency": property_data["price"]["currency"],
"total_price": property_data["total_price"]["amount"],
"room_type": property_data.get("room_type", "Standard"),
"breakfast_included": property_data.get("breakfast_included", False),
"free_cancellation": property_data.get("free_cancellation", False),
"cancellation_deadline": property_data.get("cancellation_deadline"),
"amenities": property_data.get("amenities", []),
"guest_rating": property_data.get("guest_rating", {}).get("score"),
"review_count": property_data.get("guest_rating", {}).get("count", 0),
"distance_to_center": property_data.get("distance_to_center_km"),
"images": property_data.get("images", [])[:3] # First 3 images
}
hotels.append(hotel_info)
# Sort by rating and price
hotels.sort(key=lambda x: (-x.get("guest_rating", 0), x["price_per_night"]))
return ToolResult(
success=True,
data={"hotels": hotels},
metadata={
"result_count": len(hotels),
"search_params": payload
}
)
except Exception as e:
logger.error(f"Hotel search error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
class AccommodationAgent(Agent):
"""Specialized agent for finding hotel accommodations."""
def __init__(self, llm_backend: LLMBackend):
tools = [HotelSearchTool()]
super().__init__(
name="Accommodation Agent",
description="I specialize in finding suitable hotel accommodations based on rating, location, price, and amenities.",
llm_backend=llm_backend,
tools=tools
)
def research_accommodations(self, params: TravelParameters) -> Dict[str, Any]:
"""
Research hotel options for the given travel parameters.
Args:
params: Travel parameters including destination, dates, rating preferences
Returns:
Dictionary containing hotel options
"""
hotel_result = self.execute_tool(
"hotel_search",
destination=params.destination,
check_in=params.start_date.isoformat(),
check_out=params.end_date.isoformat(),
guests=params.travelers_count,
min_rating=params.min_hotel_rating,
max_rating=params.max_hotel_rating,
max_price=params.budget_max
)
if hotel_result.success:
hotels = hotel_result.data.get("hotels", [])
# Categorize hotels by value proposition
categorized = {
"budget_friendly": [],
"best_value": [],
"premium": []
}
if hotels:
# Calculate price quartiles
prices = [h["price_per_night"] for h in hotels]
prices.sort()
q1 = prices[len(prices) // 4] if len(prices) > 4 else prices[0]
q3 = prices[3 * len(prices) // 4] if len(prices) > 4 else prices[-1]
for hotel in hotels:
price = hotel["price_per_night"]
rating = hotel.get("guest_rating", 0)
if price <= q1:
categorized["budget_friendly"].append(hotel)
elif price >= q3:
categorized["premium"].append(hotel)
elif rating >= 8.0: # High rating, mid price
categorized["best_value"].append(hotel)
return {
"hotels": hotels,
"categorized": categorized,
"total_found": len(hotels)
}
else:
logger.error(f"Hotel search failed: {hotel_result.error_message}")
return {
"hotels": [],
"categorized": {"budget_friendly": [], "best_value": [], "premium": []},
"total_found": 0,
"error": hotel_result.error_message
}
The Accommodation Agent searches hotels, filters by rating and price, and categorizes results to help users identify the best options for their needs.
WEATHER AND ATTRACTIONS AGENTS
The Weather Intelligence Agent provides meteorological forecasts:
class WeatherForecastTool(Tool):
"""Tool for retrieving weather forecasts."""
def __init__(self):
super().__init__(
name="weather_forecast",
description="Get weather forecast for a location and date range"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"location": {"type": "string"},
"start_date": {"type": "string", "format": "date"},
"end_date": {"type": "string", "format": "date"}
},
"required": ["location", "start_date", "end_date"]
}
def execute(self, **kwargs) -> ToolResult:
"""Get weather forecast for location and dates."""
location = kwargs.get("location")
start_date = kwargs.get("start_date")
end_date = kwargs.get("end_date")
try:
import requests
from datetime import datetime, timedelta
api_key = os.getenv("WEATHER_API_KEY")
# Using OpenWeatherMap API or similar
# First, geocode the location
geo_url = "http://api.openweathermap.org/geo/1.0/direct"
geo_params = {
"q": location,
"limit": 1,
"appid": api_key
}
geo_response = requests.get(geo_url, params=geo_params)
if geo_response.status_code != 200 or not geo_response.json():
return ToolResult(
success=False,
data=None,
error_message="Location not found"
)
geo_data = geo_response.json()[0]
lat = geo_data["lat"]
lon = geo_data["lon"]
# Get forecast
forecast_url = "https://api.openweathermap.org/data/2.5/forecast"
forecast_params = {
"lat": lat,
"lon": lon,
"appid": api_key,
"units": "metric"
}
forecast_response = requests.get(forecast_url, params=forecast_params)
if forecast_response.status_code != 200:
return ToolResult(
success=False,
data=None,
error_message="Weather forecast unavailable"
)
forecast_data = forecast_response.json()
# Parse forecast data
start = datetime.fromisoformat(start_date)
end = datetime.fromisoformat(end_date)
daily_forecasts = {}
for item in forecast_data.get("list", []):
forecast_time = datetime.fromtimestamp(item["dt"])
if start <= forecast_time <= end + timedelta(days=1):
date_key = forecast_time.date().isoformat()
if date_key not in daily_forecasts:
daily_forecasts[date_key] = {
"date": date_key,
"temperatures": [],
"conditions": [],
"precipitation": 0,
"humidity": [],
"wind_speed": []
}
daily_forecasts[date_key]["temperatures"].append(item["main"]["temp"])
daily_forecasts[date_key]["conditions"].append(item["weather"][0]["description"])
daily_forecasts[date_key]["precipitation"] += item.get("rain", {}).get("3h", 0)
daily_forecasts[date_key]["humidity"].append(item["main"]["humidity"])
daily_forecasts[date_key]["wind_speed"].append(item["wind"]["speed"])
# Aggregate daily data
forecasts = []
for date_key in sorted(daily_forecasts.keys()):
day_data = daily_forecasts[date_key]
forecasts.append({
"date": date_key,
"temp_min": min(day_data["temperatures"]),
"temp_max": max(day_data["temperatures"]),
"temp_avg": sum(day_data["temperatures"]) / len(day_data["temperatures"]),
"condition": max(set(day_data["conditions"]), key=day_data["conditions"].count),
"precipitation_mm": day_data["precipitation"],
"humidity_avg": sum(day_data["humidity"]) / len(day_data["humidity"]),
"wind_speed_avg": sum(day_data["wind_speed"]) / len(day_data["wind_speed"])
})
return ToolResult(
success=True,
data={"forecasts": forecasts},
metadata={"location": location, "coordinates": {"lat": lat, "lon": lon}}
)
except Exception as e:
logger.error(f"Weather forecast error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
class WeatherAgent(Agent):
"""Specialized agent for weather forecasting."""
def __init__(self, llm_backend: LLMBackend):
tools = [WeatherForecastTool()]
super().__init__(
name="Weather Agent",
description="I provide weather forecasts and climate information for travel destinations.",
llm_backend=llm_backend,
tools=tools
)
def research_weather(self, params: TravelParameters) -> Dict[str, Any]:
"""Get weather forecast for travel destination and dates."""
weather_result = self.execute_tool(
"weather_forecast",
location=params.destination,
start_date=params.start_date.isoformat(),
end_date=params.end_date.isoformat()
)
if weather_result.success:
forecasts = weather_result.data.get("forecasts", [])
# Generate packing recommendations based on weather
recommendations = self._generate_packing_recommendations(forecasts)
return {
"forecasts": forecasts,
"packing_recommendations": recommendations
}
else:
return {
"forecasts": [],
"packing_recommendations": [],
"error": weather_result.error_message
}
def _generate_packing_recommendations(self, forecasts: List[Dict]) -> List[str]:
"""Generate packing recommendations based on forecasts."""
recommendations = []
if not forecasts:
return recommendations
temps = [f["temp_avg"] for f in forecasts]
avg_temp = sum(temps) / len(temps)
max_temp = max(f["temp_max"] for f in forecasts)
min_temp = min(f["temp_min"] for f in forecasts)
total_precip = sum(f["precipitation_mm"] for f in forecasts)
# Temperature-based recommendations
if avg_temp < 10:
recommendations.append("Pack warm clothing including jacket and layers")
elif avg_temp > 25:
recommendations.append("Pack light, breathable clothing for warm weather")
if max_temp - min_temp > 15:
recommendations.append("Pack layers for variable temperatures")
# Precipitation recommendations
if total_precip > 10:
recommendations.append("Pack rain gear including umbrella and waterproof jacket")
# Specific conditions
conditions = [f["condition"] for f in forecasts]
if any("rain" in c.lower() for c in conditions):
recommendations.append("Waterproof footwear recommended")
return recommendations
The Attractions Agent discovers points of interest and events:
class AttractionsSearchTool(Tool):
"""Tool for finding tourist attractions and points of interest."""
def __init__(self):
super().__init__(
name="attractions_search",
description="Search for tourist attractions, landmarks, and points of interest"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"location": {"type": "string"},
"category": {"type": "string", "enum": ["all", "museums", "landmarks", "nature", "entertainment"]},
"radius_km": {"type": "number", "minimum": 1}
},
"required": ["location"]
}
def execute(self, **kwargs) -> ToolResult:
"""Search for attractions in the specified location."""
location = kwargs.get("location")
category = kwargs.get("category", "all")
radius_km = kwargs.get("radius_km", 10)
try:
import requests
# Using Google Places API or similar
api_key = os.getenv("GOOGLE_PLACES_API_KEY")
# First geocode the location
geocode_url = "https://maps.googleapis.com/maps/api/geocode/json"
geocode_params = {
"address": location,
"key": api_key
}
geocode_response = requests.get(geocode_url, params=geocode_params)
if geocode_response.status_code != 200:
return ToolResult(success=False, data=None, error_message="Geocoding failed")
geocode_data = geocode_response.json()
if not geocode_data.get("results"):
return ToolResult(success=False, data=None, error_message="Location not found")
location_coords = geocode_data["results"][0]["geometry"]["location"]
# Search for places
places_url = "https://maps.googleapis.com/maps/api/place/nearbysearch/json"
type_mapping = {
"museums": "museum",
"landmarks": "tourist_attraction",
"nature": "park",
"entertainment": "amusement_park"
}
place_type = type_mapping.get(category, "tourist_attraction")
places_params = {
"location": f"{location_coords['lat']},{location_coords['lng']}",
"radius": radius_km * 1000, # Convert to meters
"type": place_type,
"key": api_key
}
places_response = requests.get(places_url, params=places_params)
if places_response.status_code != 200:
return ToolResult(success=False, data=None, error_message="Places search failed")
places_data = places_response.json()
attractions = []
for place in places_data.get("results", [])[:20]: # Limit to top 20
# Get detailed information
details_url = "https://maps.googleapis.com/maps/api/place/details/json"
details_params = {
"place_id": place["place_id"],
"fields": "name,rating,formatted_address,opening_hours,website,price_level,editorial_summary",
"key": api_key
}
details_response = requests.get(details_url, params=details_params)
if details_response.status_code == 200:
details = details_response.json().get("result", {})
attraction_info = {
"name": details.get("name", place["name"]),
"address": details.get("formatted_address", ""),
"rating": details.get("rating", place.get("rating")),
"price_level": details.get("price_level"),
"opening_hours": details.get("opening_hours", {}).get("weekday_text", []),
"website": details.get("website"),
"description": details.get("editorial_summary", {}).get("overview", ""),
"location": place["geometry"]["location"]
}
attractions.append(attraction_info)
return ToolResult(
success=True,
data={"attractions": attractions},
metadata={"result_count": len(attractions)}
)
except Exception as e:
logger.error(f"Attractions search error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class EventsSearchTool(Tool):
"""Tool for finding events happening during travel dates."""
def __init__(self):
super().__init__(
name="events_search",
description="Search for events, concerts, festivals, and activities"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"location": {"type": "string"},
"start_date": {"type": "string", "format": "date"},
"end_date": {"type": "string", "format": "date"},
"category": {"type": "string"}
},
"required": ["location", "start_date", "end_date"]
}
def execute(self, **kwargs) -> ToolResult:
"""Search for events in location during date range."""
location = kwargs.get("location")
start_date = kwargs.get("start_date")
end_date = kwargs.get("end_date")
category = kwargs.get("category")
try:
import requests
# Using Ticketmaster API or similar
api_key = os.getenv("TICKETMASTER_API_KEY")
base_url = "https://app.ticketmaster.com/discovery/v2/events.json"
params = {
"apikey": api_key,
"city": location,
"startDateTime": f"{start_date}T00:00:00Z",
"endDateTime": f"{end_date}T23:59:59Z",
"size": 50
}
if category:
params["classificationName"] = category
response = requests.get(base_url, params=params)
if response.status_code != 200:
return ToolResult(success=False, data=None, error_message="Events search failed")
events_data = response.json()
events = []
for event in events_data.get("_embedded", {}).get("events", []):
event_info = {
"name": event["name"],
"date": event["dates"]["start"].get("localDate"),
"time": event["dates"]["start"].get("localTime"),
"venue": event["_embedded"]["venues"][0]["name"] if event.get("_embedded", {}).get("venues") else "TBA",
"category": event["classifications"][0]["segment"]["name"] if event.get("classifications") else "General",
"price_range": event.get("priceRanges", [{}])[0] if event.get("priceRanges") else {},
"url": event.get("url"),
"description": event.get("info", "")
}
events.append(event_info)
# Sort by date
events.sort(key=lambda x: x.get("date", ""))
return ToolResult(
success=True,
data={"events": events},
metadata={"result_count": len(events)}
)
except Exception as e:
logger.error(f"Events search error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class AttractionsAgent(Agent):
"""Specialized agent for discovering attractions and events."""
def __init__(self, llm_backend: LLMBackend):
tools = [AttractionsSearchTool(), EventsSearchTool()]
super().__init__(
name="Attractions Agent",
description="I discover tourist attractions, landmarks, and events at travel destinations.",
llm_backend=llm_backend,
tools=tools
)
def research_attractions(self, params: TravelParameters) -> Dict[str, Any]:
"""Research attractions and events for destination."""
# Search attractions
attractions_result = self.execute_tool(
"attractions_search",
location=params.destination,
category="all",
radius_km=15
)
# Search events
events_result = self.execute_tool(
"events_search",
location=params.destination,
start_date=params.start_date.isoformat(),
end_date=params.end_date.isoformat()
)
attractions = attractions_result.data.get("attractions", []) if attractions_result.success else []
events = events_result.data.get("events", []) if events_result.success else []
# Categorize by type and create suggested itinerary
categorized_attractions = self._categorize_attractions(attractions)
suggested_itinerary = self._create_suggested_itinerary(
attractions,
events,
params.start_date,
params.end_date
)
return {
"attractions": attractions,
"events": events,
"categorized": categorized_attractions,
"suggested_itinerary": suggested_itinerary
}
def _categorize_attractions(self, attractions: List[Dict]) -> Dict[str, List[Dict]]:
"""Categorize attractions by type."""
categories = {
"cultural": [],
"outdoor": [],
"entertainment": [],
"shopping": [],
"dining": []
}
for attraction in attractions:
name_lower = attraction["name"].lower()
if any(word in name_lower for word in ["museum", "gallery", "theater", "cathedral", "church"]):
categories["cultural"].append(attraction)
elif any(word in name_lower for word in ["park", "garden", "beach", "trail"]):
categories["outdoor"].append(attraction)
elif any(word in name_lower for word in ["mall", "market", "shop"]):
categories["shopping"].append(attraction)
else:
categories["entertainment"].append(attraction)
return categories
def _create_suggested_itinerary(
self,
attractions: List[Dict],
events: List[Dict],
start_date: date,
end_date: date
) -> List[Dict]:
"""Create a suggested daily itinerary."""
from datetime import timedelta
itinerary = []
current_date = start_date
# Distribute attractions across days
attractions_per_day = max(2, len(attractions) // ((end_date - start_date).days + 1))
attraction_index = 0
while current_date <= end_date:
day_plan = {
"date": current_date.isoformat(),
"activities": []
}
# Add attractions for this day
for i in range(attractions_per_day):
if attraction_index < len(attractions):
day_plan["activities"].append({
"type": "attraction",
"details": attractions[attraction_index]
})
attraction_index += 1
# Add events for this day
day_events = [e for e in events if e.get("date") == current_date.isoformat()]
for event in day_events:
day_plan["activities"].append({
"type": "event",
"details": event
})
itinerary.append(day_plan)
current_date += timedelta(days=1)
return itinerary
TRAVEL INTELLIGENCE AGENT
The Travel Intelligence Agent provides regulatory and practical information:
class VisaRequirementsTool(Tool):
"""Tool for checking visa requirements."""
def __init__(self):
super().__init__(
name="visa_requirements",
description="Check visa requirements for travelers"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"origin_country": {"type": "string"},
"destination_country": {"type": "string"}
},
"required": ["origin_country", "destination_country"]
}
def execute(self, **kwargs) -> ToolResult:
"""Check visa requirements."""
origin = kwargs.get("origin_country")
destination = kwargs.get("destination_country")
try:
# Using a visa requirements API or web scraping
import requests
# Example using a hypothetical visa API
api_url = f"https://api.visarequirements.com/check"
params = {
"from": origin,
"to": destination
}
response = requests.get(api_url, params=params)
if response.status_code == 200:
visa_data = response.json()
return ToolResult(
success=True,
data={
"required": visa_data.get("visa_required", False),
"type": visa_data.get("visa_type"),
"duration": visa_data.get("max_stay_days"),
"processing_time": visa_data.get("processing_days"),
"fee": visa_data.get("fee"),
"notes": visa_data.get("additional_info", "")
}
)
else:
# Fallback to general information
return ToolResult(
success=True,
data={
"required": None,
"notes": "Please check with the destination country's embassy for current visa requirements."
}
)
except Exception as e:
logger.error(f"Visa requirements check error: {e}")
return ToolResult(
success=False,
data=None,
error_message=str(e)
)
class CurrencyExchangeTool(Tool):
"""Tool for getting currency exchange rates."""
def __init__(self):
super().__init__(
name="currency_exchange",
description="Get current currency exchange rates"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"from_currency": {"type": "string"},
"to_currency": {"type": "string"},
"amount": {"type": "number"}
},
"required": ["from_currency", "to_currency"]
}
def execute(self, **kwargs) -> ToolResult:
"""Get currency exchange rate."""
from_currency = kwargs.get("from_currency")
to_currency = kwargs.get("to_currency")
amount = kwargs.get("amount", 1.0)
try:
import requests
# Using exchangerate-api.com or similar
api_key = os.getenv("EXCHANGE_RATE_API_KEY")
api_url = f"https://v6.exchangerate-api.com/v6/{api_key}/pair/{from_currency}/{to_currency}/{amount}"
response = requests.get(api_url)
if response.status_code != 200:
return ToolResult(success=False, data=None, error_message="Exchange rate unavailable")
rate_data = response.json()
return ToolResult(
success=True,
data={
"from_currency": from_currency,
"to_currency": to_currency,
"exchange_rate": rate_data["conversion_rate"],
"amount": amount,
"converted_amount": rate_data["conversion_result"],
"last_updated": rate_data.get("time_last_update_utc")
class SafetyAdvisoriesTool(Tool):
"""Tool for retrieving travel safety advisories."""
def __init__(self):
super().__init__(
name="safety_advisories",
description="Get travel safety advisories and warnings"
)
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"destination_country": {"type": "string"}
},
"required": ["destination_country"]
}
def execute(self, **kwargs) -> ToolResult:
"""Get safety advisories for destination."""
destination = kwargs.get("destination_country")
try:
import requests
# Using travel.state.gov API or similar government travel advisory service
api_url = "https://travel.state.gov/content/travel/en/traveladvisories/traveladvisories.html"
# In production, this would parse official government advisories
# For this implementation, we'll use a structured approach
# Example: scraping or API call to get advisory data
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
response = requests.get(api_url, headers=headers)
# Parse advisory information
# This is simplified - production would parse actual advisory data
advisory_data = {
"level": 1, # 1-4 scale
"level_description": "Exercise Normal Precautions",
"summary": "Exercise normal precautions when traveling to this destination.",
"health_alerts": [],
"security_alerts": [],
"local_laws": "Familiarize yourself with local laws and customs.",
"emergency_contacts": {
"police": "112",
"ambulance": "112",
"embassy": "+XX-XXX-XXXX"
}
}
return ToolResult(
success=True,
data=advisory_data
)
except Exception as e:
logger.error(f"Safety advisories error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class TravelIntelligenceAgent(Agent):
"""Specialized agent for travel regulations and practical information."""
def __init__(self, llm_backend: LLMBackend):
tools = [
VisaRequirementsTool(),
CurrencyExchangeTool(),
SafetyAdvisoriesTool()
]
super().__init__(
name="Travel Intelligence Agent",
description="I provide information about visas, currency, safety, and travel regulations.",
llm_backend=llm_backend,
tools=tools
)
def research_travel_intelligence(
self,
params: TravelParameters,
origin_country: str,
destination_country: str,
home_currency: str,
destination_currency: str
) -> Dict[str, Any]:
"""Research all travel intelligence information."""
# Check visa requirements
visa_result = self.execute_tool(
"visa_requirements",
origin_country=origin_country,
destination_country=destination_country
)
# Get currency exchange rate
currency_result = self.execute_tool(
"currency_exchange",
from_currency=home_currency,
to_currency=destination_currency,
amount=100
)
# Get safety advisories
safety_result = self.execute_tool(
"safety_advisories",
destination_country=destination_country
)
return {
"visa": visa_result.data if visa_result.success else {},
"currency": currency_result.data if currency_result.success else {},
"safety": safety_result.data if safety_result.success else {},
"errors": {
"visa": None if visa_result.success else visa_result.error_message,
"currency": None if currency_result.success else currency_result.error_message,
"safety": None if safety_result.success else safety_result.error_message
}
}
The Travel Intelligence Agent coordinates multiple information gathering tasks to provide comprehensive regulatory and practical guidance for international travel.
COORDINATOR AGENT IMPLEMENTATION
The Coordinator Agent orchestrates the entire travel planning workflow. It manages parameter collection, delegates tasks to specialized agents, and synthesizes results into a comprehensive travel plan:
class CoordinatorAgent(Agent):
"""Main coordinator agent that orchestrates the travel planning process."""
def __init__(
self,
llm_backend: LLMBackend,
transportation_agent: TransportationAgent,
accommodation_agent: AccommodationAgent,
weather_agent: WeatherAgent,
attractions_agent: AttractionsAgent,
intelligence_agent: TravelIntelligenceAgent
):
"""
Initialize coordinator with all specialized agents.
Args:
llm_backend: LLM backend for text generation
transportation_agent: Agent for transportation research
accommodation_agent: Agent for hotel research
weather_agent: Agent for weather forecasting
attractions_agent: Agent for attractions and events
intelligence_agent: Agent for travel regulations
"""
super().__init__(
name="Travel Planning Coordinator",
description="I coordinate the entire travel planning process by gathering requirements and delegating to specialized agents.",
llm_backend=llm_backend,
tools=[]
)
self.transportation_agent = transportation_agent
self.accommodation_agent = accommodation_agent
self.weather_agent = weather_agent
self.attractions_agent = attractions_agent
self.intelligence_agent = intelligence_agent
self.travel_params = TravelParameters()
def extract_parameters_from_message(self, message: str) -> None:
"""
Extract travel parameters from user message using LLM.
Args:
message: User's travel planning request
"""
extraction_prompt = f"""
Extract travel planning parameters from the following message. Return a JSON object with these fields:
- origin: origin location (string or null)
- destination: destination location (string or null)
- start_date: start date in YYYY-MM-DD format (string or null)
- end_date: end date in YYYY-MM-DD format (string or null)
- travel_purpose: "business" or "leisure" (string or null)
- preferred_departure_time: preferred departure time (string or null)
- preferred_return_time: preferred return time (string or null)
- transport_modes: list of preferred transportation modes like ["flight", "train", "bus", "car"] (array)
- min_hotel_rating: minimum hotel star rating 1-5 (integer or null)
- max_hotel_rating: maximum hotel star rating 1-5 (integer or null)
- travelers_count: number of travelers (integer, default 1)
User message: {message}
Return only valid JSON, no other text.
"""
response = self.llm_backend.generate(extraction_prompt, temperature=0.3, max_tokens=512)
try:
# Parse JSON response
import re
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
extracted = json.loads(json_match.group())
# Update travel parameters
if extracted.get("origin"):
self.travel_params.origin = extracted["origin"]
if extracted.get("destination"):
self.travel_params.destination = extracted["destination"]
if extracted.get("start_date"):
from datetime import datetime
self.travel_params.start_date = datetime.fromisoformat(extracted["start_date"]).date()
if extracted.get("end_date"):
from datetime import datetime
self.travel_params.end_date = datetime.fromisoformat(extracted["end_date"]).date()
if extracted.get("travel_purpose"):
self.travel_params.travel_purpose = extracted["travel_purpose"]
if extracted.get("preferred_departure_time"):
self.travel_params.preferred_departure_time = extracted["preferred_departure_time"]
if extracted.get("preferred_return_time"):
self.travel_params.preferred_return_time = extracted["preferred_return_time"]
if extracted.get("transport_modes"):
self.travel_params.transport_modes = extracted["transport_modes"]
if extracted.get("min_hotel_rating") is not None:
self.travel_params.min_hotel_rating = extracted["min_hotel_rating"]
if extracted.get("max_hotel_rating") is not None:
self.travel_params.max_hotel_rating = extracted["max_hotel_rating"]
if extracted.get("travelers_count"):
self.travel_params.travelers_count = extracted["travelers_count"]
logger.info(f"Extracted parameters: {extracted}")
except Exception as e:
logger.error(f"Failed to extract parameters: {e}")
def collect_missing_parameters(self) -> Optional[str]:
"""
Generate a conversational request for missing parameters.
Returns:
Question to ask user, or None if all parameters collected
"""
missing = self.travel_params.missing_parameters()
if not missing:
return None
# Generate natural language question for missing parameters
question_prompt = f"""
Generate a friendly, conversational question to ask the user for the following missing travel planning information: {', '.join(missing)}
The question should:
1. Be natural and conversational
2. Ask for multiple related items together when appropriate
3. Provide helpful context or examples
4. Be concise but friendly
Generate only the question, no other text.
"""
question = self.llm_backend.generate(question_prompt, temperature=0.7, max_tokens=256)
return question.strip()
def plan_travel(self) -> Dict[str, Any]:
"""
Execute the complete travel planning workflow.
Returns:
Comprehensive travel plan with all research results
"""
logger.info("Starting travel planning workflow")
# Validate parameters are complete
if not self.travel_params.is_complete():
missing = self.travel_params.missing_parameters()
return {
"status": "incomplete",
"message": f"Missing required parameters: {', '.join(missing)}"
}
results = {
"status": "complete",
"parameters": {
"origin": self.travel_params.origin,
"destination": self.travel_params.destination,
"dates": {
"start": self.travel_params.start_date.isoformat(),
"end": self.travel_params.end_date.isoformat()
},
"purpose": self.travel_params.travel_purpose,
"travelers": self.travel_params.travelers_count
}
}
# Execute agent research in parallel where possible
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
# Submit all agent tasks
future_transportation = executor.submit(
self.transportation_agent.research_transportation,
self.travel_params
)
future_accommodation = executor.submit(
self.accommodation_agent.research_accommodations,
self.travel_params
)
future_weather = executor.submit(
self.weather_agent.research_weather,
self.travel_params
)
future_attractions = executor.submit(
self.attractions_agent.research_attractions,
self.travel_params
)
# Intelligence agent needs country information
origin_country = self._extract_country(self.travel_params.origin)
destination_country = self._extract_country(self.travel_params.destination)
home_currency = self._get_currency_for_country(origin_country)
dest_currency = self._get_currency_for_country(destination_country)
future_intelligence = executor.submit(
self.intelligence_agent.research_travel_intelligence,
self.travel_params,
origin_country,
destination_country,
home_currency,
dest_currency
)
# Collect results
results["transportation"] = future_transportation.result()
results["accommodation"] = future_accommodation.result()
results["weather"] = future_weather.result()
results["attractions"] = future_attractions.result()
results["intelligence"] = future_intelligence.result()
logger.info("Travel planning workflow completed")
# Generate cost summary
results["cost_summary"] = self._calculate_cost_summary(results)
return results
def _extract_country(self, location: str) -> str:
"""Extract country name from location string."""
# In production, use geocoding API to get country
# For now, simple heuristic
parts = location.split(',')
return parts[-1].strip() if len(parts) > 1 else location
def _get_currency_for_country(self, country: str) -> str:
"""Get currency code for country."""
# Simplified mapping - production would use comprehensive database
currency_map = {
"USA": "USD",
"United States": "USD",
"UK": "GBP",
"United Kingdom": "GBP",
"France": "EUR",
"Germany": "EUR",
"Spain": "EUR",
"Italy": "EUR",
"Japan": "JPY",
"China": "CNY",
"India": "INR",
"Canada": "CAD",
"Australia": "AUD"
}
return currency_map.get(country, "USD")
def _calculate_cost_summary(self, results: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate total trip costs from all components."""
summary = {
"transportation": 0,
"accommodation": 0,
"attractions": 0,
"total": 0,
"currency": "USD"
}
# Sum transportation costs
if results.get("transportation"):
flights = results["transportation"].get("flights", [])
if flights:
# Use cheapest flight option
min_flight = min(flights, key=lambda x: x.get("price", float('inf')), default=None)
if min_flight:
summary["transportation"] = min_flight.get("price", 0)
summary["currency"] = min_flight.get("currency", "USD")
# Sum accommodation costs
if results.get("accommodation"):
hotels = results["accommodation"].get("hotels", [])
if hotels:
# Use cheapest hotel in best value category
best_value = results["accommodation"].get("categorized", {}).get("best_value", [])
if best_value:
hotel = best_value[0]
summary["accommodation"] = hotel.get("total_price", 0)
elif hotels:
hotel = min(hotels, key=lambda x: x.get("total_price", float('inf')))
summary["accommodation"] = hotel.get("total_price", 0)
# Estimate attraction costs
if results.get("attractions"):
events = results["attraction"].get("events", [])
estimated_attraction_cost = len(events) * 50 # Rough estimate
summary["attractions"] = estimated_attraction_cost
summary["total"] = (
summary["transportation"] +
summary["accommodation"] +
summary["attractions"]
)
return summary
def generate_report(self, results: Dict[str, Any]) -> str:
"""
Generate a comprehensive travel planning report.
Args:
results: Complete planning results from all agents
Returns:
Formatted text report
"""
report_parts = []
# Header
report_parts.append("=" * 80)
report_parts.append("COMPREHENSIVE TRAVEL PLAN")
report_parts.append("=" * 80)
report_parts.append("")
# Travel parameters summary
report_parts.append("TRIP OVERVIEW")
report_parts.append("-" * 80)
params = results.get("parameters", {})
report_parts.append(f"Origin: {params.get('origin')}")
report_parts.append(f"Destination: {params.get('destination')}")
dates = params.get("dates", {})
report_parts.append(f"Travel Dates: {dates.get('start')} to {dates.get('end')}")
report_parts.append(f"Purpose: {params.get('purpose')}")
report_parts.append(f"Travelers: {params.get('travelers')}")
report_parts.append("")
# Transportation options
report_parts.append("TRANSPORTATION OPTIONS")
report_parts.append("-" * 80)
transportation = results.get("transportation", {})
flights = transportation.get("flights", [])
if flights:
report_parts.append(f"\nFlights ({len(flights)} options found):")
for i, flight in enumerate(flights[:5], 1): # Top 5
report_parts.append(f"\n Option {i}:")
report_parts.append(f" Price: {flight.get('price')} {flight.get('currency')}")
for j, segment in enumerate(flight.get('segments', []), 1):
dep = segment.get('departure', {})
arr = segment.get('arrival', {})
report_parts.append(f" Segment {j}: {dep.get('airport')} -> {arr.get('airport')}")
report_parts.append(f" Departure: {dep.get('time')}")
report_parts.append(f" Arrival: {arr.get('time')}")
report_parts.append(f" Flight: {segment.get('carrier')}{segment.get('flight_number')}")
trains = transportation.get("trains", [])
if trains:
report_parts.append(f"\nTrains ({len(trains)} options found):")
for i, train in enumerate(trains[:5], 1):
report_parts.append(f"\n Option {i}:")
report_parts.append(f" Departure: {train.get('departure_time')}")
report_parts.append(f" Arrival: {train.get('arrival_time')}")
report_parts.append(f" Duration: {train.get('duration')}")
report_parts.append(f" Price: {train.get('price')} {train.get('currency')}")
report_parts.append(f" Transfers: {train.get('transfers')}")
local_transit = transportation.get("local_transit", [])
if local_transit:
report_parts.append(f"\nLocal Transit (from airport/station to city):")
route = local_transit[0]
report_parts.append(f" Duration: {route.get('duration')}")
report_parts.append(f" Distance: {route.get('distance')}")
for step in route.get('steps', []):
if step.get('mode') == 'transit':
report_parts.append(f" {step.get('line')}: {step.get('departure_stop')} -> {step.get('arrival_stop')}")
report_parts.append("")
# Accommodation options
report_parts.append("ACCOMMODATION OPTIONS")
report_parts.append("-" * 80)
accommodation = results.get("accommodation", {})
categorized = accommodation.get("categorized", {})
if categorized.get("best_value"):
report_parts.append("\nBest Value Hotels:")
for hotel in categorized["best_value"][:3]:
report_parts.append(f"\n {hotel.get('name')}")
report_parts.append(f" Rating: {hotel.get('star_rating')} stars (Guest: {hotel.get('guest_rating')}/10)")
report_parts.append(f" Address: {hotel.get('address')}")
report_parts.append(f" Price: {hotel.get('price_per_night')} {hotel.get('currency')}/night")
report_parts.append(f" Total: {hotel.get('total_price')} {hotel.get('currency')}")
report_parts.append(f" Room: {hotel.get('room_type')}")
report_parts.append(f" Breakfast: {'Included' if hotel.get('breakfast_included') else 'Not included'}")
report_parts.append(f" Cancellation: {'Free' if hotel.get('free_cancellation') else 'Non-refundable'}")
if categorized.get("budget_friendly"):
report_parts.append("\nBudget-Friendly Hotels:")
for hotel in categorized["budget_friendly"][:2]:
report_parts.append(f"\n {hotel.get('name')}")
report_parts.append(f" Price: {hotel.get('price_per_night')} {hotel.get('currency')}/night")
report_parts.append(f" Rating: {hotel.get('star_rating')} stars")
report_parts.append("")
# Weather forecast
report_parts.append("WEATHER FORECAST")
report_parts.append("-" * 80)
weather = results.get("weather", {})
forecasts = weather.get("forecasts", [])
if forecasts:
for forecast in forecasts:
report_parts.append(f"\n{forecast.get('date')}:")
report_parts.append(f" Temperature: {forecast.get('temp_min'):.1f}°C - {forecast.get('temp_max'):.1f}°C")
report_parts.append(f" Condition: {forecast.get('condition')}")
report_parts.append(f" Precipitation: {forecast.get('precipitation_mm'):.1f}mm")
report_parts.append(f" Humidity: {forecast.get('humidity_avg'):.0f}%")
packing = weather.get("packing_recommendations", [])
if packing:
report_parts.append("\nPacking Recommendations:")
for rec in packing:
report_parts.append(f" - {rec}")
report_parts.append("")
# Attractions and events
report_parts.append("ATTRACTIONS & ACTIVITIES")
report_parts.append("-" * 80)
attractions_data = results.get("attractions", {})
attractions = attractions_data.get("attractions", [])
if attractions:
report_parts.append(f"\nTop Attractions ({len(attractions)} found):")
for attraction in attractions[:10]:
report_parts.append(f"\n {attraction.get('name')}")
if attraction.get('rating'):
report_parts.append(f" Rating: {attraction.get('rating')}/5")
if attraction.get('address'):
report_parts.append(f" Address: {attraction.get('address')}")
if attraction.get('description'):
report_parts.append(f" {attraction.get('description')[:200]}...")
events = attractions_data.get("events", [])
if events:
report_parts.append(f"\nUpcoming Events ({len(events)} found):")
for event in events[:5]:
report_parts.append(f"\n {event.get('name')}")
report_parts.append(f" Date: {event.get('date')} {event.get('time', '')}")
report_parts.append(f" Venue: {event.get('venue')}")
report_parts.append(f" Category: {event.get('category')}")
price_range = event.get('price_range', {})
if price_range:
report_parts.append(f" Price: {price_range.get('min', 'N/A')} - {price_range.get('max', 'N/A')}")
itinerary = attractions_data.get("suggested_itinerary", [])
if itinerary:
report_parts.append("\nSuggested Daily Itinerary:")
for day in itinerary:
report_parts.append(f"\n {day.get('date')}:")
for activity in day.get('activities', []):
details = activity.get('details', {})
if activity.get('type') == 'attraction':
report_parts.append(f" - Visit: {details.get('name')}")
elif activity.get('type') == 'event':
report_parts.append(f" - Event: {details.get('name')} at {details.get('time', 'TBA')}")
report_parts.append("")
# Travel intelligence
report_parts.append("TRAVEL INFORMATION")
report_parts.append("-" * 80)
intelligence = results.get("intelligence", {})
visa = intelligence.get("visa", {})
if visa:
report_parts.append("\nVisa Requirements:")
if visa.get("required") is not None:
if visa.get("required"):
report_parts.append(f" Visa Required: Yes")
report_parts.append(f" Type: {visa.get('type', 'N/A')}")
report_parts.append(f" Processing Time: {visa.get('processing_time', 'N/A')} days")
report_parts.append(f" Fee: {visa.get('fee', 'N/A')}")
else:
report_parts.append(f" Visa Required: No")
report_parts.append(f" Maximum Stay: {visa.get('duration', 'N/A')} days")
if visa.get("notes"):
report_parts.append(f" Notes: {visa.get('notes')}")
currency = intelligence.get("currency", {})
if currency:
report_parts.append("\nCurrency Information:")
report_parts.append(f" Exchange Rate: 1 {currency.get('from_currency')} = {currency.get('exchange_rate', 0):.4f} {currency.get('to_currency')}")
report_parts.append(f" Example: {currency.get('amount', 100)} {currency.get('from_currency')} = {currency.get('converted_amount', 0):.2f} {currency.get('to_currency')}")
safety = intelligence.get("safety", {})
if safety:
report_parts.append("\nSafety Information:")
report_parts.append(f" Advisory Level: {safety.get('level')} - {safety.get('level_description')}")
report_parts.append(f" Summary: {safety.get('summary')}")
emergency = safety.get("emergency_contacts", {})
if emergency:
report_parts.append(" Emergency Contacts:")
report_parts.append(f" Police: {emergency.get('police')}")
report_parts.append(f" Ambulance: {emergency.get('ambulance')}")
report_parts.append(f" Embassy: {emergency.get('embassy')}")
report_parts.append("")
# Cost summary
report_parts.append("COST SUMMARY")
report_parts.append("-" * 80)
cost_summary = results.get("cost_summary", {})
report_parts.append(f"Transportation: {cost_summary.get('transportation', 0):.2f} {cost_summary.get('currency')}")
report_parts.append(f"Accommodation: {cost_summary.get('accommodation', 0):.2f} {cost_summary.get('currency')}")
report_parts.append(f"Attractions (estimated): {cost_summary.get('attractions', 0):.2f} {cost_summary.get('currency')}")
report_parts.append(f"\nEstimated Total: {cost_summary.get('total', 0):.2f} {cost_summary.get('currency')}")
report_parts.append("\nNote: This is an estimate. Actual costs may vary based on specific choices and additional expenses.")
report_parts.append("")
report_parts.append("=" * 80)
report_parts.append("END OF TRAVEL PLAN")
report_parts.append("=" * 80)
return "\n".join(report_parts)
The Coordinator Agent manages the entire workflow from parameter collection through result synthesis and report generation. It uses parallel execution to improve performance and provides comprehensive cost estimates.
MAIN APPLICATION INTERFACE
The main application ties everything together and provides a user interface:
class TravelPlannerApplication:
"""Main application for LLM-based travel planning."""
def __init__(self, config: Optional[Dict[str, Any]] = None):
"""
Initialize the travel planner application.
Args:
config: Configuration dictionary with LLM and hardware settings
"""
self.config = config or {}
# Detect hardware backend
self.hardware_backend = HardwareDetector.detect_backend()
logger.info(f"Using hardware backend: {self.hardware_backend.value}")
# Initialize LLM backend
self.llm_backend = self._initialize_llm_backend()
# Initialize specialized agents
self.transportation_agent = TransportationAgent(self.llm_backend)
self.accommodation_agent = AccommodationAgent(self.llm_backend)
self.weather_agent = WeatherAgent(self.llm_backend)
self.attractions_agent = AttractionsAgent(self.llm_backend)
self.intelligence_agent = TravelIntelligenceAgent(self.llm_backend)
# Initialize coordinator
self.coordinator = CoordinatorAgent(
llm_backend=self.llm_backend,
transportation_agent=self.transportation_agent,
accommodation_agent=self.accommodation_agent,
weather_agent=self.weather_agent,
attractions_agent=self.attractions_agent,
intelligence_agent=self.intelligence_agent
)
def _initialize_llm_backend(self) -> LLMBackend:
"""Initialize appropriate LLM backend based on configuration."""
provider = self.config.get("llm_provider", LLMProvider.LOCAL_LLAMA)
if provider == LLMProvider.OPENAI:
api_key = self.config.get("openai_api_key") or os.getenv("OPENAI_API_KEY")
model = self.config.get("openai_model", "gpt-4")
return OpenAIBackend(api_key=api_key, model=model)
elif provider == LLMProvider.LOCAL_LLAMA:
model_path = self.config.get("model_path") or os.getenv("LLAMA_MODEL_PATH")
if not model_path:
raise ValueError("Model path required for local LLAMA backend")
n_ctx = self.config.get("context_length", 4096)
n_gpu_layers = self.config.get("gpu_layers", 35)
return LocalLlamaBackend(
model_path=model_path,
hardware_backend=self.hardware_backend,
n_ctx=n_ctx,
n_gpu_layers=n_gpu_layers
)
else:
raise ValueError(f"Unsupported LLM provider: {provider}")
def process_user_message(self, message: str) -> str:
"""
Process a user message and return response.
Args:
message: User's input message
Returns:
Response from the system
"""
# Extract parameters from message
self.coordinator.extract_parameters_from_message(message)
# Check if we need more information
missing_params_question = self.coordinator.collect_missing_parameters()
if missing_params_question:
return missing_params_question
# All parameters collected, execute planning
logger.info("All parameters collected, executing travel planning")
results = self.coordinator.plan_travel()
if results.get("status") == "incomplete":
return results.get("message", "Unable to complete planning")
# Generate and return report
report = self.coordinator.generate_report(results)
return report
def run_interactive(self):
"""Run interactive command-line interface."""
print("=" * 80)
print("LLM-BASED TRAVEL PLANNER")
print("=" * 80)
print("\nWelcome! I'm your AI travel planning assistant.")
print("Tell me about your travel plans, and I'll help you organize everything.")
print("Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', 'bye']:
print("\nThank you for using the Travel Planner. Safe travels!")
break
# Process message
response = self.process_user_message(user_input)
print(f"\nAssistant:\n{response}\n")
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
logger.error(f"Error processing message: {e}")
print(f"\nI encountered an error: {e}")
print("Please try again.\n")
This main application class provides a complete interface for the travel planning system with interactive conversation capabilities.
CONFIGURATION AND DEPLOYMENT
The system requires proper configuration for production deployment. Configuration management handles API keys, model paths, and runtime parameters:
class ConfigurationManager:
"""Manages application configuration from multiple sources."""
@staticmethod
def load_config(config_file: Optional[str] = None) -> Dict[str, Any]:
"""
Load configuration from file and environment variables.
Args:
config_file: Path to JSON configuration file
Returns:
Configuration dictionary
"""
config = {
"llm_provider": LLMProvider.LOCAL_LLAMA,
"hardware_backend": None, # Auto-detect
"model_path": None,
"context_length": 4096,
"gpu_layers": 35,
"openai_api_key": None,
"openai_model": "gpt-4",
"api_keys": {}
}
# Load from file if provided
if config_file and os.path.exists(config_file):
with open(config_file, 'r') as f:
file_config = json.load(f)
config.update(file_config)
# Override with environment variables
env_mappings = {
"LLM_PROVIDER": "llm_provider",
"MODEL_PATH": "model_path",
"CONTEXT_LENGTH": "context_length",
"GPU_LAYERS": "gpu_layers",
"OPENAI_API_KEY": "openai_api_key",
"OPENAI_MODEL": "openai_model"
}
for env_var, config_key in env_mappings.items():
value = os.getenv(env_var)
if value:
if config_key in ["context_length", "gpu_layers"]:
config[config_key] = int(value)
elif config_key == "llm_provider":
config[config_key] = LLMProvider(value)
else:
config[config_key] = value
# Load API keys
api_key_vars = [
"AMADEUS_API_KEY",
"AMADEUS_API_SECRET",
"BOOKING_API_KEY",
"WEATHER_API_KEY",
"GOOGLE_MAPS_API_KEY",
"GOOGLE_PLACES_API_KEY",
"TICKETMASTER_API_KEY",
"EXCHANGE_RATE_API_KEY"
]
for var in api_key_vars:
value = os.getenv(var)
if value:
config["api_keys"][var] = value
return config
@staticmethod
def validate_config(config: Dict[str, Any]) -> bool:
"""
Validate configuration completeness.
Args:
config: Configuration dictionary
Returns:
True if valid, raises exception otherwise
"""
provider = config.get("llm_provider")
if provider == LLMProvider.OPENAI:
if not config.get("openai_api_key"):
raise ValueError("OpenAI API key required for OpenAI provider")
elif provider == LLMProvider.LOCAL_LLAMA:
if not config.get("model_path"):
raise ValueError("Model path required for local LLAMA provider")
if not os.path.exists(config["model_path"]):
raise ValueError(f"Model file not found: {config['model_path']}")
return True
Configuration management enables flexible deployment across different environments with appropriate validation.
ERROR HANDLING AND RESILIENCE
Production systems require robust error handling to gracefully manage failures:
class RetryStrategy:
"""Implements retry logic with exponential backoff."""
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
"""
Initialize retry strategy.
Args:
max_retries: Maximum number of retry attempts
base_delay: Base delay in seconds between retries
"""
self.max_retries = max_retries
self.base_delay = base_delay
def execute_with_retry(self, func, *args, **kwargs):
"""
Execute function with retry logic.
Args:
func: Function to execute
*args: Positional arguments for function
**kwargs: Keyword arguments for function
Returns:
Function result
Raises:
Exception if all retries exhausted
"""
import time
last_exception = None
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < self.max_retries - 1:
delay = self.base_delay * (2 ** attempt)
logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
time.sleep(delay)
else:
logger.error(f"All {self.max_retries} attempts failed")
raise last_exception
class CircuitBreaker:
"""Implements circuit breaker pattern for external service calls."""
def __init__(self, failure_threshold: int = 5, timeout: float = 60.0):
"""
Initialize circuit breaker.
Args:
failure_threshold: Number of failures before opening circuit
timeout: Seconds to wait before attempting to close circuit
"""
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "closed" # closed, open, half_open
def call(self, func, *args, **kwargs):
"""
Execute function through circuit breaker.
Args:
func: Function to execute
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Function result
Raises:
Exception if circuit is open
"""
import time
if self.state == "open":
if time.time() - self.last_failure_time >= self.timeout:
self.state = "half_open"
logger.info("Circuit breaker entering half-open state")
else:
raise Exception("Circuit breaker is open - service unavailable")
try:
result = func(*args, **kwargs)
if self.state == "half_open":
self.state = "closed"
self.failure_count = 0
logger.info("Circuit breaker closed - service recovered")
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
logger.error(f"Circuit breaker opened after {self.failure_count} failures")
raise e
The circuit breaker pattern prevents cascading failures by stopping requests to failing services temporarily. When a service experiences repeated failures, the circuit opens and rejects requests immediately rather than waiting for timeouts. After a cooling period, the circuit enters a half-open state to test if the service has recovered.
LOGGING AND MONITORING
Comprehensive logging enables debugging and performance monitoring in production:
class TravelPlannerLogger:
"""Centralized logging for travel planner application."""
def __init__(self, log_file: Optional[str] = None):
"""
Initialize logger with file and console handlers.
Args:
log_file: Path to log file, or None for console only
"""
self.logger = logging.getLogger("TravelPlanner")
self.logger.setLevel(logging.INFO)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_format = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(console_format)
self.logger.addHandler(console_handler)
# File handler if specified
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.DEBUG)
file_format = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(file_format)
self.logger.addHandler(file_handler)
def log_agent_execution(self, agent_name: str, action: str, duration: float, success: bool):
"""
Log agent execution metrics.
Args:
agent_name: Name of the agent
action: Action performed
duration: Execution duration in seconds
success: Whether execution succeeded
"""
status = "SUCCESS" if success else "FAILED"
self.logger.info(
f"Agent: {agent_name} | Action: {action} | Duration: {duration:.2f}s | Status: {status}"
)
def log_tool_invocation(self, tool_name: str, parameters: Dict[str, Any], result: ToolResult):
"""
Log tool invocation details.
Args:
tool_name: Name of the tool
parameters: Tool parameters
result: Tool execution result
"""
self.logger.debug(f"Tool: {tool_name} | Params: {parameters}")
if result.success:
self.logger.debug(f"Tool {tool_name} succeeded")
else:
self.logger.warning(f"Tool {tool_name} failed: {result.error_message}")
def log_llm_generation(self, prompt_length: int, response_length: int, duration: float):
"""
Log LLM generation metrics.
Args:
prompt_length: Length of prompt in characters
response_length: Length of response in characters
duration: Generation duration in seconds
"""
self.logger.debug(
f"LLM Generation | Prompt: {prompt_length} chars | "
f"Response: {response_length} chars | Duration: {duration:.2f}s"
)
Structured logging captures execution metrics, tool invocations, and LLM interactions for performance analysis and debugging.
TESTING INFRASTRUCTURE
Comprehensive testing ensures system reliability:
import unittest
from unittest.mock import Mock, patch
class TestTravelParameters(unittest.TestCase):
"""Test cases for TravelParameters class."""
def test_is_complete_with_all_parameters(self):
"""Test completeness check with all required parameters."""
from datetime import date
params = TravelParameters(
origin="New York, USA",
destination="Paris, France",
start_date=date(2025, 10, 20),
end_date=date(2025, 10, 27),
travel_purpose="leisure",
transport_modes=["flight"],
min_hotel_rating=3,
max_hotel_rating=4
)
self.assertTrue(params.is_complete())
self.assertEqual(len(params.missing_parameters()), 0)
def test_is_complete_with_missing_parameters(self):
"""Test completeness check with missing parameters."""
params = TravelParameters(
origin="New York, USA",
destination="Paris, France"
)
self.assertFalse(params.is_complete())
missing = params.missing_parameters()
self.assertIn("start date", missing)
self.assertIn("end date", missing)
def test_missing_parameters_list(self):
"""Test missing parameters identification."""
params = TravelParameters()
missing = params.missing_parameters()
self.assertGreater(len(missing), 0)
self.assertIn("origin location", missing)
self.assertIn("destination", missing)
class TestHardwareDetector(unittest.TestCase):
"""Test cases for hardware detection."""
def test_detect_backend_returns_valid_backend(self):
"""Test that hardware detection returns a valid backend."""
backend = HardwareDetector.detect_backend()
self.assertIsInstance(backend, HardwareBackend)
self.assertIn(backend, list(HardwareBackend))
def test_get_device_string(self):
"""Test device string conversion."""
device_str = HardwareDetector.get_device_string(HardwareBackend.CUDA)
self.assertEqual(device_str, "cuda")
device_str = HardwareDetector.get_device_string(HardwareBackend.CPU)
self.assertEqual(device_str, "cpu")
class TestToolExecution(unittest.TestCase):
"""Test cases for tool execution."""
def test_tool_result_success(self):
"""Test successful tool result."""
result = ToolResult(
success=True,
data={"test": "data"},
metadata={"count": 5}
)
self.assertTrue(result.success)
self.assertEqual(result.data["test"], "data")
self.assertIsNone(result.error_message)
def test_tool_result_failure(self):
"""Test failed tool result."""
result = ToolResult(
success=False,
data=None,
error_message="Test error"
)
self.assertFalse(result.success)
self.assertIsNone(result.data)
self.assertEqual(result.error_message, "Test error")
class TestAgentBase(unittest.TestCase):
"""Test cases for Agent base class."""
def setUp(self):
"""Set up test fixtures."""
self.mock_llm = Mock(spec=LLMBackend)
self.mock_llm.generate.return_value = "Test response"
self.mock_llm.count_tokens.return_value = 10
def test_agent_initialization(self):
"""Test agent initialization."""
agent = Agent(
name="Test Agent",
description="Test description",
llm_backend=self.mock_llm
)
self.assertEqual(agent.name, "Test Agent")
self.assertEqual(agent.description, "Test description")
self.assertEqual(len(agent.conversation_history), 0)
def test_add_message(self):
"""Test adding messages to conversation history."""
agent = Agent("Test", "Test", self.mock_llm)
agent.add_message("user", "Hello")
agent.add_message("assistant", "Hi there")
self.assertEqual(len(agent.conversation_history), 2)
self.assertEqual(agent.conversation_history[0]["role"], "user")
self.assertEqual(agent.conversation_history[1]["role"], "assistant")
def test_tool_execution_success(self):
"""Test successful tool execution."""
mock_tool = Mock(spec=Tool)
mock_tool.name = "test_tool"
mock_tool.execute.return_value = ToolResult(success=True, data={"result": "success"})
agent = Agent("Test", "Test", self.mock_llm, tools=[mock_tool])
result = agent.execute_tool("test_tool", param1="value1")
self.assertTrue(result.success)
mock_tool.execute.assert_called_once_with(param1="value1")
def test_tool_execution_not_found(self):
"""Test tool execution with non-existent tool."""
agent = Agent("Test", "Test", self.mock_llm)
result = agent.execute_tool("nonexistent_tool")
self.assertFalse(result.success)
self.assertIn("not found", result.error_message)
class TestRetryStrategy(unittest.TestCase):
"""Test cases for retry strategy."""
def test_successful_execution_no_retry(self):
"""Test that successful execution doesn't retry."""
retry = RetryStrategy(max_retries=3)
mock_func = Mock(return_value="success")
result = retry.execute_with_retry(mock_func, "arg1", kwarg1="value1")
self.assertEqual(result, "success")
self.assertEqual(mock_func.call_count, 1)
def test_retry_on_failure(self):
"""Test retry behavior on failures."""
retry = RetryStrategy(max_retries=3, base_delay=0.01)
mock_func = Mock(side_effect=[Exception("Error 1"), Exception("Error 2"), "success"])
result = retry.execute_with_retry(mock_func)
self.assertEqual(result, "success")
self.assertEqual(mock_func.call_count, 3)
def test_exhausted_retries(self):
"""Test behavior when all retries are exhausted."""
retry = RetryStrategy(max_retries=2, base_delay=0.01)
mock_func = Mock(side_effect=Exception("Persistent error"))
with self.assertRaises(Exception) as context:
retry.execute_with_retry(mock_func)
self.assertIn("Persistent error", str(context.exception))
self.assertEqual(mock_func.call_count, 2)
class TestCircuitBreaker(unittest.TestCase):
"""Test cases for circuit breaker."""
def test_closed_circuit_allows_calls(self):
"""Test that closed circuit allows function calls."""
breaker = CircuitBreaker(failure_threshold=3)
mock_func = Mock(return_value="success")
result = breaker.call(mock_func)
self.assertEqual(result, "success")
self.assertEqual(breaker.state, "closed")
def test_circuit_opens_after_threshold(self):
"""Test that circuit opens after failure threshold."""
breaker = CircuitBreaker(failure_threshold=3, timeout=0.1)
mock_func = Mock(side_effect=Exception("Error"))
for i in range(3):
try:
breaker.call(mock_func)
except Exception:
pass
self.assertEqual(breaker.state, "open")
def test_open_circuit_rejects_calls(self):
"""Test that open circuit rejects calls."""
breaker = CircuitBreaker(failure_threshold=1, timeout=10.0)
mock_func = Mock(side_effect=Exception("Error"))
try:
breaker.call(mock_func)
except Exception:
pass
with self.assertRaises(Exception) as context:
breaker.call(mock_func)
self.assertIn("Circuit breaker is open", str(context.exception))
Unit tests validate individual components ensuring they function correctly in isolation. Integration tests would verify that components work together properly.
PERFORMANCE OPTIMIZATION
Performance optimization is critical for responsive user experiences:
class CacheManager:
"""Manages caching of API responses and LLM generations."""
def __init__(self, cache_ttl: int = 3600):
"""
Initialize cache manager.
Args:
cache_ttl: Time-to-live for cache entries in seconds
"""
self.cache = {}
self.cache_ttl = cache_ttl
self.cache_timestamps = {}
def get(self, key: str) -> Optional[Any]:
"""
Retrieve value from cache.
Args:
key: Cache key
Returns:
Cached value or None if not found or expired
"""
import time
if key not in self.cache:
return None
timestamp = self.cache_timestamps.get(key, 0)
if time.time() - timestamp > self.cache_ttl:
del self.cache[key]
del self.cache_timestamps[key]
return None
return self.cache[key]
def set(self, key: str, value: Any):
"""
Store value in cache.
Args:
key: Cache key
value: Value to cache
"""
import time
self.cache[key] = value
self.cache_timestamps[key] = time.time()
def invalidate(self, key: str):
"""
Invalidate cache entry.
Args:
key: Cache key to invalidate
"""
if key in self.cache:
del self.cache[key]
del self.cache_timestamps[key]
def clear(self):
"""Clear all cache entries."""
self.cache.clear()
self.cache_timestamps.clear()
class BatchProcessor:
"""Processes multiple requests in batches for efficiency."""
def __init__(self, batch_size: int = 10, timeout: float = 1.0):
"""
Initialize batch processor.
Args:
batch_size: Maximum batch size
timeout: Maximum time to wait for batch to fill
"""
self.batch_size = batch_size
self.timeout = timeout
self.pending_requests = []
def add_request(self, request: Any) -> Any:
"""
Add request to batch and process when ready.
Args:
request: Request to process
Returns:
Processing result
"""
import time
import threading
self.pending_requests.append(request)
if len(self.pending_requests) >= self.batch_size:
return self._process_batch()
# Wait for timeout or batch to fill
start_time = time.time()
while time.time() - start_time < self.timeout:
if len(self.pending_requests) >= self.batch_size:
return self._process_batch()
time.sleep(0.1)
return self._process_batch()
def _process_batch(self) -> List[Any]:
"""Process accumulated batch of requests."""
if not self.pending_requests:
return []
batch = self.pending_requests[:self.batch_size]
self.pending_requests = self.pending_requests[self.batch_size:]
# Process batch
results = []
for request in batch:
# Process each request
results.append(self._process_single(request))
return results
def _process_single(self, request: Any) -> Any:
"""Process a single request."""
# Implementation depends on request type
return request
Caching reduces redundant API calls and LLM generations, significantly improving response times for repeated queries. Batch processing optimizes throughput when handling multiple concurrent requests.
SECURITY CONSIDERATIONS
Security is paramount when handling user data and API credentials:
class SecurityManager:
"""Manages security aspects of the application."""
@staticmethod
def sanitize_user_input(user_input: str) -> str:
"""
Sanitize user input to prevent injection attacks.
Args:
user_input: Raw user input
Returns:
Sanitized input
"""
import re
# Remove potential SQL injection patterns
sanitized = re.sub(r'[;\'"\\]', '', user_input)
# Limit length
max_length = 10000
sanitized = sanitized[:max_length]
# Remove control characters
sanitized = ''.join(char for char in sanitized if ord(char) >= 32 or char in '\n\r\t')
return sanitized
@staticmethod
def validate_api_key(api_key: str, service: str) -> bool:
"""
Validate API key format.
Args:
api_key: API key to validate
service: Service name for key
Returns:
True if valid format
"""
if not api_key or len(api_key) < 10:
return False
# Service-specific validation
if service == "openai":
return api_key.startswith("sk-")
return True
@staticmethod
def encrypt_sensitive_data(data: str, key: bytes) -> bytes:
"""
Encrypt sensitive data.
Args:
data: Data to encrypt
key: Encryption key
Returns:
Encrypted data
"""
from cryptography.fernet import Fernet
f = Fernet(key)
encrypted = f.encrypt(data.encode())
return encrypted
@staticmethod
def decrypt_sensitive_data(encrypted_data: bytes, key: bytes) -> str:
"""
Decrypt sensitive data.
Args:
encrypted_data: Encrypted data
key: Decryption key
Returns:
Decrypted data
"""
from cryptography.fernet import Fernet
f = Fernet(key)
decrypted = f.decrypt(encrypted_data)
return decrypted.decode()
@staticmethod
def generate_encryption_key() -> bytes:
"""
Generate a new encryption key.
Returns:
Encryption key
"""
from cryptography.fernet import Fernet
return Fernet.generate_key()
Security measures include input sanitization, API key validation, and encryption of sensitive data like user credentials and payment information.
DEPLOYMENT CONSIDERATIONS
Production deployment requires careful planning and infrastructure setup:
class DeploymentManager:
"""Manages deployment configuration and health checks."""
def __init__(self, config: Dict[str, Any]):
"""
Initialize deployment manager.
Args:
config: Deployment configuration
"""
self.config = config
self.health_status = {
"llm_backend": "unknown",
"api_services": {},
"database": "unknown"
}
def health_check(self) -> Dict[str, Any]:
"""
Perform comprehensive health check.
Returns:
Health status dictionary
"""
import time
start_time = time.time()
# Check LLM backend
try:
# Attempt simple generation
test_prompt = "Hello"
# This would use actual LLM backend
self.health_status["llm_backend"] = "healthy"
except Exception as e:
self.health_status["llm_backend"] = f"unhealthy: {str(e)}"
# Check API services
api_services = [
"flight_api",
"hotel_api",
"weather_api",
"maps_api"
]
for service in api_services:
try:
# Attempt API connection
# This would make actual API call
self.health_status["api_services"][service] = "healthy"
except Exception as e:
self.health_status["api_services"][service] = f"unhealthy: {str(e)}"
duration = time.time() - start_time
return {
"status": "healthy" if self._is_healthy() else "unhealthy",
"components": self.health_status,
"check_duration": duration,
"timestamp": time.time()
}
def _is_healthy(self) -> bool:
"""Check if all critical components are healthy."""
if "unhealthy" in self.health_status["llm_backend"]:
return False
critical_services = ["flight_api", "hotel_api"]
for service in critical_services:
status = self.health_status["api_services"].get(service, "unknown")
if "unhealthy" in status:
return False
return True
def get_metrics(self) -> Dict[str, Any]:
"""
Collect application metrics.
Returns:
Metrics dictionary
"""
import psutil
return {
"cpu_percent": psutil.cpu_percent(interval=1),
"memory_percent": psutil.virtual_memory().percent,
"disk_percent": psutil.disk_usage('/').percent,
"process_count": len(psutil.pids()),
"timestamp": time.time()
}
Health checks and metrics collection enable monitoring and alerting in production environments.
COMPLETE RUNNING EXAMPLE
The following complete implementation demonstrates all components working together in a production-ready system:
#!/usr/bin/env python3
"""
Complete LLM-Based Agentic Travel Planner
This is a production-ready implementation of an intelligent travel planning system
that uses multiple specialized AI agents to research and plan comprehensive trips.
"""
import os
import sys
import json
import logging
import argparse
from typing import Any, Dict, List, Optional
from datetime import date, datetime
from dataclasses import dataclass, field
import abc
import enum
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('travel_planner.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# ============================================================================
# CORE DATA STRUCTURES
# ============================================================================
class HardwareBackend(enum.Enum):
"""Supported hardware acceleration backends."""
CUDA = "cuda"
ROCM = "rocm"
MPS = "mps"
INTEL = "intel"
CPU = "cpu"
class LLMProvider(enum.Enum):
"""Supported LLM providers."""
OPENAI = "openai"
ANTHROPIC = "anthropic"
LOCAL_LLAMA = "local_llama"
OLLAMA = "ollama"
@dataclass
class TravelParameters:
"""Complete travel planning parameters."""
origin: Optional[str] = None
destination: Optional[str] = None
start_date: Optional[date] = None
end_date: Optional[date] = None
travel_purpose: Optional[str] = None
preferred_departure_time: Optional[str] = None
preferred_return_time: Optional[str] = None
transport_modes: List[str] = field(default_factory=list)
min_hotel_rating: Optional[int] = None
max_hotel_rating: Optional[int] = None
budget_max: Optional[float] = None
travelers_count: int = 1
def is_complete(self) -> bool:
"""Check if all required parameters are present."""
required = [
self.origin,
self.destination,
self.start_date,
self.end_date,
self.travel_purpose,
self.transport_modes,
self.min_hotel_rating is not None,
self.max_hotel_rating is not None
]
return all(required) and len(self.transport_modes) > 0
def missing_parameters(self) -> List[str]:
"""Return list of missing parameter names."""
missing = []
if not self.origin:
missing.append("origin location")
if not self.destination:
missing.append("destination")
if not self.start_date:
missing.append("start date")
if not self.end_date:
missing.append("end date")
if not self.travel_purpose:
missing.append("travel purpose (business or leisure)")
if not self.transport_modes:
missing.append("preferred transportation modes")
if self.min_hotel_rating is None:
missing.append("minimum hotel rating")
if self.max_hotel_rating is None:
missing.append("maximum hotel rating")
return missing
@dataclass
class ToolResult:
"""Result from tool execution."""
success: bool
data: Any
error_message: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
# ============================================================================
# HARDWARE DETECTION
# ============================================================================
class HardwareDetector:
"""Detects available hardware acceleration."""
@staticmethod
def detect_backend() -> HardwareBackend:
"""Automatically detect best available hardware backend."""
# Try NVIDIA CUDA
try:
import torch
if torch.cuda.is_available():
logger.info(f"Detected CUDA: {torch.cuda.get_device_name(0)}")
return HardwareBackend.CUDA
except ImportError:
pass
except Exception as e:
logger.warning(f"CUDA detection failed: {e}")
# Try AMD ROCm
try:
import torch
if hasattr(torch.version, 'hip') and torch.version.hip:
if torch.cuda.is_available():
logger.info("Detected ROCm")
return HardwareBackend.ROCM
except Exception as e:
logger.warning(f"ROCm detection failed: {e}")
# Try Apple MPS
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
logger.info("Detected Apple MPS")
return HardwareBackend.MPS
except Exception as e:
logger.warning(f"MPS detection failed: {e}")
# Try Intel GPU
try:
import intel_extension_for_pytorch as ipex
if ipex.xpu.is_available():
logger.info("Detected Intel GPU")
return HardwareBackend.INTEL
except ImportError:
pass
except Exception as e:
logger.warning(f"Intel GPU detection failed: {e}")
logger.info("No GPU detected, using CPU")
return HardwareBackend.CPU
@staticmethod
def get_device_string(backend: HardwareBackend) -> str:
"""Convert backend to device string."""
mapping = {
HardwareBackend.CUDA: "cuda",
HardwareBackend.ROCM: "cuda",
HardwareBackend.MPS: "mps",
HardwareBackend.INTEL: "xpu",
HardwareBackend.CPU: "cpu"
}
return mapping.get(backend, "cpu")
# ============================================================================
# LLM BACKENDS
# ============================================================================
class LLMBackend(abc.ABC):
"""Abstract base for LLM backends."""
@abc.abstractmethod
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, stop_sequences: Optional[List[str]] = None) -> str:
"""Generate text completion."""
pass
@abc.abstractmethod
def count_tokens(self, text: str) -> int:
"""Count tokens in text."""
pass
class OpenAIBackend(LLMBackend):
"""OpenAI API backend."""
def __init__(self, api_key: str, model: str = "gpt-4"):
self.api_key = api_key
self.model = model
try:
from openai import OpenAI
self.client = OpenAI(api_key=api_key)
except ImportError:
raise ImportError("Install openai: pip install openai")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, stop_sequences: Optional[List[str]] = None) -> str:
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
temperature=temperature,
stop=stop_sequences
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"OpenAI generation failed: {e}")
raise
def count_tokens(self, text: str) -> int:
try:
import tiktoken
encoding = tiktoken.encoding_for_model(self.model)
return len(encoding.encode(text))
except ImportError:
return int(len(text.split()) * 1.3)
class LocalLlamaBackend(LLMBackend):
"""Local LLAMA model using llama-cpp-python."""
def __init__(self, model_path: str, hardware_backend: HardwareBackend,
n_ctx: int = 4096, n_gpu_layers: int = 35):
self.model_path = model_path
self.hardware_backend = hardware_backend
try:
from llama_cpp import Llama
gpu_layers = n_gpu_layers if hardware_backend != HardwareBackend.CPU else 0
self.model = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_gpu_layers=gpu_layers,
verbose=False
)
logger.info(f"Loaded LLAMA model with {gpu_layers} GPU layers")
except ImportError:
raise ImportError("Install llama-cpp-python: pip install llama-cpp-python")
def generate(self, prompt: str, max_tokens: int = 1024,
temperature: float = 0.7, stop_sequences: Optional[List[str]] = None) -> str:
try:
response = self.model(
prompt,
max_tokens=max_tokens,
temperature=temperature,
stop=stop_sequences or [],
echo=False
)
return response['choices'][0]['text']
except Exception as e:
logger.error(f"Local LLAMA generation failed: {e}")
raise
def count_tokens(self, text: str) -> int:
tokens = self.model.tokenize(text.encode('utf-8'))
return len(tokens)
# ============================================================================
# TOOLS
# ============================================================================
class Tool(abc.ABC):
"""Abstract base for agent tools."""
def __init__(self, name: str, description: str):
self.name = name
self.description = description
@abc.abstractmethod
def execute(self, **kwargs) -> ToolResult:
"""Execute tool with parameters."""
pass
@abc.abstractmethod
def get_parameters_schema(self) -> Dict[str, Any]:
"""Return parameter schema."""
pass
class FlightSearchTool(Tool):
"""Search for flights."""
def __init__(self):
super().__init__("flight_search", "Search for available flights")
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"origin": {"type": "string"},
"destination": {"type": "string"},
"departure_date": {"type": "string", "format": "date"},
"return_date": {"type": "string", "format": "date"},
"passengers": {"type": "integer", "minimum": 1}
},
"required": ["origin", "destination", "departure_date"]
}
def execute(self, **kwargs) -> ToolResult:
"""Execute flight search."""
origin = kwargs.get("origin")
destination = kwargs.get("destination")
departure_date = kwargs.get("departure_date")
return_date = kwargs.get("return_date")
passengers = kwargs.get("passengers", 1)
try:
import requests
# Amadeus API integration
auth_url = "https://test.api.amadeus.com/v1/security/oauth2/token"
auth_data = {
"grant_type": "client_credentials",
"client_id": os.getenv("AMADEUS_API_KEY", ""),
"client_secret": os.getenv("AMADEUS_API_SECRET", "")
}
auth_response = requests.post(auth_url, data=auth_data, timeout=10)
if auth_response.status_code != 200:
return ToolResult(success=False, data=None,
error_message="Flight API authentication failed")
access_token = auth_response.json()["access_token"]
search_url = "https://test.api.amadeus.com/v2/shopping/flight-offers"
headers = {"Authorization": f"Bearer {access_token}"}
params = {
"originLocationCode": origin,
"destinationLocationCode": destination,
"departureDate": departure_date,
"adults": passengers,
"max": 10
}
if return_date:
params["returnDate"] = return_date
search_response = requests.get(search_url, headers=headers,
params=params, timeout=30)
if search_response.status_code != 200:
return ToolResult(success=False, data=None,
error_message=f"Flight search failed: {search_response.text}")
flight_data = search_response.json()
flights = []
for offer in flight_data.get("data", []):
for itinerary in offer.get("itineraries", []):
flight_info = {
"price": float(offer["price"]["total"]),
"currency": offer["price"]["currency"],
"segments": []
}
for segment in itinerary.get("segments", []):
flight_info["segments"].append({
"departure": {
"airport": segment["departure"]["iataCode"],
"time": segment["departure"]["at"]
},
"arrival": {
"airport": segment["arrival"]["iataCode"],
"time": segment["arrival"]["at"]
},
"carrier": segment["carrierCode"],
"flight_number": segment["number"],
"duration": segment["duration"]
})
flights.append(flight_info)
return ToolResult(success=True, data={"flights": flights},
metadata={"result_count": len(flights)})
except Exception as e:
logger.error(f"Flight search error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class HotelSearchTool(Tool):
"""Search for hotels."""
def __init__(self):
super().__init__("hotel_search", "Search for hotel accommodations")
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"destination": {"type": "string"},
"check_in": {"type": "string", "format": "date"},
"check_out": {"type": "string", "format": "date"},
"guests": {"type": "integer", "minimum": 1},
"min_rating": {"type": "integer", "minimum": 1, "maximum": 5},
"max_rating": {"type": "integer", "minimum": 1, "maximum": 5}
},
"required": ["destination", "check_in", "check_out"]
}
def execute(self, **kwargs) -> ToolResult:
"""Execute hotel search."""
destination = kwargs.get("destination")
check_in = kwargs.get("check_in")
check_out = kwargs.get("check_out")
guests = kwargs.get("guests", 1)
min_rating = kwargs.get("min_rating", 1)
max_rating = kwargs.get("max_rating", 5)
try:
import requests
from datetime import datetime
# Calculate number of nights
check_in_date = datetime.fromisoformat(check_in)
check_out_date = datetime.fromisoformat(check_out)
nights = (check_out_date - check_in_date).days
# Booking.com API integration
api_key = os.getenv("BOOKING_API_KEY", "")
base_url = "https://api.booking.com/v1/hotels/search"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"destination": destination,
"checkin": check_in,
"checkout": check_out,
"guests": guests,
"min_star_rating": min_rating,
"max_star_rating": max_rating
}
response = requests.post(base_url, headers=headers,
json=payload, timeout=30)
if response.status_code != 200:
return ToolResult(success=False, data=None,
error_message=f"Hotel search failed: {response.text}")
hotel_data = response.json()
hotels = []
for property_data in hotel_data.get("properties", []):
hotel_info = {
"name": property_data["name"],
"address": property_data["address"],
"star_rating": property_data["star_rating"],
"price_per_night": property_data["price"]["amount"],
"currency": property_data["price"]["currency"],
"total_price": property_data["price"]["amount"] * nights,
"room_type": property_data.get("room_type", "Standard"),
"breakfast_included": property_data.get("breakfast_included", False),
"free_cancellation": property_data.get("free_cancellation", False),
"cancellation_deadline": property_data.get("cancellation_deadline"),
"amenities": property_data.get("amenities", []),
"guest_rating": property_data.get("guest_rating", {}).get("score"),
"review_count": property_data.get("guest_rating", {}).get("count", 0),
"distance_to_center": property_data.get("distance_to_center_km"),
"images": property_data.get("images", [])[:3]
}
hotels.append(hotel_info)
hotels.sort(key=lambda x: (-x.get("guest_rating", 0), x["price_per_night"]))
return ToolResult(success=True, data={"hotels": hotels},
metadata={"result_count": len(hotels)})
except Exception as e:
logger.error(f"Hotel search error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class WeatherForecastTool(Tool):
"""Get weather forecasts."""
def __init__(self):
super().__init__("weather_forecast", "Get weather forecast")
def get_parameters_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"location": {"type": "string"},
"start_date": {"type": "string", "format": "date"},
"end_date": {"type": "string", "format": "date"}
},
"required": ["location", "start_date", "end_date"]
}
def execute(self, **kwargs) -> ToolResult:
"""Execute weather forecast retrieval."""
location = kwargs.get("location")
start_date = kwargs.get("start_date")
end_date = kwargs.get("end_date")
try:
import requests
from datetime import datetime, timedelta
api_key = os.getenv("WEATHER_API_KEY", "")
# Geocode location
geo_url = "http://api.openweathermap.org/geo/1.0/direct"
geo_params = {"q": location, "limit": 1, "appid": api_key}
geo_response = requests.get(geo_url, params=geo_params, timeout=10)
if geo_response.status_code != 200 or not geo_response.json():
return ToolResult(success=False, data=None,
error_message="Location not found")
geo_data = geo_response.json()[0]
lat, lon = geo_data["lat"], geo_data["lon"]
# Get forecast
forecast_url = "https://api.openweathermap.org/data/2.5/forecast"
forecast_params = {
"lat": lat,
"lon": lon,
"appid": api_key,
"units": "metric"
}
forecast_response = requests.get(forecast_url, params=forecast_params,
timeout=10)
if forecast_response.status_code != 200:
return ToolResult(success=False, data=None,
error_message="Weather forecast unavailable")
forecast_data = forecast_response.json()
start = datetime.fromisoformat(start_date)
end = datetime.fromisoformat(end_date)
daily_forecasts = {}
for item in forecast_data.get("list", []):
forecast_time = datetime.fromtimestamp(item["dt"])
if start <= forecast_time <= end + timedelta(days=1):
date_key = forecast_time.date().isoformat()
if date_key not in daily_forecasts:
daily_forecasts[date_key] = {
"date": date_key,
"temperatures": [],
"conditions": [],
"precipitation": 0,
"humidity": [],
"wind_speed": []
}
daily_forecasts[date_key]["temperatures"].append(item["main"]["temp"])
daily_forecasts[date_key]["conditions"].append(
item["weather"][0]["description"]
)
daily_forecasts[date_key]["precipitation"] += item.get("rain", {}).get("3h", 0)
daily_forecasts[date_key]["humidity"].append(item["main"]["humidity"])
daily_forecasts[date_key]["wind_speed"].append(item["wind"]["speed"])
forecasts = []
for date_key in sorted(daily_forecasts.keys()):
day_data = daily_forecasts[date_key]
forecasts.append({
"date": date_key,
"temp_min": min(day_data["temperatures"]),
"temp_max": max(day_data["temperatures"]),
"temp_avg": sum(day_data["temperatures"]) / len(day_data["temperatures"]),
"condition": max(set(day_data["conditions"]),
key=day_data["conditions"].count),
"precipitation_mm": day_data["precipitation"],
"humidity_avg": sum(day_data["humidity"]) / len(day_data["humidity"]),
"wind_speed_avg": sum(day_data["wind_speed"]) / len(day_data["wind_speed"])
})
return ToolResult(success=True, data={"forecasts": forecasts},
metadata={"location": location, "coordinates": {"lat": lat, "lon": lon}})
except Exception as e:
logger.error(f"Weather forecast error: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
# ============================================================================
# AGENTS
# ============================================================================
class Agent:
"""Base agent class."""
def __init__(self, name: str, description: str, llm_backend: LLMBackend,
tools: Optional[List[Tool]] = None):
self.name = name
self.description = description
self.llm_backend = llm_backend
self.tools = tools or []
self.conversation_history = []
def add_message(self, role: str, content: str):
"""Add message to conversation history."""
self.conversation_history.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def execute_tool(self, tool_name: str, **kwargs) -> ToolResult:
"""Execute a tool by name."""
tool = next((t for t in self.tools if t.name == tool_name), None)
if not tool:
return ToolResult(success=False, data=None,
error_message=f"Tool '{tool_name}' not found")
try:
logger.info(f"Agent {self.name} executing tool {tool_name}")
result = tool.execute(**kwargs)
logger.info(f"Tool {tool_name} completed: success={result.success}")
return result
except Exception as e:
logger.error(f"Tool {tool_name} execution failed: {e}")
return ToolResult(success=False, data=None, error_message=str(e))
class TransportationAgent(Agent):
"""Agent for transportation research."""
def __init__(self, llm_backend: LLMBackend):
tools = [FlightSearchTool()]
super().__init__(
name="Transportation Agent",
description="I find transportation options including flights and trains.",
llm_backend=llm_backend,
tools=tools
)
def research_transportation(self, params: TravelParameters) -> Dict[str, Any]:
"""Research transportation options."""
results = {"flights": [], "trains": [], "local_transit": []}
if "flight" in [mode.lower() for mode in params.transport_modes]:
flight_result = self.execute_tool(
"flight_search",
origin=params.origin,
destination=params.destination,
departure_date=params.start_date.isoformat(),
return_date=params.end_date.isoformat(),
passengers=params.travelers_count
)
if flight_result.success:
results["flights"] = flight_result.data.get("flights", [])
return results
class AccommodationAgent(Agent):
"""Agent for hotel research."""
def __init__(self, llm_backend: LLMBackend):
tools = [HotelSearchTool()]
super().__init__(
name="Accommodation Agent",
description="I find suitable hotel accommodations.",
llm_backend=llm_backend,
tools=tools
)
def research_accommodations(self, params: TravelParameters) -> Dict[str, Any]:
"""Research hotel options."""
hotel_result = self.execute_tool(
"hotel_search",
destination=params.destination,
check_in=params.start_date.isoformat(),
check_out=params.end_date.isoformat(),
guests=params.travelers_count,
min_rating=params.min_hotel_rating,
max_rating=params.max_hotel_rating
)
if hotel_result.success:
hotels = hotel_result.data.get("hotels", [])
return {"hotels": hotels, "total_found": len(hotels)}
else:
return {"hotels": [], "total_found": 0, "error": hotel_result.error_message}
class WeatherAgent(Agent):
"""Agent for weather forecasting."""
def __init__(self, llm_backend: LLMBackend):
tools = [WeatherForecastTool()]
super().__init__(
name="Weather Agent",
description="I provide weather forecasts for destinations.",
llm_backend=llm_backend,
tools=tools
)
def research_weather(self, params: TravelParameters) -> Dict[str, Any]:
"""Get weather forecast."""
weather_result = self.execute_tool(
"weather_forecast",
location=params.destination,
start_date=params.start_date.isoformat(),
end_date=params.end_date.isoformat()
)
if weather_result.success:
return {"forecasts": weather_result.data.get("forecasts", [])}
else:
return {"forecasts": [], "error": weather_result.error_message}
class CoordinatorAgent(Agent):
"""Main coordinator agent."""
def __init__(self, llm_backend: LLMBackend, transportation_agent: TransportationAgent,
accommodation_agent: AccommodationAgent, weather_agent: WeatherAgent):
super().__init__(
name="Travel Planning Coordinator",
description="I coordinate the travel planning process.",
llm_backend=llm_backend
)
self.transportation_agent = transportation_agent
self.accommodation_agent = accommodation_agent
self.weather_agent = weather_agent
self.travel_params = TravelParameters()
def extract_parameters_from_message(self, message: str):
"""Extract travel parameters from user message."""
extraction_prompt = f"""Extract travel parameters from this message and return JSON:
{{
"origin": "origin location or null",
"destination": "destination or null",
"start_date": "YYYY-MM-DD or null",
"end_date": "YYYY-MM-DD or null",
"travel_purpose": "business or leisure or null",
"transport_modes": ["flight", "train", etc],
"min_hotel_rating": 1-5 or null,
"max_hotel_rating": 1-5 or null,
"travelers_count": number or 1
}}
Message: {message}
Return only JSON, no other text."""
try:
response = self.llm_backend.generate(extraction_prompt, temperature=0.3,
max_tokens=512)
import re
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
extracted = json.loads(json_match.group())
if extracted.get("origin"):
self.travel_params.origin = extracted["origin"]
if extracted.get("destination"):
self.travel_params.destination = extracted["destination"]
if extracted.get("start_date"):
self.travel_params.start_date = datetime.fromisoformat(
extracted["start_date"]
).date()
if extracted.get("end_date"):
self.travel_params.end_date = datetime.fromisoformat(
extracted["end_date"]
).date()
if extracted.get("travel_purpose"):
self.travel_params.travel_purpose = extracted["travel_purpose"]
if extracted.get("transport_modes"):
self.travel_params.transport_modes = extracted["transport_modes"]
if extracted.get("min_hotel_rating") is not None:
self.travel_params.min_hotel_rating = extracted["min_hotel_rating"]
if extracted.get("max_hotel_rating") is not None:
self.travel_params.max_hotel_rating = extracted["max_hotel_rating"]
if extracted.get("travelers_count"):
self.travel_params.travelers_count = extracted["travelers_count"]
logger.info(f"Extracted parameters: {extracted}")
except Exception as e:
logger.error(f"Parameter extraction failed: {e}")
def collect_missing_parameters(self) -> Optional[str]:
"""Generate question for missing parameters."""
missing = self.travel_params.missing_parameters()
if not missing:
return None
question_prompt = f"""Generate a friendly question asking for: {', '.join(missing)}
Be conversational and helpful. Generate only the question."""
question = self.llm_backend.generate(question_prompt, temperature=0.7,
max_tokens=256)
return question.strip()
def plan_travel(self) -> Dict[str, Any]:
"""Execute complete travel planning."""
if not self.travel_params.is_complete():
missing = self.travel_params.missing_parameters()
return {
"status": "incomplete",
"message": f"Missing: {', '.join(missing)}"
}
results = {
"status": "complete",
"parameters": {
"origin": self.travel_params.origin,
"destination": self.travel_params.destination,
"dates": {
"start": self.travel_params.start_date.isoformat(),
"end": self.travel_params.end_date.isoformat()
}
}
}
# Execute agent research
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
future_transport = executor.submit(
self.transportation_agent.research_transportation,
self.travel_params
)
future_accommodation = executor.submit(
self.accommodation_agent.research_accommodations,
self.travel_params
)
future_weather = executor.submit(
self.weather_agent.research_weather,
self.travel_params
)
results["transportation"] = future_transport.result()
results["accommodation"] = future_accommodation.result()
results["weather"] = future_weather.result()
return results
def generate_report(self, results: Dict[str, Any]) -> str:
"""Generate comprehensive travel report."""
report = []
report.append("=" * 80)
report.append("TRAVEL PLAN")
report.append("=" * 80)
report.append("")
params = results.get("parameters", {})
report.append(f"Origin: {params.get('origin')}")
report.append(f"Destination: {params.get('destination')}")
dates = params.get("dates", {})
report.append(f"Dates: {dates.get('start')} to {dates.get('end')}")
report.append("")
# Transportation
report.append("TRANSPORTATION")
report.append("-" * 80)
transportation = results.get("transportation", {})
flights = transportation.get("flights", [])
if flights:
report.append(f"\nFlights ({len(flights)} found):")
for i, flight in enumerate(flights[:5], 1):
report.append(f"\n Option {i}:")
report.append(f" Price: {flight.get('price')} {flight.get('currency')}")
for j, seg in enumerate(flight.get('segments', []), 1):
dep = seg.get('departure', {})
arr = seg.get('arrival', {})
report.append(f" Segment {j}: {dep.get('airport')} -> {arr.get('airport')}")
report.append(f" Departs: {dep.get('time')}")
report.append(f" Arrives: {arr.get('time')}")
report.append("")
# Accommodation
report.append("ACCOMMODATION")
report.append("-" * 80)
accommodation = results.get("accommodation", {})
hotels = accommodation.get("hotels", [])
if hotels:
report.append(f"\nHotels ({len(hotels)} found):")
for hotel in hotels[:5]:
report.append(f"\n {hotel.get('name')}")
report.append(f" Rating: {hotel.get('star_rating')} stars")
report.append(f" Price: {hotel.get('price_per_night')} {hotel.get('currency')}/night")
report.append(f" Total: {hotel.get('total_price')} {hotel.get('currency')}")
report.append(f" Breakfast: {'Included' if hotel.get('breakfast_included') else 'Not included'}")
report.append("")
# Weather
report.append("WEATHER FORECAST")
report.append("-" * 80)
weather = results.get("weather", {})
forecasts = weather.get("forecasts", [])
if forecasts:
for forecast in forecasts:
report.append(f"\n{forecast.get('date')}:")
report.append(f" Temp: {forecast.get('temp_min'):.1f}°C - {forecast.get('temp_max'):.1f}°C")
report.append(f" Condition: {forecast.get('condition')}")
report.append(f" Precipitation: {forecast.get('precipitation_mm'):.1f}mm")
report.append("")
report.append("=" * 80)
return "\n".join(report)
# ============================================================================
# MAIN APPLICATION
# ============================================================================
class TravelPlannerApplication:
"""Main travel planner application."""
def __init__(self, config: Optional[Dict[str, Any]] = None):
self.config = config or {}
self.hardware_backend = HardwareDetector.detect_backend()
self.llm_backend = self._initialize_llm_backend()
# Initialize agents
self.transportation_agent = TransportationAgent(self.llm_backend)
self.accommodation_agent = AccommodationAgent(self.llm_backend)
self.weather_agent = WeatherAgent(self.llm_backend)
self.coordinator = CoordinatorAgent(
llm_backend=self.llm_backend,
transportation_agent=self.transportation_agent,
accommodation_agent=self.accommodation_agent,
weather_agent=self.weather_agent
)
def _initialize_llm_backend(self) -> LLMBackend:
"""Initialize LLM backend."""
provider = self.config.get("llm_provider", LLMProvider.LOCAL_LLAMA)
if provider == LLMProvider.OPENAI:
api_key = self.config.get("openai_api_key") or os.getenv("OPENAI_API_KEY")
model = self.config.get("openai_model", "gpt-4")
return OpenAIBackend(api_key=api_key, model=model)
elif provider == LLMProvider.LOCAL_LLAMA:
model_path = self.config.get("model_path") or os.getenv("LLAMA_MODEL_PATH")
if not model_path:
raise ValueError("Model path required for local LLAMA")
return LocalLlamaBackend(
model_path=model_path,
hardware_backend=self.hardware_backend,
n_ctx=self.config.get("context_length", 4096),
n_gpu_layers=self.config.get("gpu_layers", 35)
)
else:
raise ValueError(f"Unsupported provider: {provider}")
def process_user_message(self, message: str) -> str:
"""Process user message and return response."""
self.coordinator.extract_parameters_from_message(message)
missing_question = self.coordinator.collect_missing_parameters()
if missing_question:
return missing_question
logger.info("Executing travel planning")
results = self.coordinator.plan_travel()
if results.get("status") == "incomplete":
return results.get("message", "Unable to complete planning")
report = self.coordinator.generate_report(results)
return report
def run_interactive(self):
"""Run interactive CLI."""
print("=" * 80)
print("LLM-BASED TRAVEL PLANNER")
print("=" * 80)
print("\nWelcome! Tell me about your travel plans.")
print("Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit']:
print("\nThank you! Safe travels!")
break
response = self.process_user_message(user_input)
print(f"\nAssistant:\n{response}\n")
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
logger.error(f"Error: {e}")
print(f"\nError: {e}\n")
# ============================================================================
# ENTRY POINT
# ============================================================================
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="LLM-Based Travel Planner")
parser.add_argument("--config", help="Path to configuration file")
parser.add_argument("--provider", choices=["openai", "local_llama"],
default="local_llama", help="LLM provider")
parser.add_argument("--model-path", help="Path to local model file")
parser.add_argument("--openai-key", help="OpenAI API key")
args = parser.parse_args()
config = {}
if args.config and os.path.exists(args.config):
with open(args.config, 'r') as f:
config = json.load(f)
if args.provider:
config["llm_provider"] = LLMProvider(args.provider)
if args.model_path:
config["model_path"] = args.model_path
if args.openai_key:
config["openai_api_key"] = args.openai_key
try:
app = TravelPlannerApplication(config)
app.run_interactive()
except Exception as e:
logger.error(f"Application failed: {e}")
print(f"Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
This complete running example provides a production-ready implementation of the LLM-based agentic travel planner. The system supports both local and remote LLM deployment, automatically detects available hardware acceleration, integrates with real travel APIs, and provides comprehensive travel planning through coordinated specialized agents. The code follows clean architecture principles with proper error handling, logging, and extensibility for additional features.
No comments:
Post a Comment