Google ADKで簡単な生成AIチャットボットを作ってみた #google #adk #ai #llm

AId-higuchiGCPLLMPython

はじめに

生成AIチャットボットに関するこれまでの記事では LangChain, LangGraph を利用していました。例：

しかし、LangGraphの「グラフを構築する」という概念が少し難しい・とっつきにくいという印象は否めませんでした。

一方で、Googleが提供する Google ADK (Agent Development Kit)は、よりAIエージェント指向で直感的な実装が可能です。

本稿では生成AI (Google Gemini)を利用し、セッションをPostgreSQLに保存するチャットボットを、LangGraphとGoogle ADKの両方で作成し、その構造的な違いを比較してみます。

生成AIチャットボットの基本仕様

ここでは次の仕様で生成AIチャットボットを作るとします。

  • 簡易なコマンドラインインターフェイスとする
  • ユーザの入力を生成AI (Google Gemini)に渡して、回答を表示する
  • 会話セッションはPostgreSQLに保存し、会話を継続できるようにする
    • ボット起動時、会話セッションを特定するIDをUUIDで生成する
    • ユーザがUUIDを入力すると、その会話セッションに切り替える
  • ユーザが quit, exit, q を入力すると、ボットを終了する

このようにとても単純ですが、一問一答で記憶をなくすのではなく、会話が継続できる仕組みとします。

LangGraphによる実装

まずはLangGraphによる実装です。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from langchain_google_genai import ChatGoogleGenerativeAI
from psycopg import Connection
from langgraph.checkpoint.postgres import PostgresSaver

from yaspin import yaspin

import uuid
import re

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("GOOGLE_API_KEY")
llm = ChatGoogleGenerativeAI(
         model = os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
         temperature = 0.95,
         max_output_tokens = 1024,
      )

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    with yaspin(text="Processing", color="yellow") as spinner:
        res = llm.invoke(state["messages"])
        spinner.ok("✅ ")
    return {"messages": res}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

thread_id = str(uuid.uuid4())
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)

def stream_graph_updates(user_input: str, thread_id: str):
    db_uri = os.getenv("DB_URI")
    with PostgresSaver.from_conn_string(db_uri) as checkpointer:
        checkpointer.setup()
        graph = graph_builder.compile(checkpointer=checkpointer)

        events = graph.stream(
            {"messages": [("user", user_input)]},
            {"configurable": {"thread_id": thread_id}},
            stream_mode="values"
        )
        for event in events:
            message = event["messages"][-1]
            if type(message) == AIMessage:
                print(message.content)

while True:
    try:
        print("\nThread ID: " + thread_id)
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        elif bool(UUID_RE.match(user_input)):
            thread_id = user_input
        else:
            stream_graph_updates(user_input, thread_id)
    except Exception as e:
        print(f"error: {e}")
        break

State クラスによる状態管理でメッセージ履歴を自動追加、 StateGraph による chatbot ノードの追加、開始・終了ノードの追加によるグラフの構築、といった点が特徴です。また、PostgreSQLへの保存は「チェックポイント」という概念で行われています。

さまざなま処理ノードを追加したり、条件分岐ができる、というのはわかるのですが、今回のような単純なチャットボットには「お約束」や「おまじない」といった感じで少しわかりづらい、「大は小を兼ねる」といってもちょっととっつきづらい、という感想になりそうです。

Google ADKによる実装

次にLangGraphによる実装をGoogle ADKで再実装したものです。

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import DatabaseSessionService
from google.genai.types import Content, Part, GenerateContentConfig

from yaspin import yaspin

import asyncio
import uuid
import re

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("GOOGLE_API_KEY")
agent = LlmAgent(
         name = "chatbot",
         model = os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
         instruction = "あなたは親切なアシスタントです。ユーザーの質問に丁寧に答えてください。",
         generate_content_config = GenerateContentConfig(
             temperature = 0.95,
             max_output_tokens = 1024,
         ),
      )

APP_NAME = "chatbot"
USER_ID = "user_1"

thread_id = str(uuid.uuid4())
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)

async def stream_runner_updates(user_input: str, thread_id: str):
    db_uri = os.getenv("DB_URI")
    session_service = DatabaseSessionService(db_url=db_uri)
    runner = Runner(agent=agent, app_name=APP_NAME, session_service=session_service)

    session = await session_service.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=thread_id
    )
    if session is None:
        await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=thread_id
        )

    message = Content(role="user", parts=[Part(text=user_input)])
    with yaspin(text="Processing", color="yellow") as spinner:
        async for event in runner.run_async(
            user_id=USER_ID, session_id=thread_id, new_message=message
        ):
            if event.is_final_response():
                spinner.ok("✅ ")
                if event.content and event.content.parts:
                    text = "".join(p.text for p in event.content.parts if p.text)
                    print(text)

while True:
    try:
        print("\nThread ID: " + thread_id)
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        elif bool(UUID_RE.match(user_input)):
            thread_id = user_input
        else:
            asyncio.run(stream_runner_updates(user_input, thread_id))
    except Exception as e:
        print(f"error: {e}")
        break

できる限りLangGraphでの実装と構造が変わらないように、Google ADKで再実装しました。

Google ADKでは「エージェント」「ランナー」「セッション」が大きな概念になると思います。簡単に言うと次のような形です。

  • エージェント: アプリの「頭脳」に相当。
  • ランナー: アプリの「手足」に相当。ユーザとエージェントの間と取り持ち、セッションの更新も行う。
  • セッション: ユーザとエージェントのやりとりの記録。

グラフの各ステップを手続き的に連結するLangGraphに対し、Google ADKはエージェントの属性を宣言的に定義するスタイルをとります。これにより、ロジックの流れを追う必要がなくなり、コードの見通しが格段に良くなっています。

元のLangGraphの実装での構造をまったく考えずにGoogle ADKで素直に実装するなら、次の通りになります。

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import DatabaseSessionService
from google.genai.types import Content, Part, GenerateContentConfig

from yaspin import yaspin

import asyncio
import uuid
import re

import os
from dotenv import load_dotenv
load_dotenv()

APP_NAME = "chatbot"
USER_ID = "user_1"
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)

def build_agent() -> LlmAgent:
    return LlmAgent(
        name="chatbot",
        model=os.getenv("GOOGLE_MODEL_NAME", "gemini-3-flash-preview"),
        instruction="あなたは親切なアシスタントです。ユーザーの質問に丁寧に答えてください。",
        generate_content_config=GenerateContentConfig(
            temperature=0.95,
            max_output_tokens=1024,
        ),
    )

async def ensure_session(session_service: DatabaseSessionService, session_id: str) -> None:
    session = await session_service.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=session_id
    )
    if session is None:
        await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=session_id
        )

async def chat(runner: Runner, session_service: DatabaseSessionService,
               session_id: str, user_input: str) -> None:
    await ensure_session(session_service, session_id)
    message = Content(role="user", parts=[Part(text=user_input)])

    with yaspin(text="Processing", color="yellow") as spinner:
        async for event in runner.run_async(
            user_id=USER_ID, session_id=session_id, new_message=message
        ):
            if event.is_final_response():
                spinner.ok("✅ ")
                if event.content and event.content.parts:
                    text = "".join(p.text for p in event.content.parts if p.text)
                    print(text)

async def main() -> None:
    session_service = DatabaseSessionService(db_url=os.getenv("DB_URI"))
    runner = Runner(agent=build_agent(), app_name=APP_NAME, session_service=session_service)

    session_id = str(uuid.uuid4())
    while True:
        try:
            print("\nThread ID: " + session_id)
            user_input = await asyncio.to_thread(input, "User: ")
            if user_input.lower() in ["quit", "exit", "q"]:
                print("Goodbye!")
                break
            if UUID_RE.match(user_input):
                session_id = user_input
                continue
            await chat(runner, session_service, session_id, user_input)
        except Exception as e:
            print(f"error: {e}")
            break

if __name__ == "__main__":
    asyncio.run(main())

ループ内で毎回生成していたランナーとセッションをループ外で一度だけ生成しています。

「関心の分離」を意識して、エージェントの構築、セッションの整合性チェック、チャット実行の責任範囲をメソッドごとに切り出しました。これにより、アプリの規模が拡大してもメンテナンスしやすい構造になっています。

まとめ

本稿では、これまでよく利用していたLangGraphでの簡単な生成AIチャットボットのコードを、Google ADKで再実装することで、どのような違いがあるかを見てみました。

LangGraphでの「グラフをどのように制御するか」という低レイヤーな実装の悩みから解放され、Google ADKでは「エージェントにどのような役割を与えるか」という本質的な設計に集中できるようになると思います。LangGraphが持つ柔軟なカスタマイズ性を必要としないシンプルなケースでは、Google ADKのほうがコードの見通しがよく、直感的であり、メンテナンス性も高くなると言えるでしょう。

エージェントフレームワークは数多く存在するので、今後もクリエーションラインではそれぞれ確認して紹介していこうと思います。

Author

Daisuke Higuchi

    Chef・Docker・Mirantis製品などの技術要素に加えて、会議の進め方・文章の書き方などの業務改善にも取り組んでいます。「Chef活用ガイド」共著のほか、Debian Official Developerもやっています。

    Daisuke Higuchiの記事一覧

    新規CTA