前篇《 AI - ADK Runtime（运行时）：Runner、Event Loop、Session/State 是怎么“跑起来”的？》我们了解了 ADK Runtime 过程，这篇我们来了解一下 ADK Runtime Config。

当你用 ADK 写好 Agent、Tools 之后，接下来一个非常实际的问题是：

同一个 Agent，在不同场景下要“以什么运行方式”执行？

比如：

• 要不要 流式输出（边生成边返回）？
• 要不要 语音输出、以及语音用什么声音/语言？
• 用户上传的文件要不要保存下来便于审计/排查？
• 这次运行最多允许调用 LLM 多少次，防止“跑飞”？

ADK 把这一切收敛成一个统一的运行时配置对象：RunConfig。它定义了 agent 的运行时行为与选项，覆盖语音、流式、函数调用相关能力（CFC）、是否保存输入为 artifacts、以及单次运行中 LLM 调用上限等。

强调默认行为：默认不启用 streaming，输入也不会作为 artifacts 保留；你需要通过 RunConfig 去覆盖这些默认值

一、RunConfig 你可以把它理解为“这次 run 的开关面板”

RunConfig 的定位不是改 Agent 的业务逻辑，而是改“这次运行的形态”。当你构建一次 agent run 时，你可以传入 RunConfig 来定制 agent 与模型交互的方式、音频处理方式、以及响应如何流式返回。

二、多语言里 RunConfig 长得不一样，但表达的是同一件事

ADK 在 Python/TS/Go/Java 都提供 RunConfig，但各语言风格不同：

• Python：RunConfig 是 Pydantic BaseModel，并且 extra=‘forbid’（多余字段会被拒绝）。

class RunConfig(BaseModel):
    """Configs for runtime behavior of agents."""

    model_config = ConfigDict(
        extra='forbid',
    )

    speech_config: Optional[types.SpeechConfig] = None
    response_modalities: Optional[list[str]] = None
    save_input_blobs_as_artifacts: bool = False
    support_cfc: bool = False
    streaming_mode: StreamingMode = StreamingMode.NONE
    output_audio_transcription: Optional[types.AudioTranscriptionConfig] = None
    max_llm_calls: int = 500

• TypeScript：是 interface RunConfig + enum StreamingMode。

export interface RunConfig {
  speechConfig?: SpeechConfig;
  responseModalities?: Modality[];
  saveInputBlobsAsArtifacts: boolean;
  supportCfc: boolean;
  streamingMode: StreamingMode;
  outputAudioTranscription?: AudioTranscriptionConfig;
  maxLlmCalls: number;
  // ... and other properties
}

export enum StreamingMode {
  NONE = 'none',
  SSE = 'sse',
  BIDI = 'bidi',
}

• Go：是可变 struct（示例中包含 StreamingMode、是否保存输入 blobs）。

type StreamingMode string

const (
    StreamingModeNone StreamingMode = "none"
    StreamingModeSSE  StreamingMode = "sse"
)

// RunConfig controls runtime behavior.
type RunConfig struct {
    // Streaming mode, None or StreamingMode.SSE.
    StreamingMode StreamingMode
    // Whether or not to save the input blobs as artifacts
    SaveInputBlobsAsArtifacts bool
}

• Java：通常是不可变数据结构风格（文档示例用 builder），并包含 StreamingMode 枚举。

public abstract class RunConfig {

  public enum StreamingMode {
    NONE,
    SSE,
    BIDI
  }

  public abstract @Nullable SpeechConfig speechConfig();

  public abstract ImmutableList<Modality> responseModalities();

  public abstract boolean saveInputBlobsAsArtifacts();

  public abstract @Nullable AudioTranscriptionConfig outputAudioTranscription();

  public abstract int maxLlmCalls();

  // ...
}

对学习者来说：你主要要掌握的是“参数语义”，语言只是写法不同。

三、RunConfig 的参数总览

最该记住的 7 个运行时参数如下：

1. streaming_mode：决定响应是不是流式、以及流式类型

支持的模式：

• NONE：不流式，一次性返回完整结果
• SSE：Server-Sent Events（单向流：服务端 → 客户端）
• BIDI：双向流（同时双向通信）

提醒：流式模式会显著影响性能和体验：SSE 让用户能看到“边生成边输出”的响应；BIDI 支持更实时的交互体验。

2. max_llm_calls：单次 run 最多允许调用 LLM 多少次

这是一个非常“工程化”的安全阀：限制一次 agent run 中 LLM 的总调用次数，避免无限循环/失控调用。

规则：

• 大于 0 且小于 sys.maxsize：会强制限制
• <= 0：允许无限（明确“不建议用于生产”）

3. save_input_blobs_as_artifacts：是否把输入 blobs 保存成 artifacts

当你希望 调试、审计 时非常有用：开启后会把输入 blobs（例如上传文件）作为 artifacts 保存，便于回放“当时 agent 实际收到了什么”。
默认是 False

4. speech_config：语音输出怎么说（声音、语言）

这是“语音能力”的核心配置，它用于具有音频能力的 live agents。

SpeechConfig 的结构（文档给了类结构）：

• voice_config：选择声音（通过 VoiceConfig → PrebuiltVoiceConfig → voice_name）
• language_code：语言（ISO 639，例如 en-US），并注明 只在 Live API 可用

强调：SpeechConfig 的接口/定义在不同语言里是一致的（语义一致）。

SpeechConfig class

class SpeechConfig(_common.BaseModel):
    """The speech generation configuration."""

    voice_config: Optional[VoiceConfig] = Field(
        default=None,
        description="""The configuration for the speaker to use.""",
    )
    language_code: Optional[str] = Field(
        default=None,
        description="""Language code (ISO 639. e.g. en-US) for the speech synthesization.
        Only available for Live API.""",
    )

VoiceConfig class

class VoiceConfig(_common.BaseModel):
    """The configuration for the voice to use."""

    prebuilt_voice_config: Optional[PrebuiltVoiceConfig] = Field(
        default=None,
        description="""The configuration for the speaker to use.""",
    )

PrebuiltVoiceConfig class

class PrebuiltVoiceConfig(_common.BaseModel):
    """The configuration for the prebuilt speaker to use."""

    voice_name: Optional[str] = Field(
        default=None,
        description="""The name of the prebuilt voice to use.""",
    )

5. response_modalities：输出“要文本还是音频，还是都要”

response_modalities 用来定义输出模态（text/audio 等）。文关键的一句是：如果不设置，默认是 AUDIO。

• Python 可能用 [“TEXT”, “AUDIO”]
• Java/TS 用结构化的 Modality 对象列表

6. output_audio_transcription：把“生成的音频输出”自动转写成文本

这是“音频输出→文本转写”的配置项，强调它适用于具备音频响应能力的 live agents，可用于无障碍、记录留存、多模态应用等。

7. support_cfc：开启 CFC（Compositional Function Calling）

这是一个“高级/实验性”开关：开启后允许基于模型输出动态执行函数，更适合复杂工作流场景。
但它有硬性前置条件与限制：

• 只在 StreamingMode.SSE 时适用
• 启用后会调用 LIVE API（因为只有 LIVE API 支持 CFC）
• 并且这是 Experimental release，未来 API/行为可能变更

四、Validation Rules：哪些值会被认为“不合理”？

这部分很实用，属于“避免你踩坑”的规则集。
RunConfig 会校验参数以确保 agent 正常运行。Python 依赖 Pydantic 自动校验；Java/TS 依赖静态类型系统，并可能在构造函数里加显式检查。

特别是 max_llm_calls：

• 极大值（如 Python 的 sys.maxsize、Java 的 Integer.MAX_VALUE、TS 的 Number.MAX_SAFE_INTEGER）通常会被禁止，以避免问题
• 0 或负数通常会触发“无限 LLM 交互”的警告

五、4 个典型配置“配方”

下面这些都是文档在 “Validation Rules” 里给出的示例组合（我用“你会在什么场景用它”来解释）。

1. 基础配置：不流式 + 限制 LLM 调用次数

适合：一次性输出、简单任务型 agent。

from google.genai.adk import RunConfig, StreamingMode

config = RunConfig(
    streaming_mode=StreamingMode.NONE,
    max_llm_calls=100
)

2. 开启 SSE Streaming：边生成边返回

适合：对话机器人、助手类产品，提升“响应很快”的体感。

from google.genai.adk import RunConfig, StreamingMode

config = RunConfig(
    streaming_mode=StreamingMode.SSE,
    max_llm_calls=200
)

3. 开启语音能力：speech_config + 双模态输出（AUDIO + TEXT）

给了一个“综合示例”来展示：语音 + 文本、保存 artifacts、开启 CFC、SSE streaming、并把 max_llm_calls 拉到 1000。

from google.genai.adk import RunConfig, StreamingMode
from google.genai import types

config = RunConfig(
    speech_config=types.SpeechConfig(
        language_code="en-US",
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
        ),
    ),
    response_modalities=["AUDIO", "TEXT"],
    save_input_blobs_as_artifacts=True,
    support_cfc=True,
    streaming_mode=StreamingMode.SSE,
    max_llm_calls=1000,
)

这个配置启用了: Kore 声音、音频+文本输出、保存输入 blobs、实验性 CFC、SSE、LLM 调用上限 1000。

4. 只开启 CFC（实验性）：SSE + support_cfc

适合：你明确需要 CFC 的场景（并能接受 experimental 特性）。

from google.genai.adk import RunConfig, StreamingMode

config = RunConfig(
    streaming_mode=StreamingMode.SSE,
    support_cfc=True,
    max_llm_calls=150
)

六、实例：开启 SSE Streaming vs 不开启

核心点是用 RunConfig.streaming_mode 控制是否产出 partial=True 的事件。默认是不 streaming。而 Runner.run_async(…, run_config=…) 是运行入口，会不断 yield Event。

run_config = RunConfig(streaming_mode=streaming_mode)
async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=user_msg,
        run_config=run_config,
    )

我们复用《AI - 使用 Google ADK 创建你的第一个 AI Agent》agent 文里定义的最简单的 Agent，再新一个 compare_sse_vs_none.py 文件，引用 agent 文件中的 root_agent，注意引用之前必须先导入环境变量。

from agent import root_agent

compare_sse_vs_none.py 完整代码如下，可直接运行即可：

import asyncio
import time
from typing import Optional
from pathlib import Path
from dotenv import load_dotenv, find_dotenv

from google.adk.runners import InMemoryRunner
from google.genai import types

from google.adk.agents.run_config import RunConfig, StreamingMode

# 1. 先加载 .env，确保 agent.py 里能拿到网关和模型配置
try:
    _env_path = Path(__file__).parent / ".env"
    if _env_path.exists():
        load_dotenv(dotenv_path=_env_path)
    else:
        found = find_dotenv(usecwd=True)
        if found:
            load_dotenv(found)
        else:
            load_dotenv()
except Exception:
    # dotenv 是可选的，即便失败也继续
    pass

from agent import root_agent

def content_to_text(content: Optional[types.Content]) -> str:
    """把 Content.parts 里的 text 拼起来（忽略非 text part）。"""
    if not content or not getattr(content, "parts", None):
        return ""
    buf = []
    for p in content.parts:
        t = getattr(p, "text", None)
        if t:
            buf.append(t)
    return "".join(buf)


async def ensure_session(runner: InMemoryRunner, user_id: str, session_id: str) -> None:
    """run_async 要求 session 已存在：不存在则创建。:contentReference[oaicite:4]{index=4}"""
    sess = await runner.session_service.get_session(
        app_name=runner.app_name, user_id=user_id, session_id=session_id
    )
    if sess is None:
        await runner.session_service.create_session(
            app_name=runner.app_name, user_id=user_id, session_id=session_id
        )


async def run_once(label: str, streaming_mode):
    # 一个尽量输出长一点的 agent，方便观察“是否逐步输出”
    runner = InMemoryRunner(agent=root_agent, app_name="sse_demo_app")

    user_id = "u1"
    session_id = f"s_{label}"

    await ensure_session(runner, user_id, session_id)

    msg_text = "解释一下什么是 SSE，并列出 10 个你认为适合用 SSE 的场景。"
    user_msg = types.Content(role="user", parts=[types.Part(text=msg_text)])

    run_config = RunConfig(streaming_mode=streaming_mode)

    print("\n" + "=" * 80)
    print(f"CASE: {label} | streaming_mode = {streaming_mode}")
    print("=" * 80)

    last_text = ""
    t0 = time.time()

    # Runner.run_async 是主入口，会 yield Event（可能包含 partial=True 的分段事件）:contentReference[oaicite:5]{index=5}
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=user_msg,
        run_config=run_config,
    ):
        content = getattr(event, "content", None)
        if not content:
            continue

        # 只看模型输出（避免把 user message 也打印出来）
        if getattr(content, "role", None) != "model":
            continue

        text = content_to_text(content)
        if not text:
            continue

        is_partial = bool(getattr(event, "partial", False))
        dt = time.time() - t0

        if is_partial:
            # 常见情况下 partial 事件携带“累计文本”，这里做 delta 打印
            delta = text[len(last_text) :] if text.startswith(last_text) else text
            if delta:
                # 打印时不换行，模拟打字机效果
                print(delta, end="", flush=True)
            last_text = text
        else:
            # 最终事件：补齐最后一段并换行
            delta = text[len(last_text) :] if text.startswith(last_text) else text
            if delta:
                print(delta, end="", flush=True)
            print(f"\n[done in {dt:.2f}s]")
            break


async def main():
    # 不开启 streaming：默认是 NONE（无 streaming）:contentReference[oaicite:6]{index=6}
    await run_once("NO_STREAM", StreamingMode.NONE)

    # 开启 SSE streaming：会更可能产出 partial=True 的事件，让你看到“逐步输出”:contentReference[oaicite:7]{index=7}
    await run_once("SSE_STREAM", StreamingMode.SSE)


if __name__ == "__main__":
    asyncio.run(main())

运行结果，NO_STREAM 是一次性输出全部结果的

================================================================================
CASE: NO_STREAM | streaming_mode = StreamingMode.NONE
================================================================================
SSE 指的是 **Server-Sent Events**（服务器发送事件），它是一种基于 HTTP 协议的技术，用于服务器向客户端发送实时更新。SSE 是一种单向通信机制，服务器会持续向客 户端推送事件，而客户端只需进行初始化连接。它是 HTML5 标准的一部分，主要用于实时数据推送，轻量级且适合用来取代基于轮询的更新方式。

相比 WebSocket 双向通信的复杂性，SSE 更简单、轻量且支持自动重连，非常适合应用在某些需要实时更新的场景中。

### 常见特性：
1. **单向通信**：从服务器到客户端。
2. **基于 HTTP 协议**：无需额外协议，直接与浏览器兼容。
3. **自动重连**：连接断开时会自动尝试重新连接。
4. **事件支持**：可以定义不同类型的事件。

---

### 适合使用 SSE 的 10 个场景：
1. **实时通知系统**：
   - 例如用户的消息通知、系统警告或应用提醒。

2. **实时股票/金融行情**：
   - 推送股票市场价格或比特币实时价格等数据。

3. **实时新闻 Feed**：
   - 像新闻客户端一样，可以动态显示最新的新闻内容。

4. **在线聊天系统**：
   - 单向更新，例如用户列表的实时状态更新。

5. **实时评论流**：
   - 比如直播页面上的观众评论流（适用于不需频繁回传的场景）。

6. **实时仪表盘更新**：
   - 显示监控中的数据更新，比如服务器负载、网络流量等数据。

7. **社交媒体动态推送**：
   - Twitter 或 Facebook 的实时动态更新推送。

8. **实时投票统计**：
   - 各类投票系统中进行实时的票数变化显示。

9. **位置或状态跟踪**：
   - 比如订单物流更新、实时位置跟踪等。

10. **物联网设备推送**：
    - 物联网（IoT）设备中向前端推送状态更新，例如智能家居设备温度或者状态变化。

而 SSE_STREAM 是部分逐步输出的

================================================================================
CASE: SSE_STREAM | streaming_mode = StreamingMode.SSE
================================================================================
**SE（Server-Sent Events）** 是一种基于 HTTP 的技术，用来实现服务端向客户端推送实时更新。SE 通常以流的形式将消息从服务端发送到浏览器，非常适合需要持续更新或实时数据传输的应 用场景。它的核心特点是：消息是由服务端主动推送的，而客户端只需要保持一个简单的 HTTP 连接，无需频繁地发起请求。

SE 是 HTML5 标准的一部分，使用 `EventSource` API 在客户端建立连接。其优点包括：
1. 简单而轻量级；
2. 支持文本传输；
3. 连接会自动重试和恢复（支持断线重连）；
4. 占用资源少，简单实现实时应用。

以下是适合使用 SSE 的 10 个场景：

1. **实时消息推送**：如在线客服系统。在一个聊天场景中，服务端可以通过 SSE 推送在线用户列表更新、新的消息提醒等。

2. **股票价格更新**：证券交易系统可以通过 SSE 向用户推送股票的最新价格、电量趋势以及市场新闻。

3. **在线游戏状态更新**：多人在线游戏场景，某个大厅或房间的状态变化可以通过 SSE 实时展示给所有联机参与者。

4. **实时赛车或体育比赛动态**：应用程序可以实时向用户推送比分变化、选手排名变化或场上发生的事件。

5. **实时数据仪表盘**：在运营、运维仪表盘上，展示系统性能（如 CPU 使用情况、内存占用率）或用户行为指标的实时动态变化。

6. **社交媒体更新**：用户的好友动态、评论、点赞、关注提醒等可以通过 SSE 实时传递更新，无需刷新页面。

7. **新闻或公告实时更新**：在线门户网站或企业内部信息平台使用 SSE 推送头条新闻、突发事件或公告。

8. **在线教育平台的课堂通知和状态更新**：如直播课堂中，教师推送通知或手势提示、题目发布等场景。

9. **服务器监控和日志实时查看**：开发人员可以用 SSE 创建工具来实时跟踪服务器日志或系统事件，而无需手动刷新页面。

10. **多用户协作应用状态同步**：如协作文档工具（Google Docs 类似应用），实现多人间实时状态同步，如文字输入、光标移动等。

---

相比 WebSocket（另一种支持双向通信的技术），SE 的特点是**单向通信（只服务端发消息）**，适合那些只需要将实时数据从服务器发送到客户端的应用场景。因此，它是轻量级、简单但功能强大的实时通信解决方案之一。**SSE（Server-Sent Events）** 是一种基于 HTTP 的技术，用来实现服务端向客户端推送实时更新。SSE 通常以流的形式将消息从服务端发送到浏览器，非常适合需 要持续更新或实时数据传输的应用场景。它的核心特点是：消息是由服务端主动推送的，而客户端只需要保持一个简单的 HTTP 连接，无需频繁地发起请求。

SSE 是 HTML5 标准的一部分，使用 `EventSource` API 在客户端建立连接。其优点包括：
1. 简单而轻量级；
2. 支持文本传输；
3. 连接会自动重试和恢复（支持断线重连）；
4. 占用资源少，简单实现实时应用。

以下是适合使用 SSE 的 10 个场景：

1. **实时消息推送**：如在线客服系统。在一个聊天场景中，服务端可以通过 SSE 推送在线用户列表更新、新的消息提醒等。

2. **股票价格更新**：证券交易系统可以通过 SSE 向用户推送股票的最新价格、电量趋势以及市场新闻。

3. **在线游戏状态更新**：多人在线游戏场景，某个大厅或房间的状态变化可以通过 SSE 实时展示给所有联机参与者。

4. **实时赛车或体育比赛动态**：应用程序可以实时向用户推送比分变化、选手排名变化或场上发生的事件。

5. **实时数据仪表盘**：在运营、运维仪表盘上，展示系统性能（如 CPU 使用情况、内存占用率）或用户行为指标的实时动态变化。

6. **社交媒体更新**：用户的好友动态、评论、点赞、关注提醒等可以通过 SSE 实时传递更新，无需刷新页面。

7. **新闻或公告实时更新**：在线门户网站或企业内部信息平台使用 SSE 推送头条新闻、突发事件或公告。

8. **在线教育平台的课堂通知和状态更新**：如直播课堂中，教师推送通知或手势提示、题目发布等场景。

9. **服务器监控和日志实时查看**：开发人员可以用 SSE 创建工具来实时跟踪服务器日志或系统事件，而无需手动刷新页面。

10. **多用户协作应用状态同步**：如协作文档工具（Google Docs 类似应用），实现多人间实时状态同步，如文字输入、光标移动等。

---

相比 WebSocket（另一种支持双向通信的技术），SSE 的特点是**单向通信（只服务端发消息）**，适合那些只需要将实时数据从服务器发送到客户端的应用场景。因此，它是轻量级、简单但功能 强大的实时通信解决方案之一。
[done in 21.96s]

七、学完 RunConfig，你应该形成的“配置直觉”

你可以把 RunConfig 当成一套可复用的策略：

• 产品体验优先（聊天助手）：优先 SSE streaming（必要时再考虑 BIDI）
• 成本与稳定性优先：一定要设置 max_llm_calls（并避免 0/负数在生产使用）
• 可观测性/合规审计：开启 save_input_blobs_as_artifacts 保留输入
• 语音/多模态：配好 speech_config + response_modalities，并按文档提示关注 Live API 相关限制（language_code 仅 Live API 可用）
• 复杂函数编排：再开启 support_cfc，并记住它需要 SSE、会走 Live API、且是实验性