はじめに
七尾百合子さん、お誕生日 43日目 おめでとうございます！ nikkieです。
Agent Development Kit (ADK)の素振りシリーズです。
目次
- はじめに
- 目次
- Agent Development Kit の LoopAgent
- LoopAgent の sub_agents から STOP signal を出す
- 「return a "STOP" signal」の内訳
- 終わりに
Agent Development Kit の LoopAgent
ADKには3カテゴリのAgentがあります。
https://google.github.io/adk-docs/agents/#core-agent-categories
- LLM Agents
- Workflow Agents
- Custom Agents
Workflow Agentsには以下の3つがあります。
https://google.github.io/adk-docs/agents/workflow-agents/
- Sequential Agents：順次実行
- Loop Agents：反復実行
- Parallel Agents：並列実行
今回取り上げるLoopAgentは、サブエージェントたちをループで（すなわち反復的に）実行します1。
指定した反復回数に達するか、終了条件が合うまで、一連のサブエージェントたちを繰り返し実行します2
ドキュメントの例は、max_iterations引数で反復回数を指定するものです。
https://google.github.io/adk-docs/agents/workflow-agents/loop-agents/#full-example-iterative-document-improvement
The CriticAgent could be designed to return a "STOP" signal when the document reaches a satisfactory quality level, preventing further iterations.
STOP signalを返す実装が、私、気になります！
LoopAgent の sub_agents から STOP signal を出す
DeepWikiでDevinに聞きました。
その結果できた実装がこちら
adk run stop-loop
user: こんにちは [counter_agent]: Counter: 1 [counter_agent]: Counter: 2 [counter_agent]: Counter: 3 [counter_agent]: Send STOP signal
adk web
ソースコードの全容はこちらです。
.env不要で動いています。
「return a "STOP" signal」の内訳
DeepWikiを頼りにソースコードを追っての理解を書きます。
LoopAgentの実装より
https://github.com/google/adk-python/blob/v0.3.0/src/google/adk/agents/loop_agent.py#L43-L55
class LoopAgent(BaseAgent): async def _run_async_impl( self, ctx: InvocationContext ) -> AsyncGenerator[Event, None]: times_looped = 0 while not self.max_iterations or times_looped < self.max_iterations: for sub_agent in self.sub_agents: async for event in sub_agent.run_async(ctx): yield event if event.actions.escalate: return times_looped += 1 return
LoopAgentのsub_agentのrun_async()を呼んで返ったeventについて、event.actions.escalateがTrueならば、_run_async_impl()メソッドからreturnします。
結果、LoopAgentによるループを抜けます。
Runtimeのドキュメントにシーケンス図がありました。
https://google.github.io/adk-docs/runtime/#how-it-works-a-simplified-invocation
エージェント（や後述するツール）からEventが順番に返る（非同期イテレータ）のですが、escalate=TrueによりEventの流れが途絶えるように制御しているのだと思います
Devinに教わった上記実装でもactions.escalateがTrueのEventを返しています
if self.counter == self.stop_at: yield Event( # 略 actions=EventActions(escalate=True), )
記事にしていて気づいたのですが、ドキュメントの例でもescalateをTrueに設定しています。
exit_toolを渡し、このtoolがescalateをTrueにします
def exit_loop(tool_context: ToolContext): tool_context.actions.escalate = True return {} refiner_agent_in_loop = LlmAgent( name="RefinerAgent", # 略 # instruction にて critique が exactly であれば、exit_toolを使うように指示しています # （前段のcritic_agent_in_loopは対象ドキュメントの筋が通っていたら exactly と返します） tools=[exit_loop], )
終わりに
ADKのWorkflow AgentsのうちのLoopAgentは反復を担います。
指定した反復回数に達するか、終了条件が合うまで繰り返します。
終了条件が合ったことは、サブエージェントからEventのactions.escalateをTrueにすることで、親のLoopAgentに伝えることができます。
サブエージェントでEventのactions.escalateをTrueにするには、そのようなEventをyieldするほか、toolを使ってtool_context.actions.escalate = Trueとする方法がありました。
