[PYTHON/LANGGRAPH] 도구 실행시 조건부 확인 기능 추가 제로샷 에이전트 만들기

■ 도구 실행시 조건부 확인 기능을 추가한 제로샷 에이전트를 만드는 방법을 보여준다.

※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.
※ TAVILY_API_KEY 환경 변수 값은 .env 파일에 정의한다.

▶ main.py


import os
import requests
import shutil
import sqlite3
import pandas as pd
import re
import numpy  as np
import openai
import pytz
import uuid

from dotenv                                  import load_dotenv
from langchain_core.tools                    import tool
from langchain_core.runnables                import RunnableConfig
from typing                                  import Optional
from datetime                                import date
from datetime                                import datetime
from typing                                  import Union
from langchain_core.prompts                  import ChatPromptTemplate
from langchain_openai                        import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from typing_extensions                       import TypedDict
from typing                                  import Annotated
from langgraph.graph.message                 import AnyMessage
from langgraph.graph.message                 import add_messages
from langchain_core.runnables                import Runnable
from langchain_core.messages                 import ToolMessage
from langchain_core.messages                 import AIMessage
from langgraph.graph                         import StateGraph
from langgraph.prebuilt                      import ToolNode
from langchain_core.runnables                import RunnableLambda
from langgraph.prebuilt                      import tools_condition
from langgraph.graph                         import START
from langgraph.graph                         import END
from langgraph.checkpoint.memory             import MemorySaver

load_dotenv()

def downloadFile(sourceURL, targetFilePath, backupFilePath, overwrite):
    if overwrite or not os.path.exists(targetFilePath):
        response = requests.get(sourceURL)
        response.raise_for_status()
        with open(targetFilePath, "wb") as bufferedWriter:
            bufferedWriter.write(response.content)
        if backupFilePath:
            shutil.copy(targetFilePath, backupFilePath)

def loadTextFile(sourceFilePath):
    with open(sourceFilePath, "r", encoding = "utf-8") as textIOWrapper:
        return textIOWrapper.read()

def updateDate(databaseFilePath, backupFilePath):
    shutil.copy(backupFilePath, databaseFilePath)
    connection          = sqlite3.connect(databaseFilePath)
    tableNameList       = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", connection).name.tolist()
    dataFrameDictionary = {}
    for tableName in tableNameList:
        dataFrameDictionary[tableName] = pd.read_sql(f"SELECT * from {tableName}", connection)
    maximumActualDepartureTimestamp = pd.to_datetime(dataFrameDictionary["flights"]["actual_departure"].replace("\\N", pd.NaT)).max()
    localCurrentTimestamp           = pd.to_datetime("now").tz_localize(maximumActualDepartureTimestamp.tz)
    timeDelta                       = localCurrentTimestamp - maximumActualDepartureTimestamp
    dataFrameDictionary["bookings"]["book_date"] = pd.to_datetime(dataFrameDictionary["bookings"]["book_date"].replace("\\N", pd.NaT), utc = True) + timeDelta
    dateTimeColumnNameList = [
        "scheduled_departure",
        "scheduled_arrival",
        "actual_departure",
        "actual_arrival"
    ]
    for dateTimeColumName in dateTimeColumnNameList:
        dataFrameDictionary["flights"][dateTimeColumName] = pd.to_datetime(dataFrameDictionary["flights"][dateTimeColumName].replace("\\N", pd.NaT)) + timeDelta
    for tableName, dataFrame in dataFrameDictionary.items():
        dataFrame.to_sql(tableName, connection, if_exists = "replace", index = False)
    del dataFrame
    del dataFrameDictionary
    connection.commit()
    connection.close()

overwrite = False

databaseURL      = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/travel2.sqlite"
databaseFilePath = "travel2.sqlite"
backupFilePath   = "travel2.backup.sqlite"

downloadFile(databaseURL, databaseFilePath, backupFilePath, overwrite)
updateDate(databaseFilePath, backupFilePath)

faqFileURL  = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/swiss_faq.md"
faqFilePath = "swiss_faq.md"

downloadFile(faqFileURL, faqFilePath, None, overwrite)

faqText = loadTextFile(faqFilePath)

documentList = [{"page_content" : text} for text in re.split(r"(?=\n##)", faqText)]

class VectorStoreRetriever:
    def __init__(self, documentList : list, vectorListList : list, openAI):
        self.vectorListList = np.array(vectorListList)
        self.documentList   = documentList
        self.openAI         = openAI

    @classmethod
    def fromDocumentList(cls, documentList, openAI):
        createEmbeddingResponse = openAI.embeddings.create(model = "text-embedding-3-small", input = [document["page_content"] for document in documentList])
        vectorListList = [embedding.embedding for embedding in createEmbeddingResponse.data] # createEmbeddingResponse.data 속성 openai.types.embedding.Embedding 타입 리스트이다.
        return cls(documentList, vectorListList, openAI)

    def query(self, query : str, k : int = 5) -> list[dict]:
        createEmbeddingResponse = self.openAI.embeddings.create(model = "text-embedding-3-small", input = [query])
        # "@"는 파이썬에서 행렬 곱셈일 뿐이다.
        scoreNDArray    = np.array(createEmbeddingResponse.data[0].embedding) @ self.vectorListList.T
        topKIndex       = np.argpartition(scoreNDArray, -k)[-k:]
        topKIndexSorted = topKIndex[np.argsort(-scoreNDArray[topKIndex])]
        return [{**self.documentList[index], "similarity" : scoreNDArray[index]} for index in topKIndexSorted]

openAI = openai.Client()

vectorStoreRetriever = VectorStoreRetriever.fromDocumentList(documentList, openAI)

@tool
def lookupPolicy(query : str) -> str:
    """Consult the company policies to check whether certain options are permitted.
    Use this before making any flight changes performing other 'write' events."""
    global vectorStoreRetriever
    documentList = vectorStoreRetriever.query(query, k = 2)
    return "\n\n".join([document["page_content"] for document in documentList])

@tool
def fetchUserFlightInformationList(runnableConfig : RunnableConfig) -> list[dict]:
    """Fetch all tickets for the user along with corresponding flight information and seat assignments.

    Returns :
        A list of dictionaries where each dictionary contains the ticket details,
        associated flight details, and the seat assignments for each ticket belonging to the user.
    """
    global databaseFilePath
    configurableDictionary = runnableConfig.get("configurable", {})
    passengerID            = configurableDictionary.get("passenger_id", None)
    if not passengerID:
        raise ValueError("No passenger ID configured.")
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    query      = """
SELECT
    t.ticket_no,
    t.book_ref,
    f.flight_id,
    f.flight_no,
    f.departure_airport,
    f.arrival_airport,
    f.scheduled_departure,
    f.scheduled_arrival,
    bp.seat_no,
    tf.fare_conditions
FROM tickets t JOIN ticket_flights tf  ON  t.ticket_no  = tf.ticket_no
               JOIN flights f          ON  tf.flight_id = f.flight_id
               JOIN boarding_passes bp ON  bp.ticket_no = t.ticket_no
                                       AND bp.flight_id = f.flight_id
WHERE t.passenger_id = ?
"""
    cursor.execute(query, (passengerID,))
    rowTuplList       = cursor.fetchall()
    columnNameList    = [columnTuple[0] for columnTuple in cursor.description]
    rowDictionaryList = [dict(zip(columnNameList, rowTuple)) for rowTuple in rowTuplList]
    cursor.close()
    connection.close()
    return rowDictionaryList

@tool
def searchFlightList(departure_airport : Optional[str] = None, arrival_airport : Optional[str] = None, start_time : Optional[date | datetime] = None, end_time : Optional[date | datetime] = None, limit : int = 20) -> list[dict]:
    """Search for flights based on departure airport, arrival airport, and departure time range."""
    global databaseFilePath
    connection    = sqlite3.connect(databaseFilePath)
    cursor        = connection.cursor()
    query         = "SELECT * FROM flights WHERE 1 = 1"
    parameterList = []
    if departure_airport:
        query += " AND departure_airport = ?"
        parameterList.append(departure_airport)
    if arrival_airport:
        query += " AND arrival_airport = ?"
        parameterList.append(arrival_airport)
    if start_time:
        query += " AND scheduled_departure >= ?"
        parameterList.append(start_time)
    if end_time:
        query += " AND scheduled_departure <= ?"
        parameterList.append(end_time)
    query += " LIMIT ?"
    parameterList.append(limit)
    cursor.execute(query, parameterList)
    rowTupleList      = cursor.fetchall()
    columnNameList    = [columnTuple[0] for columnTuple in cursor.description]
    rowDictionaryList = [dict(zip(columnNameList, rowTuple)) for rowTuple in rowTupleList]
    cursor.close()
    connection.close()
    return rowDictionaryList

@tool
def updateTicketToNewFlight(ticket_no : str, new_flight_id : int, *, runnableConfig : RunnableConfig) -> str:
    """Update the user's ticket to a new valid flight."""
    global databaseFilePath
    configurableDictionary = runnableConfig.get("configurable", {})
    passengerID            = configurableDictionary.get("passenger_id", None)
    if not passengerID:
        raise ValueError("No passenger ID configured.")
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("SELECT departure_airport, arrival_airport, scheduled_departure FROM flights WHERE flight_id = ?", (new_flight_id,))
    newFlightRowTuple = cursor.fetchone()
    if not newFlightRowTuple:
        cursor.close()
        connection.close()
        return "Invalid new flight ID provided."
    columnNameList         = [columnTuple[0] for columnTuple in cursor.description]
    newFlightRowDictionary = dict(zip(columnNameList, newFlightRowTuple))
    timezone               = pytz.timezone("Etc/GMT-3")
    currentDatetime        = datetime.now(tz = timezone)
    departureDatetime      = datetime.strptime(newFlightRowDictionary["scheduled_departure"], "%Y-%m-%d %H:%M:%S.%f%z")
    totalSecondCount       = (departureDatetime - currentDatetime).total_seconds()
    if totalSecondCount < (3 * 3600):
        return f"Not permitted to reschedule to a flight that is less than 3 hours from the current time. Selected flight is at {departureDatetime}."
    cursor.execute("SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))
    currentFlightRowTuple = cursor.fetchone()
    if not currentFlightRowTuple:
        cursor.close()
        connection.close()
        return "No existing ticket found for the given ticket number."
    cursor.execute("SELECT * FROM tickets WHERE ticket_no = ? AND passenger_id = ?", (ticket_no, passengerID))
    currentTicketRowTuple = cursor.fetchone()
    if not currentTicketRowTuple:
        cursor.close()
        connection.close()
        return f"Current signed-in passenger with ID {passengerID} not the owner of ticket {ticket_no}"
    # 실제 애플리케이션에서는 "새로운 출발 공항이 현재 티켓과 일치하는가"와 같이 비즈니스 로직을 적용하기 위해 여기에 추가적인 검사를 추가할 가능성이 높다.
    # LLM에 대한 '타입 힌팅' 정책에서 선제적으로 노력하는 것이 가장 좋지만 필연적으로 문제가 발생할 수 있으므로 또한 API가 유효한 동작을 시행하도록 해야 한다.
    cursor.execute("UPDATE ticket_flights SET flight_id = ? WHERE ticket_no = ?", (new_flight_id, ticket_no))
    connection.commit()
    cursor.close()
    connection.close()
    return "Ticket successfully updated to new flight."

@tool
def cancelTicket(ticket_no : str, *, runnableConfig : RunnableConfig) -> str:
    """Cancel the user's ticket and remove it from the database."""
    global databaseFilePath
    configuration = runnableConfig.get("configurable", {})
    passengerID   = configuration.get("passenger_id", None)
    if not passengerID:
        raise ValueError("No passenger ID configured.")
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))
    existing_ticket = cursor.fetchone()
    if not existing_ticket:
        cursor.close()
        connection.close()
        return "No existing ticket found for the given ticket number."
    cursor.execute("SELECT flight_id FROM tickets WHERE ticket_no = ? AND passenger_id = ?", (ticket_no, passengerID))
    currentTicketRowTuple = cursor.fetchone()
    if not currentTicketRowTuple:
        cursor.close()
        connection.close()
        return f"Current signed-in passenger with ID {passengerID} not the owner of ticket {ticket_no}"
    cursor.execute("DELETE FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))
    connection.commit()
    cursor.close()
    connection.close()
    return "Ticket successfully cancelled."

@tool
def searchCarRentalList(location : Optional[str] = None, name : Optional[str] = None, price_tier : Optional[str] = None, start_date : Optional[Union[datetime, date]] = None, end_date : Optional[Union[datetime, date]] = None) -> list[dict]:
    """
    Search for car rentals based on location, name, price tier, start date, and end date.

    Args :
        location (Optional[str]) : The location of the car rental. Defaults to None.
        name (Optional[str]) : The name of the car rental company. Defaults to None.
        price_tier (Optional[str]) : The price tier of the car rental. Defaults to None.
        start_date (Optional[Union[datetime, date]]) : The start date of the car rental. Defaults to None.
        end_date (Optional[Union[datetime, date]]) : The end date of the car rental. Defaults to None.

    Returns :
        list[dict] : A list of car rental dictionaries matching the search criteria.
    """
    global databaseFilePath
    connection    = sqlite3.connect(databaseFilePath)
    cursor        = connection.cursor()
    query         = "SELECT * FROM car_rentals WHERE 1=1"
    parameterList = []
    if location:
        query += " AND location LIKE ?"
        parameterList.append(f"%{location}%")
    if name:
        query += " AND name LIKE ?"
        parameterList.append(f"%{name}%")
    # 튜토리얼에서는 원하는 날짜와 가격 범위에 맞춰 예약을 진행하실 수 있다.
    # (우리의 장난감 데이터 세트에는 데이터가 많지 않기 때문에)
    cursor.execute(query, parameterList)
    rowDictionaryList = cursor.fetchall()
    connection.close()
    return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDictionary)) for rowDictionary in rowDictionaryList]

@tool
def bookCarRental(rental_id : int) -> str:
    """
    Book a car rental by its ID.

    Args :
        rental_id (int): The ID of the car rental to book.

    Returns :
        str: A message indicating whether the car rental was successfully booked or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE car_rentals SET booked = 1 WHERE id = ?", (rental_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Car rental {rental_id} successfully booked."
    else:
        connection.close()
        return f"No car rental found with ID {rental_id}."

@tool
def updateCarRental(rental_id : int, start_date : Optional[Union[datetime, date]] = None, end_date : Optional[Union[datetime, date]] = None) -> str:
    """
    Update a car rental's start and end dates by its ID.

    Args :
        rental_id (int) : The ID of the car rental to update.
        start_date (Optional[Union[datetime, date]]) : The new start date of the car rental. Defaults to None.
        end_date (Optional[Union[datetime, date]]) : The new end date of the car rental. Defaults to None.

    Returns :
        str: A message indicating whether the car rental was successfully updated or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    if start_date:
        cursor.execute("UPDATE car_rentals SET start_date = ? WHERE id = ?", (start_date, rental_id))
    if end_date:
        cursor.execute("UPDATE car_rentals SET end_date = ? WHERE id = ?", (end_date, rental_id))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Car rental {rental_id} successfully updated."
    else:
        connection.close()
        return f"No car rental found with ID {rental_id}."

@tool
def cancelCarRental(rental_id : int) -> str:
    """
    Cancel a car rental by its ID.

    Args :
        rental_id (int) : The ID of the car rental to cancel.

    Returns :
        str : A message indicating whether the car rental was successfully cancelled or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE car_rentals SET booked = 0 WHERE id = ?", (rental_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Car rental {rental_id} successfully cancelled."
    else:
        connection.close()
        return f"No car rental found with ID {rental_id}."

@tool
def searchHotelList(location : Optional[str] = None, name : Optional[str] = None, price_tier : Optional[str] = None, checkin_date : Optional[Union[datetime, date]] = None, checkout_date : Optional[Union[datetime, date]] = None) -> list[dict]:
    """
    Search for hotels based on location, name, price tier, check-in date, and check-out date.

    Args :
        location (Optional[str]) : The location of the hotel. Defaults to None.
        name (Optional[str]) : The name of the hotel. Defaults to None.
        price_tier (Optional[str]) : The price tier of the hotel. Defaults to None. Examples: Midscale, Upper Midscale, Upscale, Luxury
        checkin_date (Optional[Union[datetime, date]]) : The check-in date of the hotel. Defaults to None.
        checkout_date (Optional[Union[datetime, date]]) : The check-out date of the hotel. Defaults to None.

    Returns :
        list[dict] : A list of hotel dictionaries matching the search criteria.
    """
    global databaseFilePath
    connection    = sqlite3.connect(databaseFilePath)
    cursor        = connection.cursor()
    query         = "SELECT * FROM hotels WHERE 1=1"
    parameterList = []
    if location:
        query += " AND location LIKE ?"
        parameterList.append(f"%{location}%")
    if name:
        query += " AND name LIKE ?"
        parameterList.append(f"%{name}%")
    # 이 튜토리얼에서는 원하는 날짜와 가격 범위에서 예약을 일치시킬 수 있도록 한다.
    cursor.execute(query, parameterList)
    rowDictionaryList = cursor.fetchall()
    connection.close()
    return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDicrionary)) for rowDicrionary in rowDictionaryList]

@tool
def bookHotel(hotel_id : int) -> str:
    """
    Book a hotel by its ID.

    Args :
        hotel_id (int) : The ID of the hotel to book.

    Returns :
        str : A message indicating whether the hotel was successfully booked or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE hotels SET booked = 1 WHERE id = ?", (hotel_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Hotel {hotel_id} successfully booked."
    else:
        connection.close()
        return f"No hotel found with ID {hotel_id}."

@tool
def updateHotel(hotel_id : int, checkin_date : Optional[Union[datetime, date]] = None, checkout_date : Optional[Union[datetime, date]] = None) -> str:
    """
    Update a hotel's check-in and check-out dates by its ID.

    Args :
        hotel_id (int) : The ID of the hotel to update.
        checkin_date (Optional[Union[datetime, date]]) : The new check-in date of the hotel. Defaults to None.
        checkout_date (Optional[Union[datetime, date]]) : The new check-out date of the hotel. Defaults to None.

    Returns :
        str : A message indicating whether the hotel was successfully updated or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    if checkin_date:
        cursor.execute("UPDATE hotels SET checkin_date = ? WHERE id = ?", (checkin_date, hotel_id))
    if checkout_date:
        cursor.execute("UPDATE hotels SET checkout_date = ? WHERE id = ?", (checkout_date, hotel_id))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Hotel {hotel_id} successfully updated."
    else:
        connection.close()
        return f"No hotel found with ID {hotel_id}."

@tool
def cancelHotel(hotel_id : int) -> str:
    """
    Cancel a hotel by its ID.

    Args :
        hotel_id (int) : The ID of the hotel to cancel.

    Returns :
        str : A message indicating whether the hotel was successfully cancelled or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE hotels SET booked = 0 WHERE id = ?", (hotel_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Hotel {hotel_id} successfully cancelled."
    else:
        connection.close()
        return f"No hotel found with ID {hotel_id}."

@tool
def searchTripRecommendationList(location : Optional[str] = None, name : Optional[str] = None, keywords : Optional[str] = None) -> list[dict]:
    """
    Search for trip recommendations based on location, name, and keywords.

    Args :
        location (Optional[str]) : The location of the trip recommendation. Defaults to None.
        name (Optional[str]) : The name of the trip recommendation. Defaults to None.
        keywords (Optional[str]) : The keywords associated with the trip recommendation. Defaults to None.

    Returns :
        list[dict] : A list of trip recommendation dictionaries matching the search criteria.
    """
    global databaseFilePath
    connection    = sqlite3.connect(databaseFilePath)
    cursor        = connection.cursor()
    query         = "SELECT * FROM trip_recommendations WHERE 1=1"
    parameterList = []
    if location:
        query += " AND location LIKE ?"
        parameterList.append(f"%{location}%")
    if name:
        query += " AND name LIKE ?"
        parameterList.append(f"%{name}%")
    if keywords:
        keyword_list       = keywords.split(",")
        keyword_conditions = " OR ".join(["keywords LIKE ?" for _ in keyword_list])
        query += f" AND ({keyword_conditions})"
        parameterList.extend([f"%{keyword.strip()}%" for keyword in keyword_list])
    cursor.execute(query, parameterList)
    rowDictionaryList = cursor.fetchall()
    connection.close()
    return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDictionary)) for rowDictionary in rowDictionaryList]

@tool
def bookExcursion(recommendation_id : int) -> str:
    """
    Book a excursion by its recommendation ID.

    Args :
        recommendation_id (int) : The ID of the trip recommendation to book.

    Returns :
        str : A message indicating whether the trip recommendation was successfully booked or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE trip_recommendations SET booked = 1 WHERE id = ?", (recommendation_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Trip recommendation {recommendation_id} successfully booked."
    else:
        connection.close()
        return f"No trip recommendation found with ID {recommendation_id}."

@tool
def updateExcursion(recommendation_id : int, details : str) -> str:
    """
    Update a trip recommendation's details by its ID.

    Args :
        recommendation_id (int) : The ID of the trip recommendation to update.
        details (str) : The new details of the trip recommendation.

    Returns :
        str : A message indicating whether the trip recommendation was successfully updated or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor = connection.cursor()
    cursor.execute("UPDATE trip_recommendations SET details = ? WHERE id = ?", (details, recommendation_id))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Trip recommendation {recommendation_id} successfully updated."
    else:
        connection.close()
        return f"No trip recommendation found with ID {recommendation_id}."

@tool
def cancelExcursion(recommendation_id : int) -> str:
    """
    Cancel a trip recommendation by its ID.

    Args :
        recommendation_id (int) : The ID of the trip recommendation to cancel.

    Returns:
        str : A message indicating whether the trip recommendation was successfully cancelled or not.
    """
    global databaseFilePath
    connection = sqlite3.connect(databaseFilePath)
    cursor     = connection.cursor()
    cursor.execute("UPDATE trip_recommendations SET booked = 0 WHERE id = ?", (recommendation_id,))
    connection.commit()
    if cursor.rowcount > 0:
        connection.close()
        return f"Trip recommendation {recommendation_id} successfully cancelled."
    else:
        connection.close()
        return f"No trip recommendation found with ID {recommendation_id}."

chatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful customer support assistant for Swiss Airlines. "
            " Use the provided tools to search for flights, company policies, and other information to assist the user's queries. "
            " When searching, be persistent. Expand your query bounds if the first search returns no results. "
            " If a search comes up empty, expand your search before giving up."
            "\n\nCurrent user:\n<User>\n{user_info}\n</User>"
            "\nCurrent time: {time}.",
        ),
        ("placeholder", "{messages}"),
    ]
).partial(time = datetime.now)

chatOpenAI = ChatOpenAI(model = "gpt-4o", temperature = 1)

safeToolList = [
    TavilySearchResults(max_results = 1),
    fetchUserFlightInformationList,
    searchFlightList,
    lookupPolicy,
    searchCarRentalList,
    searchHotelList,
    searchTripRecommendationList
]

sensitiveToolList = [
    updateTicketToNewFlight,
    cancelTicket,
    bookCarRental,
    updateCarRental,
    cancelCarRental,
    bookHotel,
    updateHotel,
    cancelHotel,
    bookExcursion,
    updateExcursion,
    cancelExcursion
]

sensitiveToolNameSet = {tool.name for tool in sensitiveToolList}

runnableSequence = chatPromptTemplate | chatOpenAI.bind_tools(safeToolList + sensitiveToolList)

class State(TypedDict):
    messages  : Annotated[list[AnyMessage], add_messages]
    user_info : str

class Assistant:
    def __init__(self, runnable : Runnable):
        self.runnable = runnable

    def __call__(self, state: State, config : RunnableConfig):
        messageList = state["messages"]
        lastMessage = messageList[-1]
        if(isinstance(lastMessage, ToolMessage) and lastMessage.content.startswith("API call denied by user.")):
            return {"messages" : [AIMessage(content = lastMessage.content)]}
        while True:
            responseAIMessage = self.runnable.invoke(state)
            # LLM이 빈 응답을 반환하는 경우 실제 응답을 위해 다시 메시지를 표시한다.
            if not responseAIMessage.tool_calls and (not responseAIMessage.content or isinstance(responseAIMessage.content, list) and not responseAIMessage.content[0].get("text")):
                messages = state["messages"] + [("user", "Respond with a real output.")]
                state    = {**state, "messages" : messages}
            else:
                break
        return {"messages" : responseAIMessage}

def handleToolError(state) -> dict:
    error        = state.get("error")
    toolCallList = state["messages"][-1].tool_calls
    return {"messages" : [ToolMessage(content = f"Error : {repr(error)}\n please fix your mistakes.", tool_call_id = toolCall["id"]) for toolCall in toolCallList]}

stateGraph = StateGraph(State)

def fetchUserInfo(state : State):
    return {"user_info" : fetchUserFlightInformationList.invoke({})}

def routeToolNode(state : State):
    nextNode = tools_condition(state)
    # 도구가 호출되지 않으면 사용자에게 돌아간다.
    if nextNode == END:
        return END
    aiMessage = state["messages"][-1]
    # 이는 단일 도구 호출을 가정한다.
    # 병렬 도구 호출을 처리하려면 ANY 조건을 사용해야 한다.
    firstToolCall = aiMessage.tool_calls[0]
    if firstToolCall["name"] in sensitiveToolNameSet:
        return "sensitive_tool_node"
    return "safe_tool_node"

# 새로운 기능 : fetch_user_info_node 노드가 먼저 실행되어 어시스턴트가 아무런 조치도 취하지 않고도 사용자의 항공편 정보를 볼 수 있다.
stateGraph.add_node("fetch_user_info_node", fetchUserInfo)
stateGraph.add_node("assistant_node"      , Assistant(runnableSequence))
stateGraph.add_node("safe_tool_node"      , ToolNode(safeToolList     ).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))
stateGraph.add_node("sensitive_tool_node" , ToolNode(sensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge(START                 , "fetch_user_info_node")
stateGraph.add_edge("fetch_user_info_node", "assistant_node"      )
stateGraph.add_conditional_edges("assistant_node", routeToolNode, ["safe_tool_node", "sensitive_tool_node", END])
stateGraph.add_edge("safe_tool_node"      , "assistant_node"      )
stateGraph.add_edge("sensitive_tool_node" , "assistant_node"      )

memorySaver = MemorySaver()

compiledStateGraph = stateGraph.compile(checkpointer = memorySaver, interrupt_before = ["sensitive_tool_node"])

tutorialQuestionList = [
    "Hi there, what time is my flight?",
    "Am i allowed to update my flight to something sooner? I want to leave later today.",
    "Update my flight to sometime next week then",
    "The next available option is great",
    "what about lodging and transportation?",
    "Yeah i think i'd like an affordable hotel for my week-long stay (7 days). And I'll want to rent a car.",
    "OK could you place a reservation for your recommended hotel? It sounds nice.",
    "yes go ahead and book anything that's moderate expense and has availability.",
    "Now for a car, what are my options?",
    "Awesome let's just get the cheapest option. Go ahead and book for 7 days",
    "Cool so now what recommendations do you have on excursions?",
    "Are they available while I'm there?",
    "interesting - i like the museums, what options are there? ",
    "OK great pick one and book it for my second day there.",
]

configurableDictionary = {
    "configurable" : {
        "thread_id"    : str(uuid.uuid4()),
        "passenger_id" : "3442 587242"
    }
}

printedSet = set()

def printEvent(addableUpdatesDict : dict, printedSet : set, maximumLength = 1500):
    currentStateList = addableUpdatesDict.get("dialog_state")
    if currentStateList:
        print("Currently in : ", currentStateList[-1])
    message = addableUpdatesDict.get("messages")
    if message:
        if isinstance(message, list):
            message = message[-1]
        if message.id not in printedSet:
            messageHTML = message.pretty_repr(html = True)
            if len(messageHTML) > maximumLength:
                messageHTML = messageHTML[:maximumLength] + " ... (truncated)"
            print(messageHTML)
            printedSet.add(message.id)

for tutorialQuestion in tutorialQuestionList:
    generator = compiledStateGraph.stream({"messages" : ("user", tutorialQuestion)}, configurableDictionary, stream_mode = "values")
    for addableUpdatesDict in generator:
        printEvent(addableUpdatesDict, printedSet)
    stateSnapshot = compiledStateGraph.get_state(configurableDictionary)
    while stateSnapshot.next:
        # 인터럽트가 발생했다! 에이전트가 도구를 사용하려고 하며, 사용자는 이를 승인하거나 거부할 수 있다.
        # 참고 : 이 코드는 모두 그래프 외부에 있다. 일반적으로 UI로 출력을 스트리밍한다.
        # 그런 다음 사용자가 입력을 제공하면 프런트엔드가 API 호출을 통해 새 실행을 트리거한다.
        try:
            print()
            userInput = input("Do you approve of the above actions? Type 'y' to continue; otherwise, explain your requested changed.\n\n")
        except:
            userInput = "y"
        if userInput.strip() == "y":
            # 그냥 계속한다.
            responseAIMessage = compiledStateGraph.invoke(None, configurableDictionary)
        else:
            # 요청된 변경 사항/마음의 변화에 대한 지침을 제공하여 도구 호출을 충족한다.
            responseAIMessage = compiledStateGraph.invoke(
                {
                    "messages": [
                        ToolMessage(
                            tool_call_id = addableUpdatesDict["messages"][-1].tool_calls[0]["id"],
                            content      = f"API call denied by user. Reasoning : '{userInput}'. Continue assisting, accounting for the user's input."
                        )
                    ]
                },
                configurableDictionary,
            )
        stateSnapshot = compiledStateGraph.get_state(configurableDictionary)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

import os

import requests

import shutil

import sqlite3

import pandas as pd

import re

import numpy as np

import openai

import pytz

import uuid

from dotenv import load_dotenv

from langchain_core.tools import tool

from langchain_core.runnables import RunnableConfig

from typing import Optional

from datetime import date

from datetime import datetime

from typing import Union

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_community.tools.tavily_search import TavilySearchResults

from typing_extensions import TypedDict

from typing import Annotated

from langgraph.graph.message import AnyMessage

from langgraph.graph.message import add_messages

from langchain_core.runnables import Runnable

from langchain_core.messages import ToolMessage

from langchain_core.messages import AIMessage

from langgraph.graph import StateGraph

from langgraph.prebuilt import ToolNode

from langchain_core.runnables import RunnableLambda

from langgraph.prebuilt import tools_condition

from langgraph.graph import START

from langgraph.graph import END

from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

def downloadFile(sourceURL, targetFilePath, backupFilePath, overwrite):

if overwrite or not os.path.exists(targetFilePath):

response = requests.get(sourceURL)

response.raise_for_status()

with open(targetFilePath, "wb") as bufferedWriter:

bufferedWriter.write(response.content)

if backupFilePath:

shutil.copy(targetFilePath, backupFilePath)

def loadTextFile(sourceFilePath):

with open(sourceFilePath, "r", encoding = "utf-8") as textIOWrapper:

return textIOWrapper.read()

def updateDate(databaseFilePath, backupFilePath):

shutil.copy(backupFilePath, databaseFilePath)

connection = sqlite3.connect(databaseFilePath)

tableNameList = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", connection).name.tolist()

dataFrameDictionary = {}

for tableName in tableNameList:

dataFrameDictionary[tableName] = pd.read_sql(f"SELECT * from {tableName}", connection)

maximumActualDepartureTimestamp = pd.to_datetime(dataFrameDictionary["flights"]["actual_departure"].replace("\\N", pd.NaT)).max()

localCurrentTimestamp = pd.to_datetime("now").tz_localize(maximumActualDepartureTimestamp.tz)

timeDelta = localCurrentTimestamp - maximumActualDepartureTimestamp

dataFrameDictionary["bookings"]["book_date"] = pd.to_datetime(dataFrameDictionary["bookings"]["book_date"].replace("\\N", pd.NaT), utc = True) + timeDelta

dateTimeColumnNameList = [

"scheduled_departure",

"scheduled_arrival",

"actual_departure",

"actual_arrival"

]

for dateTimeColumName in dateTimeColumnNameList:

dataFrameDictionary["flights"][dateTimeColumName] = pd.to_datetime(dataFrameDictionary["flights"][dateTimeColumName].replace("\\N", pd.NaT)) + timeDelta

for tableName, dataFrame in dataFrameDictionary.items():

dataFrame.to_sql(tableName, connection, if_exists = "replace", index = False)

del dataFrame

del dataFrameDictionary

connection.commit()

connection.close()

overwrite = False

databaseURL = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/travel2.sqlite"

databaseFilePath = "travel2.sqlite"

backupFilePath = "travel2.backup.sqlite"

downloadFile(databaseURL, databaseFilePath, backupFilePath, overwrite)

updateDate(databaseFilePath, backupFilePath)

faqFileURL = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/swiss_faq.md"

faqFilePath = "swiss_faq.md"

downloadFile(faqFileURL, faqFilePath, None, overwrite)

faqText = loadTextFile(faqFilePath)

documentList = [{"page_content" : text} for text in re.split(r"(?=\n##)", faqText)]

class VectorStoreRetriever:

def __init__(self, documentList : list, vectorListList : list, openAI):

self.vectorListList = np.array(vectorListList)

self.documentList = documentList

self.openAI = openAI

@classmethod

def fromDocumentList(cls, documentList, openAI):

createEmbeddingResponse = openAI.embeddings.create(model = "text-embedding-3-small", input = [document["page_content"] for document in documentList])

vectorListList = [embedding.embedding for embedding in createEmbeddingResponse.data] # createEmbeddingResponse.data 속성 openai.types.embedding.Embedding 타입 리스트이다.

return cls(documentList, vectorListList, openAI)

def query(self, query : str, k : int = 5) -> list[dict]:

createEmbeddingResponse = self.openAI.embeddings.create(model = "text-embedding-3-small", input = [query])

# "@"는 파이썬에서 행렬 곱셈일 뿐이다.

scoreNDArray = np.array(createEmbeddingResponse.data[0].embedding) @ self.vectorListList.T

topKIndex = np.argpartition(scoreNDArray, -k)[-k:]

topKIndexSorted = topKIndex[np.argsort(-scoreNDArray[topKIndex])]

return [{**self.documentList[index], "similarity" : scoreNDArray[index]} for index in topKIndexSorted]

openAI = openai.Client()

vectorStoreRetriever = VectorStoreRetriever.fromDocumentList(documentList, openAI)

@tool

def lookupPolicy(query : str) -> str:

"""Consult the company policies to check whether certain options are permitted.

Use this before making any flight changes performing other 'write' events."""

global vectorStoreRetriever

documentList = vectorStoreRetriever.query(query, k = 2)

return "\n\n".join([document["page_content"] for document in documentList])

@tool

def fetchUserFlightInformationList(runnableConfig : RunnableConfig) -> list[dict]:

"""Fetch all tickets for the user along with corresponding flight information and seat assignments.

Returns :

A list of dictionaries where each dictionary contains the ticket details,

associated flight details, and the seat assignments for each ticket belonging to the user.

"""

global databaseFilePath

configurableDictionary = runnableConfig.get("configurable", {})

passengerID = configurableDictionary.get("passenger_id", None)

if not passengerID:

raise ValueError("No passenger ID configured.")

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

query = """

SELECT

t.ticket_no,

t.book_ref,

f.flight_id,

f.flight_no,

f.departure_airport,

f.arrival_airport,

f.scheduled_departure,

f.scheduled_arrival,

bp.seat_no,

tf.fare_conditions

FROM tickets t JOIN ticket_flights tf ON t.ticket_no = tf.ticket_no

JOIN flights f ON tf.flight_id = f.flight_id

JOIN boarding_passes bp ON bp.ticket_no = t.ticket_no

AND bp.flight_id = f.flight_id

WHERE t.passenger_id = ?

"""

cursor.execute(query, (passengerID,))

rowTuplList = cursor.fetchall()

columnNameList = [columnTuple[0] for columnTuple in cursor.description]

rowDictionaryList = [dict(zip(columnNameList, rowTuple)) for rowTuple in rowTuplList]

cursor.close()

connection.close()

return rowDictionaryList

@tool

def searchFlightList(departure_airport : Optional[str] = None, arrival_airport : Optional[str] = None, start_time : Optional[date | datetime] = None, end_time : Optional[date | datetime] = None, limit : int = 20) -> list[dict]:

"""Search for flights based on departure airport, arrival airport, and departure time range."""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

query = "SELECT * FROM flights WHERE 1 = 1"

parameterList = []

if departure_airport:

query += " AND departure_airport = ?"

parameterList.append(departure_airport)

if arrival_airport:

query += " AND arrival_airport = ?"

parameterList.append(arrival_airport)

if start_time:

query += " AND scheduled_departure >= ?"

parameterList.append(start_time)

if end_time:

query += " AND scheduled_departure <= ?"

parameterList.append(end_time)

query += " LIMIT ?"

parameterList.append(limit)

cursor.execute(query, parameterList)

rowTupleList = cursor.fetchall()

columnNameList = [columnTuple[0] for columnTuple in cursor.description]

rowDictionaryList = [dict(zip(columnNameList, rowTuple)) for rowTuple in rowTupleList]

cursor.close()

connection.close()

return rowDictionaryList

@tool

def updateTicketToNewFlight(ticket_no : str, new_flight_id : int, *, runnableConfig : RunnableConfig) -> str:

"""Update the user's ticket to a new valid flight."""

global databaseFilePath

configurableDictionary = runnableConfig.get("configurable", {})

passengerID = configurableDictionary.get("passenger_id", None)

if not passengerID:

raise ValueError("No passenger ID configured.")

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("SELECT departure_airport, arrival_airport, scheduled_departure FROM flights WHERE flight_id = ?", (new_flight_id,))

newFlightRowTuple = cursor.fetchone()

if not newFlightRowTuple:

cursor.close()

connection.close()

return "Invalid new flight ID provided."

columnNameList = [columnTuple[0] for columnTuple in cursor.description]

newFlightRowDictionary = dict(zip(columnNameList, newFlightRowTuple))

timezone = pytz.timezone("Etc/GMT-3")

currentDatetime = datetime.now(tz = timezone)

departureDatetime = datetime.strptime(newFlightRowDictionary["scheduled_departure"], "%Y-%m-%d %H:%M:%S.%f%z")

totalSecondCount = (departureDatetime - currentDatetime).total_seconds()

if totalSecondCount < (3 * 3600):

return f"Not permitted to reschedule to a flight that is less than 3 hours from the current time. Selected flight is at {departureDatetime}."

cursor.execute("SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))

currentFlightRowTuple = cursor.fetchone()

if not currentFlightRowTuple:

cursor.close()

connection.close()

return "No existing ticket found for the given ticket number."

cursor.execute("SELECT * FROM tickets WHERE ticket_no = ? AND passenger_id = ?", (ticket_no, passengerID))

currentTicketRowTuple = cursor.fetchone()

if not currentTicketRowTuple:

cursor.close()

connection.close()

return f"Current signed-in passenger with ID {passengerID} not the owner of ticket {ticket_no}"

# 실제 애플리케이션에서는 "새로운 출발 공항이 현재 티켓과 일치하는가"와 같이 비즈니스 로직을 적용하기 위해 여기에 추가적인 검사를 추가할 가능성이 높다.

# LLM에 대한 '타입 힌팅' 정책에서 선제적으로 노력하는 것이 가장 좋지만 필연적으로 문제가 발생할 수 있으므로 또한 API가 유효한 동작을 시행하도록 해야 한다.

cursor.execute("UPDATE ticket_flights SET flight_id = ? WHERE ticket_no = ?", (new_flight_id, ticket_no))

connection.commit()

cursor.close()

connection.close()

return "Ticket successfully updated to new flight."

@tool

def cancelTicket(ticket_no : str, *, runnableConfig : RunnableConfig) -> str:

"""Cancel the user's ticket and remove it from the database."""

global databaseFilePath

configuration = runnableConfig.get("configurable", {})

passengerID = configuration.get("passenger_id", None)

if not passengerID:

raise ValueError("No passenger ID configured.")

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))

existing_ticket = cursor.fetchone()

if not existing_ticket:

cursor.close()

connection.close()

return "No existing ticket found for the given ticket number."

cursor.execute("SELECT flight_id FROM tickets WHERE ticket_no = ? AND passenger_id = ?", (ticket_no, passengerID))

currentTicketRowTuple = cursor.fetchone()

if not currentTicketRowTuple:

cursor.close()

connection.close()

return f"Current signed-in passenger with ID {passengerID} not the owner of ticket {ticket_no}"

cursor.execute("DELETE FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))

connection.commit()

cursor.close()

connection.close()

return "Ticket successfully cancelled."

@tool

def searchCarRentalList(location : Optional[str] = None, name : Optional[str] = None, price_tier : Optional[str] = None, start_date : Optional[Union[datetime, date]] = None, end_date : Optional[Union[datetime, date]] = None) -> list[dict]:

"""

Search for car rentals based on location, name, price tier, start date, and end date.

Args :

location (Optional[str]) : The location of the car rental. Defaults to None.

name (Optional[str]) : The name of the car rental company. Defaults to None.

price_tier (Optional[str]) : The price tier of the car rental. Defaults to None.

start_date (Optional[Union[datetime, date]]) : The start date of the car rental. Defaults to None.

end_date (Optional[Union[datetime, date]]) : The end date of the car rental. Defaults to None.

Returns :

list[dict] : A list of car rental dictionaries matching the search criteria.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

query = "SELECT * FROM car_rentals WHERE 1=1"

parameterList = []

if location:

query += " AND location LIKE ?"

parameterList.append(f"%{location}%")

if name:

query += " AND name LIKE ?"

parameterList.append(f"%{name}%")

# 튜토리얼에서는 원하는 날짜와 가격 범위에 맞춰 예약을 진행하실 수 있다.

# (우리의 장난감 데이터 세트에는 데이터가 많지 않기 때문에)

cursor.execute(query, parameterList)

rowDictionaryList = cursor.fetchall()

connection.close()

return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDictionary)) for rowDictionary in rowDictionaryList]

@tool

def bookCarRental(rental_id : int) -> str:

"""

Book a car rental by its ID.

Args :

rental_id (int): The ID of the car rental to book.

Returns :

str: A message indicating whether the car rental was successfully booked or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE car_rentals SET booked = 1 WHERE id = ?", (rental_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Car rental {rental_id} successfully booked."

else:

connection.close()

return f"No car rental found with ID {rental_id}."

@tool

def updateCarRental(rental_id : int, start_date : Optional[Union[datetime, date]] = None, end_date : Optional[Union[datetime, date]] = None) -> str:

"""

Update a car rental's start and end dates by its ID.

Args :

rental_id (int) : The ID of the car rental to update.

start_date (Optional[Union[datetime, date]]) : The new start date of the car rental. Defaults to None.

end_date (Optional[Union[datetime, date]]) : The new end date of the car rental. Defaults to None.

Returns :

str: A message indicating whether the car rental was successfully updated or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

if start_date:

cursor.execute("UPDATE car_rentals SET start_date = ? WHERE id = ?", (start_date, rental_id))

if end_date:

cursor.execute("UPDATE car_rentals SET end_date = ? WHERE id = ?", (end_date, rental_id))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Car rental {rental_id} successfully updated."

else:

connection.close()

return f"No car rental found with ID {rental_id}."

@tool

def cancelCarRental(rental_id : int) -> str:

"""

Cancel a car rental by its ID.

Args :

rental_id (int) : The ID of the car rental to cancel.

Returns :

str : A message indicating whether the car rental was successfully cancelled or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE car_rentals SET booked = 0 WHERE id = ?", (rental_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Car rental {rental_id} successfully cancelled."

else:

connection.close()

return f"No car rental found with ID {rental_id}."

@tool

def searchHotelList(location : Optional[str] = None, name : Optional[str] = None, price_tier : Optional[str] = None, checkin_date : Optional[Union[datetime, date]] = None, checkout_date : Optional[Union[datetime, date]] = None) -> list[dict]:

"""

Search for hotels based on location, name, price tier, check-in date, and check-out date.

Args :

location (Optional[str]) : The location of the hotel. Defaults to None.

name (Optional[str]) : The name of the hotel. Defaults to None.

price_tier (Optional[str]) : The price tier of the hotel. Defaults to None. Examples: Midscale, Upper Midscale, Upscale, Luxury

checkin_date (Optional[Union[datetime, date]]) : The check-in date of the hotel. Defaults to None.

checkout_date (Optional[Union[datetime, date]]) : The check-out date of the hotel. Defaults to None.

Returns :

list[dict] : A list of hotel dictionaries matching the search criteria.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

query = "SELECT * FROM hotels WHERE 1=1"

parameterList = []

if location:

query += " AND location LIKE ?"

parameterList.append(f"%{location}%")

if name:

query += " AND name LIKE ?"

parameterList.append(f"%{name}%")

# 이 튜토리얼에서는 원하는 날짜와 가격 범위에서 예약을 일치시킬 수 있도록 한다.

cursor.execute(query, parameterList)

rowDictionaryList = cursor.fetchall()

connection.close()

return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDicrionary)) for rowDicrionary in rowDictionaryList]

@tool

def bookHotel(hotel_id : int) -> str:

"""

Book a hotel by its ID.

Args :

hotel_id (int) : The ID of the hotel to book.

Returns :

str : A message indicating whether the hotel was successfully booked or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE hotels SET booked = 1 WHERE id = ?", (hotel_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Hotel {hotel_id} successfully booked."

else:

connection.close()

return f"No hotel found with ID {hotel_id}."

@tool

def updateHotel(hotel_id : int, checkin_date : Optional[Union[datetime, date]] = None, checkout_date : Optional[Union[datetime, date]] = None) -> str:

"""

Update a hotel's check-in and check-out dates by its ID.

Args :

hotel_id (int) : The ID of the hotel to update.

checkin_date (Optional[Union[datetime, date]]) : The new check-in date of the hotel. Defaults to None.

checkout_date (Optional[Union[datetime, date]]) : The new check-out date of the hotel. Defaults to None.

Returns :

str : A message indicating whether the hotel was successfully updated or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

if checkin_date:

cursor.execute("UPDATE hotels SET checkin_date = ? WHERE id = ?", (checkin_date, hotel_id))

if checkout_date:

cursor.execute("UPDATE hotels SET checkout_date = ? WHERE id = ?", (checkout_date, hotel_id))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Hotel {hotel_id} successfully updated."

else:

connection.close()

return f"No hotel found with ID {hotel_id}."

@tool

def cancelHotel(hotel_id : int) -> str:

"""

Cancel a hotel by its ID.

Args :

hotel_id (int) : The ID of the hotel to cancel.

Returns :

str : A message indicating whether the hotel was successfully cancelled or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE hotels SET booked = 0 WHERE id = ?", (hotel_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Hotel {hotel_id} successfully cancelled."

else:

connection.close()

return f"No hotel found with ID {hotel_id}."

@tool

def searchTripRecommendationList(location : Optional[str] = None, name : Optional[str] = None, keywords : Optional[str] = None) -> list[dict]:

"""

Search for trip recommendations based on location, name, and keywords.

Args :

location (Optional[str]) : The location of the trip recommendation. Defaults to None.

name (Optional[str]) : The name of the trip recommendation. Defaults to None.

keywords (Optional[str]) : The keywords associated with the trip recommendation. Defaults to None.

Returns :

list[dict] : A list of trip recommendation dictionaries matching the search criteria.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

query = "SELECT * FROM trip_recommendations WHERE 1=1"

parameterList = []

if location:

query += " AND location LIKE ?"

parameterList.append(f"%{location}%")

if name:

query += " AND name LIKE ?"

parameterList.append(f"%{name}%")

if keywords:

keyword_list = keywords.split(",")

keyword_conditions = " OR ".join(["keywords LIKE ?" for _ in keyword_list])

query += f" AND ({keyword_conditions})"

parameterList.extend([f"%{keyword.strip()}%" for keyword in keyword_list])

cursor.execute(query, parameterList)

rowDictionaryList = cursor.fetchall()

connection.close()

return [dict(zip([columnTuple[0] for columnTuple in cursor.description], rowDictionary)) for rowDictionary in rowDictionaryList]

@tool

def bookExcursion(recommendation_id : int) -> str:

"""

Book a excursion by its recommendation ID.

Args :

recommendation_id (int) : The ID of the trip recommendation to book.

Returns :

str : A message indicating whether the trip recommendation was successfully booked or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE trip_recommendations SET booked = 1 WHERE id = ?", (recommendation_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Trip recommendation {recommendation_id} successfully booked."

else:

connection.close()

return f"No trip recommendation found with ID {recommendation_id}."

@tool

def updateExcursion(recommendation_id : int, details : str) -> str:

"""

Update a trip recommendation's details by its ID.

Args :

recommendation_id (int) : The ID of the trip recommendation to update.

details (str) : The new details of the trip recommendation.

Returns :

str : A message indicating whether the trip recommendation was successfully updated or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE trip_recommendations SET details = ? WHERE id = ?", (details, recommendation_id))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Trip recommendation {recommendation_id} successfully updated."

else:

connection.close()

return f"No trip recommendation found with ID {recommendation_id}."

@tool

def cancelExcursion(recommendation_id : int) -> str:

"""

Cancel a trip recommendation by its ID.

Args :

recommendation_id (int) : The ID of the trip recommendation to cancel.

Returns:

str : A message indicating whether the trip recommendation was successfully cancelled or not.

"""

global databaseFilePath

connection = sqlite3.connect(databaseFilePath)

cursor = connection.cursor()

cursor.execute("UPDATE trip_recommendations SET booked = 0 WHERE id = ?", (recommendation_id,))

connection.commit()

if cursor.rowcount > 0:

connection.close()

return f"Trip recommendation {recommendation_id} successfully cancelled."

else:

connection.close()

return f"No trip recommendation found with ID {recommendation_id}."

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"You are a helpful customer support assistant for Swiss Airlines. "

" Use the provided tools to search for flights, company policies, and other information to assist the user's queries. "

" When searching, be persistent. Expand your query bounds if the first search returns no results. "

" If a search comes up empty, expand your search before giving up."

"\n\nCurrent user:\n<User>\n{user_info}\n</User>"

"\nCurrent time: {time}.",

("placeholder", "{messages}"),

]

).partial(time = datetime.now)

chatOpenAI = ChatOpenAI(model = "gpt-4o", temperature = 1)

safeToolList = [

TavilySearchResults(max_results = 1),

fetchUserFlightInformationList,

searchFlightList,

lookupPolicy,

searchCarRentalList,

searchHotelList,

searchTripRecommendationList

]

sensitiveToolList = [

updateTicketToNewFlight,

cancelTicket,

bookCarRental,

updateCarRental,

cancelCarRental,

bookHotel,

updateHotel,

cancelHotel,

bookExcursion,

updateExcursion,

cancelExcursion

]

sensitiveToolNameSet = {tool.name for tool in sensitiveToolList}

runnableSequence = chatPromptTemplate | chatOpenAI.bind_tools(safeToolList + sensitiveToolList)

class State(TypedDict):

messages : Annotated[list[AnyMessage], add_messages]

user_info : str

class Assistant:

def __init__(self, runnable : Runnable):

self.runnable = runnable

def __call__(self, state: State, config : RunnableConfig):

messageList = state["messages"]

lastMessage = messageList[-1]

if(isinstance(lastMessage, ToolMessage) and lastMessage.content.startswith("API call denied by user.")):

return {"messages" : [AIMessage(content = lastMessage.content)]}

while True:

responseAIMessage = self.runnable.invoke(state)

# LLM이 빈 응답을 반환하는 경우 실제 응답을 위해 다시 메시지를 표시한다.

if not responseAIMessage.tool_calls and (not responseAIMessage.content or isinstance(responseAIMessage.content, list) and not responseAIMessage.content[0].get("text")):

messages = state["messages"] + [("user", "Respond with a real output.")]

state = {**state, "messages" : messages}

else:

break

return {"messages" : responseAIMessage}

def handleToolError(state) -> dict:

error = state.get("error")

toolCallList = state["messages"][-1].tool_calls

return {"messages" : [ToolMessage(content = f"Error : {repr(error)}\n please fix your mistakes.", tool_call_id = toolCall["id"]) for toolCall in toolCallList]}

stateGraph = StateGraph(State)

def fetchUserInfo(state : State):

return {"user_info" : fetchUserFlightInformationList.invoke({})}

def routeToolNode(state : State):

nextNode = tools_condition(state)

# 도구가 호출되지 않으면 사용자에게 돌아간다.

if nextNode == END:

return END

aiMessage = state["messages"][-1]

# 이는 단일 도구 호출을 가정한다.

# 병렬 도구 호출을 처리하려면 ANY 조건을 사용해야 한다.

firstToolCall = aiMessage.tool_calls[0]

if firstToolCall["name"] in sensitiveToolNameSet:

return "sensitive_tool_node"

return "safe_tool_node"

# 새로운 기능 : fetch_user_info_node 노드가 먼저 실행되어 어시스턴트가 아무런 조치도 취하지 않고도 사용자의 항공편 정보를 볼 수 있다.

stateGraph.add_node("fetch_user_info_node", fetchUserInfo)

stateGraph.add_node("assistant_node" , Assistant(runnableSequence))

stateGraph.add_node("safe_tool_node" , ToolNode(safeToolList ).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_node("sensitive_tool_node" , ToolNode(sensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge(START , "fetch_user_info_node")

stateGraph.add_edge("fetch_user_info_node", "assistant_node" )

stateGraph.add_conditional_edges("assistant_node", routeToolNode, ["safe_tool_node", "sensitive_tool_node", END])

stateGraph.add_edge("safe_tool_node" , "assistant_node" )

stateGraph.add_edge("sensitive_tool_node" , "assistant_node" )

memorySaver = MemorySaver()

compiledStateGraph = stateGraph.compile(checkpointer = memorySaver, interrupt_before = ["sensitive_tool_node"])

tutorialQuestionList = [

"Hi there, what time is my flight?",

"Am i allowed to update my flight to something sooner? I want to leave later today.",

"Update my flight to sometime next week then",

"The next available option is great",

"what about lodging and transportation?",

"Yeah i think i'd like an affordable hotel for my week-long stay (7 days). And I'll want to rent a car.",

"OK could you place a reservation for your recommended hotel? It sounds nice.",

"yes go ahead and book anything that's moderate expense and has availability.",

"Now for a car, what are my options?",

"Awesome let's just get the cheapest option. Go ahead and book for 7 days",

"Cool so now what recommendations do you have on excursions?",

"Are they available while I'm there?",

"interesting - i like the museums, what options are there? ",

"OK great pick one and book it for my second day there.",

]

configurableDictionary = {

"configurable" : {

"thread_id" : str(uuid.uuid4()),

"passenger_id" : "3442 587242"

}

printedSet = set()

def printEvent(addableUpdatesDict : dict, printedSet : set, maximumLength = 1500):

currentStateList = addableUpdatesDict.get("dialog_state")

if currentStateList:

print("Currently in : ", currentStateList[-1])

message = addableUpdatesDict.get("messages")

if message:

if isinstance(message, list):

message = message[-1]

if message.id not in printedSet:

messageHTML = message.pretty_repr(html = True)

if len(messageHTML) > maximumLength:

messageHTML = messageHTML[:maximumLength] + " ... (truncated)"

print(messageHTML)

printedSet.add(message.id)

for tutorialQuestion in tutorialQuestionList:

generator = compiledStateGraph.stream({"messages" : ("user", tutorialQuestion)}, configurableDictionary, stream_mode = "values")

for addableUpdatesDict in generator:

printEvent(addableUpdatesDict, printedSet)

stateSnapshot = compiledStateGraph.get_state(configurableDictionary)

while stateSnapshot.next:

# 인터럽트가 발생했다! 에이전트가 도구를 사용하려고 하며, 사용자는 이를 승인하거나 거부할 수 있다.

# 참고 : 이 코드는 모두 그래프 외부에 있다. 일반적으로 UI로 출력을 스트리밍한다.

# 그런 다음 사용자가 입력을 제공하면 프런트엔드가 API 호출을 통해 새 실행을 트리거한다.

try:

print()

userInput = input("Do you approve of the above actions? Type 'y' to continue; otherwise, explain your requested changed.\n\n")

except:

userInput = "y"

if userInput.strip() == "y":

# 그냥 계속한다.

responseAIMessage = compiledStateGraph.invoke(None, configurableDictionary)

else:

# 요청된 변경 사항/마음의 변화에 대한 지침을 제공하여 도구 호출을 충족한다.

responseAIMessage = compiledStateGraph.invoke(

{

"messages": [

ToolMessage(

tool_call_id = addableUpdatesDict["messages"][-1].tool_calls[0]["id"],

content = f"API call denied by user. Reasoning : '{userInput}'. Continue assisting, accounting for the user's input."

)

]

configurableDictionary,

)

stateSnapshot = compiledStateGraph.get_state(configurableDictionary)

▶ requirements.txt


aiohappyeyeballs==2.4.4
aiohttp==3.11.11
aiosignal==1.3.2
annotated-types==0.7.0
anyio==4.7.0
attrs==24.3.0
certifi==2024.12.14
charset-normalizer==3.4.1
colorama==0.4.6
dataclasses-json==0.6.7
distro==1.9.0
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
httpx-sse==0.4.0
idna==3.10
jiter==0.8.2
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.13
langchain-community==0.3.13
langchain-core==0.3.28
langchain-openai==0.2.14
langchain-text-splitters==0.3.4
langgraph==0.2.60
langgraph-checkpoint==2.0.9
langgraph-sdk==0.1.48
langsmith==0.2.7
marshmallow==3.23.2
msgpack==1.1.0
multidict==6.1.0
mypy-extensions==1.0.0
numpy==2.2.1
openai==1.58.1
orjson==3.10.13
packaging==24.2
pandas==2.2.3
propcache==0.2.1
pydantic==2.10.4
pydantic-settings==2.7.1
pydantic_core==2.27.2
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
pytz==2024.2
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.3
requests-toolbelt==1.0.0
six==1.17.0
sniffio==1.3.1
SQLAlchemy==2.0.36
tenacity==9.0.0
tiktoken==0.8.0
tqdm==4.67.1
typing-inspect==0.9.0
typing_extensions==4.12.2
tzdata==2024.2
urllib3==2.3.0
yarl==1.18.3

aiohappyeyeballs==2.4.4

aiohttp==3.11.11

aiosignal==1.3.2

annotated-types==0.7.0

anyio==4.7.0

attrs==24.3.0

certifi==2024.12.14

charset-normalizer==3.4.1

colorama==0.4.6

dataclasses-json==0.6.7

distro==1.9.0

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.28.1

httpx-sse==0.4.0

idna==3.10

jiter==0.8.2

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.13

langchain-community==0.3.13

langchain-core==0.3.28

langchain-openai==0.2.14

langchain-text-splitters==0.3.4

langgraph==0.2.60

langgraph-checkpoint==2.0.9

langgraph-sdk==0.1.48

langsmith==0.2.7

marshmallow==3.23.2

msgpack==1.1.0

multidict==6.1.0

mypy-extensions==1.0.0

numpy==2.2.1

openai==1.58.1

orjson==3.10.13

packaging==24.2

pandas==2.2.3

propcache==0.2.1

pydantic==2.10.4

pydantic-settings==2.7.1

pydantic_core==2.27.2

python-dateutil==2.9.0.post0

python-dotenv==1.0.1

pytz==2024.2

PyYAML==6.0.2

regex==2024.11.6

requests==2.32.3

requests-toolbelt==1.0.0

six==1.17.0

sniffio==1.3.1

SQLAlchemy==2.0.36

tenacity==9.0.0

tiktoken==0.8.0

tqdm==4.67.1

typing-inspect==0.9.0

typing_extensions==4.12.2

tzdata==2024.2

urllib3==2.3.0

yarl==1.18.3

※ pip install python-dotenv langchain-community langchain-openai langgraph tavily-python pandas 명령을 실행했다.

AI LANGGRAPH LLM OPENAI PYTHON TAVILY TOOL

icodebroker

[PYTHON/LANGGRAPH] 도구 실행시 조건부 확인 기능 추가 제로샷 에이전트 만들기

분류

가장 많이 읽힌 글

보관함