LangGraph实战 从零分阶打造人工默认航空客服助手

客服助手机器人能够协助团队更高效地处置日常咨询，但要打造一个能够稳固应答各种义务且不会让用户感到烦恼的机器人并非易事。

成功本教程后，你不只会领有一特性能完备的机器人，还将深化了解LangGraph的外围思念和架构设计。这些常识将协助你在其他人工默认名目中运用相似的设计形式。

由于内容较多，本文将由浅入深，分四个阶段启动解说，每个阶段都将打造出一个具有以上形容一切才干的机器人。但受限于LLM的才干，初期阶段的机器人的运转或许存在各类疑问，但都将在后续阶段失掉处置。

你最终成功的聊天机器人将相似于以下示用意：

最终示用意

如今，让咱们开启这第一阶段段的学习之旅吧！

预备上班

在开局之前，咱们须要搭建好环境。本教程将装置一些必要的先决条件，包括下载测试用的数据库，并定义一些在后续各局部中会用到的工具。

咱们会经常使用 Claude 作为言语模型（LLM），并创立一些定制化的工具。这些工具大少数会衔接到本地的 SQLite 数据库，无需额外依赖。此外，咱们还会经过 Tavily 为代理提供网络搜查性能。

%%capture --no-stderr% pip install -U langgraph langchain-community langchain-anthropic tavily-python pandas

数据库初始化

接上去，执行上方的脚原本失掉咱们为这个教程预备的 SQLite 数据库，并更新它以反映的数据形态。详细细节不是重点。

import osimport requestsimport sqlite3import pandas as pdimport shutil# 下载数据库文件db_url = "https://storage.googleapis.com/benchmarks-artifacts/travel-db/travel2.sqlite"local_file = "travel2.sqlite"backup_file = "travel2.backup.sqlite"overwrite = Falseif 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 reimport numpy as npimport openaifrom langchain_core.tools import tool@tooldef lookup_policy(query):"""查问公司政策，以确定某些选项能否准许。"""

航班治理

定义一个工具来失掉用户的航班信息，而后定义一些工具来搜查航班和治理用户的预订信息，这些信息存储在 SQL 数据库中。

咱们经常使用 ensure_config​ 来经过性能参数传递 passenger_id。言语模型不须要显式提供这些信息，它们会在图的每次调用中提供，以确保每个用户不可访问其余乘客的预订信息。

from langchain_core.runnables import ensure_configfrom typing import Optionalimport sqlite3import pytzfrom datetime import datetime, timedelta, date@tooldef fetch_user_flight_information():"""失掉用户的所无机票信息，包括航班概略和座位调配。"""@tooldef search_flights(departure_airport=None,arrival_airport=None,start_time=None,end_time=None,limit=20,):"""依据登程机场、抵达机场和登程期间范围来搜查航班。"""@tooldef update_ticket_to_new_flight(ticket_no, new_flight_id):"""将用户的机票更新到一个新的有效航班上。"""@tooldef cancel_ticket(ticket_no):"""敞开用户的机票，并从数据库中移除。"""

租车服务

用户预订了航班后，或许须要租车服务。定义一些工具，让用户能够在目的地搜查和预订汽车。

from typing import Optional, Unionfrom datetime import datetime, date@tooldef 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]: 婚配搜查条件的租车服务列表。"""@tooldef book_car_rental(rental_id):"""经过租车ID来预订租车服务。参数：rental_id (int): 要预订的租车服务的ID。前往：str: 预订成功与否的信息。"""@tooldef 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: 更新成功与否的信息。"""@tooldef cancel_car_rental(rental_id):"""经过租车ID来敞开租车服务。参数：rental_id (int): 要敞开的租车服务的ID。前往：str: 敞开成功与否的信息。"""

酒店预订

用户须要住宿，因此定义一些工具来搜查和治理酒店预订。

@tooldef 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]: 合乎搜查条件的酒店列表。"""@tooldef book_hotel(hotel_id):"""经过酒店ID启动预订。参数：hotel_id (int): 要预订的酒店的ID。前往：str: 预订成功与否的信息。"""@tooldef 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: 更新成功与否的信息。"""@tooldef cancel_hotel(hotel_id):"""经过酒店ID敞开酒店预订。参数：hotel_id (int): 要敞开预订的酒店的ID。前往：str: 敞开成功与否的信息。"""

远足优惠

最后，定义一些工具，让用户在抵达目的地后搜查优惠并启动预订。

@tooldef search_trip_recommendations(locatinotallow=None,name=None,keywords=None,):"""依据位置、称号和主要词搜查游览介绍。参数：location (Optional[str]): 游览介绍的地点。name (Optional[str]): 游览介绍的名字。keywords (Optional[str]): 与游览介绍关系的主要词。前往：list[dict]: 合乎搜查条件的游览介绍列表。"""@tooldef book_excursion(recommendation_id):"""经过介绍ID预订远足优惠。参数：recommendation_id (int): 要预订的游览介绍的ID。前往：str: 预订成功与否的信息。"""@tooldef update_excursion(recommendation_id, details):"""经过介绍ID更新游览介绍的细节。参数：recommendation_id (int): 要更新的游览介绍的ID。details (str): 游览介绍的新细节。前往：str: 更新成功与否的信息。"""@tooldef cancel_excursion(recommendation_id):"""经过介绍ID敞开游览介绍。参数：recommendation_id (int): 要敞开的游览介绍的ID。前往：str: 敞开成功与否的信息。"""

适用工具

定义一些辅佐函数，以便在调试环节中赞美图形中的信息显示，并为工具节点减少失误处置（经过将失误减少到聊天记载中）。

from langgraph.prebuilt import ToolNodefrom langchain_core.runnables import RunnableLambdadef handle_tool_error(state):error = state.get("error")tool_calls = state["messages"][-1].tool_callsreturn {"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, AnyMessagefrom typing_extensions import TypedDictfrom typing import Annotatedclass State(TypedDict):messages: Annotated[list[AnyMessage], add_messages]

代理

而后，咱们定义助手函数。这个函数接纳图的形态，将其格局化为提醒，而后调用一个大型言语模型（LLM）来预测最佳的照应。

from langchain_core.runnables import Runnable, RunnableConfigfrom langchain_community.tools.tavily_search import TavilySearchResultsfrom langchain_anthropic import ChatAnthropicfrom langchain_core.prompts import ChatPromptTemplateclass Assistant:def __init__(self, runnable: Runnable):self.runnable = runnabledef __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.contentor isinstance(result.content, list)and not result.content[0].get("text")):messages = state["messages"] + [("user", "请给出一个实在的输入。")]state = {**state, "messages": messages}else:breakreturn {"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 SqliteSaverfrom langgraph.graph import StateGraph, ENDfrom langgraph.prebuilt import tools_condition, ToolNodebuilder = 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, displaytry:display(Image(part_1_graph.get_graph(xray=True).draw_mermaid_png()))except:# 这须要一些额外的依赖项，是可选的pass

示例对话

如今，让咱们经过一系列对话示例来测试咱们的聊天机器人。

import uuidimport 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)

第一局部回忆

咱们的便捷助手体现得还不错！它能够正当地回答一切疑问，极速地在高低文中做出回应，并成功地执行了咱们一切的义务。你可以经过检查LangSmith的示例跟踪[2]来更好地了解LLM在上述交互中是如何被提醒的。

假设这是一个便捷的问答机器人，咱们或许会对上述结果感到满意。由于咱们的客户支持机器人是代表用户采取执行，因此它的一些行为有点令人担心：

在下一节中，咱们将处置前两个疑问，等候你的继续关注！。

[1] 详细代码:

[2] LangSmith的示例跟踪:

本文转载自​​，作者：