[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 pytz
import openai
import uuid

from dotenv                                  import load_dotenv
from typing                                  import Optional
from typing_extensions                       import TypedDict
from typing                                  import Annotated
from langgraph.graph.message                 import AnyMessage
from langgraph.graph.message                 import add_messages
from typing                                  import Literal
from langchain_core.runnables                import Runnable
from langchain_openai                        import ChatOpenAI
from pydantic                                import BaseModel
from typing                                  import Callable
from langchain_core.messages                 import ToolMessage
from langgraph.graph                         import StateGraph
from langchain_core.tools                    import tool
from langchain_core.runnables                import RunnableConfig
from langgraph.graph                         import START
from langchain_core.prompts                  import ChatPromptTemplate
from datetime                                import date
from datetime                                import datetime
from langgraph.prebuilt                      import tools_condition
from langgraph.graph                         import END
from langgraph.prebuilt                      import ToolNode
from langchain_core.runnables                import RunnableLambda
from typing                                  import Union
from langchain_community.tools.tavily_search import TavilySearchResults
from pydantic                                import Field
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)

def updateDialogStateList(targetList : list[str], text : Optional[str]) -> list[str]:
    """Push or pop the state."""
    # "상태를 넣거나 꺼낸다.
    if text is None:
        return targetList
    if text == "pop":
        return targetList[:-1]
    return targetList + [text]

class State(TypedDict):
    messages          : Annotated[list[AnyMessage], add_messages]
    user_info         : str
    dialog_state_list : Annotated[list[Literal["primary_assistant_node", "flight_booking_assistant_node", "car_rental_booking_assistant_node", "hotel_booking_assistant_node", "excursion_booking_assistant_node"]], updateDialogStateList]

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

    def __call__(self, state : State, config : RunnableConfig):
        while True:
            responseAIMessage = self.runnable.invoke(state)
            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}

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

class CompleteOrEscalate(BaseModel):
    """A tool to mark the current task as completed and/or to escalate control of the dialog to the main assistant, who can re-route the dialog based on the user's needs."""
    # 현재 작업을 완료로 표시하거나 대화 제어권을 주요 어시스턴에게 위임하여, 사용자의 필요에 따라 대화 경로를 변경할 수 있는 도구이다.
    cancel : bool = True
    reason : str

    class Config:
        json_schema_extra = {
            "example 1" : {
                "cancel" : True,
                "reason" : "User changed their mind about the current task." # 사용자가 현재 작업에 대한 마음을 바꾸었다.
            },
            "example 2" : {
                "cancel" : True,
                "reason" : "I have fully completed the task." # 나는 그 임무를 완전히 완수했다.
            },
            "example 3" : {
                "cancel" : False,
                "reason" : "I need to search the user's emails or calendar for more information." # 자세한 정보를 알아보기 위해 사용자의 이메일이나 캘린더를 검색해야 한다.
            }
        }

def createEntryNode(assistantName : str, newDialogState : str) -> Callable:
    def entry_node(state : State) -> dict:
        toolCallID = state["messages"][-1].tool_calls[0]["id"]
        return {
            "messages" : [
                ToolMessage(
                    content = f"The assistant is now the {assistantName}. Reflect on the above conversation between the host assistant and the user."
                    f" The user's intent is unsatisfied. Use the provided tools to assist the user. Remember, you are {assistantName},"
                    " and the booking, update, other other action is not complete until after you have successfully invoked the appropriate tool."
                    " If the user changes their mind or needs help for other tasks, call the CompleteOrEscalate function to let the primary host assistant take control."
                    " Do not mention who you are - just act as the proxy for the assistant.",
                    # 이제 어시스턴트는 {assistantName}입니다. 호스트 어시스턴트와 사용자 간의 위의 대화를 생각해 본다.
                    # 사용자의 의도가 충족되지 않았다.
                    # 제공된 도구를 사용하여 사용자를 지원한다.
                    # 기억한다, 당신은 {assistantName}이고, 예약, 업데이트, 기타 다른 작업은 적절한 도구를 성공적으로 호출한 후에야 완료된다.
                    # 사용자가 마음을 바꾸거나 다른 작업에 대한 도움이 필요한 경우 CompleteOrEscalate 함수를 호출하여 기본 호스트 지원자가 제어권을 갖도록 한다.
                    # 자신이 누구인지 밝히지 않는다.
                    # 그저 이시스턴트의 대리인으로 행동한다.
                    tool_call_id = toolCallID
                )
            ],
            "dialog_state_list" : newDialogState
        }
    return entry_node

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)

# 사용자 정보 가져오기 노드
@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

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

# 위임된 각 워크플로는 사용자에게 직접 응답할 수 있다.
# 사용자가 응답하면 현재 활성화된 워크플로로 돌아가기를 원한다.
def routeToWorkflow(state : State) -> Literal["primary_assistant_node", "flight_booking_assistant_node", "car_rental_booking_assistant_node", "hotel_booking_assistant_node", "excursion_booking_assistant_node"]:
    """If we are in a delegated state, route directly to the appropriate assistant."""
    # 위임 상태인 경우 해당 보조자에게 직접 전달한다.
    dialogStateList = state.get("dialog_state_list")
    if not dialogStateList:
        return "primary_assistant_node"
    return dialogStateList[-1]

stateGraph.add_node("fetch_user_info_node", fetchUserInfo)

stateGraph.add_edge(START, "fetch_user_info_node")
stateGraph.add_conditional_edges("fetch_user_info_node", routeToWorkflow)

# 항공편 예약 어시스턴트 노드
flightBookingChatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a specialized assistant for handling flight updates. "
            "The primary assistant delegates work to you whenever the user needs help updating their bookings. "
            "Confirm the updated flight details with the customer and inform them of any additional fees. "
            "When searching, be persistent. Expand your query bounds if the first search returns no results. "
            "If you need more information or the customer changes their mind, escalate the task back to the main assistant. "
            "Remember that a booking isn't completed until after the relevant tool has successfully been used. "
            "\n\nCurrent user flight information :\n<Flights>\n{user_info}\n</Flights>"
            "\nCurrent time : {time}."
            "\n\nIf the user needs help, and none of your tools are appropriate for it, then "
            '"CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'
            # 당신은 항공편 업데이트를 처리하는 전문 어시스턴트이다.
            # 주요 어시스턴트는 사용자가 예약을 업데이트하는 데 도움이 필요할 때마다 사용자를 대신하여 작업한다.
            # 고객에게 업데이트된 항공편 세부 정보를 확인하고 추가 비용이 있는 경우 이를 알려준다.
            # 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.
            # 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.
            # 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.
            # 현재 사용자 항공편 정보 :\n<Flights>\n{user_info}\n</Flights>
            # 현재 시간 : {time}.
            # 사용자에게 도움이 필요하고 도구 중 어느 것도 적합하지 않은 경우, 호스트 어시스턴트에게 대화를 "CompleteOrEscalate"합니다.
            # 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.
        ),
        ("placeholder", "{messages}")
    ]
).partial(time = datetime.now)

@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."

flightBookingSafeToolList      = [searchFlightList]
flightBookingSensitiveToolList = [updateTicketToNewFlight, cancelTicket]
flightBookingToolList          = flightBookingSafeToolList + flightBookingSensitiveToolList
flightBookinRunnableSequence   = flightBookingChatPromptTemplate | chatOpenAI.bind_tools(flightBookingToolList + [CompleteOrEscalate])

def routeFlightBookingNode(state : State):
    routingNode = tools_condition(state)
    if routingNode == END:
        return END
    toolCallList    = state["messages"][-1].tool_calls
    alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)
    if alreadyCanceled:
        return "leave_skill_node"
    safeToolNameList = [tool.name for tool in flightBookingSafeToolList]
    if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):
        return "flight_booking_safe_tool_node"
    return "flight_booking_sensitive_tool_node"

stateGraph.add_node("flight_booking_entry_node"         , createEntryNode("Flight Updates & Booking Assistant", "flight_booking_assistant_node"))
stateGraph.add_node("flight_booking_assistant_node"     , Assistant(flightBookinRunnableSequence))
stateGraph.add_node("flight_booking_safe_tool_node"     , ToolNode(flightBookingSafeToolList).with_fallbacks([RunnableLambda(handleToolError)]     , exception_key = "error"))
stateGraph.add_node("flight_booking_sensitive_tool_node", ToolNode(flightBookingSensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge("flight_booking_entry_node"         , "flight_booking_assistant_node")
stateGraph.add_conditional_edges("flight_booking_assistant_node", routeFlightBookingNode, ["flight_booking_safe_tool_node", "flight_booking_sensitive_tool_node", "leave_skill_node", END])
stateGraph.add_edge("flight_booking_safe_tool_node"     , "flight_booking_assistant_node")
stateGraph.add_edge("flight_booking_sensitive_tool_node", "flight_booking_assistant_node")

# 자동차 렌탈 예약 어시스턴트 노드
carRentalBookingChatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a specialized assistant for handling car rental bookings. "
            "The primary assistant delegates work to you whenever the user needs help booking a car rental. "
            "Search for available car rentals based on the user's preferences and confirm the booking details with the customer. "
            "When searching, be persistent. Expand your query bounds if the first search returns no results. "
            "If you need more information or the customer changes their mind, escalate the task back to the main assistant. "
            "Remember that a booking isn't completed until after the relevant tool has successfully been used. "
            "\nCurrent time : {time}. "
            "\n\nIf the user needs help, and none of your tools are appropriate for it, then "
            '"CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'
            "\n\nSome examples for which you should CompleteOrEscalate :\n"
            " - 'what's the weather like this time of year?'\n"
            " - 'What flights are available?'\n"
            " - 'nevermind i think I'll book separately'\n"
            " - 'Oh wait i haven't booked my flight yet i'll do that first'\n"
            " - 'Car rental booking confirmed'"
            # 귀하는 자동차 렌탈 예약을 처리하는 전문 어시스턴트이다.
            # 주요 어시스턴트는 사용자가 렌터카 예약에 도움이 필요할 때마다 도움을 준다.
            # 사용자의 선호도에 따라 이용 가능한 렌터카를 검색하고 고객과 예약 세부 사항을 확인한다.
            # 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.
            # 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.
            # 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.
            # 현재 시간 : {time}.
            # 사용자에게 도움이 필요하고 도구 중 어느 것도 적합하지 않은 경우, 호스트 어시스턴트에게 대화를 "CompleteOrEscalate"한다.
            # 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.
            # CompleteOrEscalate해야 하는 몇 가지 예 :
            # - '이번 시기에 날씨가 어때요?'
            # - '어떤 항공편이 가능한가요?'
            # - '괜찮아요. 따로 예약할게요'
            # - '아, 잠깐만요. 아직 비행기 예약을 안 했어요. 먼저 예약할게요.'
            # - '렌터카 예약 확인'
        ),
        ("placeholder", "{messages}")
    ]
).partial(time = datetime.now)

@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.
    """
    # 위치, 이름, 가격 등급, 시작 날짜, 종료 날짜를 기준으로 렌터카를 검색한다.
    # 인자 :
    #     location (Optional[str])   : 렌터카 위치. 기본값은 없음이다.
    #     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] : 검색 기준과 일치하는 자동차 렌탈 사전 목록이다.
    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.
    """
    # 차량 ID로 렌터카를 예약한다.
    # 인자 :
    #     rental_id (int) : 예약할 렌터카의 ID이다.
    # 반환 :
    #     str : 자동차 렌탈이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # ID를 사용하여 자동차 렌탈 시작 및 종료 날짜를 업데이트한다.
    # 인자 :
    #     rental_id (int)                              : 업데이트할 렌터카 ID이다.
    #     start_date (Optional[Union[datetime, date]]) : 자동차 렌탈의 새로운 시작 날짜. 기본값은 None이다.
    #     end_date (Optional[Union[datetime, date]])   : 자동차 렌탈의 새로운 종료 날짜. 기본값은 None이다.
    # 반환 :
    #     str : 렌터카가 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # 차량 렌탈을 ID로 취소한다.
    # 인자 :
    #     rental_id (int) : 취소할 렌터카의 ID이다.
    # 반환 :
    #     str : 차량 렌탈이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.
    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}."

carRentalBookingSafeToolList      = [searchCarRentalList]
carRentalBookingSensitiveToolList = [bookCarRental, updateCarRental, cancelCarRental]
carRentalBookingToolList          = carRentalBookingSafeToolList + carRentalBookingSensitiveToolList
carRentalBookingRunnableSequence  = carRentalBookingChatPromptTemplate | chatOpenAI.bind_tools(carRentalBookingToolList + [CompleteOrEscalate])

def routeCarRentalBookingNode(state : State):
    routingNode = tools_condition(state)
    if routingNode == END:
        return END
    toolCallList    = state["messages"][-1].tool_calls
    alreadyCanceled = any(tollCall["name"] == CompleteOrEscalate.__name__ for tollCall in toolCallList)
    if alreadyCanceled:
        return "leave_skill_node"
    safeToolNameList = [tool.name for tool in carRentalBookingSafeToolList]
    if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):
        return "car_rental_booking_safe_tool_node"
    return "car_rental_booking_sensitive_tool_node"

stateGraph.add_node("car_rental_booking_entry_node"         , createEntryNode("Car Rental Assistant", "car_rental_booking_assistant_node"))
stateGraph.add_node("car_rental_booking_assistant_node"     , Assistant(carRentalBookingRunnableSequence))
stateGraph.add_node("car_rental_booking_safe_tool_node"     , ToolNode(carRentalBookingSafeToolList).with_fallbacks([RunnableLambda(handleToolError)]     , exception_key = "error"))
stateGraph.add_node("car_rental_booking_sensitive_tool_node", ToolNode(carRentalBookingSensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge("car_rental_booking_entry_node"         , "car_rental_booking_assistant_node")
stateGraph.add_conditional_edges("car_rental_booking_assistant_node", routeCarRentalBookingNode, ["car_rental_booking_safe_tool_node", "car_rental_booking_sensitive_tool_node", "leave_skill_node", END])
stateGraph.add_edge("car_rental_booking_safe_tool_node"     , "car_rental_booking_assistant_node")
stateGraph.add_edge("car_rental_booking_sensitive_tool_node", "car_rental_booking_assistant_node")

# 호텔 예약 어시스턴트 노드
hotelBookingChatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a specialized assistant for handling hotel bookings. "
            "The primary assistant delegates work to you whenever the user needs help booking a hotel. "
            "Search for available hotels based on the user's preferences and confirm the booking details with the customer. "
            "When searching, be persistent. Expand your query bounds if the first search returns no results. "
            "If you need more information or the customer changes their mind, escalate the task back to the main assistant. "
            "Remember that a booking isn't completed until after the relevant tool has successfully been used. "
            "\nCurrent time : {time}."
            '\n\nIf the user needs help, and none of your tools are appropriate for it, then "CompleteOrEscalate" the dialog to the host assistant. '
            "Do not waste the user's time. Do not make up invalid tools or functions."
            "\n\nSome examples for which you should CompleteOrEscalate :\n"
            " - 'what's the weather like this time of year?'\n"
            " - 'nevermind i think I'll book separately'\n"
            " - 'i need to figure out transportation while i'm there'\n"
            " - 'Oh wait i haven't booked my flight yet i'll do that first'\n"
            " - 'Hotel booking confirmed'"
            # 당신은 호텔 예약을 처리하는 전문 어시스턴트이다.
            # 주요 어시스턴트는 사용자가 호텔 예약에 도움이 필요할 때마다 도움을 준다.
            # 사용자의 선호도에 따라 이용 가능한 호텔을 검색하고 고객과 예약 세부 정보를 확인한다.
            # 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.
            # 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.
            # 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.
            # 현재 시간 : {time}.
            # 사용자에게 도움이 필요하고 어떤 도구도 이에 적합하지 않은 경우, 호스트 지원팀에 대화를 "완료 또는 에스컬레이션"한다.
            # 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들지 않는다.
            # CompleteOrEscalate를 수행해야 하는 몇 가지 예 :
            # - '이번 시기에 날씨가 어때요?'
            # - '괜찮아요. 따로 예약할게요'
            # - '내가 거기에 있는 동안 교통수단을 알아내야 해'
            # - '아, 잠깐만요. 아직 비행기 예약을 안 했어요. 먼저 예약할게요.'
            # - '호텔 예약이 확정되었습니다'
        ),
        ("placeholder", "{messages}")
    ]
).partial(time = datetime.now)

@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.
    """
    # 위치, 이름, 가격, 체크인 날짜, 체크아웃 날짜를 기준으로 호텔을 검색한다.
    # 인자 :
    #     location (Optional[str])   : 호텔의 위치. 기본값은 None이다.
    #     name (Optional[str])       : 호텔 이름. 기본값은 None이다.
    #     price_tier (Optional[str]) : 호텔의 가격 등급이다. 기본값은 None이다. 예 : Midscale, Upper Midscale, Upscale, Luxury
    #     checkin_date (Optional[Union[datetime, date]])  : 호텔의 체크인 날짜. 기본값은 None이다.
    #     checkout_date (Optional[Union[datetime, date]]) : 호텔의 체크아웃 날짜. 기본값은 None이다.
    # 반환 :
    #     list[dict] : 검색 기준과 일치하는 호텔 사전 목록이다.
    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.
    """
    # 호텔 ID로 호텔을 예약한다.
    # 인자 :
    #     hotel_id (int) : 예약할 호텔의 ID이다.
    # 반환 :
    #     str : 호텔이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # 호텔 ID를 사용하여 호텔의 체크인 및 체크아웃 날짜를 업데이트한다.
    # 인자 :
    #     hotel_id (int) : 업데이트할 호텔의 ID이다.
    #     checkin_date (Optional[Union[datetime, date]])  : 호텔의 새로운 체크인 날짜. 기본값은 None이다.
    #     checkout_date (Optional[Union[datetime, date]]) : 호텔의 새로운 체크아웃 날짜. 기본값은 None이다.
    # 반환 :
    #     str : 호텔이 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # 호텔 ID로 호텔 취소한다.
    # 인자 :
    #     hotel_id (int) : 취소할 호텔의 ID이다.
    # 반환 :
    #     str : 호텔이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.
    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}."

hotelBookingSafeToolList      = [searchHotelList]
hotelBookingSensitiveToolList = [bookHotel, updateHotel, cancelHotel]
hotelBookingToolList          = hotelBookingSafeToolList + hotelBookingSensitiveToolList
hotelBookingRunnableSequence  = hotelBookingChatPromptTemplate | chatOpenAI.bind_tools(hotelBookingToolList + [CompleteOrEscalate])

def routeHotelBookingNode(state : State):
    routingNode = tools_condition(state)
    if routingNode == END:
        return END
    toolCallList    = state["messages"][-1].tool_calls
    alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)
    if alreadyCanceled:
        return "leave_skill_node"
    safeToolNameList = [t.name for t in hotelBookingSafeToolList]
    if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):
        return "hotel_booking_safe_tool_node"
    return "hotel_booking_sensitive_tool_node"

stateGraph.add_node("hotel_booking_entry_node"         , createEntryNode("Hotel Booking Assistant", "hotel_booking_assistant_node"))
stateGraph.add_node("hotel_booking_assistant_node"     , Assistant(hotelBookingRunnableSequence))
stateGraph.add_node("hotel_booking_safe_tool_node"     , ToolNode(hotelBookingSafeToolList).with_fallbacks([RunnableLambda(handleToolError)]     , exception_key = "error"))
stateGraph.add_node("hotel_booking_sensitive_tool_node", ToolNode(hotelBookingSensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge("hotel_booking_entry_node"         , "hotel_booking_assistant_node")
stateGraph.add_conditional_edges("hotel_booking_assistant_node", routeHotelBookingNode, ["hotel_booking_safe_tool_node", "hotel_booking_sensitive_tool_node", "leave_skill_node", END])
stateGraph.add_edge("hotel_booking_safe_tool_node"     , "hotel_booking_assistant_node")
stateGraph.add_edge("hotel_booking_sensitive_tool_node", "hotel_booking_assistant_node")

# 여행 예약 어시스턴트 노드
excursionBookingChatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a specialized assistant for handling trip recommendations. "
            "The primary assistant delegates work to you whenever the user needs help booking a recommended trip. "
            "Search for available trip recommendations based on the user's preferences and confirm the booking details with the customer. "
            "If you need more information or the customer changes their mind, escalate the task back to the main assistant. "
            "When searching, be persistent. Expand your query bounds if the first search returns no results. "
            "Remember that a booking isn't completed until after the relevant tool has successfully been used. "
            "\nCurrent time : {time}."
            '\n\nIf the user needs help, and none of your tools are appropriate for it, then "CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'
            "\n\nSome examples for which you should CompleteOrEscalate :\n"
            " - 'nevermind i think I'll book separately'\n"
            " - 'i need to figure out transportation while i'm there'\n"
            " - 'Oh wait i haven't booked my flight yet i'll do that first'\n"
            " - 'Excursion booking confirmed!'"
            # 여행 추천을 처리하는 전문 어시스턴트이다.
            # 주요 어시스턴트는 사용자가 추천된 여행을 예약하는 데 도움이 필요할 때마다 작업을 위임한다.
            # 사용자의 선호도에 따라 이용 가능한 여행 추천을 검색하고 고객과 예약 세부 정보를 확인한다.
            # 추가 정보가 필요하거나 고객이 마음을 바꾸면 작업을 다시 주요 보조원에게 에스컬레이션한다.
            # 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 반환되지 않으면 쿼리 범위를 확장한다.
            # 관련 도구를 성공적으로 사용한 후에야 예약이 완료된다.
            # 현재 시간 : {time}.
            # 사용자에게 도움이 필요하고 도구 중 적합한 도구가 없는 경우 호스트 보조원에게 대화를 "CompleteOrEscalate"한다.
            # 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.
            # CompleteOrEscalate해야 하는 몇 가지 예:
            # - '괜찮아요. 따로 예약할 것 같아요'
            # - '거기에 있는 동안 교통수단을 알아내야 해요'
            # - '아, 아직 항공편을 예약하지 않았네요. 먼저 예약할게요'
            # - '여행 예약이 확정되었습니다!'
        ),
        ("placeholder", "{messages}")
    ]
).partial(time = datetime.now)

@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.
    """
    # 위치, 이름, 키워드를 기반으로 여행 추천을 검색한다.
    # 인자 :
    #     location (Optional[str]) : 여행 추천의 위치이다. 기본값은 None이다.
    #     name (Optional[str])     : 여행 추천의 이름이다. 기본값은 None이다.
    #     keywords (Optional[str]) : 여행 추천과 관련된 키워드. 기본값은 None이다.
    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.
    """
    # 추천 ID로 여행을 예약한다.
    # 인자 :
    #     recommendation_id (int) : 예약을 위한 여행 추천 ID이다.
    # 반환 :
    #     str : 여행 추천이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # ID를 사용하여 여행 추천 세부 정보를 업데이트한다.
    # 인자 :
    #     recommendation_id (int) : 업데이트할 여행 추천의 ID이다.
    #     details (str)           : 여행 추천의 새로운 세부 정보.
    # 반환 :
    #     str : 여행 추천이 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.
    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.
    """
    # ID로 여행 추천을 취소한다.
    # 인자 :
    #      recommendation_id (int) : 취소할 여행 추천의 ID이다.
    # 반환 :
    #      str : 여행 추천이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.
    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}."

excursionBookingSafeToolList      = [searchTripRecommendationList]
excursionBookingSensitiveToolList = [bookExcursion, updateExcursion, cancelExcursion]
excursionBookingToolList          = excursionBookingSafeToolList + excursionBookingSensitiveToolList
excursionBookingRunnableSequence  = excursionBookingChatPromptTemplate | chatOpenAI.bind_tools(excursionBookingToolList + [CompleteOrEscalate])

def routeExcursionBookingNode(state : State):
    routingNode = tools_condition(state)
    if routingNode == END:
        return END
    toolCallList    = state["messages"][-1].tool_calls
    alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)
    if alreadyCanceled:
        return "leave_skill_node"
    safeToolNameList = [tool.name for tool in excursionBookingSafeToolList]
    if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):
        return "excursion_booking_safe_tool_node"
    return "excursion_booking_sensitive_tool_node"

stateGraph.add_node("excursion_booking_entry_node"         , createEntryNode("Trip Recommendation Assistant", "excursion_booking_assistant_node"))
stateGraph.add_node("excursion_booking_assistant_node"     , Assistant(excursionBookingRunnableSequence))
stateGraph.add_node("excursion_booking_safe_tool_node"     , ToolNode(excursionBookingSafeToolList).with_fallbacks([RunnableLambda(handleToolError)]     , exception_key = "error"))
stateGraph.add_node("excursion_booking_sensitive_tool_node", ToolNode(excursionBookingSensitiveToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

stateGraph.add_edge("excursion_booking_entry_node"         , "excursion_booking_assistant_node")
stateGraph.add_conditional_edges("excursion_booking_assistant_node", routeExcursionBookingNode, ["excursion_booking_safe_tool_node", "excursion_booking_sensitive_tool_node", "leave_skill_node", END])
stateGraph.add_edge("excursion_booking_safe_tool_node"     , "excursion_booking_assistant_node")
stateGraph.add_edge("excursion_booking_sensitive_tool_node", "excursion_booking_assistant_node")

# 메인 어시스턴트 노드
primaryAssistantChatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful customer support assistant for Swiss Airlines. "
            "Your primary role is to search for flight information and company policies to answer customer queries. "
            "If a customer requests to update or cancel a flight, book a car rental, book a hotel, or get trip recommendations, "
            "delegate the task to the appropriate specialized assistant by invoking the corresponding tool. You are not able to make these types of changes yourself."
            " Only the specialized assistants are given permission to do this for the user."
            "The user is not aware of the different specialized assistants, so do not mention them; just quietly delegate through function calls. "
            "Provide detailed information to the customer, and always double-check the database before concluding that information is unavailable. "
            " 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 flight information :\n<Flights>\n{user_info}\n</Flights>"
            "\nCurrent time : {time}."
            # Swiss Airlines의 유용한 고객 지원 어시슽입니다.
            # 주요 역할은 고객 문의에 답하기 위해 항공편 정보와 회사 정책을 검색하는 것입니다.
            # 고객이 항공편을 업데이트하거나 취소하거나, 렌터카를 예약하거나, 호텔을 예약하거나, 여행 추천을 받으려고 요청하는 경우
            # 해당 도구를 호출하여 해당 전문 지원 담당자에게 작업을 위임합니다. 직접 이러한 유형의 변경을 할 수는 없습니다.
            # 전문 지원 담당자에게만 사용자를 대신하여 이 작업을 수행할 권한이 부여됩니다.
            # 사용자는 다양한 전문 지원 담당자를 알지 못하므로 언급하지 말고 함수 호출을 통해 조용히 위임하세요.
            # 고객에게 자세한 정보를 제공하고 정보를 사용할 수 없다는 결론을 내리기 전에 항상 데이터베이스를 다시 확인하세요.
            # 검색할 때는 끈기 있게 노력하세요. 첫 번째 검색에서 결과가 반환되지 않으면 쿼리 범위를 확장하세요.
            # 검색 결과가 비어 있는 경우 포기하기 전에 검색 범위를 확장하세요.
            # 현재 사용자 항공편 정보:\n<Flights>\n{user_info}\n</Flights>
            # 현재 시간: {time}.
        ),
        ("placeholder", "{messages}")
    ]
).partial(time = datetime.now)

@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])

primaryAssistantToolList = [
    TavilySearchResults(max_results = 1),
    searchFlightList,
    lookupPolicy
]

class FlightBookingAssistantModel(BaseModel):
    """Transfers work to a specialized assistant to handle flight updates and cancellations."""
    # 항공편 업데이트 및 취소를 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

    request : str = Field(description = "Any necessary followup questions the update flight assistant should clarify before proceeding.") # 항공편 업데이트 어시스턴트가 처리전 명확히 해야할 필요한 후속 질문

class CarRentalBookingAssistantModel(BaseModel):
    """Transfers work to a specialized assistant to handle car rental bookings."""
    # 렌터카 예약을 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

    location   : str = Field(description = "The location where the user wants to rent a car.") # 사용자가 자동차를 렌트하고자 하는 위치
    start_date : str = Field(description = "The start date of the car rental.") # 차량 렌탈 시작 날짜
    end_date   : str = Field(description = "The end date of the car rental."  ) # 차량 렌탈 종료 날짜
    request    : str = Field(description = "Any additional information or requests from the user regarding the car rental.") # 자동차 렌탈 예약과 관련하여 사용자로부터 추가 정보나 요청 사항

    class Config:
        json_schema_extra = {
            "example" : {
                "location"   : "Basel",
                "start_date" : "2023-07-01",
                "end_date"   : "2023-07-05",
                "request"    : "I need a compact car with automatic transmission."
            }
        }

class HotelBookingAssistantModel(BaseModel):
    """Transfer work to a specialized assistant to handle hotel bookings."""
    # 호텔 예약을 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

    location      : str = Field(description = "The location where the user wants to book a hotel.") # 사용자가 호텔을 예약하고자 하는 위치이다.
    checkin_date  : str = Field(description = "The check-in date for the hotel." ) # 호텔 체크인 날짜
    checkout_date : str = Field(description = "The check-out date for the hotel.") # 호텔 체크아웃 날짜
    request       : str = Field(description = "Any additional information or requests from the user regarding the hotel booking.") # 호텔 예약과 관련하여 사용자로부터 추가 정보나 요청 사항

    class Config:
        json_schema_extra = {
            "example" : {
                "location"      : "Zurich", # 취리히
                "checkin_date"  : "2023-08-15",
                "checkout_date" : "2023-08-20",
                "request"       : "I prefer a hotel near the city center with a room that has a view." # 저는 전망이 좋은 객실이 있는 시내 중심가 근처의 호텔을 선호한다.
            }
        }

class ExcursionBookingAssistantModel(BaseModel):
    """Transfers work to a specialized assistant to handle trip recommendation and other excursion bookings."""
    # 여행 추천 및 기타 여행 예약을 담당하는 전문 어시스턴트에게 업무를 이관한다.

    location : str = Field(description = "The location where the user wants to book a recommended trip.") # 사용자가 추천 여행을 예약하고자 하는 위치이다.
    request  : str = Field(description = "Any additional information or requests from the user regarding the trip recommendation.") # 여행 추천과 관련하여 사용자로부터 추가 정보나 요청 사항

    class Config:
        json_schema_extra = {
            "example" : {
                "location" : "Lucerne", # 루체른
                "request"  : "The user is interested in outdoor activities and scenic views." # 사용자는 야외 활동과 경치 좋은 풍경에 관심이 있다.
            }
        }

primaryAssistantRunnableSequence = primaryAssistantChatPromptTemplate | chatOpenAI.bind_tools(primaryAssistantToolList + [FlightBookingAssistantModel, CarRentalBookingAssistantModel, HotelBookingAssistantModel, ExcursionBookingAssistantModel])

def routePrimaryAssistantNode(state : State):
    routingNode = tools_condition(state)
    if routingNode == END:
        return END
    toolCallList = state["messages"][-1].tool_calls
    if toolCallList:
        if toolCallList[0]["name"] == FlightBookingAssistantModel.__name__:
            return "flight_booking_entry_node"
        elif toolCallList[0]["name"] == CarRentalBookingAssistantModel.__name__:
            return "car_rental_booking_entry_node"
        elif toolCallList[0]["name"] == HotelBookingAssistantModel.__name__:
            return "hotel_booking_entry_node"
        elif toolCallList[0]["name"] == ExcursionBookingAssistantModel.__name__:
            return "excursion_booking_entry_node"
        return "primary_assistant_tool_node"
    raise ValueError("Invalid route")

stateGraph.add_node("primary_assistant_node"     , Assistant(primaryAssistantRunnableSequence))
stateGraph.add_node("primary_assistant_tool_node", ToolNode(primaryAssistantToolList).with_fallbacks([RunnableLambda(handleToolError)], exception_key = "error"))

# 어시스턴트는 위임된 어시스턴트 중 한 명에게 라우팅하거나 도구를 직접 사용하거나 사용자에게 직접 응답할 수 있다.
stateGraph.add_conditional_edges("primary_assistant_node", routePrimaryAssistantNode, ["flight_booking_entry_node", "car_rental_booking_entry_node", "hotel_booking_entry_node", "excursion_booking_entry_node", "primary_assistant_tool_node", END])
stateGraph.add_edge("primary_assistant_tool_node", "primary_assistant_node")

# 스킬 이탈 노드
# 이 노드는 모든 전문화된 어시스턴트를 종료하기 위해 공유된다.
def popDialogState(state : State) -> dict:
    """Pop the dialog stack and return to the main assistant.

    This lets the full graph explicitly track the dialog flow and delegate control
    to specific sub-graphs.
    """
    messageList = []
    if state["messages"][-1].tool_calls:
        # 참고 : 현재 LLM이 병렬 도구 호출을 수행하는 에지 케이스를 처리하지 않는다.
        messageList.append(
            ToolMessage(
                content      = "Resuming dialog with the host assistant. Please reflect on the past conversation and assist the user as needed.",
                tool_call_id = state["messages"][-1].tool_calls[0]["id"]
            )
        )
    return {
        "dialog_state_list" : "pop",
        "messages"          : messageList
    }

stateGraph.add_node("leave_skill_node", popDialogState)

stateGraph.add_edge("leave_skill_node", "primary_assistant_node")

# 컴파일 상태 그래프 객체를 만든다.
memorySaver = MemorySaver()

compiledStateGraph = stateGraph.compile(
    checkpointer = memorySaver,
    # 사용자가 민감한 도구 사용을 승인하거나 거부하도록 한다.
    interrupt_before = [
        "flight_booking_sensitive_tool_node",
        "car_rental_booking_sensitive_tool_node",
        "hotel_booking_sensitive_tool_node",
        "excursion_booking_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."
    # "안녕하세요, 제 항공편은 몇 시인가요?"
    # "항공편을 좀 더 일찍 예약해도 될까요? 오늘 늦게 출발하고 싶어요."
    # "그럼 다음 주 어느 때로 항공편을 예약하세요"
    # "다음 가능한 옵션이 좋습니다"
    # "숙박과 교통은 어때요?"
    # "네, 일주일(7일) 동안 머물기 위해 저렴한 호텔을 원합니다. 그리고 차를 렌트하고 싶어요."
    # "좋아요, 추천 호텔을 예약해 주시겠어요? 좋은 것 같아요."
    # "네, 적당한 가격대에 예약 가능한 곳이면 무엇이든 예약하세요."
    # "이제 차는 어떤 옵션이 있나요?"
    # "좋아요, 가장 저렴한 옵션을 선택해 볼까요. 7일 동안 예약하세요"
    # "좋네요. 그럼 여행에 대한 추천이 있나요?"
    # "제가 거기에 있는 동안 이용할 수 있나요?"
    # "흥미롭네요. 박물관이 마음에 들어요. 어떤 옵션이 있나요?"
    # "좋아요, 하나를 골라서 둘째 날 예약하세요."
]

printedSet = set()

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

def printEvent(addableUpdatesDict : dict, printedSet : set, maximumLength = 1500):
    currentStateList = addableUpdatesDict.get("dialog_state_list")
    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

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

1001

1002

1003

1004

1005

1006

1007

1008

1009

1010

1011

1012

1013

1014

1015

1016

1017

1018

1019

1020

1021

1022

1023

1024

1025

1026

1027

1028

1029

1030

1031

1032

1033

1034

1035

1036

1037

1038

1039

1040

1041

1042

1043

1044

1045

1046

1047

1048

1049

1050

1051

1052

1053

1054

1055

1056

1057

1058

1059

1060

1061

1062

1063

1064

1065

1066

1067

1068

1069

1070

1071

1072

1073

1074

1075

1076

1077

1078

1079

1080

1081

1082

1083

1084

1085

1086

1087

1088

1089

1090

1091

1092

1093

1094

1095

1096

1097

1098

1099

1100

1101

1102

1103

1104

1105

1106

1107

1108

1109

1110

1111

1112

1113

1114

1115

1116

1117

1118

1119

1120

1121

1122

1123

1124

1125

1126

1127

1128

1129

1130

1131

1132

1133

1134

1135

1136

1137

1138

1139

1140

1141

1142

1143

1144

1145

1146

1147

1148

1149

1150

1151

1152

1153

1154

1155

1156

1157

1158

1159

1160

1161

1162

1163

1164

1165

1166

1167

1168

1169

1170

1171

1172

1173

1174

1175

1176

1177

1178

1179

1180

1181

1182

1183

1184

1185

1186

1187

1188

1189

1190

1191

1192

1193

1194

1195

1196

1197

1198

1199

1200

1201

1202

1203

1204

1205

1206

1207

1208

1209

1210

1211

1212

1213

1214

1215

1216

1217

1218

1219

1220

1221

1222

1223

1224

1225

1226

1227

1228

1229

1230

1231

1232

1233

1234

1235

1236

1237

1238

1239

1240

1241

1242

1243

1244

1245

1246

1247

1248

1249

1250

1251

1252

1253

1254

1255

1256

1257

1258

1259

1260

1261

1262

1263

1264

1265

1266

1267

1268

1269

1270

1271

1272

1273

1274

1275

1276

1277

1278

1279

1280

1281

import os

import requests

import shutil

import sqlite3

import pandas as pd

import re

import numpy as np

import pytz

import openai

import uuid

from dotenv import load_dotenv

from typing import Optional

from typing_extensions import TypedDict

from typing import Annotated

from langgraph.graph.message import AnyMessage

from langgraph.graph.message import add_messages

from typing import Literal

from langchain_core.runnables import Runnable

from langchain_openai import ChatOpenAI

from pydantic import BaseModel

from typing import Callable

from langchain_core.messages import ToolMessage

from langgraph.graph import StateGraph

from langchain_core.tools import tool

from langchain_core.runnables import RunnableConfig

from langgraph.graph import START

from langchain_core.prompts import ChatPromptTemplate

from datetime import date

from datetime import datetime

from langgraph.prebuilt import tools_condition

from langgraph.graph import END

from langgraph.prebuilt import ToolNode

from langchain_core.runnables import RunnableLambda

from typing import Union

from langchain_community.tools.tavily_search import TavilySearchResults

from pydantic import Field

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)

def updateDialogStateList(targetList : list[str], text : Optional[str]) -> list[str]:

"""Push or pop the state."""

# "상태를 넣거나 꺼낸다.

if text is None:

return targetList

if text == "pop":

return targetList[:-1]

return targetList + [text]

class State(TypedDict):

messages : Annotated[list[AnyMessage], add_messages]

user_info : str

dialog_state_list : Annotated[list[Literal["primary_assistant_node", "flight_booking_assistant_node", "car_rental_booking_assistant_node", "hotel_booking_assistant_node", "excursion_booking_assistant_node"]], updateDialogStateList]

class Assistant:

def __init__(self, runnable : Runnable):

self.runnable = runnable

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

while True:

responseAIMessage = self.runnable.invoke(state)

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}

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

class CompleteOrEscalate(BaseModel):

"""A tool to mark the current task as completed and/or to escalate control of the dialog to the main assistant, who can re-route the dialog based on the user's needs."""

# 현재 작업을 완료로 표시하거나 대화 제어권을 주요 어시스턴에게 위임하여, 사용자의 필요에 따라 대화 경로를 변경할 수 있는 도구이다.

cancel : bool = True

reason : str

class Config:

json_schema_extra = {

"example 1" : {

"cancel" : True,

"reason" : "User changed their mind about the current task." # 사용자가 현재 작업에 대한 마음을 바꾸었다.

"example 2" : {

"cancel" : True,

"reason" : "I have fully completed the task." # 나는 그 임무를 완전히 완수했다.

"example 3" : {

"cancel" : False,

"reason" : "I need to search the user's emails or calendar for more information." # 자세한 정보를 알아보기 위해 사용자의 이메일이나 캘린더를 검색해야 한다.

}

def createEntryNode(assistantName : str, newDialogState : str) -> Callable:

def entry_node(state : State) -> dict:

toolCallID = state["messages"][-1].tool_calls[0]["id"]

return {

"messages" : [

ToolMessage(

content = f"The assistant is now the {assistantName}. Reflect on the above conversation between the host assistant and the user."

f" The user's intent is unsatisfied. Use the provided tools to assist the user. Remember, you are {assistantName},"

" and the booking, update, other other action is not complete until after you have successfully invoked the appropriate tool."

" If the user changes their mind or needs help for other tasks, call the CompleteOrEscalate function to let the primary host assistant take control."

" Do not mention who you are - just act as the proxy for the assistant.",

# 이제 어시스턴트는 {assistantName}입니다. 호스트 어시스턴트와 사용자 간의 위의 대화를 생각해 본다.

# 사용자의 의도가 충족되지 않았다.

# 제공된 도구를 사용하여 사용자를 지원한다.

# 기억한다, 당신은 {assistantName}이고, 예약, 업데이트, 기타 다른 작업은 적절한 도구를 성공적으로 호출한 후에야 완료된다.

# 사용자가 마음을 바꾸거나 다른 작업에 대한 도움이 필요한 경우 CompleteOrEscalate 함수를 호출하여 기본 호스트 지원자가 제어권을 갖도록 한다.

# 자신이 누구인지 밝히지 않는다.

# 그저 이시스턴트의 대리인으로 행동한다.

tool_call_id = toolCallID

)

"dialog_state_list" : newDialogState

}

return entry_node

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)

# 사용자 정보 가져오기 노드

@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

def fetchUserInfo(state : State):

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

# 위임된 각 워크플로는 사용자에게 직접 응답할 수 있다.

# 사용자가 응답하면 현재 활성화된 워크플로로 돌아가기를 원한다.

def routeToWorkflow(state : State) -> Literal["primary_assistant_node", "flight_booking_assistant_node", "car_rental_booking_assistant_node", "hotel_booking_assistant_node", "excursion_booking_assistant_node"]:

"""If we are in a delegated state, route directly to the appropriate assistant."""

# 위임 상태인 경우 해당 보조자에게 직접 전달한다.

dialogStateList = state.get("dialog_state_list")

if not dialogStateList:

return "primary_assistant_node"

return dialogStateList[-1]

stateGraph.add_node("fetch_user_info_node", fetchUserInfo)

stateGraph.add_edge(START, "fetch_user_info_node")

stateGraph.add_conditional_edges("fetch_user_info_node", routeToWorkflow)

# 항공편 예약 어시스턴트 노드

flightBookingChatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"You are a specialized assistant for handling flight updates. "

"The primary assistant delegates work to you whenever the user needs help updating their bookings. "

"Confirm the updated flight details with the customer and inform them of any additional fees. "

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

"If you need more information or the customer changes their mind, escalate the task back to the main assistant. "

"Remember that a booking isn't completed until after the relevant tool has successfully been used. "

"\n\nCurrent user flight information :\n<Flights>\n{user_info}\n</Flights>"

"\nCurrent time : {time}."

"\n\nIf the user needs help, and none of your tools are appropriate for it, then "

'"CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'

# 당신은 항공편 업데이트를 처리하는 전문 어시스턴트이다.

# 주요 어시스턴트는 사용자가 예약을 업데이트하는 데 도움이 필요할 때마다 사용자를 대신하여 작업한다.

# 고객에게 업데이트된 항공편 세부 정보를 확인하고 추가 비용이 있는 경우 이를 알려준다.

# 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.

# 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.

# 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.

# 현재 사용자 항공편 정보 :\n<Flights>\n{user_info}\n</Flights>

# 현재 시간 : {time}.

# 사용자에게 도움이 필요하고 도구 중 어느 것도 적합하지 않은 경우, 호스트 어시스턴트에게 대화를 "CompleteOrEscalate"합니다.

# 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.

("placeholder", "{messages}")

]

).partial(time = datetime.now)

@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."

flightBookingSafeToolList = [searchFlightList]

flightBookingSensitiveToolList = [updateTicketToNewFlight, cancelTicket]

flightBookingToolList = flightBookingSafeToolList + flightBookingSensitiveToolList

flightBookinRunnableSequence = flightBookingChatPromptTemplate | chatOpenAI.bind_tools(flightBookingToolList + [CompleteOrEscalate])

def routeFlightBookingNode(state : State):

routingNode = tools_condition(state)

if routingNode == END:

return END

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

alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)

if alreadyCanceled:

return "leave_skill_node"

safeToolNameList = [tool.name for tool in flightBookingSafeToolList]

if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):

return "flight_booking_safe_tool_node"

return "flight_booking_sensitive_tool_node"

stateGraph.add_node("flight_booking_entry_node" , createEntryNode("Flight Updates & Booking Assistant", "flight_booking_assistant_node"))

stateGraph.add_node("flight_booking_assistant_node" , Assistant(flightBookinRunnableSequence))

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

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

stateGraph.add_edge("flight_booking_entry_node" , "flight_booking_assistant_node")

stateGraph.add_conditional_edges("flight_booking_assistant_node", routeFlightBookingNode, ["flight_booking_safe_tool_node", "flight_booking_sensitive_tool_node", "leave_skill_node", END])

stateGraph.add_edge("flight_booking_safe_tool_node" , "flight_booking_assistant_node")

stateGraph.add_edge("flight_booking_sensitive_tool_node", "flight_booking_assistant_node")

# 자동차 렌탈 예약 어시스턴트 노드

carRentalBookingChatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"You are a specialized assistant for handling car rental bookings. "

"The primary assistant delegates work to you whenever the user needs help booking a car rental. "

"Search for available car rentals based on the user's preferences and confirm the booking details with the customer. "

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

"If you need more information or the customer changes their mind, escalate the task back to the main assistant. "

"Remember that a booking isn't completed until after the relevant tool has successfully been used. "

"\nCurrent time : {time}. "

"\n\nIf the user needs help, and none of your tools are appropriate for it, then "

'"CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'

"\n\nSome examples for which you should CompleteOrEscalate :\n"

" - 'what's the weather like this time of year?'\n"

" - 'What flights are available?'\n"

" - 'nevermind i think I'll book separately'\n"

" - 'Oh wait i haven't booked my flight yet i'll do that first'\n"

" - 'Car rental booking confirmed'"

# 귀하는 자동차 렌탈 예약을 처리하는 전문 어시스턴트이다.

# 주요 어시스턴트는 사용자가 렌터카 예약에 도움이 필요할 때마다 도움을 준다.

# 사용자의 선호도에 따라 이용 가능한 렌터카를 검색하고 고객과 예약 세부 사항을 확인한다.

# 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.

# 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.

# 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.

# 현재 시간 : {time}.

# 사용자에게 도움이 필요하고 도구 중 어느 것도 적합하지 않은 경우, 호스트 어시스턴트에게 대화를 "CompleteOrEscalate"한다.

# 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.

# CompleteOrEscalate해야 하는 몇 가지 예 :

# - '이번 시기에 날씨가 어때요?'

# - '어떤 항공편이 가능한가요?'

# - '괜찮아요. 따로 예약할게요'

# - '아, 잠깐만요. 아직 비행기 예약을 안 했어요. 먼저 예약할게요.'

# - '렌터카 예약 확인'

("placeholder", "{messages}")

]

).partial(time = datetime.now)

@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.

"""

# 위치, 이름, 가격 등급, 시작 날짜, 종료 날짜를 기준으로 렌터카를 검색한다.

# 인자 :

# location (Optional[str]) : 렌터카 위치. 기본값은 없음이다.

# 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] : 검색 기준과 일치하는 자동차 렌탈 사전 목록이다.

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.

"""

# 차량 ID로 렌터카를 예약한다.

# 인자 :

# rental_id (int) : 예약할 렌터카의 ID이다.

# 반환 :

# str : 자동차 렌탈이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.

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.

"""

# ID를 사용하여 자동차 렌탈 시작 및 종료 날짜를 업데이트한다.

# 인자 :

# rental_id (int) : 업데이트할 렌터카 ID이다.

# start_date (Optional[Union[datetime, date]]) : 자동차 렌탈의 새로운 시작 날짜. 기본값은 None이다.

# end_date (Optional[Union[datetime, date]]) : 자동차 렌탈의 새로운 종료 날짜. 기본값은 None이다.

# 반환 :

# str : 렌터카가 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.

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.

"""

# 차량 렌탈을 ID로 취소한다.

# 인자 :

# rental_id (int) : 취소할 렌터카의 ID이다.

# 반환 :

# str : 차량 렌탈이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.

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}."

carRentalBookingSafeToolList = [searchCarRentalList]

carRentalBookingSensitiveToolList = [bookCarRental, updateCarRental, cancelCarRental]

carRentalBookingToolList = carRentalBookingSafeToolList + carRentalBookingSensitiveToolList

carRentalBookingRunnableSequence = carRentalBookingChatPromptTemplate | chatOpenAI.bind_tools(carRentalBookingToolList + [CompleteOrEscalate])

def routeCarRentalBookingNode(state : State):

routingNode = tools_condition(state)

if routingNode == END:

return END

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

alreadyCanceled = any(tollCall["name"] == CompleteOrEscalate.__name__ for tollCall in toolCallList)

if alreadyCanceled:

return "leave_skill_node"

safeToolNameList = [tool.name for tool in carRentalBookingSafeToolList]

if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):

return "car_rental_booking_safe_tool_node"

return "car_rental_booking_sensitive_tool_node"

stateGraph.add_node("car_rental_booking_entry_node" , createEntryNode("Car Rental Assistant", "car_rental_booking_assistant_node"))

stateGraph.add_node("car_rental_booking_assistant_node" , Assistant(carRentalBookingRunnableSequence))

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

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

stateGraph.add_edge("car_rental_booking_entry_node" , "car_rental_booking_assistant_node")

stateGraph.add_conditional_edges("car_rental_booking_assistant_node", routeCarRentalBookingNode, ["car_rental_booking_safe_tool_node", "car_rental_booking_sensitive_tool_node", "leave_skill_node", END])

stateGraph.add_edge("car_rental_booking_safe_tool_node" , "car_rental_booking_assistant_node")

stateGraph.add_edge("car_rental_booking_sensitive_tool_node", "car_rental_booking_assistant_node")

# 호텔 예약 어시스턴트 노드

hotelBookingChatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"You are a specialized assistant for handling hotel bookings. "

"The primary assistant delegates work to you whenever the user needs help booking a hotel. "

"Search for available hotels based on the user's preferences and confirm the booking details with the customer. "

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

"If you need more information or the customer changes their mind, escalate the task back to the main assistant. "

"Remember that a booking isn't completed until after the relevant tool has successfully been used. "

"\nCurrent time : {time}."

'\n\nIf the user needs help, and none of your tools are appropriate for it, then "CompleteOrEscalate" the dialog to the host assistant. '

"Do not waste the user's time. Do not make up invalid tools or functions."

"\n\nSome examples for which you should CompleteOrEscalate :\n"

" - 'what's the weather like this time of year?'\n"

" - 'nevermind i think I'll book separately'\n"

" - 'i need to figure out transportation while i'm there'\n"

" - 'Oh wait i haven't booked my flight yet i'll do that first'\n"

" - 'Hotel booking confirmed'"

# 당신은 호텔 예약을 처리하는 전문 어시스턴트이다.

# 주요 어시스턴트는 사용자가 호텔 예약에 도움이 필요할 때마다 도움을 준다.

# 사용자의 선호도에 따라 이용 가능한 호텔을 검색하고 고객과 예약 세부 정보를 확인한다.

# 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 나오지 않으면 쿼리 범위를 확장한다.

# 더 많은 정보가 필요하거나 고객이 마음을 바꾸면 해당 작업을 메인 어시스턴트에게 다시 전달한다.

# 해당 도구를 성공적으로 사용한 후에야 예약이 완료된다.

# 현재 시간 : {time}.

# 사용자에게 도움이 필요하고 어떤 도구도 이에 적합하지 않은 경우, 호스트 지원팀에 대화를 "완료 또는 에스컬레이션"한다.

# 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들지 않는다.

# CompleteOrEscalate를 수행해야 하는 몇 가지 예 :

# - '이번 시기에 날씨가 어때요?'

# - '괜찮아요. 따로 예약할게요'

# - '내가 거기에 있는 동안 교통수단을 알아내야 해'

# - '아, 잠깐만요. 아직 비행기 예약을 안 했어요. 먼저 예약할게요.'

# - '호텔 예약이 확정되었습니다'

("placeholder", "{messages}")

]

).partial(time = datetime.now)

@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.

"""

# 위치, 이름, 가격, 체크인 날짜, 체크아웃 날짜를 기준으로 호텔을 검색한다.

# 인자 :

# location (Optional[str]) : 호텔의 위치. 기본값은 None이다.

# name (Optional[str]) : 호텔 이름. 기본값은 None이다.

# price_tier (Optional[str]) : 호텔의 가격 등급이다. 기본값은 None이다. 예 : Midscale, Upper Midscale, Upscale, Luxury

# checkin_date (Optional[Union[datetime, date]]) : 호텔의 체크인 날짜. 기본값은 None이다.

# checkout_date (Optional[Union[datetime, date]]) : 호텔의 체크아웃 날짜. 기본값은 None이다.

# 반환 :

# list[dict] : 검색 기준과 일치하는 호텔 사전 목록이다.

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.

"""

# 호텔 ID로 호텔을 예약한다.

# 인자 :

# hotel_id (int) : 예약할 호텔의 ID이다.

# 반환 :

# str : 호텔이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.

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.

"""

# 호텔 ID를 사용하여 호텔의 체크인 및 체크아웃 날짜를 업데이트한다.

# 인자 :

# hotel_id (int) : 업데이트할 호텔의 ID이다.

# checkin_date (Optional[Union[datetime, date]]) : 호텔의 새로운 체크인 날짜. 기본값은 None이다.

# checkout_date (Optional[Union[datetime, date]]) : 호텔의 새로운 체크아웃 날짜. 기본값은 None이다.

# 반환 :

# str : 호텔이 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.

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.

"""

# 호텔 ID로 호텔 취소한다.

# 인자 :

# hotel_id (int) : 취소할 호텔의 ID이다.

# 반환 :

# str : 호텔이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.

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}."

hotelBookingSafeToolList = [searchHotelList]

hotelBookingSensitiveToolList = [bookHotel, updateHotel, cancelHotel]

hotelBookingToolList = hotelBookingSafeToolList + hotelBookingSensitiveToolList

hotelBookingRunnableSequence = hotelBookingChatPromptTemplate | chatOpenAI.bind_tools(hotelBookingToolList + [CompleteOrEscalate])

def routeHotelBookingNode(state : State):

routingNode = tools_condition(state)

if routingNode == END:

return END

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

alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)

if alreadyCanceled:

return "leave_skill_node"

safeToolNameList = [t.name for t in hotelBookingSafeToolList]

if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):

return "hotel_booking_safe_tool_node"

return "hotel_booking_sensitive_tool_node"

stateGraph.add_node("hotel_booking_entry_node" , createEntryNode("Hotel Booking Assistant", "hotel_booking_assistant_node"))

stateGraph.add_node("hotel_booking_assistant_node" , Assistant(hotelBookingRunnableSequence))

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

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

stateGraph.add_edge("hotel_booking_entry_node" , "hotel_booking_assistant_node")

stateGraph.add_conditional_edges("hotel_booking_assistant_node", routeHotelBookingNode, ["hotel_booking_safe_tool_node", "hotel_booking_sensitive_tool_node", "leave_skill_node", END])

stateGraph.add_edge("hotel_booking_safe_tool_node" , "hotel_booking_assistant_node")

stateGraph.add_edge("hotel_booking_sensitive_tool_node", "hotel_booking_assistant_node")

# 여행 예약 어시스턴트 노드

excursionBookingChatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"You are a specialized assistant for handling trip recommendations. "

"The primary assistant delegates work to you whenever the user needs help booking a recommended trip. "

"Search for available trip recommendations based on the user's preferences and confirm the booking details with the customer. "

"If you need more information or the customer changes their mind, escalate the task back to the main assistant. "

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

"Remember that a booking isn't completed until after the relevant tool has successfully been used. "

"\nCurrent time : {time}."

'\n\nIf the user needs help, and none of your tools are appropriate for it, then "CompleteOrEscalate" the dialog to the host assistant. Do not waste the user\'s time. Do not make up invalid tools or functions.'

"\n\nSome examples for which you should CompleteOrEscalate :\n"

" - 'nevermind i think I'll book separately'\n"

" - 'i need to figure out transportation while i'm there'\n"

" - 'Oh wait i haven't booked my flight yet i'll do that first'\n"

" - 'Excursion booking confirmed!'"

# 여행 추천을 처리하는 전문 어시스턴트이다.

# 주요 어시스턴트는 사용자가 추천된 여행을 예약하는 데 도움이 필요할 때마다 작업을 위임한다.

# 사용자의 선호도에 따라 이용 가능한 여행 추천을 검색하고 고객과 예약 세부 정보를 확인한다.

# 추가 정보가 필요하거나 고객이 마음을 바꾸면 작업을 다시 주요 보조원에게 에스컬레이션한다.

# 검색할 때는 끈기 있게 노력한다. 첫 번째 검색에서 결과가 반환되지 않으면 쿼리 범위를 확장한다.

# 관련 도구를 성공적으로 사용한 후에야 예약이 완료된다.

# 현재 시간 : {time}.

# 사용자에게 도움이 필요하고 도구 중 적합한 도구가 없는 경우 호스트 보조원에게 대화를 "CompleteOrEscalate"한다.

# 사용자의 시간을 낭비하지 않는다. 잘못된 도구나 기능을 만들어내지 않는다.

# CompleteOrEscalate해야 하는 몇 가지 예:

# - '괜찮아요. 따로 예약할 것 같아요'

# - '거기에 있는 동안 교통수단을 알아내야 해요'

# - '아, 아직 항공편을 예약하지 않았네요. 먼저 예약할게요'

# - '여행 예약이 확정되었습니다!'

("placeholder", "{messages}")

]

).partial(time = datetime.now)

@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.

"""

# 위치, 이름, 키워드를 기반으로 여행 추천을 검색한다.

# 인자 :

# location (Optional[str]) : 여행 추천의 위치이다. 기본값은 None이다.

# name (Optional[str]) : 여행 추천의 이름이다. 기본값은 None이다.

# keywords (Optional[str]) : 여행 추천과 관련된 키워드. 기본값은 None이다.

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.

"""

# 추천 ID로 여행을 예약한다.

# 인자 :

# recommendation_id (int) : 예약을 위한 여행 추천 ID이다.

# 반환 :

# str : 여행 추천이 성공적으로 예약되었는지 여부를 나타내는 메시지이다.

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.

"""

# ID를 사용하여 여행 추천 세부 정보를 업데이트한다.

# 인자 :

# recommendation_id (int) : 업데이트할 여행 추천의 ID이다.

# details (str) : 여행 추천의 새로운 세부 정보.

# 반환 :

# str : 여행 추천이 성공적으로 업데이트되었는지 여부를 나타내는 메시지이다.

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.

"""

# ID로 여행 추천을 취소한다.

# 인자 :

# recommendation_id (int) : 취소할 여행 추천의 ID이다.

# 반환 :

# str : 여행 추천이 성공적으로 취소되었는지 여부를 나타내는 메시지이다.

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}."

excursionBookingSafeToolList = [searchTripRecommendationList]

excursionBookingSensitiveToolList = [bookExcursion, updateExcursion, cancelExcursion]

excursionBookingToolList = excursionBookingSafeToolList + excursionBookingSensitiveToolList

excursionBookingRunnableSequence = excursionBookingChatPromptTemplate | chatOpenAI.bind_tools(excursionBookingToolList + [CompleteOrEscalate])

def routeExcursionBookingNode(state : State):

routingNode = tools_condition(state)

if routingNode == END:

return END

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

alreadyCanceled = any(toolCall["name"] == CompleteOrEscalate.__name__ for toolCall in toolCallList)

if alreadyCanceled:

return "leave_skill_node"

safeToolNameList = [tool.name for tool in excursionBookingSafeToolList]

if all(toolCall["name"] in safeToolNameList for toolCall in toolCallList):

return "excursion_booking_safe_tool_node"

return "excursion_booking_sensitive_tool_node"

stateGraph.add_node("excursion_booking_entry_node" , createEntryNode("Trip Recommendation Assistant", "excursion_booking_assistant_node"))

stateGraph.add_node("excursion_booking_assistant_node" , Assistant(excursionBookingRunnableSequence))

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

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

stateGraph.add_edge("excursion_booking_entry_node" , "excursion_booking_assistant_node")

stateGraph.add_conditional_edges("excursion_booking_assistant_node", routeExcursionBookingNode, ["excursion_booking_safe_tool_node", "excursion_booking_sensitive_tool_node", "leave_skill_node", END])

stateGraph.add_edge("excursion_booking_safe_tool_node" , "excursion_booking_assistant_node")

stateGraph.add_edge("excursion_booking_sensitive_tool_node", "excursion_booking_assistant_node")

# 메인 어시스턴트 노드

primaryAssistantChatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

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

"Your primary role is to search for flight information and company policies to answer customer queries. "

"If a customer requests to update or cancel a flight, book a car rental, book a hotel, or get trip recommendations, "

"delegate the task to the appropriate specialized assistant by invoking the corresponding tool. You are not able to make these types of changes yourself."

" Only the specialized assistants are given permission to do this for the user."

"The user is not aware of the different specialized assistants, so do not mention them; just quietly delegate through function calls. "

"Provide detailed information to the customer, and always double-check the database before concluding that information is unavailable. "

" 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 flight information :\n<Flights>\n{user_info}\n</Flights>"

"\nCurrent time : {time}."

# Swiss Airlines의 유용한 고객 지원 어시슽입니다.

# 주요 역할은 고객 문의에 답하기 위해 항공편 정보와 회사 정책을 검색하는 것입니다.

# 고객이 항공편을 업데이트하거나 취소하거나, 렌터카를 예약하거나, 호텔을 예약하거나, 여행 추천을 받으려고 요청하는 경우

# 해당 도구를 호출하여 해당 전문 지원 담당자에게 작업을 위임합니다. 직접 이러한 유형의 변경을 할 수는 없습니다.

# 전문 지원 담당자에게만 사용자를 대신하여 이 작업을 수행할 권한이 부여됩니다.

# 사용자는 다양한 전문 지원 담당자를 알지 못하므로 언급하지 말고 함수 호출을 통해 조용히 위임하세요.

# 고객에게 자세한 정보를 제공하고 정보를 사용할 수 없다는 결론을 내리기 전에 항상 데이터베이스를 다시 확인하세요.

# 검색할 때는 끈기 있게 노력하세요. 첫 번째 검색에서 결과가 반환되지 않으면 쿼리 범위를 확장하세요.

# 검색 결과가 비어 있는 경우 포기하기 전에 검색 범위를 확장하세요.

# 현재 사용자 항공편 정보:\n<Flights>\n{user_info}\n</Flights>

# 현재 시간: {time}.

("placeholder", "{messages}")

]

).partial(time = datetime.now)

@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])

primaryAssistantToolList = [

TavilySearchResults(max_results = 1),

searchFlightList,

lookupPolicy

]

class FlightBookingAssistantModel(BaseModel):

"""Transfers work to a specialized assistant to handle flight updates and cancellations."""

# 항공편 업데이트 및 취소를 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

request : str = Field(description = "Any necessary followup questions the update flight assistant should clarify before proceeding.") # 항공편 업데이트 어시스턴트가 처리전 명확히 해야할 필요한 후속 질문

class CarRentalBookingAssistantModel(BaseModel):

"""Transfers work to a specialized assistant to handle car rental bookings."""

# 렌터카 예약을 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

location : str = Field(description = "The location where the user wants to rent a car.") # 사용자가 자동차를 렌트하고자 하는 위치

start_date : str = Field(description = "The start date of the car rental.") # 차량 렌탈 시작 날짜

end_date : str = Field(description = "The end date of the car rental." ) # 차량 렌탈 종료 날짜

request : str = Field(description = "Any additional information or requests from the user regarding the car rental.") # 자동차 렌탈 예약과 관련하여 사용자로부터 추가 정보나 요청 사항

class Config:

json_schema_extra = {

"example" : {

"location" : "Basel",

"start_date" : "2023-07-01",

"end_date" : "2023-07-05",

"request" : "I need a compact car with automatic transmission."

}

class HotelBookingAssistantModel(BaseModel):

"""Transfer work to a specialized assistant to handle hotel bookings."""

# 호텔 예약을 처리하기 위해 전문 어시스턴트에게 업무를 이관한다.

location : str = Field(description = "The location where the user wants to book a hotel.") # 사용자가 호텔을 예약하고자 하는 위치이다.

checkin_date : str = Field(description = "The check-in date for the hotel." ) # 호텔 체크인 날짜

checkout_date : str = Field(description = "The check-out date for the hotel.") # 호텔 체크아웃 날짜

request : str = Field(description = "Any additional information or requests from the user regarding the hotel booking.") # 호텔 예약과 관련하여 사용자로부터 추가 정보나 요청 사항

class Config:

json_schema_extra = {

"example" : {

"location" : "Zurich", # 취리히

"checkin_date" : "2023-08-15",

"checkout_date" : "2023-08-20",

"request" : "I prefer a hotel near the city center with a room that has a view." # 저는 전망이 좋은 객실이 있는 시내 중심가 근처의 호텔을 선호한다.

}

class ExcursionBookingAssistantModel(BaseModel):

"""Transfers work to a specialized assistant to handle trip recommendation and other excursion bookings."""

# 여행 추천 및 기타 여행 예약을 담당하는 전문 어시스턴트에게 업무를 이관한다.

location : str = Field(description = "The location where the user wants to book a recommended trip.") # 사용자가 추천 여행을 예약하고자 하는 위치이다.

request : str = Field(description = "Any additional information or requests from the user regarding the trip recommendation.") # 여행 추천과 관련하여 사용자로부터 추가 정보나 요청 사항

class Config:

json_schema_extra = {

"example" : {

"location" : "Lucerne", # 루체른

"request" : "The user is interested in outdoor activities and scenic views." # 사용자는 야외 활동과 경치 좋은 풍경에 관심이 있다.

}

primaryAssistantRunnableSequence = primaryAssistantChatPromptTemplate | chatOpenAI.bind_tools(primaryAssistantToolList + [FlightBookingAssistantModel, CarRentalBookingAssistantModel, HotelBookingAssistantModel, ExcursionBookingAssistantModel])

def routePrimaryAssistantNode(state : State):

routingNode = tools_condition(state)

if routingNode == END:

return END

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

if toolCallList:

if toolCallList[0]["name"] == FlightBookingAssistantModel.__name__:

return "flight_booking_entry_node"

elif toolCallList[0]["name"] == CarRentalBookingAssistantModel.__name__:

return "car_rental_booking_entry_node"

elif toolCallList[0]["name"] == HotelBookingAssistantModel.__name__:

return "hotel_booking_entry_node"

elif toolCallList[0]["name"] == ExcursionBookingAssistantModel.__name__:

return "excursion_booking_entry_node"

return "primary_assistant_tool_node"

raise ValueError("Invalid route")

stateGraph.add_node("primary_assistant_node" , Assistant(primaryAssistantRunnableSequence))

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

# 어시스턴트는 위임된 어시스턴트 중 한 명에게 라우팅하거나 도구를 직접 사용하거나 사용자에게 직접 응답할 수 있다.

stateGraph.add_conditional_edges("primary_assistant_node", routePrimaryAssistantNode, ["flight_booking_entry_node", "car_rental_booking_entry_node", "hotel_booking_entry_node", "excursion_booking_entry_node", "primary_assistant_tool_node", END])

stateGraph.add_edge("primary_assistant_tool_node", "primary_assistant_node")

# 스킬 이탈 노드

# 이 노드는 모든 전문화된 어시스턴트를 종료하기 위해 공유된다.

def popDialogState(state : State) -> dict:

"""Pop the dialog stack and return to the main assistant.

This lets the full graph explicitly track the dialog flow and delegate control

to specific sub-graphs.

"""

messageList = []

if state["messages"][-1].tool_calls:

# 참고 : 현재 LLM이 병렬 도구 호출을 수행하는 에지 케이스를 처리하지 않는다.

messageList.append(

ToolMessage(

content = "Resuming dialog with the host assistant. Please reflect on the past conversation and assist the user as needed.",

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

)

return {

"dialog_state_list" : "pop",

"messages" : messageList

}

stateGraph.add_node("leave_skill_node", popDialogState)

stateGraph.add_edge("leave_skill_node", "primary_assistant_node")

# 컴파일 상태 그래프 객체를 만든다.

memorySaver = MemorySaver()

compiledStateGraph = stateGraph.compile(

checkpointer = memorySaver,

# 사용자가 민감한 도구 사용을 승인하거나 거부하도록 한다.

interrupt_before = [

"flight_booking_sensitive_tool_node",

"car_rental_booking_sensitive_tool_node",

"hotel_booking_sensitive_tool_node",

"excursion_booking_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."

# "안녕하세요, 제 항공편은 몇 시인가요?"

# "항공편을 좀 더 일찍 예약해도 될까요? 오늘 늦게 출발하고 싶어요."

# "그럼 다음 주 어느 때로 항공편을 예약하세요"

# "다음 가능한 옵션이 좋습니다"

# "숙박과 교통은 어때요?"

# "네, 일주일(7일) 동안 머물기 위해 저렴한 호텔을 원합니다. 그리고 차를 렌트하고 싶어요."

# "좋아요, 추천 호텔을 예약해 주시겠어요? 좋은 것 같아요."

# "네, 적당한 가격대에 예약 가능한 곳이면 무엇이든 예약하세요."

# "이제 차는 어떤 옵션이 있나요?"

# "좋아요, 가장 저렴한 옵션을 선택해 볼까요. 7일 동안 예약하세요"

# "좋네요. 그럼 여행에 대한 추천이 있나요?"

# "제가 거기에 있는 동안 이용할 수 있나요?"

# "흥미롭네요. 박물관이 마음에 들어요. 어떤 옵션이 있나요?"

# "좋아요, 하나를 골라서 둘째 날 예약하세요."

]

printedSet = set()

configurableDictionary = {

"configurable" : {

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

"passenger_id" : "3442 587242"

}

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

currentStateList = addableUpdatesDict.get("dialog_state_list")

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] 전문 워크플로우를 사용하는 제로샷 에이전트 만들기

분류

가장 많이 읽힌 글

보관함