跳转到内容

运行时

什么是运行时？

ADK Runtime是支撑智能体应用在用户交互期间运行的基础引擎。该系统负责接收您定义的智能体、工具和回调函数，并根据用户输入协调它们的执行，管理信息流、状态变更以及与LLM或存储等外部服务的交互。

将运行时(Runtime)视为您智能体应用的"引擎"。您定义各个组件(智能体、工具)，而运行时则负责处理它们如何连接并协同运行以满足用户请求。

核心理念：事件循环

ADK Runtime的核心是一个事件循环。这个循环促进了Runner组件与你定义的"执行逻辑"(包括你的智能体、它们进行的LLM调用、回调和工具)之间的双向通信。

intro_components.png

简单来说：

  1. Runner接收用户查询并请求主Agent开始处理。
  2. Agent（及其关联逻辑）会持续运行，直到有需要报告的内容（如响应、使用工具的请求或状态变更）——随后它会生成一个Event。
  3. Runner接收此Event，处理任何关联操作（如通过Services保存状态变更），并将事件继续转发（例如转发至用户界面）。
  4. 只有在Runner处理完事件后，Agent的逻辑才会恢复执行，此时可能会看到Runner提交的变更所产生的影响。
  5. 这个循环会持续重复，直到智能体对当前用户查询没有更多事件可生成。

这个事件驱动的循环是控制ADK如何执行您智能体代码的基本模式。

核心机制：事件循环 - 内部工作原理

事件循环是定义Runner与您的自定义代码（智能体、工具、回调函数，在设计文档中统称为"执行逻辑"或"逻辑组件"）之间交互的核心运行模式。它建立了明确的职责划分：

运行器角色（协调器）

Runner作为单次用户调用的核心协调器。它在循环中的职责包括：

  1. 初始化：接收终端用户的查询（new_message），通常通过SessionService将其附加到会话历史记录中。
  2. 启动: 通过调用主智能体的执行方法(例如 agent_to_run.run_async(...))来开始事件生成过程。
  3. Receive & Process: Waits for the agent logic to yield an Event. Upon receiving an event, the Runner promptly processes it. This involves:
    • 使用配置好的Services（SessionService、ArtifactService、MemoryService）来提交event.actions中指示的变更（如state_delta、artifact_delta）。
    • 执行其他内部簿记任务。
  4. Yield Upstream: 将处理过的事件继续向上传递（例如传递给调用应用程序或用户界面进行渲染）。
  5. 迭代： 向智能体逻辑发出信号，表示对当前产生事件的处理已完成，使其能够继续运行并生成下一个事件。

概念性运行循环:

# Simplified view of Runner's main loop logic
def run(new_query, ...) -> Generator[Event]:
    # 1. Append new_query to session event history (via SessionService)
    session_service.append_event(session, Event(author='user', content=new_query))

    # 2. Kick off event loop by calling the agent
    agent_event_generator = agent_to_run.run_async(context)

    async for event in agent_event_generator:
        # 3. Process the generated event and commit changes
        session_service.append_event(session, event) # Commits state/artifact deltas etc.
        # memory_service.update_memory(...) # If applicable
        # artifact_service might have already been called via context during agent run

        # 4. Yield event for upstream processing (e.g., UI rendering)
        yield event
        # Runner implicitly signals agent generator can continue after yielding

执行逻辑的角色（智能体、工具、回调）

智能体、工具和回调函数中的代码负责实际的计算和决策。它们与循环的交互包括：

  1. 执行: 根据当前的InvocationContext运行其逻辑，包括会话状态在恢复执行时的状态。
  2. Yield: 当逻辑需要通信时（发送消息、调用工具、报告状态变更），它会构建一个包含相关内容和操作的Event，然后将这个事件yield回Runner。
  3. 暂停: 关键点在于，智能体逻辑的执行会在yield语句后立即暂停。它会等待Runner完成第3步(处理和提交)。
  4. 恢复执行: 仅当Runner处理完生成的事件后，智能体逻辑才会从紧跟在yield语句后的位置继续执行。
  5. 查看更新状态：恢复后，智能体逻辑现在可以可靠地访问会话状态(ctx.session.state)，该状态反映了由Runner从先前产生的事件提交的更改。

概念执行逻辑：

# Simplified view of logic inside Agent.run_async, callbacks, or tools

# ... previous code runs based on current state ...

# 1. Determine a change or output is needed, construct the event
# Example: Updating state
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="State updated.")])
    # ... other event fields ...
)

# 2. Yield the event to the Runner for processing & commit
yield event_with_state_change
# <<<<<<<<<<<< EXECUTION PAUSES HERE >>>>>>>>>>>>

# <<<<<<<<<<<< RUNNER PROCESSES & COMMITS THE EVENT >>>>>>>>>>>>

# 3. Resume execution ONLY after Runner is done processing the above event.
# Now, the state committed by the Runner is reliably reflected.
# Subsequent code can safely assume the change from the yielded event happened.
val = ctx.session.state['field_1']
# here `val` is guaranteed to be "value_2" (assuming Runner committed successfully)
print(f"Resumed execution. Value of field_1 is now: {val}")

# ... subsequent code continues ...
# Maybe yield another event later...

在Runner与您的执行逻辑之间，通过Event对象协调的协作式yield/pause/resume循环，构成了ADK运行时的核心。

运行时的关键组件

在ADK运行时环境中，多个组件协同工作以执行智能体调用。了解它们的作用有助于阐明事件循环的运行机制：

  1. Runner

    • 角色： 单个用户查询的主要入口点和协调器（run_async）。
    • 功能： 管理整个事件循环，接收由执行逻辑产生的事件，与服务协调处理和提交事件动作（状态/工件变更），并将处理过的事件向上游转发（例如到用户界面）。它本质上基于产生的事件逐步驱动对话轮次。（定义于google.adk.runners.runner.py）。

  2. 执行逻辑组件

    • 角色：包含自定义代码和核心智能体能力的部分。
    • 组件：
    • Agent (BaseAgent, LlmAgent, 等): 您的主要逻辑单元，用于处理信息并决定采取的行动。它们实现了产生事件的_run_async_impl方法。
    • Tools (BaseTool, FunctionTool, AgentTool, 等): 智能体(通常是LlmAgent)用来与外界交互或执行特定任务的外部函数或能力。它们执行并返回结果，结果会被包装在事件中。
    • Callbacks (函数): 用户定义的附加到智能体上的函数(例如before_agent_callback, after_model_callback)，这些函数会挂接到执行流程中的特定点，可能修改行为或状态，其效果会被捕获在事件中。
    • 功能： 执行实际的思考、计算或外部交互。它们通过生成Event对象并暂停运行直到Runner处理这些对象来传递结果或需求。

  3. Event

    • 角色：在Runner和执行逻辑之间来回传递的消息。
    • 功能：表示一个原子事件（用户输入、智能体文本、工具调用/结果、状态变更请求、控制信号）。它既携带事件的内容，也包含预期的副作用（如state_delta之类的actions）。（定义在google.adk.events.event.py中）。

  4. Services

    • 角色：负责管理持久化或共享资源的后端组件。主要在事件处理期间由Runner使用。
    • 组件：
    • SessionService (BaseSessionService, InMemorySessionService等): 管理Session对象，包括保存/加载会话，将state_delta应用到会话状态，以及将事件追加到event history中。
    • ArtifactService (BaseArtifactService, InMemoryArtifactService, GcsArtifactService等): 管理二进制工件数据的存储和检索。虽然在执行逻辑中通过上下文调用save_artifact，但事件中的artifact_delta会向Runner/SessionService确认该操作。
    • MemoryService (BaseMemoryService, etc.): (可选) 管理用户跨会话的长期语义记忆。
    • 功能：提供持久化层。Runner与它们交互，确保在继续执行逻辑之前，由event.actions发出的变更被可靠地存储。

  5. Session

    • 角色： 一个数据容器，用于保存用户与应用之间特定对话的状态和历史记录。
    • 功能：存储当前的state字典、所有过去的events列表(event history)，以及相关工件的引用。这是交互的主要记录，由SessionService管理。(定义在google.adk.sessions.session.py中)。

  6. Invocation

    • 角色： 一个概念性术语，表示从Runner接收到用户查询开始，到智能体逻辑完成对该查询的事件生成期间发生的所有事情。
    • 功能： 一次调用可能涉及多个智能体运行（如果使用智能体转移或AgentTool）、多次LLM调用、工具执行和回调执行，所有这些都通过InvocationContext中的单个invocation_id关联在一起。

这些玩家通过事件循环持续交互以处理用户的请求。

工作原理：简化调用示例

让我们追踪一个涉及LLM智能体调用工具的典型用户查询的简化流程:

intro_components.png

逐步分解

  1. 用户输入： 用户发送一个查询（例如："法国的首都是什么？"）。
  2. 运行器启动： Runner.run_async 开始执行。它会与 SessionService 交互以加载相关的 Session，并将用户查询作为第一个 Event 添加到会话历史中。同时会准备一个 InvocationContext（ctx）。
  3. 智能体执行: Runner 调用 agent.run_async(ctx) 在指定的根智能体上(例如一个 LlmAgent)。
  4. LLM调用（示例）： Agent_Llm智能体判断它需要信息，可能是通过调用某个工具。它会准备一个对LLM的请求。假设LLM决定调用MyTool。
  5. 生成 FunctionCall 事件： Agent_Llm 从大语言模型接收到 FunctionCall 响应，将其封装为 Event(author='Agent_Llm', content=Content(parts=[Part(function_call=...)])) 事件，并通过 yield 生成该事件。
  6. 智能体暂停: Agent_Llm的执行会在yield后立即暂停。
  7. 运行器进程： Runner接收FunctionCall事件。它将该事件传递给SessionService以记录在历史中。然后Runner将事件向上游传递给User（或应用程序）。
  8. 智能体恢复: Runner 发出信号表示事件已处理完毕，Agent_Llm 恢复执行。
  9. 工具执行: Agent_Llm 的内部流程现在继续执行请求的 MyTool。它调用 tool.run_async(...)。
  10. 工具返回结果： MyTool 执行并返回其结果（例如 {'result': 'Paris'}）。
  11. Yield FunctionResponse 事件： 智能体 (Agent_Llm) 将工具结果封装成一个包含 FunctionResponse 部分的 Event（例如 Event(author='Agent_Llm', content=Content(role='user', parts=[Part(function_response=...)]))）。如果工具修改了状态（state_delta）或保存了工件（artifact_delta），该事件可能还包含 actions。智能体通过 yield 生成这个事件。
  12. 智能体暂停: Agent_Llm 再次暂停。
  13. 运行器进程： Runner接收FunctionResponse事件。它将该事件传递给SessionService，后者会应用任何state_delta/artifact_delta并将事件添加到历史记录中。Runner将事件向上游传递。
  14. 智能体恢复: Agent_Llm 恢复运行，此时已获知工具结果并提交所有状态变更。
  15. 最终LLM调用（示例）： Agent_Llm 将工具结果发送回 LLM 以生成自然语言响应。
  16. 生成最终文本事件： Agent_Llm 从 LLM 接收最终文本，将其包装在 Event(author='Agent_Llm', content=Content(parts=[Part(text=...)])) 中，然后 yield 该事件。
  17. 智能体暂停: Agent_Llm 暂停.
  18. 运行器进程： Runner接收最终的文本事件，将其传递给SessionService进行历史记录，并向上游传递给User。这很可能被标记为is_final_response()。
  19. 智能体恢复与完成: Agent_Llm 恢复运行。已完成本次调用的任务，其run_async生成器结束。
  20. Runner 完成: Runner 检测到智能体的生成器已耗尽，并完成本次调用的循环。

这种yield/pause/process/resume循环确保状态变更能够被一致地应用，并且执行逻辑总是在产生事件后基于最新提交的状态进行操作。

重要运行时行为

理解ADK运行时如何处理状态、流式处理和异步操作的一些关键方面，对于构建可预测且高效的智能体至关重要。

状态更新与提交时机

  • 规则: 当你的代码(在智能体、工具或回调中)修改会话状态时(例如context.state['my_key'] = 'new_value')，这个变更最初会记录在当前InvocationContext的本地范围内。只有当携带对应state_delta的Event在你的代码中通过yield返回并被Runner处理后，这个变更才确保会被持久化(由SessionService保存)。

  • 含义： 从yield恢复后运行的代码可以可靠地假设，在yielded event中发出的状态变更已被提交。

# Inside agent logic (conceptual)

# 1. Modify state
ctx.session.state['status'] = 'processing'
event1 = Event(..., actions=EventActions(state_delta={'status': 'processing'}))

# 2. Yield event with the delta
yield event1
# --- PAUSE --- Runner processes event1, SessionService commits 'status' = 'processing' ---

# 3. Resume execution
# Now it's safe to rely on the committed state
current_status = ctx.session.state['status'] # Guaranteed to be 'processing'
print(f"Status after resuming: {current_status}")

会话状态的"脏读"

  • 定义： 虽然提交发生在yield之后，但在同一调用内稍后运行且在状态变更事件实际被yield和处理之前的代码，通常可以看到本地的未提交更改。这有时被称为"脏读"。
  • 示例：
# Code in before_agent_callback
callback_context.state['field_1'] = 'value_1'
# State is locally set to 'value_1', but not yet committed by Runner

# ... agent runs ...

# Code in a tool called later *within the same invocation*
# Readable (dirty read), but 'value_1' isn't guaranteed persistent yet.
val = tool_context.state['field_1'] # 'val' will likely be 'value_1' here
print(f"Dirty read value in tool: {val}")

# Assume the event carrying the state_delta={'field_1': 'value_1'}
# is yielded *after* this tool runs and is processed by the Runner.
  • 影响：
  • 优势：允许在单个复杂步骤中（例如，在下一次LLM轮次前的多个回调或工具调用）的不同逻辑部分通过状态进行协调，而无需等待完整的yield/commit周期。
  • 注意事项：在关键逻辑中过度依赖脏读可能存在风险。如果调用在携带state_delta的事件被Runner生成和处理之前就失败了，那么未提交的状态变更将会丢失。对于关键的状态转换，请确保它们与一个被成功处理的事件相关联。

流式输出与非流式输出 (partial=True)

这主要涉及如何处理来自大语言模型的响应，尤其是在使用流式生成API时。

  • 流式传输: 大语言模型逐个令牌或以小块形式生成响应。
  • 框架（通常在BaseLlmFlow内）会为单个概念性响应生成多个Event对象。这些事件大多数会带有partial=True参数。
  • 当Runner接收到带有partial=True的事件时，通常会立即将其转发给上游（用于UI显示），但跳过处理其actions（如state_delta）。
  • 最终，框架会为该响应生成一个最终事件，标记为非部分事件（partial=False 或通过 turn_complete=True隐式表示）。
  • Runner 仅完全处理这个最终事件，提交任何关联的state_delta或artifact_delta。
  • 非流式: 大语言模型一次性生成完整响应。框架会产生一个标记为非部分事件的单一事件，由Runner完整处理。
  • 重要性： 确保状态变更基于LLM的完整响应以原子方式且仅应用一次，同时仍允许UI在文本生成过程中逐步显示内容。

异步是首要方式 (run_async)

  • 核心设计： ADK Runtime 从根本上基于 Python 的 asyncio 库构建，用于高效处理并发操作（如等待 LLM 响应或工具执行）而不会阻塞。
  • 主入口点： Runner.run_async 是执行智能体调用的主要方法。所有核心可运行组件（智能体、特定流程）内部都使用 async def 方法。
  • 同步便捷方法(run): 同步的Runner.run方法主要为方便使用而存在(例如在简单脚本或测试环境中)。不过在内部实现上，Runner.run通常只是调用Runner.run_async并为您管理异步事件循环的执行。
  • 开发者体验： 通常你应该使用asyncio来设计你的应用逻辑（例如使用ADK的web服务器）。
  • 同步回调/工具： 该框架旨在无缝处理作为工具或回调提供的async def和常规def函数。长时间运行的同步工具或回调，特别是执行阻塞I/O的操作，可能会阻塞主asyncio事件循环。该框架可能使用类似asyncio.to_thread的机制来缓解这种情况，通过在单独的线程池中运行此类阻塞同步代码，防止其阻碍其他异步任务。然而，CPU密集型的同步代码仍会阻塞其运行的线程。

理解这些行为有助于您编写更健壮的ADK应用程序，并调试与状态一致性、流式更新和异步执行相关的问题。