客戶支持機器人可以通過處理常規問題來節省團隊的時間，但構建一個能夠可靠處理各種任務的機器人卻不容易，尤其是要避免讓用戶感到困惑或失望。
在本教程中，你將為一家航空公司構建一個客戶支持機器人，幫助用戶研究和安排旅行計劃。你將學習如何使用 LangGraph 的中斷功能、檢查點和更複雜的狀態管理來組織助手的工具，並幫助用戶進行航班預訂、酒店預訂、租車以及旅行活動安排。本教程假設你已經熟悉LangGraph 入門教程中介紹的概念。
到本教程結束時，你將構建一個能夠正常運行的機器人，並理解 LangGraph 的關鍵概念和架構。你還可以將這些設計模式應用到其他 AI 項目中。
最終的聊天機器人架構大致如下圖所示：
現在讓我們開始吧！
先決條件#
首先，設置你的開發環境。我們將安裝本教程所需的依賴項，下載測試數據庫，並定義在每個部分中會重複使用的工具。
我們將使用 Claude 作為我們的大型語言模型（LLM），並定義一些自定義工具。雖然大部分工具將連接到本地的 sqlite 數據庫（不需要額外的依賴），我們也會通過 Tavily 為代理提供通用的網頁搜索功能。
%%capture --no-stderr
% pip install -U langgraph langchain-community langchain-anthropic tavily-python pandas
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")
_set_env("TAVILY_API_KEY")
填充數據庫#
運行以下腳本來獲取我們為本教程準備的 sqlite 數據庫，並將其更新為最新狀態。具體細節並不重要。
import os
import shutil
import sqlite3
import pandas as pd
import requests
db_url = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/travel2.sqlite"
local_file = "travel2.sqlite"
# The backup lets us restart for each tutorial section
backup_file = "travel2.backup.sqlite"
overwrite = False
if overwrite or not os.path.exists(local_file):
response = requests.get(db_url)
response.raise_for_status() # Ensure the request was successful
with open(local_file, "wb") as f:
f.write(response.content)
# Backup - we will use this to "reset" our DB in each section
shutil.copy(local_file, backup_file)
# Convert the flights to present time for our tutorial
def update_dates(file):
shutil.copy(backup_file, file)
conn = sqlite3.connect(file)
cursor = conn.cursor()
tables = pd.read_sql(
"SELECT name FROM sqlite_master WHERE type='table';", conn
).name.tolist()
tdf = {}
for t in tables:
tdf[t] = pd.read_sql(f"SELECT * from {t}", conn)
example_time = pd.to_datetime(
tdf["flights"]["actual_departure"].replace("\\N", pd.NaT)
).max()
current_time = pd.to_datetime("now").tz_localize(example_time.tz)
time_diff = current_time - example_time
tdf["bookings"]["book_date"] = (
pd.to_datetime(tdf["bookings"]["book_date"].replace("\\N", pd.NaT), utc=True)
+ time_diff
)
datetime_columns = [
"scheduled_departure",
"scheduled_arrival",
"actual_departure",
"actual_arrival",
]
for column in datetime_columns:
tdf["flights"][column] = (
pd.to_datetime(tdf["flights"][column].replace("\\N", pd.NaT)) + time_diff
)
for table_name, df in tdf.items():
df.to_sql(table_name, conn, if_exists="replace", index=False)
del df
del tdf
conn.commit()
conn.close()
return file
db = update_dates(local_file)
工具#
接下來，我們定義助手的工具，用於搜索航空公司的政策手冊，以及搜索和管理航班、酒店、租車和旅行活動的預訂。這些工具將在整個教程中重複使用。具體的實現細節並不重要，因此你可以直接運行以下代碼並跳到第 1 部分。
查找公司政策#
助手可以檢索政策信息來回答用戶的問題。請注意，政策的執行仍然必須通過工具或 API 來完成，因為大型語言模型（LLM）可能會忽略這些規定。
import re
import numpy as np
import openai
from langchain_core.tools import tool
response = requests.get(
"https://storage.googleapis.com/benchmarks-artifacts/travel-db/swiss_faq.md"
)
response.raise_for_status()
faq_text = response.text
docs = [{"page_content": txt} for txt in re.split(r"(?=\n##)", faq_text)]
class VectorStoreRetriever:
def __init__(self, docs: list, vectors: list, oai_client):
self._arr = np.array(vectors)
self._docs = docs
self._client = oai_client
@classmethod
def from_docs(cls, docs, oai_client):
embeddings = oai_client.embeddings.create(
model="text-embedding-3-small", input=[doc["page_content"] for doc in docs]
)
vectors = [emb.embedding for emb in embeddings.data]
return cls(docs, vectors, oai_client)
def query(self, query: str, k: int = 5) -> list[dict]:
embed = self._client.embeddings.create(
model="text-embedding-3-small", input=[query]
)
# "@" is just a matrix multiplication in python
scores = np.array(embed.data[0].embedding) @ self._arr.T
top_k_idx = np.argpartition(scores, -k)[-k:]
top_k_idx_sorted = top_k_idx[np.argsort(-scores[top_k_idx])]
return [
{**self._docs[idx], "similarity": scores[idx]} for idx in top_k_idx_sorted
]
retriever = VectorStoreRetriever.from_docs(docs, openai.Client())
@tool
def lookup_policy(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."""
docs = retriever.query(query, k=2)
return "\n\n".join([doc["page_content"] for doc in docs])
航班#
定義 (fetch_user_flight_information) 工具，以便讓代理查看當前用戶的航班信息。然後定義工具來搜索航班並管理存儲在 SQL 數據庫中的乘客預訂信息。
我們可以通過訪問 RunnableConfig 來檢查訪問此應用程序的用戶的passenger_id。大型語言模型（LLM）不需要顯式提供這些信息，它們會在圖的每次調用時提供，確保每個用戶無法訪問其他乘客的預訂信息。
import sqlite3
from datetime import date, datetime
from typing import Optional
import pytz
from langchain_core.runnables import RunnableConfig
@tool
def fetch_user_flight_information(config: 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.
"""
configuration = config.get("configurable", {})
passenger_id = configuration.get("passenger_id", None)
if not passenger_id:
raise ValueError("No passenger ID configured.")
conn = sqlite3.connect(db)
cursor = conn.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, (passenger_id,))
rows = cursor.fetchall()
column_names = [column[0] for column in cursor.description]
results = [dict(zip(column_names, row)) for row in rows]
cursor.close()
conn.close()
return results
@tool
def search_flights(
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."""
conn = sqlite3.connect(db)
cursor = conn.cursor()
query = "SELECT * FROM flights WHERE 1 = 1"
params = []
if departure_airport:
query += " AND departure_airport = ?"
params.append(departure_airport)
if arrival_airport:
query += " AND arrival_airport = ?"
params.append(arrival_airport)
if start_time:
query += " AND scheduled_departure >= ?"
params.append(start_time)
if end_time:
query += " AND scheduled_departure <= ?"
params.append(end_time)
query += " LIMIT ?"
params.append(limit)
cursor.execute(query, params)
rows = cursor.fetchall()
column_names = [column[0] for column in cursor.description]
results = [dict(zip(column_names, row)) for row in rows]
cursor.close()
conn.close()
return results
@tool
def update_ticket_to_new_flight(
ticket_no: str, new_flight_id: int, *, config: RunnableConfig
) -> str:
"""Update the user's ticket to a new valid flight."""
configuration = config.get("configurable", {})
passenger_id = configuration.get("passenger_id", None)
if not passenger_id:
raise ValueError("No passenger ID configured.")
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(
"SELECT departure_airport, arrival_airport, scheduled_departure FROM flights WHERE flight_id = ?",
(new_flight_id,),
)
new_flight = cursor.fetchone()
if not new_flight:
cursor.close()
conn.close()
return "Invalid new flight ID provided."
column_names = [column[0] for column in cursor.description]
new_flight_dict = dict(zip(column_names, new_flight))
timezone = pytz.timezone("Etc/GMT-3")
current_time = datetime.now(tz=timezone)
departure_time = datetime.strptime(
new_flight_dict["scheduled_departure"], "%Y-%m-%d %H:%M:%S.%f%z"
)
time_until = (departure_time - current_time).total_seconds()
if time_until < (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 {departure_time}."
cursor.execute(
"SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,)
)
current_flight = cursor.fetchone()
if not current_flight:
cursor.close()
conn.close()
return "No existing ticket found for the given ticket number."
# Check the signed-in user actually has this ticket
cursor.execute(
"SELECT * FROM tickets WHERE ticket_no = ? AND passenger_id = ?",
(ticket_no, passenger_id),
)
current_ticket = cursor.fetchone()
if not current_ticket:
cursor.close()
conn.close()
return f"Current signed-in passenger with ID {passenger_id} not the owner of ticket {ticket_no}"
# In a real application, you'd likely add additional checks here to enforce business logic,
# like "does the new departure airport match the current ticket", etc.
# While it's best to try to be *proactive* in 'type-hinting' policies to the LLM
# it's inevitably going to get things wrong, so you **also** need to ensure your
# API enforces valid behavior
cursor.execute(
"UPDATE ticket_flights SET flight_id = ? WHERE ticket_no = ?",
(new_flight_id, ticket_no),
)
conn.commit()
cursor.close()
conn.close()
return "Ticket successfully updated to new flight."
@tool
def cancel_ticket(ticket_no: str, *, config: RunnableConfig) -> str:
"""Cancel the user's ticket and remove it from the database."""
configuration = config.get("configurable", {})
passenger_id = configuration.get("passenger_id", None)
if not passenger_id:
raise ValueError("No passenger ID configured.")
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(
"SELECT flight_id FROM ticket_flights WHERE ticket_no = ?", (ticket_no,)
)
existing_ticket = cursor.fetchone()
if not existing_ticket:
cursor.close()
conn.close()
return "No existing ticket found for the given ticket number."
# Check the signed-in user actually has this ticket
cursor.execute(
"SELECT flight_id FROM tickets WHERE ticket_no = ? AND passenger_id = ?",
(ticket_no, passenger_id),
)
current_ticket = cursor.fetchone()
if not current_ticket:
cursor.close()
conn.close()
return f"Current signed-in passenger with ID {passenger_id} not the owner of ticket {ticket_no}"
cursor.execute("DELETE FROM ticket_flights WHERE ticket_no = ?", (ticket_no,))
conn.commit()
cursor.close()
conn.close()
return "Ticket successfully cancelled."
租車工具#
用戶預訂航班後，通常會希望安排交通工具。定義一些 “租車” 工具，讓用戶可以在目的地搜索並預訂車輛。
from datetime import date, datetime
from typing import Optional, Union
@tool
def search_car_rentals(
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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
query = "SELECT * FROM car_rentals WHERE 1=1"
params = []
if location:
query += " AND location LIKE ?"
params.append(f"%{location}%")
if name:
query += " AND name LIKE ?"
params.append(f"%{name}%")
# For our tutorial, we will let you match on any dates and price tier.
# (since our toy dataset doesn't have much data)
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return [
dict(zip([column[0] for column in cursor.description], row)) for row in results
]
@tool
def book_car_rental(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("UPDATE car_rentals SET booked = 1 WHERE id = ?", (rental_id,))
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Car rental {rental_id} successfully booked."
else:
conn.close()
return f"No car rental found with ID {rental_id}."
@tool
def update_car_rental(
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.
"""
conn = sqlite3.connect(db)
cursor = conn.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)
)
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Car rental {rental_id} successfully updated."
else:
conn.close()
return f"No car rental found with ID {rental_id}."
@tool
def cancel_car_rental(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("UPDATE car_rentals SET booked = 0 WHERE id = ?", (rental_id,))
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Car rental {rental_id} successfully cancelled."
else:
conn.close()
return f"No car rental found with ID {rental_id}."
酒店#
用戶需要睡覺！定義一些工具來搜索和管理酒店預訂。
@tool
def search_hotels(
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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
query = "SELECT * FROM hotels WHERE 1=1"
params = []
if location:
query += " AND location LIKE ?"
params.append(f"%{location}%")
if name:
query += " AND name LIKE ?"
params.append(f"%{name}%")
# For the sake of this tutorial, we will let you match on any dates and price tier.
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return [
dict(zip([column[0] for column in cursor.description], row)) for row in results
]
@tool
def book_hotel(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("UPDATE hotels SET booked = 1 WHERE id = ?", (hotel_id,))
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Hotel {hotel_id} successfully booked."
else:
conn.close()
return f"No hotel found with ID {hotel_id}."
@tool
def update_hotel(
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.
"""
conn = sqlite3.connect(db)
cursor = conn.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),
)
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Hotel {hotel_id} successfully updated."
else:
conn.close()
return f"No hotel found with ID {hotel_id}."
@tool
def cancel_hotel(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("UPDATE hotels SET booked = 0 WHERE id = ?", (hotel_id,))
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Hotel {hotel_id} successfully cancelled."
else:
conn.close()
return f"No hotel found with ID {hotel_id}."
旅行活動#
最後，定義一些工具，允許用戶在到達目的地後搜索可以做的事情（並進行預訂）。
@tool
def search_trip_recommendations(
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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
query = "SELECT * FROM trip_recommendations WHERE 1=1"
params = []
if location:
query += " AND location LIKE ?"
params.append(f"%{location}%")
if name:
query += " AND name LIKE ?"
params.append(f"%{name}%")
if keywords:
keyword_list = keywords.split(",")
keyword_conditions = " OR ".join(["keywords LIKE ?" for _ in keyword_list])
query += f" AND ({keyword_conditions})"
params.extend([f"%{keyword.strip()}%" for keyword in keyword_list])
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return [
dict(zip([column[0] for column in cursor.description], row)) for row in results
]
@tool
def book_excursion(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(
"UPDATE trip_recommendations SET booked = 1 WHERE id = ?", (recommendation_id,)
)
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Trip recommendation {recommendation_id} successfully booked."
else:
conn.close()
return f"No trip recommendation found with ID {recommendation_id}."
@tool
def update_excursion(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(
"UPDATE trip_recommendations SET details = ? WHERE id = ?",
(details, recommendation_id),
)
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Trip recommendation {recommendation_id} successfully updated."
else:
conn.close()
return f"No trip recommendation found with ID {recommendation_id}."
@tool
def cancel_excursion(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.
"""
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(
"UPDATE trip_recommendations SET booked = 0 WHERE id = ?", (recommendation_id,)
)
conn.commit()
if cursor.rowcount > 0:
conn.close()
return f"Trip recommendation {recommendation_id} successfully cancelled."
else:
conn.close()
return f"No trip recommendation found with ID {recommendation_id}."
工具函數#
定義輔助函數，以便在調試時美化打印圖中的消息，並為我們的工具節點提供錯誤處理功能（通過將錯誤添加到聊天記錄中）。
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableLambda
from langgraph.prebuilt import ToolNode
def handle_tool_error(state) -> dict:
error = state.get("error")
tool_calls = state["messages"][-1].tool_calls
return {
"messages": [
ToolMessage(
content=f"Error: {repr(error)}\n please fix your mistakes.",
tool_call_id=tc["id"],
)
for tc in tool_calls
]
}
def create_tool_node_with_fallback(tools: list) -> dict:
return ToolNode(tools).with_fallbacks(
[RunnableLambda(handle_tool_error)], exception_key="error"
)
def _print_event(event: dict, _printed: set, max_length=1500):
current_state = event.get("dialog_state")
if current_state:
print("Currently in: ", current_state[-1])
message = event.get("messages")
if message:
if isinstance(message, list):
message = message[-1]
if message.id not in _printed:
msg_repr = message.pretty_repr(html=True)
if len(msg_repr) > max_length:
msg_repr = msg_repr[:max_length] + " ... (truncated)"
print(msg_repr)
_printed.add(message.id)
第 1 部分：零次學習代理#
在構建時，最好從最簡單的可工作實現開始，並使用類似LangSmith 的評估工具來衡量其有效性。如果其他條件相同，應優先選擇簡單且可擴展的解決方案，而不是複雜的方案。在本例中，單一圖表方法有其局限性。機器人可能會在沒有用戶確認的情況下採取不必要的行動，難以處理複雜查詢，且在回答時缺乏針對性。我們將在後續部分解決這些問題。
在本部分中，我們將定義一個簡單的零次學習代理作為助手，給該代理所有的工具，並提示其謹慎使用這些工具來協助用戶。
簡單的雙節點圖表如下所示：
首先定義狀態。
狀態#
將我們的 StateGraph 狀態定義為一個類型化字典，包含一個只能追加的消息列表。這些消息構成了聊天記錄，也是我們簡單助手所需的全部狀態信息。
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
代理#
接下來，定義助手函數。該函數接受圖狀態，將其格式化為提示，然後調用大型語言模型（LLM）來預測最佳響應。
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
class Assistant:
def __init__(self, runnable: Runnable):
self.runnable = runnable
def __call__(self, state: State, config: RunnableConfig):
while True:
configuration = config.get("configurable", {})
passenger_id = configuration.get("passenger_id", None)
state = {**state, "user_info": passenger_id}
result = self.runnable.invoke(state)
# If the LLM happens to return an empty response, we will re-prompt it
# for an actual response.
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
else:
break
return {"messages": result}
# Haiku is faster and cheaper, but less accurate
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
# You could swap LLMs, though you will likely want to update the prompts when
# doing so!
# from langchain_openai import ChatOpenAI
# llm = ChatOpenAI(model="gpt-4-turbo-preview")
primary_assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful customer support assistant for Swiss Airlines. "
" Use the provided tools to search for flights, company policies, and other information to assist the user's queries. "
" When searching, be persistent. Expand your query bounds if the first search returns no results. "
" If a search comes up empty, expand your search before giving up."
"\n\nCurrent user:\n<User>\n{user_info}\n</User>"
"\nCurrent time: {time}.",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
part_1_tools = [
TavilySearchResults(max_results=1),
fetch_user_flight_information,
search_flights,
lookup_policy,
update_ticket_to_new_flight,
cancel_ticket,
search_car_rentals,
book_car_rental,
update_car_rental,
cancel_car_rental,
search_hotels,
book_hotel,
update_hotel,
cancel_hotel,
search_trip_recommendations,
book_excursion,
update_excursion,
cancel_excursion,
]
part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)
定義圖#
現在，創建圖。這個圖是本節的最終助手。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import tools_condition
builder = StateGraph(State)
# Define nodes: these do the work
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))
# Define edges: these determine how the control flow moves
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
"assistant",
tools_condition,
)
builder.add_edge("tools", "assistant")
# The checkpointer lets the graph persist its state
# this is a complete memory for the entire graph.
memory = MemorySaver()
part_1_graph = builder.compile(checkpointer=memory)
from IPython.display import Image, display
try:
display(Image(part_1_graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
示例對話#
現在是時候嘗試我們強大的聊天機器人了！讓我們用下面的對話輪次來運行它。如果遇到 “RecursionLimit”，那意味著機器人無法在分配的步驟內獲得答案。這沒關係！在本教程的後續部分，我們還有更多技巧可以使用。
import shutil
import uuid
# Let's create an example conversation a user might have with the assistant
tutorial_questions = [
"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.",
]
# Update with the backup file so we can restart from the original place in each section
db = update_dates(db)
thread_id = str(uuid.uuid4())
config = {
"configurable": {
# The passenger_id is used in our flight tools to
# fetch the user's flight information
"passenger_id": "3442 587242",
# Checkpoints are accessed by thread_id
"thread_id": thread_id,
}
}
_printed = set()
for question in tutorial_questions:
events = part_1_graph.stream(
{"messages": ("user", question)}, config, stream_mode="values"
)
for event in events:
_print_event(event, _printed)
第 1 部分回顧#
我們的簡單助手表現還不錯！它能夠合理地回答所有問題，快速在上下文中響應，並成功執行了所有任務。你可以查看一個 LangSmith 的示例跟踪，以更好地了解 LLM 在上述互動中的提示方式。
如果這只是個簡單的問答機器人，我們對上述結果可能會感到滿意。然而，由於我們的客戶支持機器人是代表用戶採取行動的，它的一些行為有點令人擔憂：
- 助手在我們專注於住宿時預訂了租車，然後不得不取消並重新預訂：哎呀！為了避免不必要的費用，應該讓用戶在預訂之前有最終的決定權。
- 助手在搜索推薦時遇到困難。我們可以通過為工具添加更詳細的指令和示例來改進這一點，但如果對每個工具都這麼做，可能會導致提示變得冗長，壓垮機器人。
- 助手不得不進行顯式搜索才能獲取用戶的相關信息。如果我們立即獲取用戶的旅行詳情，助手就能直接做出響應，從而節省大量時間。
在下一節中，我們將解決上述問題中的前兩個。
第 2 部分：添加確認#
當助手代表用戶採取行動時，幾乎總是應該讓用戶最終決定是否執行這些操作。否則，助手所犯的任何小錯誤（或任何它受到的提示注入）都可能對用戶造成實際損害。
在本節中，我們將使用 interrupt_before 來在執行任何工具之前暫停圖表的運行，並將控制權交還給用戶。
你的圖表將類似於以下內容：
與之前一樣，從定義狀態開始：
狀態 & 助手#
我們的圖表狀態和調用 LLM 的方式與第 1 部分幾乎相同，除了以下幾點不同：
- 我們增加了一個
user_info字段，該字段將由我們的圖表主動填充 - 我們可以直接在
Assistant對象中使用狀態，而不是使用可配置的參數
from typing import Annotated
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
user_info: str
class Assistant:
def __init__(self, runnable: Runnable):
self.runnable = runnable
def __call__(self, state: State, config: RunnableConfig):
while True:
result = self.runnable.invoke(state)
# If the LLM happens to return an empty response, we will re-prompt it
# for an actual response.
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
else:
break
return {"messages": result}
# Haiku is faster and cheaper, but less accurate
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
# You could also use OpenAI or another model, though you will likely have
# to adapt the prompts
# from langchain_openai import ChatOpenAI
# llm = ChatOpenAI(model="gpt-4-turbo-preview")
assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful customer support assistant for Swiss Airlines. "
" Use the provided tools to search for flights, company policies, and other information to assist the user's queries. "
" When searching, be persistent. Expand your query bounds if the first search returns no results. "
" If a search comes up empty, expand your search before giving up."
"\n\nCurrent user:\n<User>\n{user_info}\n</User>"
"\nCurrent time: {time}.",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
part_2_tools = [
TavilySearchResults(max_results=1),
fetch_user_flight_information,
search_flights,
lookup_policy,
update_ticket_to_new_flight,
cancel_ticket,
search_car_rentals,
book_car_rental,
update_car_rental,
cancel_car_rental,
search_hotels,
book_hotel,
update_hotel,
cancel_hotel,
search_trip_recommendations,
book_excursion,
update_excursion,
cancel_excursion,
]
part_2_assistant_runnable = assistant_prompt | llm.bind_tools(part_2_tools)
定義圖#
現在，創建圖表。針對第一部分中的問題，做出以下兩項改動：
- 在使用工具之前添加一個中斷。
- 在第一個節點中明確填充用戶狀態，以便助手不必使用工具就能了解用戶信息。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.prebuilt import tools_condition
builder = StateGraph(State)
def user_info(state: State):
return {"user_info": fetch_user_flight_information.invoke({})}
# NEW: The fetch_user_info node runs first, meaning our assistant can see the user's flight information without
# having to take an action
builder.add_node("fetch_user_info", user_info)
builder.add_edge(START, "fetch_user_info")
builder.add_node("assistant", Assistant(part_2_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_2_tools))
builder.add_edge("fetch_user_info", "assistant")
builder.add_conditional_edges(
"assistant",
tools_condition,
)
builder.add_edge("tools", "assistant")
memory = MemorySaver()
part_2_graph = builder.compile(
checkpointer=memory,
# NEW: The graph will always halt before executing the "tools" node.
# The user can approve or reject (or even alter the request) before
# the assistant continues
interrupt_before=["tools"],
)
from IPython.display import Image, display
try:
display(Image(part_2_graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
示例對話#
現在是時候嘗試我們新修訂的聊天機器人了！讓我們根據以下對話輪次來運行它。
import shutil
import uuid
# Update with the backup file so we can restart from the original place in each section
db = update_dates(db)
thread_id = str(uuid.uuid4())
config = {
"configurable": {
# The passenger_id is used in our flight tools to
# fetch the user's flight information
"passenger_id": "3442 587242",
# Checkpoints are accessed by thread_id
"thread_id": thread_id,
}
}
_printed = set()
# We can reuse the tutorial questions from part 1 to see how it does.
for question in tutorial_questions:
events = part_2_graph.stream(
{"messages": ("user", question)}, config, stream_mode="values"
)
for event in events:
_print_event(event, _printed)
snapshot = part_2_graph.get_state(config)
while snapshot.next:
# We have an interrupt! The agent is trying to use a tool, and the user can approve or deny it
# Note: This code is all outside of your graph. Typically, you would stream the output to a UI.
# Then, you would have the frontend trigger a new run via an API call when the user has provided input.
user_input = input(
"Do you approve of the above actions? Type 'y' to continue;"
" otherwise, explain your requested changed.\n\n"
)
if user_input.strip() == "y":
# Just continue
result = part_2_graph.invoke(
None,
config,
)
else:
# Satisfy the tool invocation by
# providing instructions on the requested changes / change of mind
result = part_2_graph.invoke(
{
"messages": [
ToolMessage(
tool_call_id=event["messages"][-1].tool_calls[0]["id"],
content=f"API call denied by user. Reasoning: '{user_input}'. Continue assisting, accounting for the user's input.",
)
]
},
config,
)
snapshot = part_2_graph.get_state(config)
第二部分回顧#
現在我們的助手已經能夠通過跳過一個步驟來直接回覆我們的航班詳情了。我們也完全掌控了執行的每一個操作。這一切都依賴於 LangGraph 的interrupts（中斷）和checkpointers（檢查點）。中斷功能會暫停圖的執行，而它的狀態會安全地保存在你配置的檢查點中。用戶可以通過運行正確的配置隨時重新啟動。
查看這個LangSmith 執行示例，你可以更好地理解圖是如何運行的。注意這個示例，通常你可以通過調用(None, config)來恢復一個流程。狀態會從檢查點加載，就像從未被中斷一樣。
這個圖運行得相當不錯！不過我們其實並不需要參與每一個助手的操作……
在下一部分中，我們將重新組織我們的圖結構，以便僅在那些實際寫入數據庫的 “敏感” 操作上觸發中斷。
第三部分：條件中斷#
在本節中，我們將通過將工具分類為安全（只讀）或敏感（數據修改）來優化我們的中斷策略。我們只會對敏感工具應用中斷，允許機器人自主處理簡單查詢。
這樣做平衡了用戶控制與對話流暢性，但隨著我們增加更多工具，單一的圖結構可能會變得過於複雜，難以維持 “平面” 結構。我們將在下一節解決這個問題。
第三部分的圖結構大致如下圖所示：
狀態#
與往常一樣，首先定義圖的狀態。我們的狀態和 LLM 調用與第二部分完全相同。
from typing import Annotated
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
user_info: str
class Assistant:
def __init__(self, runnable: Runnable):
self.runnable = runnable
def __call__(self, state: State, config: RunnableConfig):
while True:
result = self.runnable.invoke(state)
# If the LLM happens to return an empty response, we will re-prompt it
# for an actual response.
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
else:
break
return {"messages": result}
# Haiku is faster and cheaper, but less accurate
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
# You can update the LLMs, though you may need to update the prompts
# from langchain_openai import ChatOpenAI
# llm = ChatOpenAI(model="gpt-4-turbo-preview")
assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful customer support assistant for Swiss Airlines. "
" Use the provided tools to search for flights, company policies, and other information to assist the user's queries. "
" When searching, be persistent. Expand your query bounds if the first search returns no results. "
" If a search comes up empty, expand your search before giving up."
"\n\nCurrent user:\n<User>\n{user_info}\n</User>"
"\nCurrent time: {time}.",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
part_3_safe_tools = [
TavilySearchResults(max_results=1),
fetch_user_flight_information,
search_flights,
lookup_policy,
search_car_rentals,
search_hotels,
search_trip_recommendations,
]
# These tools all change the user's reservations.
# The user has the right to control what decisions are made
part_3_sensitive_tools = [
update_ticket_to_new_flight,
cancel_ticket,
book_car_rental,
update_car_rental,
cancel_car_rental,
book_hotel,
update_hotel,
cancel_hotel,
book_excursion,
update_excursion,
cancel_excursion,
]
sensitive_tool_names = {t.name for t in part_3_sensitive_tools}
# Our LLM doesn't have to know which nodes it has to route to. In its 'mind', it's just invoking functions.
part_3_assistant_runnable = assistant_prompt | llm.bind_tools(
part_3_safe_tools + part_3_sensitive_tools
)
定義圖#
現在，創建圖。我們的圖與第二部分幾乎相同，區別在於我們將工具分成了兩個獨立的節點。我們只會在那些實際對用戶預訂進行更改的工具之前進行中斷。
from typing import Literal
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.prebuilt import tools_condition
builder = StateGraph(State)
def user_info(state: State):
return {"user_info": fetch_user_flight_information.invoke({})}
# NEW: The fetch_user_info node runs first, meaning our assistant can see the user's flight information without
# having to take an action
builder.add_node("fetch_user_info", user_info)
builder.add_edge(START, "fetch_user_info")
builder.add_node("assistant", Assistant(part_3_assistant_runnable))
builder.add_node("safe_tools", create_tool_node_with_fallback(part_3_safe_tools))
builder.add_node(
"sensitive_tools", create_tool_node_with_fallback(part_3_sensitive_tools)
)
# Define logic
builder.add_edge("fetch_user_info", "assistant")
def route_tools(state: State) -> Literal["safe_tools", "sensitive_tools", "__end__"]:
next_node = tools_condition(state)
# If no tools are invoked, return to the user
if next_node == END:
return END
ai_message = state["messages"][-1]
# This assumes single tool calls. To handle parallel tool calling, you'd want to
# use an ANY condition
first_tool_call = ai_message.tool_calls[0]
if first_tool_call["name"] in sensitive_tool_names:
return "sensitive_tools"
return "safe_tools"
builder.add_conditional_edges(
"assistant",
route_tools,
)
builder.add_edge("safe_tools", "assistant")
builder.add_edge("sensitive_tools", "assistant")
memory = MemorySaver()
part_3_graph = builder.compile(
checkpointer=memory,
# NEW: The graph will always halt before executing the "tools" node.
# The user can approve or reject (or even alter the request) before
# the assistant continues
interrupt_before=["sensitive_tools"],
)
from IPython.display import Image, display
try:
display(Image(part_3_graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
示例對話#
現在是時候嘗試我們剛剛改進的聊天機器人了！讓我們通過以下對話輪次來運行它。這次，我們會有更少的確認步驟。
import shutil
import uuid
# Update with the backup file so we can restart from the original place in each section
db = update_dates(db)
thread_id = str(uuid.uuid4())
config = {
"configurable": {
# The passenger_id is used in our flight tools to
# fetch the user's flight information
"passenger_id": "3442 587242",
# Checkpoints are accessed by thread_id
"thread_id": thread_id,
}
}
tutorial_questions = [
"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.",
]
_printed = set()
# We can reuse the tutorial questions from part 1 to see how it does.
for question in tutorial_questions:
events = part_3_graph.stream(
{"messages": ("user", question)}, config, stream_mode="values"
)
for event in events:
_print_event(event, _printed)
第三部分回顧#
效果更好了！我們的代理現在運行得很好 —— 查看 LangSmith 的執行記錄，來檢查它的工作！你可能對這個設計已經感到滿意。代碼是封裝好的，並且它的表現符合預期。
這個設計的一個問題是，我們把太多壓力放在了單個提示詞上。如果我們想添加更多工具，或者每個工具變得更加複雜（更多篩選條件、更多業務邏輯限制行為等），那麼工具的使用效果以及機器人的整體表現很可能會開始下降。
在接下來的部分中，我們將展示如何通過根據用戶意圖路由到專門的代理或子圖，來更好地控制不同的用戶體驗。
第 4 部分：專門的工作流程#
在前面的章節中，我們看到了依賴於單一提示和大型語言模型（LLM）處理各種用戶意圖的 “廣泛” 聊天機器人能夠帶我們走得多遠。然而，這種方法很難為已知的意圖創建可預測的出色用戶體驗。
另一種方法是，您的圖可以檢測用戶意圖並選擇合適的工作流程或 “技能” 來滿足用戶的需求。每個工作流程可以專注於其領域，從而可以進行獨立優化，而不會降低整個助手的整體性能。
在本節中，我們將用戶體驗劃分為單獨的子圖，最終結構如下所示：
在上圖中，每個方塊都包含一個具備執行能力、專注的工作流程。主助手處理用戶的初始查詢，圖根據查詢內容將請求路由到相應的 “專家”。
狀態#
我們希望能夠隨時跟踪哪個子圖在控制當前會話。雖然我們_可以_通過對消息列表進行一些運算來實現這一點，但更簡單的方法是將其作為專門的堆棧進行跟踪。
在下面的 State 中添加一個 dialog_state 列表。每當一個 node 運行並返回一個 dialog_state 值時，將調用 update_dialog_stack 函數來確定如何應用更新。
from typing import Annotated, Literal, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
"""Push or pop the state."""
if right is None:
return left
if right == "pop":
return left[:-1]
return left + [right]
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
user_info: str
dialog_state: Annotated[
list[
Literal[
"assistant",
"update_flight",
"book_car_rental",
"book_hotel",
"book_excursion",
]
],
update_dialog_stack,
]
助手#
這次我們將為每個工作流程創建一個助手。也就是說：
- 航班預訂助手
- 酒店預訂助手
- 租車助手
- 行程安排助手
- 最後，一個 “主助手” 用於在這些助手之間進行路由
如果你留意了，你可能會發現這是我們多代理示例中的監督者設計模式的實例。
下面定義用於驅動每個助手的 Runnable 對象。每個 Runnable 都有一個提示、LLM（大型語言模型）以及適用於該助手的工具的模式。
每個專門化的 / 委派的助手還可以調用 CompleteOrEscalate 工具，來指示控制流應回到主助手。這發生在助手成功完成其任務，或用戶改變主意，或者需要處理超出該工作流程範圍的問題時。
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableConfig
from pydantic import BaseModel, Field
class Assistant:
def __init__(self, runnable: Runnable):
self.runnable = runnable
def __call__(self, state: State, config: RunnableConfig):
while True:
result = self.runnable.invoke(state)
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
else:
break
return {"messages": result}
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": {
"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.",
},
}
# Flight booking assistant
flight_booking_prompt = 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.',
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
update_flight_safe_tools = [search_flights]
update_flight_sensitive_tools = [update_ticket_to_new_flight, cancel_ticket]
update_flight_tools = update_flight_safe_tools + update_flight_sensitive_tools
update_flight_runnable = flight_booking_prompt | llm.bind_tools(
update_flight_tools + [CompleteOrEscalate]
)
# Hotel Booking Assistant
book_hotel_prompt = 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'",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
book_hotel_safe_tools = [search_hotels]
book_hotel_sensitive_tools = [book_hotel, update_hotel, cancel_hotel]
book_hotel_tools = book_hotel_safe_tools + book_hotel_sensitive_tools
book_hotel_runnable = book_hotel_prompt | llm.bind_tools(
book_hotel_tools + [CompleteOrEscalate]
)
# Car Rental Assistant
book_car_rental_prompt = 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'",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
book_car_rental_safe_tools = [search_car_rentals]
book_car_rental_sensitive_tools = [
book_car_rental,
update_car_rental,
cancel_car_rental,
]
book_car_rental_tools = book_car_rental_safe_tools + book_car_rental_sensitive_tools
book_car_rental_runnable = book_car_rental_prompt | llm.bind_tools(
book_car_rental_tools + [CompleteOrEscalate]
)
# Excursion Assistant
book_excursion_prompt = 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!'",
),
("placeholder", "{messages}"),
]
).partial(time=datetime.now())
book_excursion_safe_tools = [search_trip_recommendations]
book_excursion_sensitive_tools = [book_excursion, update_excursion, cancel_excursion]
book_excursion_tools = book_excursion_safe_tools + book_excursion_sensitive_tools
book_excursion_runnable = book_excursion_prompt | llm.bind_tools(
book_excursion_tools + [CompleteOrEscalate]
)
# Primary Assistant
class ToFlightBookingAssistant(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 ToBookCarRental(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 ToHotelBookingAssistant(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 ToBookExcursion(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.",
}
}
# The top-level assistant performs general Q&A and delegates specialized tasks to other assistants.
# The task delegation is a simple form of semantic routing / does simple intent detection
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
primary_assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful customer support assistant for Swiss Airlines. "
"Your primary role is