LangGraph實戰：可控的AI航空客服助手

ermulong

發布于 2024-5-20 15:13

瀏覽

0收藏

當智能助手代表用戶執行操作時，用戶幾乎總是應該對是否執行這些操作擁有最終決定權。否則，即使是智能助手的一點小失誤，或是它未能抵御的任何指令注入，都可能對用戶造成實際損害。

在這部分，我們將利用LangGraph的interrupt_before功能，在執行任何工具之前，暫停流程并把控制權交還給用戶。

您的流程圖可能看起來像這樣：

LangGraph實戰：可控的AI航空客服助手-AI.x社區

流程圖示例

和之前一樣，我們首先定義狀態：

狀態與智能助手

我們的流程圖狀態和LLM調用與第一部分基本相同，除了：

我們新增了一個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:
            passenger_id = config.get("passenger_id", None)
            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}

# Haiku模型更快更經濟，但準確性較低
# llm = ChatAnthropic(model="claude-3-haiku-20240307")
llm = ChatAnthropic(model="claude-3-sonnet-20240229", temperature=1)
# 你也可以使用OpenAI或其他模型，盡管你可能需要調整提示
# from langchain_openai import ChatOpenAI

# llm = ChatOpenAI(model="gpt-4-turbo-preview")

assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是一個樂于助人的瑞士航空客戶支持智能助手。"
            "利用提供的工具搜索航班、公司政策和其他信息，以幫助解答用戶的疑問。"
            "在搜索時，要持之以恒。如果首次搜索沒有結果，就擴大你的搜索范圍。"
            "如果搜索依然一無所獲，繼續擴大搜索范圍，不要輕言放棄。"
            "\n\n當前用戶：\n<User>\n{user_info}\n</User>"
            "\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)

定義流程圖

現在，創建流程圖。根據第一部分的反饋，我們做出兩個改變：

在使用工具之前加入一個中斷點。
在第一個節點中明確填充用戶狀態，這樣智能助手就不必通過使用工具來了解用戶信息。

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(State)

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

# 新增：fetch_user_info節點首先執行，這意味著我們的智能助手可以在
# 不采取任何行動的情況下查看用戶的航班信息
builder.add_node("fetch_user_info", user_info)
builder.set_entry_point("fetch_user_info")
builder.add_node("assistant", Assistant(part_2_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_2_tools))
builder.add_edge("fetch_user_info", "assistant")
builder.add_conditional_edges(
    "assistant", tools_condition, {"action": "action", END: END}
)
builder.add_edge("action", "assistant")

memory = SqliteSaver.from_conn_string(":memory:")
part_2_graph = builder.compile(
    checkpointer=memory,
    # 新增：流程圖在執行“action”節點之前總是暫停。
    # 用戶可以在智能助手繼續之前批準或拒絕（甚至修改請求）
    interrupt_before=["action"],
)

from IPython.display import Image, display

try:
    display(Image(part_2_graph.get_graph(xray=True).draw_mermaid_png()))
except:
    # 這需要一些額外的依賴項，并且是可選的
    pass

LangGraph實戰：可控的AI航空客服助手-AI.x社區

流程圖示例2

示例對話

現在，讓我們通過以下對話示例來測試我們新修訂的聊天機器人。

import shutil
import uuid

# 使用備份文件更新，以便我們可以從每個部分的起始點重新啟動
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_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:
        # 我們遇到了一個中斷！代理正試圖使用一個工具，而用戶可以批準或拒絕它
        # 注意：這段代碼位于你的流程圖之外。通常，你會將輸出流式傳輸到用戶界面。
        # 然后，你會在用戶輸入時，通過API調用觸發一個新的運行。
        user_input = input(
            "你同意上述操作嗎？輸入'y'以繼續；"
            "否則，請說明你請求的更改。\n\n"
        )
        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"],
                            cnotallow=f"API調用被用戶拒絕。理由：'{user_input}'. 繼續協助，考慮用戶的輸入。",
                        )
                    ]
                },
                config,
            )
        snapshot = part_2_graph.get_state(config)