Gemini 2.0 FlashとLangGraphで構築するAIエージェント

この記事は Google Cloud Japan Advent Calendar 2024 (Gemini編) の21日目の記事です。

LangGraph と Vertex AI の新しい Gemini API を組み合わせて、Gemini 2.0でAIエージェントを構築する方法を紹介します。

今回は、通貨 API からリアルタイムおよび過去の為替レートを取得し、データの正確性を検証し、分析結果をまとめたレポートを生成するプロセスを自動化するエージェントを作成します。

なぜ LangGraph を使うのか？

LangGraph は、複雑な AI アプリケーションを構築するための強力なフレームワークです。特に、以下のような利点があります。

• ワークフローの明確化: 複雑なタスクを、明確に定義されたノードとエッジで構成されるワークフローとして表現できます。これにより、アプリケーションの構造が理解しやすくなり、保守性も向上します。
• 柔軟な実行制御: 各ノードの実行順序や条件を細かく制御できます。これにより、複雑なロジックを実装しやすくなります。
• 状態管理: ワークフロー全体で状態を管理できます。これにより、各ノードが前のノードの結果を参照したり、情報を共有したりすることが容易になります。
• ツール統合: LangChain のツールを簡単に統合できます。これにより、外部 API やデータソースへのアクセスが容易になります。
• 並列処理: 複数のノードを並列に実行できます。これにより、処理速度を向上させることができます。

実行環境

今回は企業利用向けのJupyter Notebookのホスティングサービスである、Colab Enterpriseを利用しました。Colab Enterprise は、Colaboratory の手軽さに加えて、セキュリティやネットワーク管理の機能が強化されており、よりエンタープライズユースケースに適したサービスとなっています。
また、操作感に関しては通常のColaboratoryと変わらないため、慣れている方であればすぐに使い始められます。

必要なパッケージのインストール

まず、必要なパッケージをインストールします。

%pip install -q -U \
    langgraph \
    langchain-google-vertexai \
    google-generativeai \
    requests

ランタイムの再起動

インストールしたパッケージを Jupyter ランタイムで使用するために、ランタイムを再起動する必要があります。以下のセルを実行してください。

import IPython

app = IPython.Application.instance()
app.kernel.do_shutdown(True)

⚠️ カーネルが再起動します。完了するまでお待ちください。 ⚠️

Google Cloud プロジェクトの設定と Vertex AI SDK の初期化

Vertex AI を使用するには、既存の Google Cloud プロジェクトがあり、Vertex AI API が有効になっている必要があります。プロジェクトと開発環境の設定について詳しくはこちらを参照してください。

import os
from google import genai

PROJECT_ID = "[your-project-id]"  # @param {type:"string", isTemplate: true}
LOCATION = "us-central1"  # @param {type:"string", isTemplate: true}
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

通貨分析エージェントの構築と実行

ライブラリのインポート

次に、LangGraph、LangChain、Vertex AI、およびエージェントの機能に必要なその他のユーティリティに必要なライブラリをインポートします。

import logging
from typing import Annotated, TypedDict
from IPython.display import Image, Markdown, display

# LangChainコアコンポーネント
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool

# Google Generative AI SDK
from google import genai

# LangGraphコンポーネント
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
import requests

エージェントの状態とメモリの初期化

ここでは、ワークフロー中に情報を保存するためにエージェントのメモリを初期化します。また、タスクの説明と会話履歴を含むエージェントの状態のスキーマを定義します。

• AgentState: エージェントの状態を定義する TypedDict です。task はタスクの説明、messages は会話履歴を格納します。
• MemorySaver: LangGraph のメモリ管理クラスです。ワークフローの状態を保存します。

# エージェントの状態を定義
class AgentState(TypedDict):
    task: str
    messages: Annotated[list, add_messages]

# エージェントのメモリを初期化
memory = MemorySaver()

# ログレベルをERRORに設定して警告をフィルタリング
logger = logging.getLogger()
logger.setLevel(logging.ERROR)

ツールの定義

続いて、エージェントが外部 API から為替レートを取得するために使用するカスタムツールを定義します。
このツールは、指定された通貨ペアと日付の為替レートを取得します。
今回は、リアルタイム・過去の為替レートを提供する、無料のAPIサービスであるfrankfurterを利用します。

LangGraph アプリケーションは、このツールノードを使用して関数を呼び出し、外部から情報を取得します。

• @tool: LangChain の tool デコレータを使用して、この関数をツールとして登録します。
• get_exchange_rate: 指定された通貨ペアと日付の為替レートを返す関数です。

@tool
def get_exchange_rate(
    currency_from: str = "USD",
    currency_to: str = "EUR",
    currency_date: str = "latest",
):
    """指定された日付の2つの通貨間の為替レートを取得します。

    Args:
        currency_from (str, optional): 基本通貨。デフォルトは"USD"。
        currency_to (str, optional): 対象通貨。デフォルトは"EUR"。
        currency_date (str, optional): 為替レートの日付 (YYYY-MM-DD)。
                                      最新レートの場合は"latest"を使用。デフォルトは"latest"。
    """
    response = requests.get(
        f"https://api.frankfurter.app/{currency_date}",
        params={"from": currency_from, "to": currency_to},
    )
    return response.json()

各ノードのプロンプトと関数の定義

ここでは、LangGraph ワークフローの各ノードで使用されるプロンプトと関数を定義します。これらのプロンプトは、Gemini モデルが為替レートの取得、正確性の検証、レポートの生成などの特定のアクションを実行する際にガイドします。

エージェントの機能を個別のノードに分割することで、特定のノードが実行するアクションと、特定のノードで LLM にどの程度任せるかを正確に選択できます。各ノードがグローバルな状態からの情報を使用し、return ステートメントでグローバルな状態変数を更新する方法に注目してください。

#######################################
# エージェントノード: 為替レートを検索
#######################################
def agent_node(state: AgentState):
    messages = state["messages"]
    prompt_text = "\n".join([msg.content for msg in messages])
    
    response = client.models.generate_content(
        model="gemini-2.0-flash-exp",
        contents=prompt_text
    )

    return {"messages": [HumanMessage(content=response.text)]}

#########################################
# ツールノード：API関数に基づいて定義
#########################################
tools = [get_exchange_rate]
tool_node = ToolNode(tools)

# 追加のツール呼び出しが必要かどうかを判断
def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"
    return "review"

#########################################
# レビューノード：分析と結果の検証
#########################################
REVIEW_PROMPT = """あなたは金融専門家です。
変換されたレートを分析し、数値の正確性を検証してください。
これまでの会話履歴は以下の通りです:
{messages}"""
def review_node(state: AgentState):
    messages = [
        HumanMessage(content=REVIEW_PROMPT.format(messages=state["messages"])),
    ]
    response = client.models.generate_content(
        model="gemini-2.0-flash-exp",
        contents=messages[0].content
    )

    return {"messages": [HumanMessage(content=response.text)]}

########################################
# レポートノード：サマリーレポートを生成
########################################
REPORT_PROMPT = """現在の為替レートを扱う金融ニュースレポーター向けのサマリーレポートを作成してください。
時間の経過に伴うレートを比較した結果の表を含めてください。
通貨記号は出力で使用しないでください。出力のレンダリングが中断される可能性があります。
これまでの会話履歴は以下の通りです:
{messages}"""
def report_node(state: AgentState):
    messages = [
        HumanMessage(content=REPORT_PROMPT.format(messages=state["messages"])),
    ]
    response = client.models.generate_content(
        model="gemini-2.0-flash-exp",
        contents=messages[0].content
    )
    return {"messages": [HumanMessage(content=response.text)]}

グラフの定義とコンパイル

このセクションでは、LangGraph ワークフローの構造を定義し、ノードを論理的な順序で接続します。次に、エージェントのメモリを状態管理に組み込んで、グラフをコンパイルします。

# LangGraphワークフローを初期化し、エージェントの状態スキーマを指定
workflow = StateGraph(AgentState)

# ワークフローにノードを追加し、各ノードを対応する関数に関連付け
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.add_node("review", review_node)
workflow.add_node("report", report_node)

# ノード間の実行フローを定義し、ワークフローのロジックを作成
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", "review"])
workflow.add_edge("tools", "agent")
workflow.add_edge("review", "report")
workflow.add_edge("report", END)

# LangGraphワークフローをコンパイルし、メモリベースの状態管理を有効化
graph = workflow.compile(checkpointer=memory)

コンパイルされたグラフの表示

コンパイルされた LangGraph ワークフローをダイアグラムで視覚化できます。これにより、エージェントの実行フローを明確に把握できます。

Image(graph.get_graph().draw_mermaid_png())

エージェントの実行

これで、システムプロンプトとユーザープロンプトを使用してワークフローを開始し、現在および過去の為替レートを取得する準備ができました。

※このプロンプトは自由に変更することは可能です。是非、国、通貨、期間、その他の変更をして、エージェントがどのように応答するか試してみてください。

# エージェントにユーザーの質問に答える方法を指示するシステムプロンプトを定義します。
SYSTEM_PROMPT = """提供されたツールを使用して、ユーザーの質問に答えてください。また、3か月前,1年前, 3年前のすべての通貨の為替レートを調べ、値を比較してください。"""

# 為替レートに関するユーザーの最初の質問を定義します。
USER_PROMPT = """今日のUSDから（EUR、JPY）の為替レートはいくらですか？また、100 USDはそれらの通貨でいくらになりますか？"""

これで LangGraph エージェントを実行する準備ができました。エージェントはグラフのノードをステップごとに処理し、API と対話し、データを検証し、レポートを準備します。

# 状態管理のための固有のIDでLangGraphスレッドを初期化します。
thread_config = {"configurable": {"thread_id": "1"}}

# LangGraphワークフローを実行し、各ノードの結果をストリーミングします。
for state in graph.stream(
    {
        "messages": [
            SystemMessage(content=SYSTEM_PROMPT),
            HumanMessage(content=USER_PROMPT),
        ],
    },
    thread_config,
):
    # 現在のノードの名前とその出力を各ステップで表示します。
    for node_name, node_output in state.items():
        print(f"エージェントノード: {node_name}\n")
        print("エージェントの結果:")
        print(str(node_output)[:1000])  # 表示のため出力の切り捨て
    print("\n====================\n")

出力のレンダリング

最後に、エージェントが生成した最終レポートをレンダリングし、通貨分析の結果を人間が読みやすい形式で表示します。

display(Markdown(state["report"]["messages"][0].content))

最後に

今回は、LangGraph と Gemini API を使用して通貨分析のAI エージェントを実装しました。
LangGraph のオーケストレーションフレームワークを活用することで、LLM単体では実現できない、データ検証、レポート生成を統合して自動化するマルチステージワークフローを構築できます。
AI エージェントの実装をよりシステマティックに実現できるLangGraphとGemini 2.0 flash、是非お試しください！

参考資料

From API to Report: Building a Currency Analysis Agent with LangGraph and Gemini
Intro to Gemini 2.0 Flash