Table of Contents
- 1.1 LangGraph Code Review – Part 1
- Import libraries and configuration
- Data structure and state definition
- 1.2 LangGraph Code Review – Part 2
- The TravelBookingWorkflow class
- Individual workflow nodes
- 1.3 Executing a LangGraph workflow
- Ranking and selection of options
- The human-in-the-loop phase
- Reservation and confirmation
- Complete workflow execution
- 1.4 Connecting LangGraph to an external API
- Integration with official Booking.com API
- Alternative via RapidAPI
- Parsing API responses
1. Introduction to Agentic AI and LangGraph
Generative AI tools are now easily capable of performing complex multi-step processes involving sophisticated assessments and real-world transactions. When we entrust an AI to carry out such tasks on our behalf, we describe this operation as agentic.
The challenge is to orchestrate these processes so that:
- You remain in control of the results.
- The state of key dynamic values is reliably maintained between steps.
This course focuses on the LangChain LangGraph library. The “Graph” part of LangGraph implies that we consider individual actions as nodes, and that the information flowing between the nodes is like the edges (edges) or lines that connect these nodes. Within LangGraph, these workflows are defined by objects of the StateGraph class.
The educational scenario: booking a business trip
To illustrate this, a workflow is created to manage flight and hotel reservations for a senior executive within a fictitious company. The workflow describes the travel dates, destination, and budget limits, then LangGraph connects to the Booking.com travel API, identifies the best combination as needed, and completes the reservations.
Important note: The goal is not to learn how to book flights and hotels. The journey scenario serves as an excuse to understand the structural mechanisms that drive LangGraph, so you can understand how to apply it to your own projects.
Note on APIs: The code examples in simulation mode faithfully reproduce the structure of the workflow. A separate section shows the actual code one would use with real access to the Booking.com API.
2. Building a multi agent workflow
1.1 LangGraph Code Review – Part 1
Importing libraries and configuring
We start in a Jupyter Notebook by importing all the necessary libraries. If a library is not yet installed, you must do so first with pip install.
To avoid having to hardcode the access keys in the source code, we load the security keys from a local file named .env using the dotenv library.
A .env file looks like this – it’s simply a plain text file that defines keys with their values:
OPENAI_API_KEY=sk-XXXXXXXXXXXXXXXXXXXXXXXXXXXX
BOOKING_COM_API_KEY=your_booking_api_key_here
BOOKING_COM_SECRET=your_booking_secret_here
RAPIDAPI_KEY=your_rapidapi_key_here
These examples are of course all fictitious. Values must be defined for as many individual keys as necessary.
import os
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional, TypedDict
from dataclasses import dataclass
import asyncio
from dotenv import load_dotenv
# Imports LangGraph et LangChain
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.schema import BaseMessage, HumanMessage, AIMessage
# Chargement des variables d'environnement
load_dotenv()
Next, we import the LangGraph-specific libraries. As we use the OpenAI API, we import LangChain’s own OpenAI library. We then load the security key values from the real .env file, which will populate the OpenAI key variable. We use it to initialize an OpenAI API client which will use the gpt-4-turbo model. There’s nothing stopping you from selecting a different model or one from Anthropic or Google.
# Initialisation du client OpenAI
llm = ChatOpenAI(
model="gpt-4-turbo-preview",
api_key=os.getenv("OPENAI_API_KEY"),
temperature=0.1
)
Data structure and state definition
The data structure section is where we initialize the variables that will be used throughout the process. This includes values representing basic travel information, such as locations and dates. We hardcode certain preferences, including the maximum budgets for the flight and the hotel stay, the seat class sought, and the minimum hotel rating. The FlightOption and HotelOption variables will optionally contain data returned by our API calls. This is the type of data that will be handled by LangGraph, regardless of the number of steps the operation involves.
# Structures de données
@dataclass
class TravelRequest:
departure_city: str
destination_city: str
departure_date: str
return_date: str
passengers: int = 1
@dataclass
class ComplianceRules:
max_budget_flight: float = 1000.0
max_budget_hotel: float = 200.0 # par nuit
preferred_class: str = "economy" # economy, business, first
min_hotel_rating: float = 3.5
@dataclass
class FlightOption:
flight_id: str
airline: str
departure_time: str
arrival_time: str
price: float
class_type: str
duration_minutes: int
@dataclass
class HotelOption:
hotel_id: str
name: str
rating: float
price_per_night: float
amenities: List[str]
location: str
The following state definition section is central to how LangGraph will manage data within our operation. All nodes running here will use these definitions, both for reading and creating data.
We build a schema to define the structure and data types which will use the TypedDict type. But pydantic and dataclass are also possible. dataclass is used for default values, as is done with the preferences above. The contents of this state definition are namespaces which will each contain dynamic data. The List keyword tells Python that the associated brackets will contain a list of elements of type FlightOption or HotelOption.
# Définition du state pour LangGraph
class TravelState(TypedDict):
request: Optional[TravelRequest]
compliance_rules: Optional[ComplianceRules]
raw_flights: List[FlightOption]
raw_hotels: List[HotelOption]
filtered_flights: List[FlightOption]
filtered_hotels: List[HotelOption]
top_options: Dict
user_selection: Optional[Dict]
booking_confirmation: Optional[Dict]
messages: List[BaseMessage]
error: Optional[str]
1.2 LangGraph Code Review – Part 2
The TravelBookingWorkflow class
If, as we just saw, the state defines how the data will be stored, the TravelBookingWorkflow class is the orchestrator that makes things actually happen. This first function initializes a set of compliance rules based on our previous definitions, then defines the order in which we will execute the steps in the form of a workflow graph.
There are actually 10 other functions in this class, each representing a LangGraph node.
class TravelBookingWorkflow:
def __init__(self):
self.compliance_rules = ComplianceRules()
self.workflow = self._create_workflow()
def _create_workflow(self) -> StateGraph:
"""Crée le workflow LangGraph"""
workflow = StateGraph(TravelState)
# Ajout des nœuds
workflow.add_node("parse_request", self.parse_travel_request)
workflow.add_node("search_flights", self.search_flights)
workflow.add_node("search_hotels", self.search_hotels)
workflow.add_node("apply_compliance", self.apply_compliance_filter)
workflow.add_node("rank_options", self.rank_options)
workflow.add_node("human_approval", self.get_human_approval)
workflow.add_node("book_travel", self.book_selected_options)
workflow.add_node("send_confirmation", self.send_confirmation)
# Définition du flux
workflow.set_entry_point("parse_request")
workflow.add_edge("parse_request", "search_flights")
workflow.add_edge("search_flights", "search_hotels")
workflow.add_edge("search_hotels", "apply_compliance")
workflow.add_edge("apply_compliance", "rank_options")
workflow.add_edge("rank_options", "human_approval")
workflow.add_edge("human_approval", "book_travel")
workflow.add_edge("book_travel", "send_confirmation")
workflow.add_edge("send_confirmation", END)
return workflow.compile()
The _create_workflow function declares each of the remaining nodes and behind the scenes creates the structure to ensure that data flows appropriately between the nodes. In other words, this is where the executable workflow is created.
Role of each node:
parse_request→ clarifies requested travel parameterssearch_flights→ searches for available flights matching the parameterssearch_hotels→ searches for available hotels matching the parametersapply_compliance→ ensures that compliance rules (budget limits) are respectedrank_options→ prepares a ranking of available options to facilitate selectionhuman_approval→ interrupts workflow to request human approvalbook_travel→ actually makes reservationssend_confirmation→ sends confirmation that everything is ready
The TravelState dictionary is what carries all intermediate data between stages. The Defining the flow section adds edges between nodes, so LangGraph knows exactly what to do when each node finishes its task. This all culminates after send_confirmation completes and the process terminates.
The individual nodes of the workflow
Node parse_travel_request
The parse_travel_request function would normally consist of a prompt where the user could enter basic travel preferences, but here it is simplified by hardcoding it. We apply the compliance rules for the budget limits that we defined previously. The AIMessage code records a system-generated message summarizing what has been analyzed since the travel request.
def parse_travel_request(self, state: TravelState) -> TravelState:
"""Analyse et valide la demande de voyage"""
try:
request = TravelRequest(
departure_city="New York",
destination_city="London",
departure_date="2025-09-15",
return_date="2025-09-22",
passengers=1
)
state["request"] = request
state["compliance_rules"] = self.compliance_rules
state["messages"].append(
AIMessage(content=f"Parsed travel request: {request.departure_city} to {request.destination_city}")
)
except Exception as e:
state["error"] = f"Failed to parse request: {str(e)}"
return state
Node search_flights
The search_flights function would normally consist of an API call to Booking.com which would return data on all available flights. For this demo, we use simulated flight data. The parameters in this example include flight duration. Even if we didn’t explicitly integrate it into the code, we could have used our AI model to take duration into account when ranking options. To simplify, we will use Python to do this a little later. AIMessage appears again to save the messages generated by this step. This will happen at the end of each node until the operation completes.
def search_flights(self, state: TravelState) -> TravelState:
"""Recherche des vols (simulé)"""
try:
mock_flights = [
FlightOption(
flight_id="BA123",
airline="British Airways",
departure_time="08:00",
arrival_time="19:30",
price=750.00,
class_type="economy",
duration_minutes=450
),
FlightOption(
flight_id="VS456",
airline="Virgin Atlantic",
departure_time="14:15",
arrival_time="01:45+1",
price=680.00,
class_type="economy",
duration_minutes=435
),
FlightOption(
flight_id="AA789",
airline="American Airlines",
departure_time="22:30",
arrival_time="09:15+1",
price=1200.00,
class_type="business",
duration_minutes=425
),
FlightOption(
flight_id="DL101",
airline="Delta",
departure_time="11:00",
arrival_time="22:30",
price=820.00,
class_type="economy",
duration_minutes=450
)
]
state["raw_flights"] = mock_flights
state["messages"].append(
AIMessage(content=f"Found {len(mock_flights)} flight options")
)
except Exception as e:
state["error"] = f"Flight search failed: {str(e)}"
return state
Node search_hotels
We go through virtually the same process to simulate hotel options, including ranking details like hotel ratings, amenities, and location.
def search_hotels(self, state: TravelState) -> TravelState:
"""Recherche des hôtels (simulé)"""
try:
mock_hotels = [
HotelOption(
hotel_id="HTL001",
name="London Grand Hotel",
rating=4.5,
price_per_night=180.00,
amenities=["WiFi", "Gym", "Restaurant", "Room Service"],
location="Central London"
),
HotelOption(
hotel_id="HTL002",
name="Budget Stay Inn",
rating=3.2,
price_per_night=85.00,
amenities=["WiFi", "24hr Reception"],
location="East London"
),
HotelOption(
hotel_id="HTL003",
name="Luxury Palace",
rating=4.8,
price_per_night=350.00,
amenities=["WiFi", "Spa", "Concierge", "Pool", "Restaurant"],
location="Mayfair"
),
HotelOption(
hotel_id="HTL004",
name="Comfort Suites",
rating=4.0,
price_per_night=150.00,
amenities=["WiFi", "Gym", "Business Center"],
location="City Center"
)
]
state["raw_hotels"] = mock_hotels
state["messages"].append(
AIMessage(content=f"Found {len(mock_hotels)} hotel options")
)
except Exception as e:
state["error"] = f"Hotel search failed: {str(e)}"
return state
Node apply_compliance_filter
Now that we have populated these FlightOption and HotelOption lists, we apply the compliance_rules state to filter the options that are not within our budget limits. We do this separately for flights and hotels.
We could have determined compliance by asking our AI model, but the results would have required some sort of human review to ensure they weren’t hallucinations. Since this is so easy to do using traditional software checks, we prefer to stick with what we know works.
def apply_compliance_filter(self, state: TravelState) -> TravelState:
"""Applique les règles de conformité pour filtrer les options"""
try:
rules = state["compliance_rules"]
# Filtrage des vols
filtered_flights = [
flight for flight in state["raw_flights"]
if (flight.price <= rules.max_budget_flight and
(rules.preferred_class == "any" or flight.class_type == rules.preferred_class))
]
# Filtrage des hôtels
filtered_hotels = [
hotel for hotel in state["raw_hotels"]
if (hotel.price_per_night <= rules.max_budget_hotel and
hotel.rating >= rules.min_hotel_rating)
]
state["filtered_flights"] = filtered_flights
state["filtered_hotels"] = filtered_hotels
state["messages"].append(
AIMessage(content=f"After compliance filtering: {len(filtered_flights)} flights, {len(filtered_hotels)} hotels")
)
except Exception as e:
state["error"] = f"Compliance filtering failed: {str(e)}"
return state
1.3 Running a LangGraph workflow
Ranking and selection of options
In the same way that we just used Python functionalities to apply compliance rules, we also use code to classify flight and hotel options. These are standard tools for comparing and sorting specified values, and you are free to make this as detailed as necessary.
def rank_options(self, state: TravelState) -> TravelState:
"""Classe et sélectionne les 3 meilleures options par catégorie"""
try:
# Classement des vols par prix et durée
flights = sorted(state["filtered_flights"],
key=lambda x: (x.price, x.duration_minutes))[:3]
# Classement des hôtels par note et prix
hotels = sorted(state["filtered_hotels"],
key=lambda x: (-x.rating, x.price_per_night))[:3]
state["top_options"] = {
"flights": flights,
"hotels": hotels
}
state["messages"].append(
AIMessage(content="Ranked top 3 options for flights and hotels")
)
except Exception as e:
state["error"] = f"Ranking failed: {str(e)}"
return state
The human-in-the-loop phase
The next node will pause the workflow to give us the chance to choose from the best flights and hotels available. top_options is the variable containing the results filtered in the previous section.
- The first break presents a numbered list of flights, including flight time, price and duration.
- The second break does the same thing for hotel options.
This code requests user input for each selection. This is all ordinary Python code, but the contents of the flight_choice and hotel_choice objects generated by this input will be read into the user_selection variable and processed by LangGraph as part of the graph state.
def get_human_approval(self, state: TravelState) -> TravelState:
"""Présente les options à l'humain pour approbation"""
try:
top_options = state["top_options"]
print("\n" + "="*60)
print("TOP FLIGHT OPTIONS:")
print("="*60)
for i, flight in enumerate(top_options["flights"], 1):
print(f"{i}. {flight.airline} ({flight.flight_id})")
print(f" Time: {flight.departure_time} - {flight.arrival_time}")
print(f" Price: ${flight.price:.2f} | Class: {flight.class_type}")
print(f" Duration: {flight.duration_minutes // 60}h {flight.duration_minutes % 60}m")
print()
print("="*60)
print("TOP HOTEL OPTIONS:")
print("="*60)
for i, hotel in enumerate(top_options["hotels"], 1):
print(f"{i}. {hotel.name}")
print(f" Rating: {hotel.rating}⭐ | Location: {hotel.location}")
print(f" Price: ${hotel.price_per_night:.2f}/night")
print(f" Amenities: {', '.join(hotel.amenities[:3])}...")
print()
# Récupération de la sélection utilisateur
flight_choice = int(input("Select flight option (1-3): ")) - 1
hotel_choice = int(input("Select hotel option (1-3): ")) - 1
state["user_selection"] = {
"flight": top_options["flights"][flight_choice],
"hotel": top_options["hotels"][hotel_choice]
}
state["messages"].append(
HumanMessage(content=f"Selected flight {flight_choice+1} and hotel {hotel_choice+1}")
)
except Exception as e:
# Pour la démo, sélection automatique des premières options en cas d'erreur
state["user_selection"] = {
"flight": state["top_options"]["flights"][0],
"hotel": state["top_options"]["hotels"][0]
}
state["messages"].append(
AIMessage(content="Auto-selected first options due to input error")
)
return state
Reservation and confirmation
Now that we have gone through the human-in-the-loop phase, LangGraph will use our selections to use the Booking.com API and actually complete the reservations. In the demo, the code simply simulates the return of a transaction in progress.
def book_selected_options(self, state: TravelState) -> TravelState:
"""Réserve les options de voyage sélectionnées (simulé)"""
try:
selection = state["user_selection"]
request = state["request"]
print("\n" + "="*50)
print("BOOKING IN PROGRESS...")
print("="*50)
import time
time.sleep(2) # Simule le temps de traitement
# Génération de la confirmation de réservation
booking_ref_flight = f"FLT{datetime.now().strftime('%Y%m%d%H%M')}"
booking_ref_hotel = f"HTL{datetime.now().strftime('%Y%m%d%H%M')}"
confirmation = {
"flight_booking": {
"reference": booking_ref_flight,
"flight": selection["flight"],
"status": "CONFIRMED",
"total_price": selection["flight"].price * request.passengers
},
"hotel_booking": {
"reference": booking_ref_hotel,
"hotel": selection["hotel"],
"status": "CONFIRMED",
"check_in": request.departure_date,
"check_out": request.return_date,
"nights": 7,
"total_price": selection["hotel"].price_per_night * 7
}
}
state["booking_confirmation"] = confirmation
state["messages"].append(
AIMessage(content="Booking completed successfully")
)
except Exception as e:
state["error"] = f"Booking failed: {str(e)}"
return state
The endpoint will retrieve the contents of the booking_confirmation variable that was populated in the previous step and passed to the state, then use it to display this confirmation information for the flight and hotel stay. This will include final pricing information. We also simulate the sending of confirmation notices by email and SMS.
def send_confirmation(self, state: TravelState) -> TravelState:
"""Envoie la confirmation de réservation"""
try:
confirmation = state["booking_confirmation"]
print("\n" + "="*60)
print("BOOKING CONFIRMATION")
print("="*60)
# Confirmation du vol
flight_booking = confirmation["flight_booking"]
print(f"FLIGHT BOOKING CONFIRMED")
print(f"Reference: {flight_booking['reference']}")
print(f"Flight: {flight_booking['flight'].airline} {flight_booking['flight'].flight_id}")
print(f"Route: {state['request'].departure_city} -> {state['request'].destination_city}")
print(f"Departure: {flight_booking['flight'].departure_time}")
print(f"Total Price: ${flight_booking['total_price']:.2f}")
print()
# Confirmation de l'hôtel
hotel_booking = confirmation["hotel_booking"]
print(f"HOTEL BOOKING CONFIRMED")
print(f"Reference: {hotel_booking['reference']}")
print(f"Hotel: {hotel_booking['hotel'].name}")
print(f"Location: {hotel_booking['hotel'].location}")
print(f"Check-in: {hotel_booking['check_in']}")
print(f"Check-out: {hotel_booking['check_out']}")
print(f"Nights: {hotel_booking['nights']}")
print(f"Total Price: ${hotel_booking['total_price']:.2f}")
print()
total_cost = flight_booking['total_price'] + hotel_booking['total_price']
print(f"TOTAL TRIP COST: ${total_cost:.2f}")
print("="*60)
# Simulation d'envoi email/SMS
print("\nConfirmation email sent to user")
print("SMS notification sent")
state["messages"].append(
AIMessage(content=f"Confirmation sent. Total trip cost: ${total_cost:.2f}")
)
except Exception as e:
state["error"] = f"Confirmation sending failed: {str(e)}"
return state
Complete workflow execution
So far, the code has been executed cell by cell in the Jupyter Notebook, but nothing has actually been executed as a workflow. The run_workflow node will control how the execution will take place. It will take the current contents of the TravelState and use it to populate the initial state variable, then use that to actually run the workflow. If something goes wrong, it will display an error message. Otherwise it will set the booking_confirmation state as the final state.
def run_workflow(self, travel_request: Optional[str] = None):
"""Exécute le workflow complet de réservation de voyage"""
initial_state = TravelState(
request=None,
compliance_rules=None,
raw_flights=[],
raw_hotels=[],
filtered_flights=[],
filtered_hotels=[],
top_options={},
user_selection=None,
booking_confirmation=None,
messages=[HumanMessage(content=travel_request or "Book travel from NYC to London")],
error=None
)
try:
# Exécution du workflow
final_state = self.workflow.invoke(initial_state)
if final_state.get("error"):
print(f"Workflow failed: {final_state['error']}")
return None
return final_state["booking_confirmation"]
except Exception as e:
print(f"Workflow execution failed: {str(e)}")
return None
Main entry point:
We take the TravelBookingWorkflow class in which most of the code was written, and use it to initialize the booking_system parameter, then this section executes the workflow.
def main():
"""Lance le workflow de réservation de voyage"""
print("Starting Travel Booking Workflow")
print("="*50)
# Initialisation du workflow
booking_system = TravelBookingWorkflow()
# Exécution du workflow
result = booking_system.run_workflow(
"I need to book a trip from New York to London from September 15-22, 2025"
)
if result:
print("\nTravel booking workflow completed successfully!")
else:
print("\nTravel booking workflow failed!")
if __name__ == "__main__":
main()
Execution result:
When the cell is executed, the first step of the operation is completed successfully. We see the three best flight options, including their price, seat class and duration. We also see the hotel options. Each option is numbered. By entering, for example, the number 2 for flight and the number 1 for hotel, texts confirm that the simulated transaction is in progress. After a few seconds, a detailed confirmation indicates that the transaction was successful and the total cost has slightly exceeded the $2,000 mark.
1.4 Connecting LangGraph to an external API
Although we cannot demonstrate it in action, we can show how the LangGraph workflow could have worked with a real API request. Although the query itself is really just a line of code, it can take quite a bit of code adaptation to make it work.
For this code, the first ~200 lines of the simulation code that we have just executed are identical (import of libraries, authentication to the OpenAI API, initialization of variables, definition of the data state, definition of the graph which governs the flow of data between nodes, and the parameters of the request where we define our travel needs). Searching for flights is also the same.
Integration with the official Booking.com API
We start with the search_hotels node. This code uses a try structure to present some alternative executions in case the first one fails.
def search_hotels(self, state: TravelState) -> TravelState:
"""Recherche des hôtels via l'API Booking.com"""
try:
request = state["request"]
# Méthode 1 : API officielle Booking.com for Business
hotels = self._search_booking_com_official(request)
# Méthode 2 : Alternative – endpoint RapidAPI Booking.com (fallback)
if not hotels:
hotels = self._search_booking_com_rapidapi(request)
# Méthode 3 : Fallback vers données simulées si les API échouent
if not hotels:
hotels = self._get_mock_hotels()
print("Using mock data - API requests failed")
state["raw_hotels"] = hotels
state["messages"].append(
AIMessage(content=f"Found {len(hotels)} hotel options from Booking.com")
)
except Exception as e:
state["error"] = f"Hotel search failed: {str(e)}"
state["raw_hotels"] = self._get_mock_hotels()
return state
The payload of the _search_booking_com_official variable will be consumed in the next node. We start by populating base_url with the real Booking.com endpoint. We then import the API key and the Booking.com secret which should be found in an .env file on the local system.
Unless you want to add this manually, you need to convert the destination (already defined in the destination_city variable) to an ID that Booking.com can understand. We then give values to each of the parameters that the API expects.
def _search_booking_com_official(self, request: TravelRequest) -> List[HotelOption]:
"""Recherche via l'API officielle Booking.com for Business"""
try:
# Endpoint de l'API Booking.com for Business
base_url = "https://distribution-xml.booking.com/json/bookings.getHotels"
# Credentials API depuis les variables d'environnement
api_key = os.getenv("BOOKING_COM_API_KEY")
secret = os.getenv("BOOKING_COM_SECRET")
if not api_key or not secret:
print("Booking.com API credentials not found in .env")
return []
# Conversion du nom de ville en ID de destination
dest_id = self._get_destination_id(request.destination_city)
# Paramètres API
params = {
"hotel_ids": "",
"destination_id": dest_id,
"checkin": request.departure_date,
"checkout": request.return_date,
"adults": request.passengers,
"children": 0,
"language": "en-us",
"currency": "USD",
"extras": "hotel_details,room_details,room_photos"
}
headers = {
"User-Agent": "TravelBookingBot/1.0",
"Content-Type": "application/json"
}
# Ajout de l'authentification
auth_params = {
"username": api_key,
"password": secret
}
params.update(auth_params)
print(f"Searching Booking.com for hotels in {request.destination_city}...")
response = requests.get(base_url, params=params, headers=headers, timeout=30)
if response.status_code == 200:
data = response.json()
return self._parse_booking_com_response(data)
else:
print(f"Booking.com API error: {response.status_code}")
return []
except Exception as e:
print(f"Official Booking.com API error: {str(e)}")
return []
Important note: You must be careful with the naming of variables. If we use the same variable name
api_keyfor OpenAI authentication at the beginning of the notebook and for Booking.com here, this may cause conflicts.
Mapping of Booking.com destination IDs:
def _get_destination_id(self, city_name: str) -> str:
"""Récupère l'ID de destination Booking.com pour une ville"""
city_mapping = {
"london": "-2601889",
"paris": "-1456928",
"new york": "-2090174",
"tokyo": "-246227",
"barcelona": "-372490",
"amsterdam": "-2140479",
"rome": "-126693",
"berlin": "-1746443"
}
return city_mapping.get(city_name.lower(), "-2601889") # Par défaut: Londres
Alternative via RapidAPI
The first fallback alternative is to use the third-party service RapidAPI to complete the Booking.com transaction for us. We assume here that we have a valid account with RapidAPI and an authentication key.
def _search_booking_com_rapidapi(self, request: TravelRequest) -> List[HotelOption]:
"""Recherche via l'endpoint RapidAPI Booking.com (alternative)"""
try:
url = "https://booking-com.p.rapidapi.com/v1/hotels/search"
rapidapi_key = os.getenv("RAPIDAPI_KEY")
if not rapidapi_key:
print("RapidAPI key not found in .env")
return []
# Récupération de l'ID de destination via RapidAPI
dest_id = self._get_destination_id_rapidapi(request.destination_city)
headers = {
"X-RapidAPI-Key": rapidapi_key,
"X-RapidAPI-Host": "booking-com.p.rapidapi.com",
"Content-Type": "application/json"
}
params = {
"dest_id": dest_id,
"order_by": "popularity",
"filter_by_currency": "USD",
"adults_number": request.passengers,
"checkin_date": request.departure_date,
"checkout_date": request.return_date,
"locale": "en-gb",
"units": "metric",
"room_number": 1,
"children_number": 0
}
print(f"Searching via RapidAPI for hotels in {request.destination_city}...")
response = requests.get(url, headers=headers, params=params, timeout=30)
if response.status_code == 200:
data = response.json()
return self._parse_rapidapi_response(data)
else:
print(f"RapidAPI Booking.com error: {response.status_code}")
return []
except Exception as e:
print(f"RapidAPI Booking.com error: {str(e)}")
return []
Note: RapidAPI expects parameters in a slightly different format than Booking.com. It’s obviously important to get it exactly right.
To obtain the destination ID via RapidAPI, we use the location search endpoint:
def _get_destination_id_rapidapi(self, city_name: str) -> str:
"""Récupère l'ID de destination pour l'endpoint RapidAPI"""
try:
url = "https://booking-com.p.rapidapi.com/v1/hotels/locations"
headers = {
"X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY"),
"X-RapidAPI-Host": "booking-com.p.rapidapi.com"
}
params = {
"locale": "en-gb",
"name": city_name
}
response = requests.get(url, headers=headers, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
if data and len(data) > 0:
return str(data[0]["dest_id"])
# Fallback vers le mapping codé en dur
return self._get_destination_id(city_name)
except Exception as e:
print(f"Destination ID lookup failed: {str(e)}")
return self._get_destination_id(city_name)
Parsing API responses
This node will parse hotel availability data obtained from Booking.com and structure it into a usable data structure. It’s worth noting how data should be treated differently depending on whether it comes from RapidAPI rather than Booking.com – although ultimately everything should look the same internally.
Parsing the official Booking.com response:
def _parse_booking_com_response(self, data: dict) -> List[HotelOption]:
"""Parse la réponse de l'API officielle Booking.com"""
hotels = []
try:
hotel_list = data.get("result", [])
for hotel_data in hotel_list[:10]: # Limite à 10 résultats
hotel = HotelOption(
hotel_id=str(hotel_data.get("hotel_id", "")),
name=hotel_data.get("hotel_name", "Unknown Hotel"),
rating=float(hotel_data.get("class", 0)),
price_per_night=float(hotel_data.get("min_total_price", 0)),
amenities=hotel_data.get("facilities", [])[:5],
location=hotel_data.get("address", "Unknown Location")
)
hotels.append(hotel)
except Exception as e:
print(f"Error parsing Booking.com response: {str(e)}")
return hotels
Parsing the RapidAPI response:
The management of price data is done separately from the rest for RapidAPI, and the rating must be converted from the 10 scale to the 5 star scale:
def _parse_rapidapi_response(self, data: dict) -> List[HotelOption]:
"""Parse la réponse RapidAPI Booking.com"""
hotels = []
try:
hotel_list = data.get("result", [])
for hotel_data in hotel_list[:10]:
# Analyse du prix (peut être dans différents formats)
price_info = hotel_data.get("price_breakdown", {})
price_per_night = 0
if "gross_price" in price_info:
price_per_night = float(price_info["gross_price"])
elif "all_inclusive_price" in price_info:
price_per_night = float(price_info["all_inclusive_price"])
# Analyse des commodités
amenities = []
facilities = hotel_data.get("hotel_facilities", [])
for facility in facilities[:5]:
if isinstance(facility, dict):
amenities.append(facility.get("name", ""))
else:
amenities.append(str(facility))
hotel = HotelOption(
hotel_id=str(hotel_data.get("hotel_id", "")),
name=hotel_data.get("hotel_name", "Unknown Hotel"),
rating=float(hotel_data.get("review_score", 0) / 2), # Conversion vers échelle 5 étoiles
price_per_night=price_per_night,
amenities=amenities,
location=hotel_data.get("address", "Unknown Location")
)
hotels.append(hotel)
except Exception as e:
print(f"Error parsing RapidAPI response: {str(e)}")
return hotels
Whichever approach was ultimately successful, the code would then execute the graph. What this showed us was how LangGraph can be used to create a robust and reliable workflow capable of managing the movement of data to and from third-party APIs, and orchestrating the execution of a complex set of programmatic objectives.
Utility functions for use in Jupyter Notebook:
def create_custom_workflow(max_flight_budget=1000, max_hotel_budget=200, preferred_class="economy"):
"""Crée un workflow avec des règles de conformité personnalisées"""
workflow = TravelBookingWorkflow()
workflow.compliance_rules = ComplianceRules(
max_budget_flight=max_flight_budget,
max_budget_hotel=max_hotel_budget,
preferred_class=preferred_class
)
return workflow
def quick_search(departure_city, destination_city, departure_date, return_date, passengers=1):
"""Fonction de recherche rapide de voyage"""
workflow = TravelBookingWorkflow()
return workflow.run_workflow(f"Book travel from {departure_city} to {destination_city}")
# Exemple avec des paramètres personnalisés :
# workflow = create_custom_workflow(max_flight_budget=1500, preferred_class="business")
# result = workflow.run_workflow("NYC to Tokyo business class trip")
3. Full source code – Simulation version
File 01/demos/langgraph_workflow_simulate.py contains the full version with simulated data:
# Travel Booking Workflow with LangGraph
# Run this in a Jupyter Notebook with OpenAI API access via .env file
import os
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional, TypedDict
from dataclasses import dataclass
import asyncio
from dotenv import load_dotenv
# LangGraph and LangChain imports
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.schema import BaseMessage, HumanMessage, AIMessage
# Load environment variables
load_dotenv()
# Initialize OpenAI client
llm = ChatOpenAI(
model="gpt-4-turbo-preview",
api_key=os.getenv("OPENAI_API_KEY"),
temperature=0.1
)
# Data structures
@dataclass
class TravelRequest:
departure_city: str
destination_city: str
departure_date: str
return_date: str
passengers: int = 1
@dataclass
class ComplianceRules:
max_budget_flight: float = 1000.0
max_budget_hotel: float = 200.0 # per night
preferred_class: str = "economy"
min_hotel_rating: float = 3.5
@dataclass
class FlightOption:
flight_id: str
airline: str
departure_time: str
arrival_time: str
price: float
class_type: str
duration_minutes: int
@dataclass
class HotelOption:
hotel_id: str
name: str
rating: float
price_per_night: float
amenities: List[str]
location: str
# State definition for LangGraph
class TravelState(TypedDict):
request: Optional[TravelRequest]
compliance_rules: Optional[ComplianceRules]
raw_flights: List[FlightOption]
raw_hotels: List[HotelOption]
filtered_flights: List[FlightOption]
filtered_hotels: List[HotelOption]
top_options: Dict
user_selection: Optional[Dict]
booking_confirmation: Optional[Dict]
messages: List[BaseMessage]
error: Optional[str]
class TravelBookingWorkflow:
def __init__(self):
self.compliance_rules = ComplianceRules()
self.workflow = self._create_workflow()
def _create_workflow(self) -> StateGraph:
workflow = StateGraph(TravelState)
workflow.add_node("parse_request", self.parse_travel_request)
workflow.add_node("search_flights", self.search_flights)
workflow.add_node("search_hotels", self.search_hotels)
workflow.add_node("apply_compliance", self.apply_compliance_filter)
workflow.add_node("rank_options", self.rank_options)
workflow.add_node("human_approval", self.get_human_approval)
workflow.add_node("book_travel", self.book_selected_options)
workflow.add_node("send_confirmation", self.send_confirmation)
workflow.set_entry_point("parse_request")
workflow.add_edge("parse_request", "search_flights")
workflow.add_edge("search_flights", "search_hotels")
workflow.add_edge("search_hotels", "apply_compliance")
workflow.add_edge("apply_compliance", "rank_options")
workflow.add_edge("rank_options", "human_approval")
workflow.add_edge("human_approval", "book_travel")
workflow.add_edge("book_travel", "send_confirmation")
workflow.add_edge("send_confirmation", END)
return workflow.compile()
def parse_travel_request(self, state: TravelState) -> TravelState:
try:
request = TravelRequest(
departure_city="New York",
destination_city="London",
departure_date="2025-09-15",
return_date="2025-09-22",
passengers=1
)
state["request"] = request
state["compliance_rules"] = self.compliance_rules
state["messages"].append(
AIMessage(content=f"Parsed travel request: {request.departure_city} to {request.destination_city}")
)
except Exception as e:
state["error"] = f"Failed to parse request: {str(e)}"
return state
def search_flights(self, state: TravelState) -> TravelState:
try:
mock_flights = [
FlightOption(flight_id="BA123", airline="British Airways",
departure_time="08:00", arrival_time="19:30",
price=750.00, class_type="economy", duration_minutes=450),
FlightOption(flight_id="VS456", airline="Virgin Atlantic",
departure_time="14:15", arrival_time="01:45+1",
price=680.00, class_type="economy", duration_minutes=435),
FlightOption(flight_id="AA789", airline="American Airlines",
departure_time="22:30", arrival_time="09:15+1",
price=1200.00, class_type="business", duration_minutes=425),
FlightOption(flight_id="DL101", airline="Delta",
departure_time="11:00", arrival_time="22:30",
price=820.00, class_type="economy", duration_minutes=450)
]
state["raw_flights"] = mock_flights
state["messages"].append(
AIMessage(content=f"Found {len(mock_flights)} flight options")
)
except Exception as e:
state["error"] = f"Flight search failed: {str(e)}"
return state
def search_hotels(self, state: TravelState) -> TravelState:
try:
mock_hotels = [
HotelOption(hotel_id="HTL001", name="London Grand Hotel",
rating=4.5, price_per_night=180.00,
amenities=["WiFi", "Gym", "Restaurant", "Room Service"],
location="Central London"),
HotelOption(hotel_id="HTL002", name="Budget Stay Inn",
rating=3.2, price_per_night=85.00,
amenities=["WiFi", "24hr Reception"],
location="East London"),
HotelOption(hotel_id="HTL003", name="Luxury Palace",
rating=4.8, price_per_night=350.00,
amenities=["WiFi", "Spa", "Concierge", "Pool", "Restaurant"],
location="Mayfair"),
HotelOption(hotel_id="HTL004", name="Comfort Suites",
rating=4.0, price_per_night=150.00,
amenities=["WiFi", "Gym", "Business Center"],
location="City Center")
]
state["raw_hotels"] = mock_hotels
state["messages"].append(
AIMessage(content=f"Found {len(mock_hotels)} hotel options")
)
except Exception as e:
state["error"] = f"Hotel search failed: {str(e)}"
return state
def apply_compliance_filter(self, state: TravelState) -> TravelState:
try:
rules = state["compliance_rules"]
filtered_flights = [
flight for flight in state["raw_flights"]
if (flight.price <= rules.max_budget_flight and
(rules.preferred_class == "any" or flight.class_type == rules.preferred_class))
]
filtered_hotels = [
hotel for hotel in state["raw_hotels"]
if (hotel.price_per_night <= rules.max_budget_hotel and
hotel.rating >= rules.min_hotel_rating)
]
state["filtered_flights"] = filtered_flights
state["filtered_hotels"] = filtered_hotels
state["messages"].append(
AIMessage(content=f"After compliance filtering: {len(filtered_flights)} flights, {len(filtered_hotels)} hotels")
)
except Exception as e:
state["error"] = f"Compliance filtering failed: {str(e)}"
return state
def rank_options(self, state: TravelState) -> TravelState:
try:
flights = sorted(state["filtered_flights"],
key=lambda x: (x.price, x.duration_minutes))[:3]
hotels = sorted(state["filtered_hotels"],
key=lambda x: (-x.rating, x.price_per_night))[:3]
state["top_options"] = {"flights": flights, "hotels": hotels}
state["messages"].append(
AIMessage(content="Ranked top 3 options for flights and hotels")
)
except Exception as e:
state["error"] = f"Ranking failed: {str(e)}"
return state
def get_human_approval(self, state: TravelState) -> TravelState:
try:
top_options = state["top_options"]
print("\n" + "="*60)
print("TOP FLIGHT OPTIONS:")
print("="*60)
for i, flight in enumerate(top_options["flights"], 1):
print(f"{i}. {flight.airline} ({flight.flight_id})")
print(f" Time: {flight.departure_time} - {flight.arrival_time}")
print(f" Price: ${flight.price:.2f} | Class: {flight.class_type}")
print(f" Duration: {flight.duration_minutes // 60}h {flight.duration_minutes % 60}m")
print()
print("="*60)
print("TOP HOTEL OPTIONS:")
print("="*60)
for i, hotel in enumerate(top_options["hotels"], 1):
print(f"{i}. {hotel.name}")
print(f" Rating: {hotel.rating} | Location: {hotel.location}")
print(f" Price: ${hotel.price_per_night:.2f}/night")
print(f" Amenities: {', '.join(hotel.amenities[:3])}...")
print()
flight_choice = int(input("Select flight option (1-3): ")) - 1
hotel_choice = int(input("Select hotel option (1-3): ")) - 1
state["user_selection"] = {
"flight": top_options["flights"][flight_choice],
"hotel": top_options["hotels"][hotel_choice]
}
state["messages"].append(
HumanMessage(content=f"Selected flight {flight_choice+1} and hotel {hotel_choice+1}")
)
except Exception as e:
state["user_selection"] = {
"flight": state["top_options"]["flights"][0],
"hotel": state["top_options"]["hotels"][0]
}
state["messages"].append(
AIMessage(content="Auto-selected first options due to input error")
)
return state
def book_selected_options(self, state: TravelState) -> TravelState:
try:
selection = state["user_selection"]
request = state["request"]
print("\n" + "="*50)
print("BOOKING IN PROGRESS...")
print("="*50)
import time
time.sleep(2)
booking_ref_flight = f"FLT{datetime.now().strftime('%Y%m%d%H%M')}"
booking_ref_hotel = f"HTL{datetime.now().strftime('%Y%m%d%H%M')}"
confirmation = {
"flight_booking": {
"reference": booking_ref_flight,
"flight": selection["flight"],
"status": "CONFIRMED",
"total_price": selection["flight"].price * request.passengers
},
"hotel_booking": {
"reference": booking_ref_hotel,
"hotel": selection["hotel"],
"status": "CONFIRMED",
"check_in": request.departure_date,
"check_out": request.return_date,
"nights": 7,
"total_price": selection["hotel"].price_per_night * 7
}
}
state["booking_confirmation"] = confirmation
state["messages"].append(
AIMessage(content="Booking completed successfully")
)
except Exception as e:
state["error"] = f"Booking failed: {str(e)}"
return state
def send_confirmation(self, state: TravelState) -> TravelState:
try:
confirmation = state["booking_confirmation"]
flight_booking = confirmation["flight_booking"]
hotel_booking = confirmation["hotel_booking"]
print("\n" + "="*60)
print("BOOKING CONFIRMATION")
print("="*60)
print(f"FLIGHT BOOKING CONFIRMED")
print(f"Reference: {flight_booking['reference']}")
print(f"Flight: {flight_booking['flight'].airline} {flight_booking['flight'].flight_id}")
print(f"Route: {state['request'].departure_city} -> {state['request'].destination_city}")
print(f"Departure: {flight_booking['flight'].departure_time}")
print(f"Total Price: ${flight_booking['total_price']:.2f}")
print()
print(f"HOTEL BOOKING CONFIRMED")
print(f"Reference: {hotel_booking['reference']}")
print(f"Hotel: {hotel_booking['hotel'].name}")
print(f"Location: {hotel_booking['hotel'].location}")
print(f"Check-in: {hotel_booking['check_in']}")
print(f"Check-out: {hotel_booking['check_out']}")
print(f"Nights: {hotel_booking['nights']}")
print(f"Total Price: ${hotel_booking['total_price']:.2f}")
total_cost = flight_booking['total_price'] + hotel_booking['total_price']
print(f"\nTOTAL TRIP COST: ${total_cost:.2f}")
print("="*60)
print("\nConfirmation email sent to user")
print("SMS notification sent")
state["messages"].append(
AIMessage(content=f"Confirmation sent. Total trip cost: ${total_cost:.2f}")
)
except Exception as e:
state["error"] = f"Confirmation sending failed: {str(e)}"
return state
def run_workflow(self, travel_request: Optional[str] = None):
initial_state = TravelState(
request=None, compliance_rules=None,
raw_flights=[], raw_hotels=[],
filtered_flights=[], filtered_hotels=[],
top_options={}, user_selection=None,
booking_confirmation=None,
messages=[HumanMessage(content=travel_request or "Book travel from NYC to London")],
error=None
)
try:
final_state = self.workflow.invoke(initial_state)
if final_state.get("error"):
print(f"Workflow failed: {final_state['error']}")
return None
return final_state["booking_confirmation"]
except Exception as e:
print(f"Workflow execution failed: {str(e)}")
return None
def main():
print("Starting Travel Booking Workflow")
print("="*50)
booking_system = TravelBookingWorkflow()
result = booking_system.run_workflow(
"I need to book a trip from New York to London from June 15-22, 2024"
)
if result:
print("\nTravel booking workflow completed successfully!")
else:
print("\nTravel booking workflow failed!")
if __name__ == "__main__":
main()
def create_custom_workflow(max_flight_budget=1000, max_hotel_budget=200, preferred_class="economy"):
workflow = TravelBookingWorkflow()
workflow.compliance_rules = ComplianceRules(
max_budget_flight=max_flight_budget,
max_budget_hotel=max_hotel_budget,
preferred_class=preferred_class
)
return workflow
def quick_search(departure_city, destination_city, departure_date, return_date, passengers=1):
workflow = TravelBookingWorkflow()
return workflow.run_workflow(f"Book travel from {departure_city} to {destination_city}")
4. Full source code – API version
File 01/demos/langgraph_workflow_api.py contains the version with the real API calls. The key differences from the simulation version are in the search_hotels method, which integrates the three fallback methods (official Booking.com → RapidAPI → simulated data), as well as in the _parse_booking_com_response and _parse_rapidapi_response parsing methods.
The parse_travel_request, search_flights, apply_compliance_filter, rank_options, get_human_approval, book_selected_options, send_confirmation and run_workflow nodes are identical to those of the simulation version.
Additional methods specific to the API version are:
def _search_booking_com_official(self, request): # Appel API officiel Booking.com
def _search_booking_com_rapidapi(self, request): # Fallback via RapidAPI
def _get_destination_id(self, city_name): # Mapping ville → ID Booking.com
def _get_destination_id_rapidapi(self, city_name): # Lookup ID via RapidAPI
def _parse_booking_com_response(self, data): # Parsing réponse Booking.com
def _parse_rapidapi_response(self, data): # Parsing réponse RapidAPI
def _get_mock_hotels(self): # Données de fallback simulées
5. Summary and key points
LangGraph Architecture
| Concept | Description |
|---|---|
| Node | Individual workflow action unit (Python function) |
| Edge | Connection between two nodes defining the order of execution |
| StateGraph | LangGraph class that orchestrates the entire workflow |
| State | Dictionary (TypedDict) carrying all data between nodes |
| END | Special marker indicating the end of the workflow |
Key Concepts of Agentic AI
- Agentic AI: AI that performs multi-step tasks autonomously on behalf of the user
- Human-in-the-loop: Mechanism allowing a human to intervene and approve at certain stages of the workflow
- State management: Reliable management of the state of dynamic data throughout the process
- Compliance rules: Compliance rules verified by traditional software code rather than by AI (to avoid hallucinations)
Travel booking workflow flow
parse_request
↓
search_flights
↓
search_hotels
↓
apply_compliance ← Filtre budget + classe + note minimale
↓
rank_options ← Tri par prix/durée (vols) et par note/prix (hôtels)
↓
human_approval ← PAUSE : sélection interactive par l'utilisateur
↓
book_travel ← Appel API Booking.com (ou simulation)
↓
send_confirmation ← Affichage + notification email/SMS
↓
END
Fallback strategy for API calls
Tentative 1 : API officielle Booking.com for Business
↓ (échec)
Tentative 2 : RapidAPI endpoint Booking.com
↓ (échec)
Tentative 3 : Données simulées (mock data)
Best practices highlighted in this course
- Do not use AI for compliance checks: Budget rules and other objective criteria are better verified with traditional code, to avoid hallucinations.
- Use a
.envfile to store API keys and never hardcode them in source code. - Unique variable naming: Avoid using the same variable name (e.g.
api_key) for several different authentications in the same notebook. try/exceptstructure with fallback: Always provide alternatives in the event of an external API failure.- TypedDict for state: Use
TypedDict(orpydantic, ordataclass) to clearly define the LangGraph state schema. - AIMessage for traceability: Log messages at the end of each node to maintain a complete trace of the workflow execution.
Installing dependencies
pip install langgraph langchain langchain-openai python-dotenv requests
Search Terms
agentic · ai · langgraph · agents · orchestration · artificial · intelligence · generative · api · workflow · booking · source · version