runLive method

Stream<Event> runLive({ required LiveRequestQueue liveRequestQueue, String? userId, String? sessionId, Session? session, RunConfig? runConfig, AdkAbortSignal? abortSignal, })

Executes live bidirectional agent streaming with liveRequestQueue.

Implementation

Stream<Event> runLive({
  required LiveRequestQueue liveRequestQueue,
  String? userId,
  String? sessionId,
  Session? session,
  RunConfig? runConfig,
  AdkAbortSignal? abortSignal,
}) async* {
  final RunConfig config = runConfig ?? RunConfig();
  if (abortSignal?.aborted ?? false) {
    return;
  }
  config.responseModalities ??= <String>['AUDIO'];
  if (agent.subAgents.isNotEmpty) {
    if (config.responseModalities!.contains('AUDIO') &&
        config.outputAudioTranscription == null) {
      config.outputAudioTranscription = <String, Object?>{};
    }
    config.inputAudioTranscription ??= <String, Object?>{};
  }

  if (session == null) {
    if (userId == null || sessionId == null) {
      throw ArgumentError(
        'Either session or both userId and sessionId are required.',
      );
    }
    session = await _getOrCreateSession(
      userId: userId,
      sessionId: sessionId,
      getSessionConfig: config.getSessionConfig,
    );
    if (abortSignal?.aborted ?? false) {
      return;
    }
  }

  final InvocationContext context = _newInvocationContext(
    session,
    liveRequestQueue: liveRequestQueue,
    runConfig: config,
    abortSignal: abortSignal,
  );

  context.agent = _findAgentToRun(context.session, agent);
  if (context.isAborted) {
    return;
  }

  final Set<BaseToolset> toolsets = _collectToolsets(agent);
  try {
    await for (final Event event in _execWithPlugin(
      invocationContext: context,
      session: session,
      execute: (InvocationContext ctx) => ctx.agent.runLive(ctx),
      isLiveCall: true,
    )) {
      yield event;
    }
  } finally {
    await _cleanupToolsets(toolsets, ignoreErrors: true);
  }
}