LangGraph实战 可控的AI航空客服助手

当智能助手代表用户执行操作时，用户简直总是应该对能否执行这些操作领有最终选择权。否则，即使是智能助手的一点小错误，或是它未能抵御的任何指令注入，都或者对用户形成实践侵害。

在这局部，咱们将应用LangGraph的interrupt_before性能，在执行任何工具之前，暂停流程并把控制权交还给用户。

您的流程图或者看起来像这样：

流程图示例

和之前一样，咱们首先定义形态：

形态与智能助手

咱们的流程图形态和LLM调用与第一局部基本相反，除了：

from typing import Annotatedfrom langchain_anthropic import ChatAnthropicfrom langchain_community.tools.tavily_search import TavilySearchResultsfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import Runnable, RunnableConfigfrom typing_extensions import TypedDictfrom langgraph.graph.message import AnyMessage, add_messagesclass State(TypedDict):messages: Annotated[list[AnyMessage], add_messages]user_info: strclass Assistant:def __init__(self, runnable: Runnable):self.runnable = runnabledef __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.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)# 你也可以经常使用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 SqliteSaverfrom langgraph.graph import END, StateGraphfrom langgraph.prebuilt import ToolNode, tools_conditionbuilder = 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, displaytry:display(Image(part_2_graph.get_graph(xray=True).draw_mermaid_png()))except:# 这须要一些额外的依赖项，并且是可选的pass

流程图示例2

示例对话

如今，让咱们经过以下对话示例来测试咱们新订正的聊天机器人。

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

第二局部回忆

如今，咱们的智能助手能够节俭一步来照应咱们的航班概略。咱们还齐全控制了执行的操作。这一切都是经过LangGraph的interrupts和checkpointers成功的。终止暂停了流程图的执行，其形态经常使用性能的审核点器安保地耐久化。用户随后可以在任何时刻经过经常使用正确的性能运转它来进行它。

检查一个LangSmith示例跟踪，以更好地理解流程图是如何运转的。留意从这个跟踪中，你理论经过经常使用(None, config)调用流程图来复原流程。形态从审核点加载，就像它从未被终止过一样。

这个流程图上班得很好！但当须要咱们介入每一个智能助手的执行的要求，十分影响经常使用体验，并且助手在执行查问等举措时并不会影响实践业务。