AWSで生成AIエージェントを操る！話題のLangGraphにBedrockで入門しよう

Last updated at 2024-08-20Posted at 2024-08-20

LangGraphとは？

LLMで「AIエージェント」を開発するための便利なライブラリがLangGraphです。
LLMアプリ開発のフレームワークで有名なLangChain社が開発しています。

LangChainだけを使ってAIエージェントを作ることももちろんできますが、LLMの推論に基づいて行動を計画させる場合、なかなか思いどおりに動いてくれない場合は多々あります。

これを解決するのがLangGraphで、エージェントの動作をグラフ（点と線の組み合わせ）で定義することができます。

グラフって何？

「グラフ理論」のグラフです。クラスメソッドさんの以下記事が概要を掴みやすいです。

公式ハンズオンをAWSでやってみよう

LangGraphの公式ページにサンプルコードやチュートリアルがいくつも掲載されているのですが、OpenAIやAnthropicのAPIをそのまま使う例となっています。

日本の職業エンジニアはクラウド上でアプリ開発するケースも多いと思いますので、AWSの生成AIサービス「Amazon Bedrock」を使った内容にアレンジして体験してみましょう。

※Bedrockの概要は以下を参照ください。

サンプルコードの概念図

今回はLangGraph公式サイトのトップページにあるExampleをもとに、下図のようなグラフを作成します。

ざっくり処理の流れは次のとおり。

ユーザーから質問が来たら、最初に「エージェント」という親玉が次のアクションを考える（LLMの推論を利用）
もし検索ツールを使う必要がある場合、「ツール」という子分に処理が渡されて、検索結果を親玉に返す
検索結果を使って回答を作れたら、親玉は最終返答をユーザーに返して処理を終了する

サンプルコード

順番に説明したあと、コードの全量を再掲します。

# Pythonの外部ライブラリをインポート
from typing import Annotated, Literal, TypedDict
from langchain_core.messages import HumanMessage
from langchain_aws import ChatBedrockConverse  # BedrockのConverse APIを利用
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

最初にPythonの外部ライブラリをインポートします。LangGraphは単体でも利用できますが、LangChainと組み合わせると便利です。

Bedrockの色んなモデルを共通の形式で扱える「Converse API」を使えるようにしています。

# 「検索」ツールを定義（中身は仮のもの、テキストを返すだけ）
@tool
def search(query: str):
    """Call to surf the web."""
    if "東京" in query:
        return "東京は今日も最高気温35度越えの猛暑です。"
    return "日本は今日、全国的に晴れです。"


# 使えるツール一覧を定義
tools = [search]

# 「ツールを実行する」関数（風のインスタンス）を定義
tool_node = ToolNode(tools)

# LLMとTool useを定義（ローカルへAWSのIAMアクセスキーを設定しておいてください）
model = ChatBedrockConverse(
    model="anthropic.claude-3-5-sonnet-20240620-v1:0"
).bind_tools(tools)

次に search という名前の検索ツールを定義し、これを使用できる tool_node という関数風のインスタンス（いわゆるRunnable）を作成しています。

また、LLMとしてBedrockを設定し、こいつにも同じツールを「バインド」することで、LLMがツールの利用可否を判断できるようにしています。

# 「次に進む」という関数を定義
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    # LLMが「ツールを使うべし」と判断したら、「tools」ノードに進む
    if last_message.tool_calls:
        return "tools"
    # そうでなければワークフローを終了する
    return END


# 「LLMを呼ぶ」という関数を定義
def call_model(state: MessagesState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

次は should_continue という関数を定義しています。名前のとおり「次に進め」という機能をもつのですが、具体的には「LLMがツール使いたいときは tools ノードへ進んでね」という条件分岐が入っています。

また、親玉がLLMを呼んで推論するための call_model という関数も定義しています。

処理の都合上、関数を先に定義しているので、続きのコードを読まないと何がしたいのか分かりづらいですね。先に進みましょう。

# グラフを定義
workflow = StateGraph(MessagesState)

# グラフにノードを2つ追加（エージェントとツール）
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# グラフの開始地点を「エージェント」ノードに設定
workflow.set_entry_point("agent")

ここでLangGraphのメイン主役「グラフ」を作成しています。
workflow という名前をつけて、そこにノードを2つ追加しました。各ノードに引数として与えているのは、ノード名と関数（そのノードが何をするか）です。

また、グラフの始点も定義しています。

# グラフに「条件付きエッジ」を追加（エージェントが呼ばれた後は「次へ進む」を実行）
workflow.add_conditional_edges("agent", should_continue)

# グラフに「エッジ」を追加（ツール → エージェントの経路）
workflow.add_edge("tools", "agent")

# ステート保持用のメモリを初期化
checkpointer = MemorySaver()

次は「エッジ」という概念が登場しました。簡単に言うと「ノードどうしを繋ぐ線」のことです。

通常のエッジだけでなく、特定の条件を満たしたときにだけ通る「Conditional edge」も作成しています。

これでグラフの全体像が完成しました！

# ここまで作ったワークフローをコンパイルして、LangChainのRunnableで呼び出せるようにする
app = workflow.compile(checkpointer=checkpointer)

# グラフの構造をターミナルに表示
app.get_graph().print_ascii()

# ワークフローを実行
final_state = app.invoke(
    {"messages": [HumanMessage(content="東京の天気は？")]},
    config={"configurable": {"thread_id": 42}},
)
result = final_state["messages"][-1].content

# 結果をターミナルに表示
print(result)

定義してきたグラフをコンパイルしたら、実行する前にグラフ構造を分かりやすいようにプリントしています（grandalfというライブラリを利用）。

その後、「東京の天気は？」というユーザーメッセージを使って、このワークフローを実行しています。

実行してみよう！

このファイルを実行してみましょう。

langgraph-bedrock.py

# Pythonの外部ライブラリをインポート
from typing import Annotated, Literal, TypedDict
from langchain_core.messages import HumanMessage
from langchain_aws import ChatBedrockConverse  # BedrockのConverse APIを利用
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode


# 「検索」ツールを定義（中身は仮のもの、テキストを返すだけ）
@tool
def search(query: str):
    """Call to surf the web."""
    if "東京" in query:
        return "東京は今日も最高気温35度越えの猛暑です。"
    return "日本は今日、全国的に晴れです。"


# 使えるツール一覧を定義
tools = [search]

# ツール実行用のインスタンスを生成
tool_node = ToolNode(tools)

# LLMとTool useを定義（ローカルへAWSのIAMアクセスキーを設定しておいてください）
model = ChatBedrockConverse(
    model="anthropic.claude-3-5-sonnet-20240620-v1:0"
).bind_tools(tools)


# 「次に進む」という関数を定義
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    # LLMが「ツールを使うべし」と判断したら、「tools」ノードに進む
    if last_message.tool_calls:
        return "tools"
    # そうでなければワークフローを終了する
    return END


# 「LLMを呼ぶ」という関数を定義
def call_model(state: MessagesState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}


# グラフを定義
workflow = StateGraph(MessagesState)

# グラフにノードを2つ追加（エージェントとツール）
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# グラフの開始地点を「エージェント」ノードに設定
workflow.set_entry_point("agent")

# グラフに「条件付きエッジ」を追加（エージェントが呼ばれた後は「次へ進む」を実行）
workflow.add_conditional_edges("agent", should_continue)

# グラフに「エッジ」を追加（ツール → エージェントの経路）
workflow.add_edge("tools", "agent")

# ステート保持用のメモリを初期化
checkpointer = MemorySaver()

# ここまで作ったワークフローをコンパイルして、LangChainのRunnableで呼び出せるようにする
app = workflow.compile(checkpointer=checkpointer)

# グラフの構造をターミナルに表示
app.get_graph().print_ascii()

# ワークフローを実行
final_state = app.invoke(
    {"messages": [HumanMessage(content="東京の天気は？")]},
    config={"configurable": {"thread_id": 42}},
)
result = final_state["messages"][-1].content

# 結果をターミナルに表示
print(result)

以下を事前に実施してください。

必要なPython外部ライブラリのインストール

ターミナル

pip install -U langchain langchain_aws langgraph boto3 grandalf

AWSアカウントでBedrockのClaude 3.5 Sonnetを有効化し、ローカルにAWS CLIで認証情報を設定（ aws configure ）

実行コマンドは以下です。

ターミナル

python3 langgraph-bedrock.py

実行結果の例は以下です。

ターミナル

        +-----------+         
        | __start__ |         
        +-----------+         
              *               
              *               
              *               
          +-------+           
          | agent |           
          +-------+           
         *         .          
       **           ..        
      *               .       
+-------+         +---------+ 
| tools |         | __end__ | 
+-------+         +---------+ 
検索結果から、東京の今日の天気について以下の情報が得られました：

東京は今日も非常に暑い天気となっているようです。具体的には：

1. 猛暑が続いています。
2. 最高気温が35度を超えています。

これは夏の厳しい暑さを示しており、熱中症などの健康リスクが高い状態です。外出する際は以下の点に注意することをおすすめします：

1. こまめな水分補給
2. 日陰の利用や日傘の使用
3. 涼しい場所での休憩
4. 通気性の良い服装の着用

また、室内でも適切な温度管理を行い、エアコンや扇風機を上手く使用して体調管理に気をつけることが大切です。特に高齢者や子供、体調のすぐれない方は注意が必要です。

最新の気象情報や熱中症警報などにも注意を払い、体調管理に気をつけてお過ごしください。

おまけ

実はAWSの公式ワークショップでも、Bedrockを用いてLangGraphを動かすサンプルコードが公開されています。

※ノートブックだと途中からエラーでそのまま動かなかったので、単一のPythonコードに移植するとうまく動きました。ご参考まで

宣伝

AWSのBedrockが気になった方は、先日、入門書を書きましたのでご活用ください！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

AWSで生成AIエージェントを操る！ 話題のLangGraphにBedrockで入門しよう

LangGraphとは？

グラフって何？

公式ハンズオンをAWSでやってみよう

サンプルコードの概念図

サンプルコード

実行してみよう！

おまけ

宣伝

AWSで生成AIエージェントを操る！話題のLangGraphにBedrockで入門しよう