【会話型エージェント】Agent Development Kit（ADK）を自社サービスに組み込んでみた

こんにちは！KIYONOのエンジニアです。

本記事は最近ホットなAgent Development Kitを使って既存のサービスにエージェント機能を組み込む方法について解説いたします。

Agent Development Kit（ADK）ですが、使用する利点の一つはその手軽さにあります。
公式ドキュメントではクイックスタートが公開されており、わずか5分程度で、キャプチャのようなチャット画面を立ち上げることができます。
adk webコマンドを一つ実行するだけで、ここまで完成度の高いUIとエージェントとの会話環境が構築できる点にはびっくりでした。

このように、フレームワークの標準的な使い方であれば、フレームワーク側が用意している仕組みに沿って実装するだけでよく、非常に簡単に開発を進めることができます。

しかし、自社の既存プロダクトにエージェント機能を組み込みたい場合や、フレームワーク標準のチャットUIではなく自社開発のUIに組み込みたい場合には、
ある程度フレームワーク内のオブジェクト構造を理解し、それらを活用した実装が求められます。

本記事では、そういった自前のUIにADKを組み込む際の実装方法について詳しく解説していきます。

また、弊社では他にも

といった、AI Agentに関する記事を発信しているので、興味がある方は合わせてこちらもご覧いただければと思います。

ADKのエージェントの作り方
Runnerを介したエージェントとの対話
最後に

ADKのエージェントの作り方

先ほど紹介した、公式docのクイックスタートを例に説明します。
下記のように、ルートディレクトリ配下にエージェント名のフォルダ（multi_tool_agent）をおきます。

parent_folder/
   multi_tool_agent/
      __init__.py
      agent.py
      .env

このフォルダにはエージェントを定義するagent.pyをおき、下記のように実装します。

importdatetime
fromzoneinfoimportZoneInfo
fromgoogle.adk.agentsimportAgent

defget_weather(city:str)->dict:
"""Retrieves the current weather report for a specified city.

  Args:
     city (str): The name of the city for which to retrieve the weather report.

 Returns:
     dict: status and result or error msg.
 """
 # 割愛


defget_current_time(city: str) -> dict: 
"""Returns the current time in a specified city.

　　Args:
　　　　city (str): The name of the city for which to retrieve the current time.

　　Returns:
　　　　dict: status and result or error msg.
"""

 #割愛


root_agent = Agent(
   name = "weather_time_agent",
   model = "gemini-2.0-flash"
   description = (
"Agent to answer questions about the time and weather in a city."
   )
   instruction = (
"You are a helpful agent who can answer user questions about the time and weather in a city."
   ),
   tools = [ # 上部で定義した関数
       get_weagher,
       geet_current_time
   ]
)

このように実装することで、フレームワーク側で自動でAgentを読み取り、
multi_tool_agentを指定して会話をすると、キャプチャのようにget_weatherとcurrent_timeを使用するweather_time_agentと会話することができます。

このように、フレームワーク側で、ユーザからの入力をエージェントやLLMに渡す仕組みを作っているため、私たちはその部分を気にすることなく、楽に実装ができるようになっています。

そうではなく、ADKを用いて、自前のフロントエンドからエージェントにプロンプトを渡したい場合、RunnerやSessionという概念が重要になってきます。

Runnerを介したエージェントとの対話

公式ドキュメントによると
Runnerとは、

「エージェント実行の全体をオーケストレーション（指揮）する中心的なコンポーネント」

です。
つまり、AIの先導者ですね。

Runnerが

ユーザーからのクエリ（メッセージ）を受け取り、
エージェントに処理を開始させ、
エージェントからの出力を取得します

他にも、

セッションの状態やイベント履歴を保存
ファイルなどのバイナリアーティファクトを管理
長期的なユーザーメモリを管理

といったことを行う、ユーザーとエージェントをつなぎ、やりとりが正しく進むように全体を管理する非常に重要なオブジェクトです。

なので、ユーザーからの入力は、直接Agentオブジェクトに渡すのではなく、Runnerオブジェクトを介す必要があります。

class LocalApp:
   def __init__(self, user_id: str, access_token: str):
      self._agent = root_agent
      self._user_id = user_id
      self._access_token = access_token

      # Runnerを定義
      self._runner = Runner(
         app_name=self._agent.name,
         agent=self._agent,
         artifact_service=InMemoryArtifactService(),
         session_service=InMemorySessionService(),
         memory_service=InMemoryMemoryService(),
      )
      self._session = None


   def setup_session(self):
      self._session = self._runner.session_service.create_session(
         app_name=self._agent.name,
         user_id=self._user_id,
         state={},
         session_id=self._access_token,
      )


   def stream(self, query):
      if self._session is None:
      self.setup_session()

      content = UserContent(parts=[Part.from_text(text=query)])
      # ユーザからの入力をRunnerに渡す
      events = self._runner.run(
         user_id=self._user_id,
         session_id=self._session.id,
         new_message=content,
      )

      result = []
      for event in events:
         if DEBUG:
            print(f'----\n{event}\n----')
         if event.content and event.content.parts:
            response = '\n'.join([p.text for p in event.content.parts if p.text])
            if response:
            result.append(response)
      return result

このようにRunnerを作成し、runner.run()という関数を呼び出すことによって、エージェントにユーザーの入力を渡すことができます。

Runnerを作成する際にはsession_idを渡す必要があります。このsession_idが同じであれば、Runnerを新規で作成しても、同じセッションとして扱われます。逆に言えば、同じセッションにも関わらず、異なるsession_idで新たにRunnerを作成してしまうと、セッションが切れてしまい、会話履歴等が継続されないため注意が必要です。

上記のLocalAppを、ユーザの入力が送られたタイミングで作成します。
djangoのViewset内で、下記のようにpostメソッドが叩かれたタイミングで、access_tokenをsession_idとして、LocalAppのインスタンスを作成します。
そして、prompt（ユーザーからの入力）を引数としてstream()を実行することによって、エージェントからの応答を取得することができます。

@action(detail=False, methods=["post"])
def query(self, request):
   company_id = self.request.user.company_id
   user_id = str(self.request.user.id)

   prompt = request.data.get("prompt")
   if not prompt:
      return Response({"error": "prompt is required"}, status=400)

   local_app = LocalApp(
      user_id=user_id,
      access_token=request.data.get("access_token"),
   )
   result = local_app.stream(prompt)

   return Response({"message": "\n".join(result)})