yowayimono
客户支持
编辑
2025-02-11
后端
00
请注意，本文编写于 88 天前，最后修改于 88 天前，其中某些信息可能已经过时。

目录

前置条件

客户支持机器人可以通过处理常规问题来解放团队的时间，但构建一个能够可靠处理多样化任务且不会让用户感到沮丧的机器人却颇具挑战。

在本教程中，你将为一家航空公司构建一个客户支持机器人，帮助用户进行旅行安排的研究和预订。你将学习如何使用 LangGraph 的中断和检查点功能，以及更复杂的状态管理，来组织助手的工具并管理用户的航班预订、酒店预订、租车和活动项目。假设你已经熟悉 LangGraph 入门教程中的概念。

完成本教程后，你将拥有一个可以正常运行的聊天机器人，并对 LangGraph 的关键概念和架构有深入的理解。你可以将这些设计模式应用到其他 AI 项目中。

你的最终聊天机器人将如下图所示：

开始吧！

前置条件

首先，设置你的开发环境。我们将安装本教程所需的依赖项，下载测试数据库，并定义将在每个部分中重复使用的工具。

安装依赖项并设置环境

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

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("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

为 LangGraph 开发设置 LangSmith

注册 LangSmith，以便快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许你使用追踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——阅读此处了解如何入门。

下载并准备数据库

python
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"
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()
    with open(local_file, "wb") as f:
        f.write(response.content)
    shutil.copy(local_file, backup_file)

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

    for table_name, df in tdf.items():
        df.to_sql(table_name, conn, if_exists="replace", index=False)
    conn.commit()
    conn.close()

    return file

db = update_dates(local_file)

定义工具

接下来，定义助手将使用的工具，包括搜索航空公司政策和管理航班、酒店、租车和活动预订的工具。我们将在教程的每个部分中重复使用这些工具。

查询公司政策

助手可以检索政策信息以回答用户的问题。请注意，这些政策的执行仍需在工具/API 中完成，因为 LLM 可能会忽略这些政策。

python
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]
        )
        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:
    """查询公司政策，检查是否允许执行某些操作。
    在执行任何航班更改或其他“写入”操作之前使用此工具。
    """
    docs = retriever.query(query, k=2)
    return "\n\n".join([doc["page_content"] for doc in docs])

航班相关工具

定义工具以检索用户的航班信息，并允许助手搜索和管理乘客的预订，这些信息存储在 SQLite 数据库中。

我们还可以通过运行时配置中的 RunnableConfig 访问 passenger_id，这样每个用户就无法访问其他乘客的预订信息。

兼容性

本教程需要 langchain-core>=0.2.16 才能使用注入的 RunnableConfig。在此之前，你需要使用 ensure_config 从上下文中获取配置。

python
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]:
    """获取用户的机票信息以及对应的航班信息和座位分配。

    返回：
        一个包含字典的列表，每个字典包含用户的机票详情、
        相关的航班信息以及属于该用户的每张机票的座位分配。
    """
    configuration = config.get("configurable", {})
    passenger_id = configuration.get("passenger_id", None)
    if not passenger_id:
        raise ValueError("未配置乘客 ID。")

    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]:
    """根据出发机场、到达机场和出发时间范围搜索航班。"""
    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:
    """将用户的机票更新为新的有效航班。"""
    configuration = config.get("configurable", {})
    passenger_id = configuration.get("passenger_id", None)
    if not passenger_id:
        raise ValueError("未配置乘客 ID。")

    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 "提供的新航班 ID 无效。"
    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"不允许重新安排到距当前时间少于3小时的航班。选定的航班时间为 {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 "未找到给定机票号的现有机票。"

    # 检查当前登录用户是否实际拥有这张机票
    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"当前登录的乘客（ID {passenger_id}）不是机票 {ticket_no} 的所有者"

    # 在实际应用中，你可能需要添加更多检查以强制执行业务逻辑，
    # 例如“新出发机场是否与当前机票匹配”等。
    # 虽然最好通过‘类型提示’提前向 LLM 传达政策，
    # 但你**也**需要确保你的 API 强制执行有效的操作
    cursor.execute(
        "UPDATE ticket_flights SET flight_id = ? WHERE ticket_no = ?",
        (new_flight_id, ticket_no),
    )
    conn.commit()

    cursor.close()
    conn.close()
    return "机票已成功更新到新航班。"

@tool
def cancel_ticket(ticket_no: str, *, config: RunnableConfig) -> str:
    """取消用户的机票并从数据库中删除。"""
    configuration = config.get("configurable", {})
    passenger_id = configuration.get("passenger_id", None)
    if not passenger_id:
        raise ValueError("未配置乘客 ID。")
    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 "未找到给定机票号的现有机票。"

    # 检查当前登录用户是否实际拥有这张机票
    cursor.execute(
        "SELECT ticket_no 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"当前登录的乘客（ID {passenger_id}）不是机票 {ticket_no} 的所有者"

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

    if cursor.rowcount > 0:
        conn.close()
        return "机票已成功取消。"
    else:
        conn.close()
        return "未找到与给定机票号匹配的机票。"

租车工具

用户预订航班后，他们可能需要安排交通。定义一些“租车”工具，让用户可以搜索和预订目的地的租车服务。

python
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]:
    """
    根据位置、名称、价格等级、开始日期和结束日期搜索租车服务。

    参数：
        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]: 匹配搜索条件的租车字典列表。
    """
    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}%")
    # 为了简化教程，我们假设数据库中已有数据
    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:
    """
    根据租车 ID 预订租车服务。

    参数：
        rental_id (int): 要预订的租车服务 ID。

    返回：
        str: 表示租车服务是否成功预订的消息。
    """
    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"租车服务 {rental_id} 已成功预订。"
    else:
        conn.close()
        return f"未找到 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:
    """
    根据租车 ID 更新租车服务的开始和结束日期。

    参数：
        rental_id (int): 要更新的租车服务 ID。
        start_date (Optional[Union[datetime, date]]): 新的开始日期。默认为 None。
        end_date (Optional[Union[datetime, date]]): 新的结束日期。默认为 None。

    返回：
        str: 表示租车服务是否成功更新的消息。
    """
    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"租车服务 {rental_id} 已成功更新。"
    else:
        conn.close()
        return f"未找到 ID 为 {rental_id} 的租车服务。"

@tool
def cancel_car_rental(rental_id: int) -> str:
    """
    根据租车 ID 取消租车服务。

    参数：
        rental_id (int): 要取消的租车服务 ID。

    返回：
        str: 表示租车服务是否成功取消的消息。
    """
    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"租车服务 {rental_id} 已成功取消。"
    else:
        conn.close()
        return f"未找到 ID 为 {rental_id} 的租车服务。"

酒店工具

用户需要住宿！定义一些工具来搜索和管理酒店预订。

python
@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]:
    """
    根据位置、名称、价格等级、入住日期和退房日期搜索酒店。

    参数：
        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]: 匹配搜索条件的酒店字典列表。
    """
    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}%")
    # 为了简化教程，我们假设数据库中已有数据
    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:
    """
    根据酒店 ID 预订酒店。

    参数：
        hotel_id (int): 要预订的酒店 ID。

    返回：
        str: 表示酒店是否成功预订的消息。
    """
    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_id} 已成功预订。"
    else:
        conn.close()
        return f"未找到 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:
    """
    根据酒店 ID 更新酒店的入住和退房日期。

    参数：
        hotel_id (int): 要更新的酒店 ID。
        checkin_date (Optional[Union[datetime, date]]): 新的入住日期。默认为 None。
        checkout_date (Optional[Union[datetime, date]]): 新的退房日期。默认为 None。

    返回：
        str: 表示酒店是否成功更新的消息。
    """
    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_id} 已成功更新。"
    else:
        conn.close()
        return f"未找到 ID 为 {hotel_id} 的酒店。"

@tool
def cancel_hotel(hotel_id: int) -> str:
    """
    根据酒店 ID 取消酒店预订。

    参数：
        hotel_id (int): 要取消的酒店 ID。

    返回：
        str: 表示酒店是否成功取消的消息。
    """
    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_id} 已成功取消。"
    else:
        conn.close()
        return f"未找到 ID 为 {hotel_id} 的酒店。"

活动工具

最后，定义一些工具，让用户可以搜索目的地的活动并进行预订。

python
@tool
def search_trip_recommendations(
    location: Optional[str] = None,
    name: Optional[str] = None,
    keywords: Optional[str] = None,
) -> list[dict]:
    """
    根据位置、名称和关键词搜索旅行建议。

    参数：
        location (Optional[str]): 旅行建议的位置。默认为 None。
        name (Optional[str]): 旅行建议的名称。默认为 None。
        keywords (Optional[str]): 与旅行建议相关的关键词。默认为 None。

    返回：
        list[dict]: 匹配搜索条件的旅行建议字典列表。
    """
    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:
    """
    根据建议 ID 预订活动。

    参数：
        recommendation_id (int): 要预订的旅行建议 ID。

    返回：
        str: 表示旅行建议是否成功预订的消息。
    """
    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"旅行建议 {recommendation_id} 已成功预订。"
    else:
        conn.close()
        return f"未找到 ID 为 {recommendation_id} 的旅行建议。"

@tool
def update_excursion(recommendation_id: int, details: str) -> str:
    """
    根据建议 ID 更新旅行建议的详细信息。

    参数：
        recommendation_id (int): 要更新的旅行建议 ID。
        details (str): 新的详细信息。

    返回：
        str: 表示旅行建议是否成功更新的消息。
    """
    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"旅行建议 {recommendation_id} 已成功更新。"
    else:
        conn.close()
        return f"未找到 ID 为 {recommendation_id} 的旅行建议。"

@tool
def cancel_excursion(recommendation_id: int) -> str:
    """
    根据建议 ID 取消旅行建议的预订。

    参数：
        recommendation_id (int): 要取消的旅行建议 ID。

    返回：
        str: 表示旅行建议是否成功取消的消息。
    """
    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"旅行建议 {recommendation_id} 已成功取消。"
    else:
        conn.close()
        return f"未找到 ID 为 {recommendation_id} 的旅行建议。"

实用工具

定义辅助函数以在图中格式化消息，并为工具节点添加错误处理（通过将错误添加到聊天历史中）。

python
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"错误: {repr(error)}\n请修正你的错误。",
                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("当前状态：", 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 的状态为一个包含类型注解的字典，其中包含一个追加的聊天历史记录。这些消息构成了聊天机器人的全部状态，也是我们简单助手所需的全部内容。

python
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 以预测最佳响应。

python
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)
            # 如果 LLM 没有返回任何内容，我们将重新提示它提供实际的响应。
            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", "请提供实际的输出。")]
                state = {**state, "messages": messages}
            else:
                break
        return {"messages": result}

# 使用 Claude-3-sonnet 模型
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\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)

定义图

现在，创建图。图是本部分的助手。

python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import tools_condition

builder = StateGraph(State)

builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,
)
builder.add_edge("tools", "assistant")

memory = MemorySaver()
part_1_graph = builder.compile(checkpointer=memory)

示例对话

现在，让我们尝试一下我们强大的聊天机器人！让我们在以下对话中运行它。如果它遇到“递归限制”，这意味着代理无法在分配的步骤内获得答案。没关系！在本教程的后续部分中，我们将有更多技巧。

python
import shutil
import uuid

# 用备份文件更新，以便我们可以从原始位置重新开始每个部分
db = update_dates(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 追踪以更好地了解 LLM 在上述交互中的表现。

如果这是一个简单的问答机器人，我们可能会对结果感到满意。由于我们的客户支持机器人代表用户执行操作，有些行为可能令人担忧：

  • 代理在没有用户确认的情况下预订了租车，然后不得不取消并重新预订。用户应该在预订前有最终决定权，以避免不必要的操作。
  • 代理在搜索推荐时遇到了困难。我们可以通过改进提示和示例来改善这一点，但随着我们添加更多工具，提示可能会变得过大，使代理感到不知所措。
  • 代理不得不执行显式搜索以获取用户的相关信息。我们可以通过立即获取用户的相关旅行信息来节省时间，这样代理可以直接回复。

在下一节中，我们将解决前两个问题。

第二部分：添加确认

当代理代表用户执行操作时，用户应在几乎所有情况下对是否继续执行操作有最终决定权。否则，代理的任何小错误（或它屈服于的任何提示注入）都可能对用户造成实际损害。

在本部分中，我们将使用 interrupt_before 在使用工具之前暂停图并返回控制权给用户。

定义状态和代理

我们的图状态和 LLM 调用几乎与第一部分相同，但有以下例外：

  • 我们添加了一个 user_info 字段，图将立即填充该字段，这样代理就不必通过工具来获取用户信息。
python
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 not result.tool_calls and (
                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}

llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)

assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是瑞士航空的客户支持助手。使用提供的工具搜索航班、公司政策和其他信息以协助用户的查询。当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n\n当前用户：\n<User>\n{user_info}\n</User>\n\n当前时间：{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)

定义图

现在，创建图。对第一部分进行两项更改：

  1. 添加一个 fetch_user_info 节点，该节点首先运行，这样我们的代理可以在不采取行动的情况下查看用户的航班信息。
  2. 在使用“工具”节点之前添加一个中断，以便用户可以在执行操作之前批准或拒绝。
python
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({})}

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,
    interrupt_before=["tools"],
)

示例对话

现在，让我们尝试一下我们新修订的聊天机器人！让我们在以下对话中运行它。

python
import shutil
import uuid

db = update_dates(db)
thread_id = str(uuid.uuid4())

config = {
    "configurable": {
        "passenger_id": "3442 587242",
        "thread_id": thread_id,
    }
}

_printed = set()
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:
        try:
            user_input = input(
                "你是否批准上述操作？输入 'y' 继续；否则，解释你请求的更改。\n\n"
            )
        except:
            user_input = "y"
        if user_input.strip() == "y":
            result = part_2_graph.invoke(None, config)
        else:
            result = part_2_graph.invoke(
                {
                    "messages": [
                        ToolMessage(
                            tool_call_id=event["messages"][-1].tool_calls[0]["id"],
                            content=f"API 调用被用户拒绝。原因：'{user_input}'. 继续协助，考虑用户的输入。",
                        )
                    ]
                },
                config,
            )
        snapshot = part_2_graph.get_state(config)

第二部分回顾

现在，我们的代理能够节省一步来回复航班信息，并且我们完全控制了执行哪些操作。这一切都是通过 LangGraph 的中断和检查点实现的。你可以查看示例 LangSmith 追踪以更好地了解图的运行情况。注意，在这个追踪中，你通常通过调用图并提供正确的配置来恢复流程。

这个图运行得很好！我们可能对这个设计感到满意。代码是自包含的，行为也符合预期。

在下一节中，我们将通过仅在实际修改用户预订的敏感工具上应用中断，来改进我们的中断策略，从而平衡用户控制和对话流程。

第三部分：条件中断

在本部分中，我们将完善我们的中断策略，将工具分类为安全（只读）或敏感（数据修改）。我们只在敏感工具上应用中断，允许机器人自主处理简单的查询。

这平衡了用户控制和对话流程，但随着我们添加更多工具，我们的单个图可能会变得过于复杂，无法适用于这种“扁平”结构。

定义状态

和往常一样，从定义图状态开始。我们的状态和 LLM 调用与第二部分相同。

python
from typing import Annotated, Literal

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]:
    """推送或弹出状态。"""
    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,
    ]

助手

这次我们将为每个工作流创建一个助手。这意味着：

  • 航班预订助手
  • 酒店预订助手
  • 租车预订助手
  • 活动助手
  • 以及一个“主要助手”来在这些之间路由

如果注意到，这类似于我们在多代理示例中提到的监督者设计模式。

定义可运行对象以驱动每个助手。每个可运行对象都有一个提示、一个 LLM 和一组特定于该助手的工具。每个专门的/委托的助手还可以调用 CompleteOrEscalate 工具，以指示控制流应传递回主要助手，无论是因为任务已完成还是因为用户需要超出该特定工作流范围的帮助。

python
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", "请提供实际的输出。")]
                state = {**state, "messages": messages}
            else:
                break
        return {"messages": result}

# 航班预订助手

flight_booking_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是处理航班更新的专门助手。当用户需要更新预订时，主要助手会将工作委托给你。\n确认更新后的航班详细信息，并告知客户任何额外费用。\n当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n如果需要更多信息或客户改变主意，将任务传递回主要助手。\n记住，预订完成之前不算完成，直到你成功调用了相关工具。\n\n当前时间：{time}。\n\n如果用户需要帮助，且你的工具不适用，请使用 'CompleteOrEscalate' 将对话传递回主机助手。不要浪费用户的时间。不要编造无效的工具或功能。",
        ),
        ("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]
)

# 酒店预订助手

book_hotel_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是处理酒店预订的专门助手。当用户需要预订酒店时，主要助手会将工作委托给你。\n搜索符合用户偏好的可用酒店，并确认预订详细信息。\n当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n如果需要更多信息或客户改变主意，将任务传递回主要助手。\n记住，预订完成之前不算完成，直到你成功调用了相关工具。\n\n当前时间：{time}。\n\n如果用户需要帮助，且你的工具不适用，请使用 'CompleteOrEscalate' 将对话传递回主机助手。\n不要浪费用户的时间。不要编造无效的工具或功能。",
        ),
        ("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]
)

# 租车助手

book_car_rental_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是处理租车预订的专门助手。当用户需要预订租车时，主要助手会将工作委托给你。\n搜索符合用户偏好的可用租车选项，并确认预订详细信息。\n当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n如果需要更多信息或客户改变主意，将任务传递回主要助手。\n记住，预订完成之前不算完成，直到你成功调用了相关工具。\n\n当前时间：{time}。\n\n如果用户需要帮助，且你的工具不适用，请使用 'CompleteOrEscalate' 将对话传递回主机助手。\n不要浪费用户的时间。不要编造无效的工具或功能。",
        ),
        ("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]
)

# 活动助手

book_excursion_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是处理旅行推荐的专门助手。当用户需要预订推荐的旅行时，主要助手会将工作委托给你。\n搜索符合用户偏好的可用旅行推荐，并确认预订详细信息。\n如果需要更多信息或客户改变主意，将任务传递回主要助手。\n当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n记住，预订完成之前不算完成，直到你成功调用了相关工具。\n\n当前时间：{time}。\n\n如果用户需要帮助，且你的工具不适用，请使用 'CompleteOrEscalate' 将对话传递回主机助手。\n不要浪费用户的时间。不要编造无效的工具或功能。",
        ),
        ("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]
)

# 主要助手

class ToFlightBookingAssistant(BaseModel):
    """将工作传递给专门的助手以处理航班更新和取消。"""
    request: str = Field(
        description="更新航班助手在继续之前应澄清的任何后续问题。"
    )

class ToBookCarRental(BaseModel):
    """将工作传递给专门的助手以处理租车预订。"""
    location: str = Field(
        description="用户想要租车的地点。"
    )
    start_date: str = Field(description="租车的开始日期。")
    end_date: str = Field(description="租车的结束日期。")
    request: str = Field(
        description="用户关于租车预订的任何其他信息或请求。"
    )

class ToHotelBookingAssistant(BaseModel):
    """将工作传递给专门的助手以处理酒店预订。"""
    location: str = Field(
        description="用户想要预订酒店的地点。"
    )
    checkin_date: str = Field(description="酒店的入住日期。")
    checkout_date: str = Field(description="酒店的退房日期。")
    request: str = Field(
        description="用户关于酒店预订的任何其他信息或请求。"
    )

class ToBookExcursion(BaseModel):
    """将工作传递给专门的助手以处理旅行推荐和其他活动预订。"""
    location: str = Field(
        description="用户想要预订推荐旅行的地点。"
    )
    request: str = Field(
        description="用户关于旅行推荐的任何其他信息或请求。"
    )

primary_assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是瑞士航空的客户支持助手。你的主要角色是搜索航班信息和公司政策以回答客户查询。\n如果客户要求更新或取消航班、预订租车、预订酒店或获取旅行推荐，请将任务委托给相应的专门助手，方法是调用相应的工具。你无法自己执行这些类型的变化。\n只有专门的助手被允许代表用户执行这些操作。\n用户不知道不同的专门助手，所以不要提及它们；只需通过函数调用静默委托。\n提供详细的客户信息，并始终在得出信息不可用之前再次检查数据库。\n当搜索时，要持之以恒。如果第一次搜索没有结果，请扩大搜索范围，然后再放弃。\n\n当前用户航班信息：\n<Flights>\n{user_info}\n</Flights>\n\n当前时间：{time}。",
        ),
        ("placeholder", "{messages}"),
    ]
).partial(time=datetime.now)

primary_assistant_tools = [
    TavilySearchResults(max_results=1),
    search_flights,
    lookup_policy,
    ToFlightBookingAssistant,
    ToBookCarRental,
    ToHotelBookingAssistant,
    ToBookExcursion,
]
assistant_runnable = primary_assistant_prompt | llm.bind_tools(primary_assistant_tools)

创建助手

我们几乎准备好创建图了。和以前一样，我们将在状态中预先填充用户的当前信息。

python
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({})}

builder.add_node("fetch_user_info", user_info)
builder.add_edge(START, "fetch_user_info")

现在，开始添加我们的专门工作流。每个迷你工作流看起来非常相似：

  1. enter_*：使用上面定义的 create_entry_node 实用函数添加一个 ToolMessage，指示新的专门助手现在控制。
  2. *_assistant：提示 + LLM 组合，接收当前状态并决定使用工具、向用户提问或结束工作流（返回到主要助手）。
  3. *_safe_tools：助手可以使用的“只读”工具，无需用户确认。
  4. *_sensitive_tools：具有“写入”访问权限的工具，需要用户确认（我们将在编译图时为这些工具分配 interrupt_before）。
  5. leave_skill：弹出 dialog_state 以指示主要助手重新控制。

由于它们的相似性，我们可以定义一个工厂函数来生成这些。由于这是一个教程，我们将明确地定义它们。

航班预订助手

python
builder.add_node(
    "enter_update_flight",
    create_entry_node("航班更新助手", "update_flight"),
)
builder.add_node("update_flight", Assistant(update_flight_runnable))
builder.add_edge("enter_update_flight", "update_flight")
builder.add_node(
    "update_flight_sensitive_tools",
    create_tool_node_with_fallback(update_flight_sensitive_tools),
)
builder.add_node(
    "update_flight_safe_tools",
    create_tool_node_with_fallback(update_flight_safe_tools),
)

def route_update_flight(state: State):
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    safe_toolnames = [t.name for t in update_flight_safe_tools]
    if all(tc["name"] in safe_toolnames for tc in tool_calls):
        return "update_flight_safe_tools"
    return "update_flight_sensitive_tools"

builder.add_edge("update_flight_sensitive_tools", "update_flight")
builder.add_edge("update_flight_safe_tools", "update_flight")
builder.add_conditional_edges(
    "update_flight",
    route_update_flight,
    ["update_flight_sensitive_tools", "update_flight_safe_tools", "leave_skill", END],
)

租车助手

python
builder.add_node(
    "enter_book_car_rental",
    create_entry_node("租车助手", "book_car_rental"),
)
builder.add_node("book_car_rental", Assistant(book_car_rental_runnable))
builder.add_edge("enter_book_car_rental", "book_car_rental")
builder.add_node(
    "book_car_rental_safe_tools",
    create_tool_node_with_fallback(book_car_rental_safe_tools),
)
builder.add_node(
    "book_car_rental_sensitive_tools",
    create_tool_node_with_fallback(book_car_rental_sensitive_tools),
)

def route_book_car_rental(state: State):
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    safe_toolnames = [t.name for t in book_car_rental_safe_tools]
    if all(tc["name"] in safe_toolnames for tc in tool_calls):
        return "book_car_rental_safe_tools"
    return "book_car_rental_sensitive_tools"

builder.add_edge("book_car_rental_sensitive_tools", "book_car_rental")
builder.add_edge("book_car_rental_safe_tools", "book_car_rental")
builder.add_conditional_edges(
    "book_car_rental",
    route_book_car_rental,
    [
        "book_car_rental_safe_tools",
        "book_car_rental_sensitive_tools",
        "leave_skill",
        END,
    ],
)

酒店预订助手

python
builder.add_node(
    "enter_book_hotel",
    create_entry_node("酒店预订助手", "book_hotel"),
)
builder.add_node("book_hotel", Assistant(book_hotel_runnable))
builder.add_edge("enter_book_hotel", "book_hotel")
builder.add_node(
    "book_hotel_safe_tools",
    create_tool_node_with_fallback(book_hotel_safe_tools),
)
builder.add_node(
    "book_hotel_sensitive_tools",
    create_tool_node_with_fallback(book_hotel_sensitive_tools),
)

def route_book_hotel(state: State):
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    tool_names = [t.name for t in book_hotel_safe_tools]
    if all(tc["name"] in tool_names for tc in tool_calls):
        return "book_hotel_safe_tools"
    return "book_hotel_sensitive_tools"

builder.add_edge("book_hotel_sensitive_tools", "book_hotel")
builder.add_edge("book_hotel_safe_tools", "book_hotel")
builder.add_conditional_edges(
    "book_hotel",
    route_book_hotel,
    ["leave_skill", "book_hotel_safe_tools", "book_hotel_sensitive_tools", END],
)

活动助手

python
builder.add_node(
    "enter_book_excursion",
    create_entry_node("旅行推荐助手", "book_excursion"),
)
builder.add_node("book_excursion", Assistant(book_excursion_runnable))
builder.add_edge("enter_book_excursion", "book_excursion")
builder.add_node(
    "book_excursion_safe_tools",
    create_tool_node_with_fallback(book_excursion_safe_tools),
)
builder.add_node(
    "book_excursion_sensitive_tools",
    create_tool_node_with_fallback(book_excursion_sensitive_tools),
)

def route_book_excursion(state: State):
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    tool_names = [t.name for t in book_excursion_safe_tools]
    if all(tc["name"] in tool_names for tc in tool_calls):
        return "book_excursion_safe_tools"
    return "book_excursion_sensitive_tools"

builder.add_edge("book_excursion_sensitive_tools", "book_excursion")
builder.add_edge("book_excursion_safe_tools", "book_excursion")
builder.add_conditional_edges(
    "book_excursion",
    route_book_excursion,
    ["book_excursion_safe_tools", "book_excursion_sensitive_tools", "leave_skill", END],
)

主要助手

python
builder.add_node("primary_assistant", Assistant(assistant_runnable))
builder.add_node(
    "primary_assistant_tools",
    create_tool_node_with_fallback(primary_assistant_tools),
)

def route_primary_assistant(state: State):
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    if tool_calls:
        if tool_calls[0]["name"] == ToFlightBookingAssistant.__name__:
            return "enter_update_flight"
        elif tool_calls[0]["name"] == ToBookCarRental.__name__:
            return "enter_book_car_rental"
        elif tool_calls[0]["name"] == ToHotelBookingAssistant.__name__:
            return "enter_book_hotel"
        elif tool_calls[0]["name"] == ToBookExcursion.__name__:
            return "enter_book_excursion"
        return "primary_assistant_tools"
    raise ValueError("无效的路由")

builder.add_conditional_edges(
    "primary_assistant",
    route_primary_assistant,
    [
        "enter_update_flight",
        "enter_book_car_rental",
        "enter_book_hotel",
        "enter_book_excursion",
        "primary_assistant_tools",
        END,
    ],
)
builder.add_edge("primary_assistant_tools", "primary_assistant")

定义图

现在，定义一个函数以在任何专门工作流退出时弹出 dialog_state 并返回到主要助手。

python
def pop_dialog_state(state: State) -> dict:
    """弹出对话栈并返回到主机助手。

    这允许图显式跟踪对话流并委托控制给特定的子图。
    """
    messages = []
    if state["messages"][-1].tool_calls:
        messages.append(
            ToolMessage(
                content="恢复与主机助手的对话。请根据过去的对话协助用户。",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        )
    return {
        "dialog_state": "pop",
        "messages": messages,
    }

builder.add_node("leave_skill", pop_dialog_state)
builder.add_edge("leave_skill", "primary_assistant")

编译图

python
memory = MemorySaver()
part_4_graph = builder.compile(
    checkpointer=memory,
    interrupt_before=[
        "update_flight_sensitive_tools",
        "book_car_rental_sensitive_tools",
        "book_hotel_sensitive_tools",
        "book_excursion_sensitive_tools",
    ],
)

对话

让我们在以下对话中运行它。这次，我们将有更少的确认。

python
import shutil
import uuid

db = update_dates(db)
thread_id = str(uuid.uuid4())

config = {
    "configurable": {
        "passenger_id": "3442 587242",
        "thread_id": thread_id,
    }
}

_printed = set()
for question in tutorial_questions:
    events = part_4_graph.stream(
        {"messages": ("user", question)}, config, stream_mode="values"
    )
    for event in events:
        _print_event(event, _printed)
    snapshot = part_4_graph.get_state(config)
    while snapshot.next:
        try:
            user_input = input(
                "你是否批准上述操作？输入 'y' 继续；否则，解释你请求的更改。\n\n"
            )
        except:
            user_input = "y"
        if user_input.strip() == "y":
            result = part_4_graph.invoke(None, config)
        else:
            result = part_4_graph.invoke(
                {
                    "messages": [
                        ToolMessage(
                            tool_call_id=event["messages"][-1].tool_calls[0]["id"],
                            content=f"API 调用被用户拒绝。原因：'{user_input}'. 继续协助，考虑用户的输入。",
                        )
                    ]
                },
                config,
            )
        snapshot = part_4_graph.get_state(config)

结论

你现在已经开发了一个能够处理多种任务的客户支持机器人，使用的是专注于特定工作流的子图。更重要的是，你已经学会了如何使用 LangGraph 的核心功能（如中断和检查点）来根据产品需求设计和重构应用程序。

上述示例可能尚未针对你的特定需求进行优化——LLM 会犯错，每个流程都可以通过更好的提示和实验来提高可靠性。一旦你构建了初始的客户支持机器人，下一步就是添加评估功能，以便更有信心地改进系统。查看文档和教程以了解更多关于如何进一步优化和扩展客户支持机器人的信息。

本文作者:yowayimono

本文链接:

版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！