【GCP ADK徹底解説シリーズ③】LoopAgentを活用したワークフローのループ処理

from google.adk.agents import LlmAgent, LoopAgent, SequentialAgent
from google.adk.tools import FunctionTool, ToolContext
import json

# 役割1: 出張プラン提案エージェント
proposer_agent = LlmAgent(
    name="trip_proposer",
    instruction="""あなたは出張手配のプロです。目的地、日程、予算、そして改善ヒント「{refinement_hint}」を基に、具体的な出張プラン（交通手段、ホテル）とその合計金額を見積もってください。
    出力は必ず以下のJSON形式で返してください:
    {"plan_details": "ひかり500号, 東横INN新大阪中央口", "total_cost": 45000}""",
    model="gemini-2.5-flash",
    output_key="proposal"
)

# 役割2: プラン検証（予算チェック）のためのツール
def validate_business_trip_plan(tool_context: ToolContext, proposal: str, budget: int) -> str:
    """提案された出張プランを経費規定（予算）内で検証し、成功ならループを停止、失敗なら理由を返す。"""
    try:
        proposal_data = json.loads(proposal)
        cost = proposal_data.get("total_cost", float('inf'))
        if cost <= budget:
            # 検証成功: エスカレーションを発生させてループを正常終了させる
            tool_context.actions.escalate = True
            return f"検証成功: 予算内に収まりました。最終プラン: {proposal}"
        else:
            # 検証失敗: 失敗理由を返す
            return f"検証失敗: 予算オーバーです。規定額{budget}円に対し、{cost}円でした。"
    except (json.JSONDecodeError, AttributeError):
        return "検証失敗: 提案のJSON形式が正しくありません。"

validator_tool = FunctionTool(name="plan_validator", func=validate_business_trip_plan)

# 検証ツールを呼び出すエージェント
validator_agent = LlmAgent(
    name="validator",
    instruction="提案されたプラン「{proposal}」を経費規定「{budget}」で検証ツールを呼び出してチェックしてください。",
    tools=[validator_tool],
    model="gemini-2.5-flash",
    output_key="validation_result"
)

# 役割3: 改善案の策定エージェント
refiner_agent = LlmAgent(
    name="cost_reduction_advisor",
    instruction="""出張コスト削減アドバイザーです。元の要望と失敗理由を基に、次の提案で試すべき具体的な改善ヒント（例：『LCCの利用を検討』、『ホテルのランクを一つ下げる』など）を生成してください。
    - 元の要望: {prompt}
    - 失敗理由: {validation_result}""",
    model="gemini-2.5-flash",
    output_key="refinement_hint" # 次のループのproposer_agentがこれを参照する
)

self_improving_cycle = LoopAgent(
    name="self_improving_cycle",
    sub_agents=[
        proposer_agent,  # 改善ヒントを基に再提案
        validator_agent, # 検証
        refiner_agent    # 次の改善ヒントを生成
    ],
    max_iterations=5 # 無限ループを避けるための安全装置
)

from google.adk.runtime import Runner, Session
from google.adk.runtime.types import Content, Part

runner = Runner(agent=self_improving_cycle, session_service=SessionService())
session = Session(user_id="user_nagaoka")

# 共有ステートに初期値を設定
session.state = {
    "prompt": "大阪への2泊3日の出張", 
    "budget": 50000, 
    "refinement_hint": "まずは標準的な新幹線とビジネスホテルで見積もってください。"
}

final_result = runner.run(
    session_id=session.id,
    new_message=Content(parts=[Part(text=session.state["prompt"])])
)

# 最終結果の確認
final_state = session.state
validation_result = final_state.get("validation_result", "")

if "検証成功" in validation_result:
    print("予算内での最適なプランが見つかりました！")
    print(validation_result)
else:
    print(f"予算内で最適なプランを見つけられませんでした。最終試行の結果: {validation_result}")