LangGraph實戰：從零分階打造人工智能航空客服助手
作者：AI小智
由于內容較多，本文將由淺入深，分四個階段進行講解，每個階段都將打造出一個具備以上描述所有能力的機器人。但受限于LLM的能力，初期階段的機器人的運行可能存在各類問題，但都將在后續階段得到解決。
客服助手機器人能夠幫助團隊更高效地處理日常咨詢，但要打造一個能夠穩定應對各種任務且不會讓用戶感到煩惱的機器人并非易事。
完成本教程后，你不僅會擁有一個功能完備的機器人，還將深入理解LangGraph的核心理念和架構設計。這些知識將幫助你在其他人工智能項目中運用相似的設計模式。
由于內容較多，本文將由淺入深，分四個階段進行講解，每個階段都將打造出一個具備以上描述所有能力的機器人。但受限于LLM的能力，初期階段的機器人的運行可能存在各類問題，但都將在后續階段得到解決。
你最終完成的聊天機器人將類似于以下示意圖：
最終示意圖
現在，讓我們開啟這第一階段段的學習之旅吧！
準備工作
在開始之前，我們需要搭建好環境。本教程將安裝一些必要的先決條件，包括下載測試用的數據庫，并定義一些在后續各部分中會用到的工具。
我們會使用 Claude 作為語言模型（LLM），并創建一些定制化的工具。這些工具大多數會連接到本地的 SQLite 數據庫，無需額外依賴。此外，我們還會通過 Tavily 為代理提供網絡搜索功能。
%%capture --no-stderr
% pip install -U langgraph langchain-community langchain-anthropic tavily-python pandas數據庫初始化
接下來，執行下面的腳本來獲取我們為這個教程準備的 SQLite 數據庫，并更新它以反映當前的數據狀態。具體細節不是重點。
import os
import requests
import sqlite3
import pandas as pd
import shutil
# 下載數據庫文件
db_url = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/travel2.sqlite"
local_file = "travel2.sqlite"
backup_file = "travel2.backup.sqlite"
overwrite = False
if not os.path.exists(local_file) or overwrite:
response = requests.get(db_url)
response.raise_for_status() # 確保請求成功
with open(local_file, "wb") as file:
file.write(response.content)
# 創建數據庫備份，以便在每個教程部分開始時重置數據庫狀態
shutil.copy(local_file, backup_file)
# 將航班數據更新為當前時間，以適應我們的教程
conn = sqlite3.connect(local_file)
cursor = conn.cursor()
# 讀取數據庫中的所有表
tables = pd.read_sql(
"SELECT name FROM sqlite_master WHERE type='table';", conn
).name.tolist()
tdf = {}
for table_name in tables:
tdf[table_name] = pd.read_sql(f"SELECT * from {table_name}", 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
# 更新預訂日期和航班時間
for column in ["book_date", "scheduled_departure", "scheduled_arrival", "actual_departure", "actual_arrival"]:
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)
conn.commit()
conn.close()
# 在本教程中，我們將使用這個本地文件作為數據庫
db = local_file工具定義
現在，我們來定義一些工具，以便助手可以搜索航空公司的政策手冊，以及搜索和管理航班、酒店、租車和遠足活動的預訂。這些工具將在教程的各個部分中重復使用，具體的實現細節不是關鍵。
查詢公司政策
助手需要檢索政策信息來回答用戶的問題。請注意，這些政策的實施還需要在工具或 API 中進行，因為語言模型可能會忽略這些信息。以下工具受限于篇幅將僅提供定義及描述，詳細代碼[1]可在github上獲取。
import re
import numpy as np
import openai
from langchain_core.tools import tool
@tool
def lookup_policy(query):
"""查詢公司政策，以確定某些選項是否允許。"""航班管理
定義一個工具來獲取用戶的航班信息，然后定義一些工具來搜索航班和管理用戶的預訂信息，這些信息存儲在 SQL 數據庫中。
我們使用 ensure_config 來通過配置參數傳遞 passenger_id。語言模型不需要顯式提供這些信息，它們會在圖的每次調用中提供，以確保每個用戶無法訪問其他乘客的預訂信息。
from langchain_core.runnables import ensure_config
from typing import Optional
import sqlite3
import pytz
from datetime import datetime, timedelta, date
@tool
def fetch_user_flight_information():
"""獲取用戶的所有機票信息，包括航班詳情和座位分配。"""
@tool
def search_flights(
departure_airport=None,
arrival_airport=None,
start_time=None,
end_time=None,
limit=20,
):
"""根據出發機場、到達機場和出發時間范圍來搜索航班。"""
@tool
def update_ticket_to_new_flight(ticket_no, new_flight_id):
"""將用戶的機票更新到一個新的有效航班上。"""
@tool
def cancel_ticket(ticket_no):
"""取消用戶的機票，并從數據庫中移除。"""租車服務
用戶預訂了航班后，可能需要租車服務。定義一些工具，讓用戶能夠在目的地搜索和預訂汽車。
from typing import Optional, Union
from datetime import datetime, date
@tool
def search_car_rentals(
locatinotallow=None,
name=None,
price_tier=None,
start_date=None,
end_date=None,
):
"""
根據位置、公司名稱、價格等級、開始日期和結束日期來搜索租車服務。
參數：
location (Optional[str]): 租車服務的位置。
name (Optional[str]): 租車公司的名稱。
price_tier (Optional[str]): 租車的價格等級。
start_date (Optional[Union[datetime, date]]): 租車的開始日期。
end_date (Optional[Union[datetime, date]]): 租車的結束日期。
返回：
list[dict]: 匹配搜索條件的租車服務列表。
"""
@tool
def book_car_rental(rental_id):
"""
通過租車ID來預訂租車服務。
參數：
rental_id (int): 要預訂的租車服務的ID。
返回：
str: 預訂成功與否的消息。
"""
@tool
def update_car_rental(
rental_id,
start_date=None,
end_date=None,
):
"""
通過租車ID來更新租車服務的開始和結束日期。
參數：
rental_id (int): 要更新的租車服務的ID。
start_date (Optional[Union[datetime, date]]): 新的租車開始日期。
end_date (Optional[Union[datetime, date]]): 新的租車結束日期。
返回：
str: 更新成功與否的消息。
"""
@tool
def cancel_car_rental(rental_id):
"""
通過租車ID來取消租車服務。
參數：
rental_id (int): 要取消的租車服務的ID。
返回：
str: 取消成功與否的消息。
"""酒店預訂
用戶需要住宿，因此定義一些工具來搜索和管理酒店預訂。
@tool
def search_hotels(
locatinotallow=None,
name=None,
price_tier=None,
checkin_date=None,
checkout_date=None,
):
"""
根據位置、名稱、價格等級、入住日期和退房日期來搜索酒店。
參數：
location (Optional[str]): 酒店的位置。
name (Optional[str]): 酒店的名稱。
price_tier (Optional[str]): 酒店的價格等級。
checkin_date
# 入住日期和退房日期，用于搜索酒店
checkin_date (Optional[Union[datetime, date]]): 酒店的入住日期。
checkout_date (Optional[Union[datetime, date]]): 酒店的退房日期。
返回：
list[dict]: 符合搜索條件的酒店列表。
"""
@tool
def book_hotel(hotel_id):
"""
通過酒店ID進行預訂。
參數：
hotel_id (int): 要預訂的酒店的ID。
返回：
str: 預訂成功與否的消息。
"""
@tool
def update_hotel(
hotel_id,
checkin_date=None,
checkout_date=None,
):
"""
通過酒店ID更新酒店預訂的入住和退房日期。
參數：
hotel_id (int): 要更新預訂的酒店的ID。
checkin_date (Optional[Union[datetime, date]]): 新的入住日期。
checkout_date (Optional[Union[datetime, date]]): 新的退房日期。
返回：
str: 更新成功與否的消息。
"""
@tool
def cancel_hotel(hotel_id):
"""
通過酒店ID取消酒店預訂。
參數：
hotel_id (int): 要取消預訂的酒店的ID。
返回：
str: 取消成功與否的消息。
"""遠足活動
最后，定義一些工具，讓用戶在到達目的地后搜索活動并進行預訂。
@tool
def search_trip_recommendations(
locatinotallow=None,
name=None,
keywords=None,
):
"""
根據位置、名稱和關鍵詞搜索旅行推薦。
參數：
location (Optional[str]): 旅行推薦的地點。
name (Optional[str]): 旅行推薦的名字。
keywords (Optional[str]): 與旅行推薦相關的關鍵詞。
返回：
list[dict]: 符合搜索條件的旅行推薦列表。
"""
@tool
def book_excursion(recommendation_id):
"""
通過推薦ID預訂遠足活動。
參數：
recommendation_id (int): 要預訂的旅行推薦的ID。
返回：
str: 預訂成功與否的消息。
"""
@tool
def update_excursion(recommendation_id, details):
"""
通過推薦ID更新旅行推薦的細節。
參數：
recommendation_id (int): 要更新的旅行推薦的ID。
details (str): 旅行推薦的新細節。
返回：
str: 更新成功與否的消息。
"""
@tool
def cancel_excursion(recommendation_id):
"""
通過推薦ID取消旅行推薦。
參數：
recommendation_id (int): 要取消的旅行推薦的ID。
返回：
str: 取消成功與否的消息。
"""實用工具
定義一些輔助函數，以便在調試過程中美化圖形中的消息顯示，并為工具節點添加錯誤處理（通過將錯誤添加到聊天記錄中）。
from langgraph.prebuilt import ToolNode
from langchain_core.runnables import RunnableLambda
def handle_tool_error(state):
error = state.get("error")
tool_calls = state["messages"][-1].tool_calls
return {
"messages": [
ToolMessage(
cnotallow=f"錯誤: {repr(error)}\n請修正你的錯誤。",
tool_call_id=tc["id"],
)
for tc in tool_calls
]
}
def create_tool_node_with_fallback(tools):
return ToolNode(tools).with_fallbacks(
[RunnableLambda(handle_tool_error)], exception_key="error"
)
def _print_event(event, _printed, max_length=1500):
current_state = event.get("dialog_state")
if current_state:
print(f"當前狀態: ", 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] + " ... （內容已截斷）"
print(msg_repr)
_printed.add(message.id)第一部分：零樣本代理
在構建任何系統時，最佳實踐是從最簡單的可行方案開始，并通過使用類似LangSmith這樣的評估工具來測試其有效性。在條件相同的情況下，我們傾向于選擇簡單且可擴展的解決方案，而不是復雜的方案。然而，單一圖譜方法存在一些限制，比如機器人可能在未經用戶確認的情況下執行不希望的操作，處理復雜查詢時可能遇到困難，或者在回答時缺乏針對性。這些問題我們會在后續進行改進。 在這部分，我們將定義一個簡單的零樣本代理作為用戶的助手，并將所有工具賦予給它。我們的目標是引導它明智地使用這些工具來幫助用戶。 我們的簡單兩節點圖如下所示：
第一部分圖解
首先，我們定義狀態。
狀態
我們將StateGraph的狀態定義為一個包含消息列表的類型化字典。這些消息構成了聊天的記錄，也就是我們簡單助手所需要的全部狀態信息。
from langgraph.graph.message import add_messages, AnyMessage
from typing_extensions import TypedDict
from typing import Annotated
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]代理
然后，我們定義助手函數。這個函數接收圖的狀態，將其格式化為提示，然后調用一個大型語言模型（LLM）來預測最佳的響應。
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
class Assistant:
def __init__(self, runnable: Runnable):
self.runnable = runnable
def __call__(self, state: State, config: RunnableConfig):
while True:
passenger_id = config.get("passenger_id", None)
state = {**state, "user_info": passenger_id}
result = self.runnable.invoke(state)
# 如果大型語言模型返回了一個空響應，我們將重新提示它給出一個實際的響應。
if (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
messages = state["messages"] + [("user", "請給出一個真實的輸出。")]
state = {**state, "messages": messages}
else:
break
return {"messages": result}
# Haiku模型更快、成本更低，但準確性稍差
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
primary_assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"你是一個為瑞士航空提供幫助的客戶支持助手。"
"使用提供的工具來搜索航班、公司政策和其他信息以幫助回答用戶的查詢。"
"在搜索時，要有毅力。如果第一次搜索沒有結果，就擴大你的查詢范圍。"
"如果搜索結果為空，不要放棄，先擴大搜索范圍。"
"\n\n當前用戶：\n<User>\n{user_info}\n</User>"
"\n當前時間：{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.sqlite import SqliteSaver
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import tools_condition, ToolNode
builder = StateGraph(State)
# 定義節點：這些節點執行具體的工作
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_1_tools))
# 定義邊：這些邊決定了控制流程如何移動
builder.set_entry_point("assistant")
builder.add_conditional_edges(
"assistant",
tools_condition,
# "action"調用我們的工具之一。END導致圖終止（并向用戶做出響應）
{"action": "action", END: END},
)
builder.add_edge("action", "assistant")
# 檢查點器允許圖保存其狀態
# 這是整個圖的完整記憶。
memory = SqliteSaver.from_conn_string(":memory:")
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:
# 這需要一些額外的依賴項，是可選的
pass示例對話
現在，讓我們通過一系列對話示例來測試我們的聊天機器人。
import uuid
import shutil
# 假設這是用戶與助手之間可能發生的對話示例
tutorial_questions = [
"你好，我的航班是什么時候？",
"我可以把我的航班改簽到更早的時間嗎？我想今天晚些時候離開。",
"那就把我的航班改簽到下周某個時間吧",
"下一個可用的選項很好",
"住宿和交通方面有什么建議？",
"我想在為期一周的住宿中選擇一個經濟實惠的酒店（7天），并且我還想租一輛車。",
"好的，你能為你推薦的酒店預訂嗎？聽起來不錯。",
"是的，去預訂任何中等價位且有可用性的酒店。",
"對于汽車，我有哪些選擇？",
"太棒了，我們只選擇最便宜的選項。預訂7天。",
"那么，你對我的旅行有什么建議？",
"在我在那里的時候，有哪些活動是可用的？",
"有趣 - 我喜歡博物館，有哪些選擇？",
"好的，那就為我在那里的第二天預訂一個。",
]
# 使用備份文件以便我們可以從每個部分的原始位置重新啟動
shutil.copy(backup_file, db)
thread_id = str(uuid.uuid4())
config = {
"configurable": {
# passenger_id 在我們的航班工具中使用
# 以獲取用戶的航班信息
"passenger_id": "3442 587242",
# 檢查點通過 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)
責任編輯：武曉燕
來源：
AI小智
