LangGraphのexamplesからエージェントの作り方を学ぶ

こんにちは。ZENKIGENデータサイエンスチームの栗原です。

エージェントは、LLMを利用したアプリケーション開発における現在の主流の一つになっていると思います。
LLMに、用意したツールや特定領域に特化したモデルなどの存在を提示し、それぞれの状況でどれを利用すべきかLLMに判断させることで、複雑なタスクを自動でこなしていくことができます。

最近ではSakana AIから、アイデア創出、実験の実行と結果の要約、論文の執筆、ピアレビューといった科学研究のサイクルを自動的に遂行するAIシステム「The AI Scientist」が発表され話題になりました。
今後もエージェントを活用した開発はますます増えていくことでしょう。
https://sakana.ai/ai-scientist-jp/

今回はLangGraphを用いたエージェント開発を学びたいと思います。
LangGraphはエージェント開発において便利なライブラリです。
https://github.com/langchain-ai/langgraph
LangGraphのGitHubリポジトリには、examplesとして、LangGraphを使ったさまざまな実装が共有されています。
このexamplesの中から Build a Customer Support Bot のnotebookを参考に、エージェントの作り方を学びたいと思います。

本notebookはPart1からPart4で構成されています。
すべて航空会社のカスタマーサポートボットを作成するもので後半に行くほど複雑にはなりますが、より柔軟にタスクを適切にこなせるようになります。

本記事では、LangGraphのnotebookから実装を掻い摘んで紹介します。
実際に動かす場合は、notebookをご参照ください。

Part 1. 単一エージェント

Part 1.では、単一のエージェントでカスタマーサポートボットを作成します。
以下図のように、User からのメッセージを Assistant が受け取り、最終的に返答を返します。
その際に、Assistant が用意されたツールを利用すべきと判断したらツールを実行し、その結果を元に返答を返します。

単一エージェントによる航空会社のカスタマーサポートボット。1つのエージェントが全てのツール群を把握し、実行を管理する。

本ボットを実装する上での、LangGraphを用いる部分を取り上げます。
Part 1.では以下のグラフが構築されます。LangGraphでは、ノードとエッジでグラフを構築し、動作を制御します。

Part 1.で構築されるグラフ。 "assistant" や "tools" がノード、実線矢印がノーマルエッジ、破線矢印が条件付きエッジ。

ノードにはエージェントやツール（を呼び出す関数）が割り当てられ、エッジでノード同士が結ばれます。
エッジには「ノーマルエッジ」（ノードとノードを結び、状態の遷移を行う通常のエッジ）と、「条件付きエッジ」（条件により到達するノードを切り替えるエッジ）があります。

それでは、グラフを定義している実装を見てみましょう。

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

builder = StateGraph(State)

# ノードの定義
# "assistant"というノードに Assistant の役割を付与
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
# "tools"というノードに create_tool_node_with_fallback の役割を付与
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))

# エッジの定義
# 最初に呼び出されるノードを指定
# START -> "assistant" にエッジを張る
builder.add_edge(START, "assistant")
# 条件付きエッジの追加
builder.add_conditional_edges(
    "assistant", # 始端は "assistant" ノード
    tools_condition, # tools_condition によって終端を制御
)
# "tools" -> "assistant" にエッジを張る
builder.add_edge("tools", "assistant")

part_1_graph = builder.compile() # 本記事の説明において不要な部分を一部改変しています

グラフの描画方法

from IPython.display import Image, display

display(Image(part_1_graph.get_graph().draw_mermaid_png()))

assistant は ユーザーからのメッセージを受けて、どのツールを利用すべきか判断し、利用すべきと判断したツールがあれば、tools ノードに遷移、実行結果を返します。
ツールの利用が必要ない、もしくは、ツールを実行した結果を受け取りもうツールの実行は不要と判断したら、__end__ ノードに遷移します。

本ボットを実行した結果がnotebookで共有されており、その一部（を日本語訳したもの）をトグルに記載します。

notebook記載の対話出力例（の日本語訳）

===== Human Message =====
こんにちは、私のフライトは何時ですか？

===== Ai Message =====
こんにちは、フライトの時間を確認するには、まずお客様の航空券情報を調べる必要があります。
お手数ですが、航空券番号または予約番号をお知らせください。
その情報がわかりましたら、フライトの詳細をお調べいたします。

(中略)

===== Ai Message =====
素晴らしい、お客様の航空券7240005432906569を
5月8日12:09pmのパリ発バーゼル行きLX0112便に変更することができました。
新しい航空券の詳細が確認されました。

更新されたご旅行プランについて、他にサポートが必要な場合はお知らせください！

===== Human Message =====
宿泊と交通手段は？

===== Ai Message =====
お客様の日程に合わせてバーゼルで送迎するレンタカーのオプションをご紹介します。
レンタカーの場所、日程、価格帯を調整する必要があればお知らせください。

また、バーゼルでの自由時間があれば、バーゼル近郊のツアーや小旅行、お勧めの旅行なども喜んでお調べします。
その他、旅行プランに必要なことがあればお知らせください！

===== Human Message =====
1週間（7日間）の滞在だから、手頃なホテルがいいです。レンタカーも借りたいです。

===== Ai Message =====
レンタカーについては、Europcarのエコノミー・オプションがバーゼルでの送迎に利用できます。
もし私に予約してほしいとか、他の価格帯や会社が必要であれば、お知らせください。

もちろん、バーゼルでの1週間の滞在に必要なものがあればお知らせください！
ホテル、車、ツアー、その他のアクティビティーの予約を喜んでお手伝いします。

（以降省略）

本notebookでは、「Human Message」側の発話を事前に用意しており、「Ai Message」の返答に合わせて入力していないので、一部「Human Message」側が不自然な部分もあるのですが、宿泊に関するユーザーからのメッセージに対して適切に対応できていな点が見受けられます。
一つのエージェントで多くのツール（今回用意されているツール一覧を以下に記載します）を把握し、実行を管理するのは難しく、適切なツールを実行できない（もしくは不適切なツールを実行してしまう）ことが起こり得ます。

今回用意されているツール群

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 2. Human-in-the-loop

Part 1.では Assistant に挙動を全て任せて、User からのメッセージに対し Assistant がツールの実行をしていました。
Part 2.ではPart 1.の課題に対し、Assistant が実行しようとするアクションに対し人間に確認、実行の可否を尋ねる機構を導入します。

Userにアクションの実行許可を得るフローが追加
Part 2.のグラフは以下のようになります。

Part 2.のグラフ
グラフを定義する実装を見てみましょう。
Part 1.との違いは fetch_user_info ノードを追加しユーザー情報を事前に確認するようにしたことと（ここは Human-in-the-loop とは無関係です）、グラフをコンパイルする際に interrupt_before を指定しブレークポイントを設定したことです。

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

# 変更箇所 : fetch_user_info ノードを追加、最初に実行されるようにし、ユーザー情報を事前に確認できるようにする。
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")

part_2_graph = builder.compile(
    # 変更箇所 : tools を実行する前に必ず停止し、ユーザーに実行の可否を求める機構を追加。
    interrupt_before=["tools"],
)

では、改良されたボットの実行結果を見てみましょう。

notebook記載の対話出力例（の日本語訳）

(省略)

===== Human Message =====
宿泊と交通手段は？

===== Ai Message =====
[{'text': 'もちろん、来週のバーゼルでの最新旅行日程の宿泊と交通の手配をお手伝いさせてください：', 'type': 'text'}, {'id': 'toolu_01PBJ6rZ2P9tvVLWPt5Nrck7', 'input': {'checkin_date': '2024-05-01', 'checkout_date': '2024-05-02', 'location': 'Basel'}, 'name': 'search_hotels', 'type': 'tool_use'}]
Tool Calls:
  search_hotels (toolu_01PBJ6rZ2P9tvVLWPt5Nrck7)
 Call ID: toolu_01PBJ6rZ2P9tvVLWPt5Nrck7
  Args:
    checkin_date: 2024-05-01
    checkout_date: 2024-05-02
    location: Basel
上記のアクションを承認しますか？
続行する場合は'y'を入力し、そうでない場合は変更を要求する内容を説明してください。

y

（省略）

実行しようとしている内容（ツールsearch_hotelsを呼び出し、その際の引数として、checkin_date, checkout_date, locationを渡すこと）を提示し、その実行の承認をユーザーに得るフローが追加されました。

Part 3. 条件付き割り込み

Part 2.でユーザーの承認を得る機構の導入を実現しましたが、現状全てのアクションに対しユーザーの承認を得るフローとなっており、これは少々冗長でUX観点であまりよくないでしょう。
そこで、Part 3.では、ツールを "Safe" と "Sensitive" に分け、"Sensitive" なツールの利用が必要と判断された場合のみ User の承認を得るようにします。

"Sensitive Tools"の実行時にUserの承認を得る。
Part 3.のグラフは以下のようになります。

Part 3.のグラフ
グラフを定義する実装を見てみましょう。
Part 2.との違いは、これまでツール群を part_2_tools でひとまとめにしていたものを、 part_3_safe_tools と part_3_sensitive_tools に分け、part_3_sensitive_tools に属するツールを呼び出す際のみ User の許可を得る機構にしたことです。

part_3_safe_tools

part_3_safe_tools = [
    TavilySearchResults(max_results=1),
    fetch_user_flight_information,
    search_flights,
    lookup_policy,
    search_car_rentals,
    search_hotels,
    search_trip_recommendations,
]

part_3_sensitive_tools

part_3_sensitive_tools = [
    update_ticket_to_new_flight,
    cancel_ticket,
    book_car_rental,
    update_car_rental,
    cancel_car_rental,
    book_hotel,
    update_hotel,
    cancel_hotel,
    book_excursion,
    update_excursion,
    cancel_excursion,
]

from typing import Literal

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_3_assistant_runnable))
builder.add_node("safe_tools", create_tool_node_with_fallback(part_3_safe_tools))
builder.add_node(
    "sensitive_tools", create_tool_node_with_fallback(part_3_sensitive_tools)
)
builder.add_edge("fetch_user_info", "assistant")

# 変更箇所 : "safe_tools" を呼ぶか、"sensitive_tools" を呼ぶかルーティングする関数を追加
def route_tools(state: State) -> Literal["safe_tools", "sensitive_tools", "__end__"]:
    next_node = tools_condition(state)
    # ツールが呼ばれない場合は "END" ノードへ遷移する
    if next_node == END:
        return END
    ai_message = state["messages"][-1]
    
    first_tool_call = ai_message.tool_calls[0]
    # assistant が呼ぶと判断したツール名が sensitive_tool_names 内にあるツール名だった場合、"sensitive_tools" ノードへ遷移。
    # そうでなければ、"safe_tools" ノードへ遷移。
    if first_tool_call["name"] in sensitive_tool_names:
        return "sensitive_tools"
    return "safe_tools"


builder.add_conditional_edges(
    "assistant",
    route_tools, # 上記 route_tools により遷移先ノードを制御
)
builder.add_edge("safe_tools", "assistant")
builder.add_edge("sensitive_tools", "assistant")

part_3_graph = builder.compile(
    interrupt_before=["sensitive_tools"], # "sensitive_tools" を実行する前には必ず停止し、ユーザーに実行の可否を求める。
)

では、改良されたボットの実行結果を見てみましょう。

notebook記載の対話出力例（の日本語訳）

(省略)
===== Ai Message =====
その他、車種や料金の詳細が必要な場合や、これらのレンタカーを予約してほしい場合はお知らせください。

===== Human Message =====
1番安いプランで7日間予約してください。

===== Ai Message =====
[{'text': "もちろん、5月2日から5月9日までの7日間のバーゼル滞在のために、最も安いレンタカーオプションを予約しましょう。", 'type': 'text'}, {'id': 'toolu_01VPFtRDMwb1BWodMSLuXDsr', 'input': {'end_date': '2024-05-09', 'location': 'Basel', 'price_tier': 'Economy', 'start_date': '2024-05-02'}, 'name': 'search_car_rentals', 'type': 'tool_use'}]
Tool Calls:
  search_car_rentals (toolu_01VPFtRDMwb1BWodMSLuXDsr)
 Call ID: toolu_01VPFtRDMwb1BWodMSLuXDsr
  Args:
    end_date: 2024-05-09
    location: Basel
    price_tier: Economy
    start_date: 2024-05-02

===== Tool Message =====
Name: search_car_rentals

[{"id": 1, "name": "Europcar", "location": "Basel", "price_tier": "Economy", "start_date": "2024-04-14", "end_date": "2024-04-11", "booked": 1}, {"id": 2, "name": "Avis", "location": "Basel", "price_tier": "Luxury", "start_date": "2024-04-10", "end_date": "2024-04-20", "booked": 0}, {"id": 7, "name": "Enterprise", "location": "Basel", "price_tier": "Premium", "start_date": "2024-04-22", "end_date": "2024-04-20", "booked": 0}, {"id": 9, "name": "Thrifty", "location": "Basel", "price_tier": "Midsize", "start_date": "2024-04-17", "end_date": "2024-04-26", "booked": 0}]

===== Ai Message =====
[{'text': '最も安いオプションはEuropcarのエコノミークラス・レンタルです。', 'type': 'text'}, {'id': 'toolu_01NczhWtTH5TtoZ7RvJAPS11', 'input': {'rental_id': 1}, 'name': 'book_car_rental', 'type': 'tool_use'}]
Tool Calls:
  book_car_rental (toolu_01NczhWtTH5TtoZ7RvJAPS11)
 Call ID: toolu_01NczhWtTH5TtoZ7RvJAPS11
  Args:
    rental_id: 1
上記のアクションを承認しますか？
続行する場合は'y'を入力し、そうでない場合は変更を要求する内容を説明してください。

y

（省略）

レンタカー予約のためにレンタカーを検索する search_car_rentals のツールの実行の際は承認フローは入らず、実際に予約をする book_car_rental の実行の際は承認フローが入っていることがわかります。

Part 4. マルチエージェント

Part 3.までは一つのエージェント（Assistant）に全てのツールを管理させ、一つのプロンプトで実行を制御していました。
これはエージェントとって大きな負担であり、ユーザーの意図を適切に汲み取り、最適なツールを実行することは困難になり得ます。
そこで、Part 4.では、それぞれの機能ごとにエージェントを作り、責任を分担することで、より適切なタスク遂行を可能とします。

"Flights", "Hotels", "Car Rentals", "Excursions" それぞれがエージェントとなり、責任を分担する
Part 4.のグラフは以下のようになります。かなり複雑になりましたが、primary_assistantの下に、flight、car_rental, hotel, excursionそれぞれのサブグラフを構築し、ツール実行の責任を分担させた構造となっています。

Part 4.のグラフ
グラフを定義する実装を見てみましょう。全て見ていくと量が多いので、primary_assistantとサブグラフの一つflightを見てみます。
まず、primary_assistantのグラフ実装部分です。

# Primary assistant
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,
) -> Literal[
    "primary_assistant_tools",
    "enter_update_flight",
    "enter_book_hotel",
    "enter_book_excursion",
    "__end__",
]:
    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("Invalid route")

# 条件付きエッジにより、利用するツールを切り替え
builder.add_conditional_edges(
    "primary_assistant",
    route_primary_assistant,
    {
        "enter_update_flight": "enter_update_flight",
        "enter_book_car_rental": "enter_book_car_rental",
        "enter_book_hotel": "enter_book_hotel",
        "enter_book_excursion": "enter_book_excursion",
        "primary_assistant_tools": "primary_assistant_tools",
        END: END,
    },
)
builder.add_edge("primary_assistant_tools", "primary_assistant")


def route_to_workflow(
    state: State,
) -> Literal[
    "primary_assistant",
    "update_flight",
    "book_car_rental",
    "book_hotel",
    "book_excursion",
]:
    dialog_state = state.get("dialog_state")
    if not dialog_state:
        return "primary_assistant"
    return dialog_state[-1]


builder.add_conditional_edges("fetch_user_info", route_to_workflow)

part_4_graph = builder.compile(
    # それぞれのエージェントが持つ sensitive_tools を実行する際は停止し、ユーザーに実行の可否を求める。
    interrupt_before=[
        "update_flight_sensitive_tools",
        "book_car_rental_sensitive_tools",
        "book_hotel_sensitive_tools",
        "book_excursion_sensitive_tools",
    ],
)

次に、flightのサブグラフを見てみましょう。ノードの定義とノード間のエッジ張りで特に変わった点はないと思います。

# Flight booking assistant
builder.add_node(
    "enter_update_flight",
    create_entry_node("Flight Updates & Booking Assistant", "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,
) -> Literal[
    "update_flight_sensitive_tools",
    "update_flight_safe_tools",
    "leave_skill",
    "__end__",
]:
    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)

car_rental、hotel、excursionも同様にサブグラフを構築し、primary_assistantと接続してユーザーからの依頼に対応します。
グラフとしては複雑になりましたが、責任の所在が明確になり、よりユーザーの意図を正確に汲み取ったタスク実行が可能になりました。
また、開発観点においてもそれぞれのエージェントが独立しているので、改善もしやすくなっています。

---

以上、LangGraphのカスタマーサポートボットを作成するexamplesからエージェントの作り方を学びました。
他にもReflection（選択した自らの行動の質を自ら評価するために過去のステップを観察させるプロセス）の組み方や、対話シミュレーションをする方法（構築したエージェントの評価のために毎回手動でシステムと対話し出力を評価するのではなく、自動で対話させ評価する仕組み）の組み方など、さまざまなexamplesが公開されているので、非常に勉強になります。

お知らせ

少しでも弊社にご興味を持っていただけた方は、お気軽にご連絡頂けますと幸いです。まずはカジュアルにお話を、という形でも、副業を検討したいという形でも歓迎しています。
https://hrmos.co/pages/zenkigen/jobs?jobType=FULL
https://speakerdeck.com/zenkigenforrecruit/detailed-version-recruitment-materials-for-data-scientists