ランタイム¶

ADKでサポートPython v0.1.0Go v0.1.0Java v0.1.0

ADKランタイムは、ユーザーとの対話中にエージェントアプリケーションを動かす基盤となるエンジンです。定義されたエージェント、ツール、コールバックを受け取り、ユーザーの入力に応じてそれらの実行を調整し、情報の流れ、状態の変化、LLMやストレージなどの外部サービスとの対話を管理するシステムです。

ランタイムをエージェントアプリケーションの「エンジン」と考えてください。部品（エージェント、ツール）を定義すると、ランタイムはそれらがどのように接続され、ユーザーの要求を満たすために一緒に実行されるかを処理します。

コアアイデア：イベントループ¶

ADKランタイムは、その中心でイベントループ上で動作します。このループは、Runnerコンポーネントと定義された「実行ロジック」（エージェント、それらが作成するLLM呼び出し、コールバック、ツールを含む）との間の双方向通信を容易にします。

簡単に言うと：

Runnerはユーザーのクエリを受信し、メインのAgentに処理を開始するように依頼します。
Agent（および関連するロジック）は、報告するもの（応答、ツールの使用要求、状態の変更など）があるまで実行され、その後Eventを生成または発行します。
RunnerはこのEventを受信し、関連するアクション（Servicesを介した状態変更の保存など）を処理し、イベントを आगे（ユーザーインターフェイスなど）に転送します。
Runnerがイベントを処理した後にのみ、Agentのロジックは一時停止した場所から再開し、Runnerによってコミットされた変更の影響を潜在的に確認します。
このサイクルは、エージェントが現在のユーザーのクエリに対して生成するイベントがなくなるまで繰り返されます。

このイベント駆動型ループは、ADKがエージェントコードを実行する方法を支配する基本的なパターンです。

ハートビート：イベントループ - 内部の仕組み¶

イベントループは、Runnerとカスタムコード（エージェント、ツール、コールバック、設計ドキュメントではまとめて「実行ロジック」または「ロジックコンポーネント」と呼ばれる）との間の相互作用を定義するコアの運用パターンです。責任の明確な分担を確立します。

Note

特定のメソッド名とパラメータ名は、SDK言語によって若干異なる場合があります（例：Pythonのagent_to_run.run_async(...)、Goのagent.Run(...)、Javaのagent_to_run.runAsync(...)）。詳細については、言語固有のAPIドキュメントを参照してください。

Runnerの役割（オーケストレーター）¶

Runnerは、単一のユーザー呼び出しの中心的なコーディネーターとして機能します。ループにおけるその責任は次のとおりです。

開始： エンドユーザーのクエリ（new_message）を受信し、通常はSessionServiceを介してセッション履歴に追加します。
キックオフ： メインエージェントの実行メソッド（例：agent_to_run.run_async(...)）を呼び出して、イベント生成プロセスを開始します。
受信と処理： エージェントロジックがEventをyieldまたはemitするのを待ちます。イベントを受信すると、Runnerは迅速に処理します。これには次のものが含まれます。
- 構成済みのServices（SessionService、ArtifactService、MemoryService）を使用して、event.actionsに示されている変更（state_delta、artifact_deltaなど）をコミットします。
- その他の内部的な簿記を実行します。
アップストリームへの生成： 処理されたイベントを आगे（呼び出し元のアプリケーションやレンダリング用のUIなど）に転送します。
反復： 生成されたイベントの処理が完了したことをエージェントロジックに通知し、再開して次のイベントを生成できるようにします。

概念的なRunnerループ：

PythonGoJava

# Runnerのメインループのロジックの簡略化されたビュー
def run(new_query, ...) -> Generator[Event]:
    # 1. SessionServiceを介してセッションイベント履歴にnew_queryを追加します
    session_service.append_event(session, Event(author='user', content=new_query))

    # 2. エージェントを呼び出してイベントループを開始します
    agent_event_generator = agent_to_run.run_async(context)

    async for event in agent_event_generator:
        # 3. 生成されたイベントを処理し、変更をコミットします
        session_service.append_event(session, event) # 状態/アーティファクトのデルタなどをコミットします
        # memory_service.update_memory(...) # 該当する場合
        # artifact_serviceは、エージェントの実行中にコンテキストを介してすでに呼び出されている可能性があります

        # 4. アップストリーム処理のためにイベントを生成します（例：UIレンダリング）
        yield event
        # Runnerは、生成後にエージェントジェネレーターが続行できることを暗黙的に通知します

// GoでのRunnerのメインループのロジックの簡略化された概念的なビュー
func (r *Runner) RunConceptual(ctx context.Context, session *session.Session, newQuery *genai.Content) iter.Seq2[*Event, error] {
    return func(yield func(*Event, error) bool) {
        // 1. SessionServiceを介してセッションイベント履歴にnew_queryを追加します
        // ...
        userEvent := session.NewEvent(ctx.InvocationID()) // 概念的なビューのために簡略化
        userEvent.Author = "user"
        userEvent.LLMResponse = model.LLMResponse{Content: newQuery}

        if _, err := r.sessionService.Append(ctx, &session.AppendRequest{Event: userEvent}); err != nil {
            yield(nil, err)
            return
        }

        // 2. エージェントを呼び出してイベントストリームを開始します
        // agent.Runもiter.Seq2[*Event, error]を返すと仮定します
        agentEventsAndErrs := r.agent.Run(ctx, &agent.RunRequest{Session: session, Input: newQuery})

        for event, err := range agentEventsAndErrs {
            if err != nil {
                if !yield(event, err) { // エラーがあってもイベントを生成してから停止します
                    return
                }
                return // エージェントがエラーで終了しました
            }

            // 3. 生成されたイベントを処理し、変更をコミットします
            // 実際のコードで見られるように、部分的なイベントのみをセッションサービスにコミットします
            if !event.LLMResponse.Partial {
                if _, err := r.sessionService.Append(ctx, &session.AppendRequest{Event: event}); err != nil {
                    yield(nil, err)
                    return
                }
            }
            // memory_service.update_memory(...) // 該当する場合
            // artifact_serviceは、エージェントの実行中にコンテキストを介してすでに呼び出されている可能性があります

            // 4. アップストリーム処理のためにイベントを生成します
            if !yield(event, nil) {
                return // アップストリームのコンシューマーが停止しました
            }
        }
        // エージェントは正常に終了しました
    }
}

// JavaでのRunnerのメインループのロジックの簡略化された概念的なビュー。
public Flowable<Event> runConceptual(
    Session session,                  
    InvocationContext invocationContext, 
    Content newQuery                
    ) {

    // 1. SessionServiceを介してセッションイベント履歴にnew_queryを追加します
    // ...
    sessionService.appendEvent(session, userEvent).blockingGet();

    // 2. エージェントを呼び出してイベントストリームを開始します
    Flowable<Event> agentEventStream = agentToRun.runAsync(invocationContext);

    // 3. 生成された各イベントを処理し、変更をコミットし、「生成」または「発行」します
    return agentEventStream.map(event -> {
        // これにより、セッションオブジェクトが変更されます（イベントが追加され、stateDeltaが適用されます）。
        // appendEventの戻り値（Single<Event>）は、概念的には
        // 処理後のイベント自体です。
        sessionService.appendEvent(session, event).blockingGet(); // 簡略化されたブロッキング呼び出し

        // memory_service.update_memory(...) // 該当する場合 - 概念的
        // artifact_serviceは、エージェントの実行中にコンテキストを介してすでに呼び出されている可能性があります

        // 4. アップストリーム処理のためにイベントを「生成」します
        //    RxJavaでは、マップでイベントを返すと、次の演算子またはサブスクライバーに効果的に生成されます。
        return event;
    });
}

実行ロジックの役割（エージェント、ツール、コールバック）¶

エージェント、ツール、コールバック内のコードは、実際の計算と意思決定を担当します。ループとの相互作用には、次のものが含まれます。

実行： 実行が再開されたときのセッション状態を含む、現在のInvocationContextに基づいてロジックを実行します。
生成： ロジックが通信する必要がある場合（メッセージの送信、ツールの呼び出し、状態の変更の報告）、関連するコンテンツとアクションを含むEventを構築し、このイベントをRunnerにyieldします。
一時停止： 重要なことに、エージェントロジックの実行は、yieldステートメント（またはRxJavaのreturn）の直後に即座に一時停止します。Runnerがステップ3（処理とコミット）を完了するのを待ちます。
再開： Runnerが生成されたイベントを処理した後にのみ、エージェントロジックはyieldの直後のステートメントから実行を再開します。
更新された状態の確認： 再開すると、エージェントロジックは、以前に生成されたイベントからRunnerによってコミットされた変更を反映するセッション状態（ctx.session.state）に確実にアクセスできます。

概念的な実行ロジック：

PythonGoJava

# Agent.run_async、コールバック、またはツール内のロジックの簡略化されたビュー

# ... 以前のコードは現在の状態に基づいて実行されます ...

# 1. 変更または出力が必要かどうかを判断し、イベントを構築します
# 例：状態の更新
update_data = {'field_1': 'value_2'}
event_with_state_change = Event(
    author=self.name,
    actions=EventActions(state_delta=update_data),
    content=types.Content(parts=[types.Part(text="状態が更新されました。")])
    # ... その他のイベントフィールド ...
)

# 2. 処理とコミットのためにRunnerにイベントを生成します
yield event_with_state_change
# <<<<<<<<<<<< 実行はここで一時停止します >>>>>>>>>>>>

# <<<<<<<<<<<< RUNNERがイベントを処理してコミットします >>>>>>>>>>>>

# 3. 上記のイベントの処理が完了した後にのみ実行を再開します。
# これで、Runnerによってコミットされた状態が確実に反映されます。
# 後続のコードは、生成されたイベントからの変更が発生したと安全に想定できます。
val = ctx.session.state['field_1']
# ここで`val`は「value_2」であることが保証されます（Runnerが正常にコミットしたと仮定）
print(f"実行を再開しました。field_1の値は現在：{val}")

# ... 後続のコードは続行されます ...
# 後で別のイベントを生成する可能性があります...

# Agent.Run、コールバック、またはツール内のロジックの簡略化されたビュー

# ... 以前のコードは現在の状態に基づいて実行されます ...

# 1. 変更または出力が必要かどうかを判断し、イベントを構築します
# 例：状態の更新
updateData := map[string]interface{}{"field_1": "value_2"}
eventWithStateChange := &Event{
    Author: self.Name(),
    Actions: &EventActions{StateDelta: updateData},
    Content: genai.NewContentFromText("状態が更新されました。", "model"),
    // ... その他のイベントフィールド ...
}

# 2. 処理とコミットのためにRunnerにイベントを生成します
# Goでは、これはイベントをチャネルに送信することによって行われます。
eventsChan <- eventWithStateChange
# <<<<<<<<<<<< 実行はここで一時停止します（概念的に） >>>>>>>>>>>>
# チャネルの反対側のRunnerがイベントを受信して処理します。
# エージェントのゴルーチンは続行する可能性がありますが、論理フローは次の入力またはステップを待ちます。

# <<<<<<<<<<<< RUNNERがイベントを処理してコミットします >>>>>>>>>>>>

# 3. 上記のイベントの処理が完了した後にのみ実行を再開します。
# 実際のGo実装では、これはおそらくエージェントが
# 次のステップを示す新しいRunRequestまたはコンテキストを受信することによって処理されます。
# 更新された状態は、その新しいリクエストのセッションオブジェクトの一部になります。
# この概念的な例では、状態を確認するだけです。
val := ctx.State.Get("field_1")
# ここで`val`は、Runnerが
# エージェントを再度呼び出す前にセッション状態を更新したため、「value_2」であることが保証されます。
fmt.Printf("実行を再開しました。field_1の値は現在： %v\n", val)

# ... 後続のコードは続行されます ...
# 後でチャネルに別のイベントを送信する可能性があります...

// Agent.runAsync、コールバック、またはツール内のロジックの簡略化されたビュー
// ... 以前のコードは現在の状態に基づいて実行されます ...

// 1. 変更または出力が必要かどうかを判断し、イベントを構築します
// 例：状態の更新
ConcurrentMap<String, Object> updateData = new ConcurrentHashMap<>();
updateData.put("field_1", "value_2");

EventActions actions = EventActions.builder().stateDelta(updateData).build();
Content eventContent = Content.builder().parts(Part.fromText("状態が更新されました。")).build();

Event eventWithStateChange = Event.builder()
    .author(self.name())
    .actions(actions)
    .content(Optional.of(eventContent))
    // ... その他のイベントフィールド ...
    .build();

// 2. イベントを「生成」します。RxJavaでは、これはストリームに発行することを意味します。
//    Runner（またはアップストリームのコンシューマー）がこのFlowableをサブスクライブします。
//    Runnerがこのイベントを受信すると、それを処理します（例：sessionService.appendEventを呼び出します）。
//    Java ADKの「appendEvent」は、「ctx」（InvocationContext）内に保持されている「Session」オブジェクトを変更します。

// <<<<<<<<<<<< 概念的な一時停止ポイント >>>>>>>>>>>>
// RxJavaでは、「eventWithStateChange」の発行が発生し、その後ストリームは
// Runnerがこのイベントを処理した*後*のロジックを表す「flatMap」または「concatMap」演算子で続行する可能性があります。

// 「Runnerが処理を完了した後にのみ実行を再開する」をモデル化するには：
// Runnerの`appendEvent`は通常、非同期操作自体です（Single<Event>を返します）。
// エージェントのフローは、コミットされた状態に依存する後続のロジックが
// その`appendEvent`が完了した*後*に実行されるように構成する必要があります。

// これは、Runnerが通常それを調整する方法です。
// Runner：
//   agent.runAsync(ctx)
//     .concatMapEager(eventFromAgent ->
//         sessionService.appendEvent(ctx.session(), eventFromAgent) // これによりctx.session().state()が更新されます
//             .toFlowable() // 処理後にイベントを発行します
//     )
//     .subscribe(processedEvent -> { /* UIがprocessedEventをレンダリングします */ });

// したがって、エージェント自身のロジック内で、生成されたイベント*後*に何かを行う必要がある場合
// 処理され、その状態の変更がctx.session().state()に反映されると、
// その後続のロジックは通常、リアクティブチェーンの別のステップになります。

// この概念的な例では、イベントを発行し、その後「再開」を
// Flowableチェーンの後続の操作としてシミュレートします。

return Flowable.just(eventWithStateChange) // ステップ2：イベントを生成します
    .concatMap(yieldedEvent -> {
        // <<<<<<<<<<<< RUNNERが概念的にイベントを処理してコミットします >>>>>>>>>>>>
        // この時点で、実際のランナーでは、ctx.session().appendEvent(yieldedEvent)が
        // Runnerによって呼び出され、ctx.session().state()が更新されます。
        // これをモデル化しようとしているエージェントの概念的なロジックの*内部*にいるため、
        // Runnerのアクションが暗黙的に「ctx.session()」を更新したと仮定します。

        // 3. 実行を再開します。
        // これで、Runnerによってコミットされた状態（sessionService.appendEventを介して）が
        // ctx.session().state()に確実に反映されます。
        Object val = ctx.session().state().get("field_1");
        // ここで`val`は、Runnerによって呼び出された`sessionService.appendEvent`が
        // `ctx`オブジェクト内のセッション状態を更新したため、「value_2」であることが保証されます。

        System.out.println("実行を再開しました。field_1の値は現在：" + val);

        // ... 後続のコードは続行されます ...
        // この後続のコードが別のイベントを生成する必要がある場合は、ここでそれを行います。

Runnerと実行ロジック間のこの協調的な生成/一時停止/再開サイクルは、Eventオブジェクトによって仲介され、ADKランタイムの中核を形成します。