参考資料
基本事項
LlmAgent クラス
- LLM による応答を得るための基本となるクラス
- セッション管理機能はなく、与えられた InvocationContext の情報を用いてワンショットで応答を返す
- InvocationContext に含まれる session_service からセッション情報（過去の会話履歴など）を取得して利用する
- InvocationContext は LlmAgent を呼び出すごとに新しく生成する前提
Runner クラス
- ローカルで LlmAgent を対話的に利用する際に利用する
- agent と session_service を保持しており、session 情報を含んだ InvocationContext を作って agent を呼び出す
- Runner クラスを利用した簡易的な会話 App の例：
class LocalApp: def __init__(self, agent): self._agent = agent self._user_id = 'local_app' self._runner = Runner( app_name=self._agent.name, agent=self._agent, artifact_service=InMemoryArtifactService(), session_service=InMemorySessionService(), memory_service=InMemoryMemoryService(), ) self._session = self._runner.session_service.create_session( app_name=self._agent.name, user_id=self._user_id, state={}, session_id=uuid.uuid1().hex, ) async def stream(self, query): content = UserContent(parts=[Part.from_text(text=query)]) async_events = self._runner.run_async( user_id=self._user_id, session_id=self._session.id, new_message=content, ) result = [] async for event in async_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: print(response) result.append(response) return result
- LocalApp クラスのインスタンスからセッション内のイベント情報（Event クラスのリスト）を取り出した例：
session = client._runner.session_service.sessions\
[client._session.app_name][client._session.user_id][client._session.id]
session.events
[出力結果]
[Event(content=UserContent(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='\nこんにちは！おすすめのコーヒーはありますか？\n')], role='user'), grounding_metadata=None, partial=None, turn_complete=None, error_code=None, error_message=None, interrupted=None, invocation_id='e-2a8d30d4-e6fb-4056-8cce-3b2dd13e1dd1', author='user', actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}), long_running_tool_ids=None, branch=None, id='X7a8bStM', timestamp=1745202464.531635),
Event(content=Content(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='とばりちゃんが答えるよ！おすすめのコーヒーですか？夜の帳ブレンドは、深煎りでコクがあり、ほんのりビターな大人の味わいで、疲れた心に染み渡りますよ。それから、月光の浅煎りは、フルーティーな香りが特徴で、すっきりとした味わいなので、リフレッシュしたい時におすすめです！どちらも、その日の気分に合わせてお選びいただけます。\n')], role='model'), grounding_metadata=None, partial=None, turn_complete=None, error_code=None, error_message=None, interrupted=None, invocation_id='e-2a8d30d4-e6fb-4056-8cce-3b2dd13e1dd1', author='TobariChan_agent', actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}), long_running_tool_ids=None, branch=None, id='M2sVH857', timestamp=1745202464.532078),
Event(content=UserContent(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='\n夜の帳ブレンドは大人の味わいなんですね！\n')], role='user'), grounding_metadata=None, partial=None, turn_complete=None, error_code=None, error_message=None, interrupted=None, invocation_id='e-6c99af33-75d0-41ea-8b34-5b98670884f5', author='user', actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}), long_running_tool_ids=None, branch=None, id='N2gl4EQp', timestamp=1745202491.251667),
Event(content=Content(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='とばりちゃんが答えるよ！そうなんです！夜の帳ブレンドは深煎りなので、少しビターで香ばしい、落ち着いた味わいが特徴です。一日の終わりにゆっくりと味わって、くつろいでいただけたら嬉しいです。もちろん、甘いデザートとの相性も抜群ですよ！\n')], role='model'), grounding_metadata=None, partial=None, turn_complete=None, error_code=None, error_message=None, interrupted=None, invocation_id='e-6c99af33-75d0-41ea-8b34-5b98670884f5', author='TobariChan_agent', actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}), long_running_tool_ids=None, branch=None, id='aPw30kDB', timestamp=1745202491.252363)]
- 特に、Event 内の author 要素が応答を生成したエージェント名を表す
サブエージェントを持つエージェントにおけるルーティング
- ユーザーの入力を処理するエージェントは、最後のイベントの author を選択する というシンプルなルール。Runner が InvocationContext を作る際に決定する。
- つまり、session_service に強制的に最後のイベントを追加すれば、次に応答するエージェントを強制変更できる。
- エージェントが自発的にルーティングする際は、transfer_to_agent ツールを Function calling で呼び出す
- ルーティングの判断指示は、プロンプトに記載されている。基本的には、description で判断するように指示しているので、適切な description の設定が重要と思われる。
If you are the best to answer the question according to your description, you can answer it.
- 転送可能なサブエージェントをツール登録するところ
エージェントが受け取るシステムインストラクションを確認する方法
AgentTool の構造
- AgentTool() で生成されたツールのインスタンスは、呼び出しごとに新しいテンポラリーセッションを生成するので、過去のセッション情報を引き継がずにワンショットの関数として実行される。
- ただし、アーティファクトとステート（エージェント間で共有される任意の Key-value データ）は呼び出し元のエージェントと共有される
ToolContext について
- ToolContext で app_name, session_id, user_id を取得する方法
Session 関連
ToolContext で変更した state が session_service 管理の session 本体に反映される流れ
Runner が
run_async()を実行 ->_run_one_step_async->_postprocess_async->_postprocess_handle_function_calls_async->handle_function_calls_async->handle_function_calls_asyncの流れで function call の関数が実行される。関数に渡される tool_context には、ここ で invocation_context から引き継いだ session が入る
関数内での state 変更は
self._event_actions.state_deltaに記録される。handle_function_calls_asyncは、function call の結果をfunction_responseで受け取って、__build_response_eventでイベントを生成する。__build_response_eventは、ここで、tool_context._event_actionsに記録されたstate_deltaを含むイベントを生成する（tool_context.actions == tool_context._event_actionsに注意）その後、Runner の run_async に戻った所で、session_service 本体にイベントが追加されて結果が反映される。
state_deltaイベントを受け取った結果をどう処理するかは、session_service の実装に依存する。
※ LlmAgent と Session は独立したもので、これらをまとめる責任は Runner にある点に注意。LlmAgent はあくまで state_delta をイベントして記録するだけで、それを Session に反映するのは、Runner 側で実施する。
Agent Engine 関係
デプロイしたエージェントの正体
- 下記で取得される remote_agent は、デプロイしたエージェント（LlmAgent クラスのオブジェクト）がクラウド上での Runner （的なもの）にラップされている
remote_agent = vertexai.agent_engines.get(
'projects/[Project Number]/locations/us-central1/reasoningEngines/[App Name]')
reasoning_engines.AdkAppクラスが「クラウド上での Runner（的なもの）」に相当する。実際にラップする処理はここで行われる。公式ドキュメントでエージェント（LlmAgent クラスのオブジェクト）を reasoning_engines.AdkApp でラップして動かす手順が紹介されているが、これはあくまで、クラウド上の動作をローカルでエミュレーションしているだけ。
非公式ではあるが、ユーザー側で先に AdkApp でラップしたものを agent_engines.create() でデプロイしても構わない。
デフォルトの Artifact service
Agent Engine にデプロイする際は、インスタンス間で情報共有されない InMemoryArtifactService は使えないが、実際にはこれがデフォルトになっており、これを変更する公式オプションが説明されていない。
When deploying
my_agent(LlmAgent instance) withagent_engine.create(agent_engine=my_agent), it's wrapped byreasoning_engines.AdkAppasagent_engine = reasoning_engines.AdkApp(agent=agent_engine)[1]Then
reasoning_engines.AdkAppusesartifact_service_builderoption to setup its artifact service [2] and the default is InMemoryArtifactService [3].So the problem is that
agent_engine.create()doesn't have an option to modify this flow to change the artifact service fromInMemoryArtifactServiceto other ones such asGcsArtifactService.
- GcsArtifactService に変更する非公式手順はこちら
remote_agent と会話する簡易アプリの実装例
class RemoteApp: def __init__(self, remote_agent, user_id='default_user'): self._remote_agent = remote_agent self._user_id = user_id self._session = remote_agent.create_session(user_id=self._user_id) def _stream(self, query): events = self._remote_agent.stream_query( user_id=self._user_id, session_id=self._session['id'], message=query, ) result = [] for event in events: if DEBUG: print(f'----\n{event}\n----') if ('content' in event and 'parts' in event['content']): response = '\n'.join( [p['text'] for p in event['content']['parts'] if 'text' in p] ) if response: print(response) result.append(response) return result def stream(self, query): # TODO: avoid infinite loop in case of permanent error while True: result = self._stream(query) if result: break if DEBUG: print('----\nRetrying...\n----') time.sleep(3) return result
- セッションは
remote_agent.create_session(user_id=self._user_id)で生成しているが、この実体は、VertexAiSessionService なので後述の方法で操作できる。 - remote_agent._session には、初期化時点でのセッション情報が入っており、app_name, user_id, session_id が参照できる。
セッション情報の操作
- remote_client から app_name, user_id, session_id を取得する
app_name = remote_client._session['app_name'] user_id = remote_client._session['user_id'] session_id = remote_client._session['id']
- VertexAiSessionService のインスタンスを利用して操作する
from google.adk.sessions import VertexAiSessionService session_service = VertexAiSessionService( project = PROJECT_ID, location = LOCATION, ) session = session_service.get_session( app_name=app_name, user_id=user_id, session_id=session_id, )
- 次は、マルチエージェント（サブエージェント）構成の際に、セッションに最後のイベントを強制追加して、エージェントを強制転送する関数の例
from google.adk.sessions import VertexAiSessionService def force_transfer_agent(remote_client, next_agent): session_service = VertexAiSessionService( project = PROJECT_ID, location = LOCATION, ) session = session_service.get_session( app_name=remote_client._session['app_name'], user_id=remote_client._session['user_id'], session_id=remote_client._session['id'], ) last_event = copy.deepcopy(session.events[-1]) last_event.author = new_agent last_event.content.parts = [Part.from_text(text='I got transferred.')] last_event.timestamp = time.time() return session_service.append_event(session, last_event)
1 回の LLM 呼び出し
async def _run_one_step_async( self, invocation_context: InvocationContext, ) -> AsyncGenerator[Event, None]: """One step means one LLM call.""" llm_request = LlmRequest() # LLM API に渡す内容 # ここで llm_request を用意する # Preprocess before calling the LLM. async for event in self._preprocess_async(invocation_context, llm_request): yield event if invocation_context.end_invocation: return # Calls the LLM. model_response_event = Event( id=Event.new_id(), invocation_id=invocation_context.invocation_id, author=invocation_context.agent.name, branch=invocation_context.branch, ) async for llm_response in self._call_llm_async( invocation_context, llm_request, model_response_event # LLM API 呼び出し ): # Postprocess after calling the LLM. async for event in self._postprocess_async( invocation_context, llm_request, llm_response, model_response_event ): yield event
- llm_request （特に system instruction）に詰め込む情報の定義
