Method _run_node_live

src/google/adk/runners.py:587–644

Run a non-agent BaseNode in live mode.

(
      self,
      *,
      session: Session,
      live_request_queue: LiveRequestQueue,
      run_config: Optional[RunConfig] = None,
  )

          )

  async def _run_node_live(
      self,
      *,
      session: Session,
      live_request_queue: LiveRequestQueue,
      run_config: Optional[RunConfig] = None,
  ) -> AsyncGenerator[Event, None]:
    """Run a non-agent BaseNode in live mode."""
    from .agents.context import Context
    from .workflow._dynamic_node_scheduler import DynamicNodeScheduler
    from .workflow._node_runner import NodeRunner
    from .workflow._workflow import _LoopState
    from .workflow._workflow import Workflow

    ic = self._new_invocation_context_for_live(
        session,
        live_request_queue=live_request_queue,
        run_config=run_config or RunConfig(),
    )
    ic._event_queue = asyncio.Queue()

    root_ctx = Context(ic)
    root_agent = self.agent
    is_workflow = isinstance(root_agent, Workflow)

    if not is_workflow:
      root_node_runner = NodeRunner(node=root_agent, parent_ctx=root_ctx)

    done_sentinel = object()

    async def _drive_root_node():
      try:
        if is_workflow:
          scheduler = DynamicNodeScheduler(state=_LoopState())
          root_ctx._workflow_scheduler = scheduler
          ctx = await scheduler(
              root_ctx,
              root_agent,
              None,
              run_id='1',
          )
        else:
          ctx = await root_node_runner.run(
              node_input=None,
          )
        if ctx.error:
          raise ctx.error
      finally:
        await ic._event_queue.put((done_sentinel, None))

    task = asyncio.create_task(_drive_root_node())

    try:
      async with aclosing(self._consume_event_queue(ic, done_sentinel)) as agen:
        async for event in agen:
          yield event
    finally:
      await self._cleanup_root_task(task, self.agent.name)