> ## Documentation Index
> Fetch the complete documentation index at: https://docs.veadk.xyz/llms.txt
> Use this file to discover all available pages before exploring further.

# 运行时

运行时负责组织模型调用、工具执行和结果返回。VeADK 默认使用 Google ADK；需要编码类执行后端时，可以选择 `codex` 或 `piagent`。切换前应检查工具、模型配置和回调的兼容性

以下示例均需先完成[模型配置](/productions/veadk/preview/zh/components/agent/model)。火山引擎与 BytePlus 分别使用对应平台的模型端点和 API Key

## 默认运行时

不设置 `runtime` 时等同于 `runtime="adk"`，适用于标准模型调用、结构化输出和工具协作

```python main.py lines theme={null}
import asyncio
from veadk import Agent, Runner


agent = Agent(name="assistant", runtime="adk")

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain what an agent runtime does.', session_id="runtime-demo"
    ))

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

运行 `python main.py` 可看到最终文本回复。需要处理事件和流式输出时使用 `Runner.run_async`

## 同步工具并行执行

默认 ADK 运行时中的同步函数可能阻塞其他异步任务。设置 `tool_thread_pool_config` 后，同步工具可在线程池执行；当模型同一轮发出多个调用时，可并行处理

```python parallel_tools.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from google.adk.agents.run_config import ToolThreadPoolConfig

def lookup_stock(product: str) -> dict:
    """Return the demo stock count for a product."""
    return {"product": product, "stock": {"notebook": 12, "pen": 40}.get(product, 0)}

agent = Agent(
    name="stock_assistant",
    tools=[lookup_stock],
    tool_thread_pool_config=ToolThreadPoolConfig(max_workers=4),
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Check stock for notebook and pen.', session_id="runtime-demo"
    ))

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

运行 `python parallel_tools.py` 可获得两个商品的库存。示例展示配置方式，模型是否同一轮发起多个调用由模型决定；线程池不会主动拆分任务。工具若共享可变数据或连接，需保证并发访问安全

| 参数 | 类型 | 默认值 | 说明 |
| :- | :- | :- | :- |
| `ToolThreadPoolConfig.max_workers` | `int` | `4` | 最大工作线程数，必须大于等于 1 |

`RunConfig.tool_thread_pool_config` 优先于 `Agent` 上的配置。该设置只影响 ADK 同步工具，不控制 `codex` 或 `piagent` 的执行机制。并行 `run_code` 的会话隔离及 `VEADK_RUN_CODE_ISOLATE_PARALLEL_CALLS` 见[代码沙箱](/productions/veadk/preview/zh/components/tools/code-sandbox#并行调用隔离)

## 切换执行后端

| `runtime` | 用途 | 额外准备 |
| :- | :- | :- |
| `adk` | 标准智能体执行，默认值 | 无额外运行时安装 |
| `codex` | 通过 Codex SDK 执行编码任务 | 安装 `veadk-python[codex]` |
| `piagent` | 通过 PiAgent RPC 执行任务 | 提供可执行文件或允许首次下载安装 |

### 使用 Codex 运行时

安装包含 SDK 与 CLI 二进制的可选依赖组：

```bash lines theme={null}
pip install "veadk-python[codex]"
```

模型仍通过 `model_name`、`model_api_base` 和 `model_api_key` 配置。默认采用隔离工作区、`workspace_write`、关闭网络和拒绝提权；以下示例显式保留这些设置

```python codex_runtime.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from veadk.runtime.codex import CodexRuntimeConfig

agent = Agent(
    name="coding_assistant",
    runtime="codex",
    codex_runtime_config=CodexRuntimeConfig(
        sandbox="workspace_write",
        approval_mode="deny_all",
        network_access=False,
    ),
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Calculate the sum of integers from 1 to 100.', session_id="runtime-demo"
    ))

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

运行 `python codex_runtime.py`。函数工具与 MCP 工具可通过 `Agent.tools` 注册，本地 ADK `SkillToolset` 中的技能可交给运行时使用；旧 `skills_mode` 的限制见下方兼容性表

### Codex 安全配置

`Agent.codex_runtime_config` 接受 `CodexRuntimeConfig` 或同字段字典

| 参数 | 类型 | 默认值 | 说明 |
| :- | :- | :- | :- |
| `approval_mode` | `"deny_all" \| "auto_review"` | `"deny_all"` | `deny_all` 拒绝提权请求；`auto_review` 自动批准所有此类请求，不经过人工审核 |
| `sandbox` | `"read_only" \| "workspace_write" \| "full_access"` | `"workspace_write"` | 只读、工作区写入或完整主机访问 |
| `network_access` | `bool` | `False` | 仅在 `workspace_write` 下控制网络 |
| `workspace_root` | `str \| None` | `None` | 工作区根路径；未指定时创建隔离临时目录 |
| `reuse_workspace` | `bool` | `False` | 配合显式 `workspace_root` 在调用间复用工作区 |
| `reasoning_effort` | `"minimal" \| "low" \| "medium" \| "high" \| "xhigh"` | `"medium"` | 推理投入程度，实际支持取决于模型 |
| `personality` | `"none" \| "friendly" \| "pragmatic"` | `"pragmatic"` | 回答风格 |
| `max_tool_iterations` | `int` | `32` | 整轮调用中桥接 ADK 工具的迭代预算，范围 1–256，不是所有原生操作的统一上限 |
| `tool_timeout_seconds` | `float \| None` | `120.0` | 单次桥接工具超时秒数，数值必须大于 0；`None` 不设置该超时 |

<Warning>
  `auto_review` 会自动批准提权和文件修改请求，不是模型审核步骤。`full_access` 放宽主机访问，`reuse_workspace=True` 可能在调用间共享文件；仅在明确需要且可信的环境中设置。`network_access=False` 不能限制 `full_access`，该组合会在配置时被拒绝。`read_only` 也不使用这个网络开关
</Warning>

以下环境变量优先于构造配置：

| 环境变量 | 默认值 | 说明 |
| :- | :- | :- |
| `VEADK_CODEX_SANDBOX` | 未设置 | 覆盖 `sandbox` |
| `VEADK_CODEX_APPROVAL_MODE` | 未设置 | 覆盖 `approval_mode` |
| `VEADK_CODEX_WORKSPACE_ROOT` | 未设置 | 覆盖 `workspace_root` |
| `VEADK_CODEX_NETWORK_ACCESS` | 未设置 | 覆盖 `network_access`；`1`、`true`、`yes`、`on` 为开启，不区分大小写 |

### Codex 可观测性

运行时将生命周期通知、函数和 MCP 工具调用转换为 ADK 事件，可供会话、链路和前端消费。运行日志中的 `invocation_id`、`call_id`、`tool`、`status` 与 `duration_ms` 用于关联调用；token 用量通过 `codex_event_type=token_usage` 事件提供

运行时日志不等同于每次模型调用的完整链路。事件、会话或自行配置的导出器可能包含任务与工具内容，接入日志系统时应控制访问和保留范围

### 配置临时错误重试

| 环境变量 | 类型 | 默认值 | 说明 |
| :- | :- | :- | :- |
| `CODEX_SHIM_NUM_RETRIES` | `int` | `2` | 限流、服务端错误、过载和超时等临时错误的最大重试次数；`0` 关闭 |
| `CODEX_SHIM_TIMEOUT` | `float` | `0` | 单次模型后端请求超时秒数；`0` 使用客户端默认值 |

### 使用 PiAgent

<Warning>
  PiAgent 在本机工作目录执行任务，内置工具可能读写文件和执行命令。配置目录隔离不等于操作系统沙箱；请使用可信项目及受限运行环境，并按需限制工具
</Warning>

```python piagent_runtime.py lines theme={null}
import asyncio
from veadk import Agent, Runner


agent = Agent(name="coding_assistant", runtime="piagent")

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain the steps for checking a CSV file without modifying files.', session_id="runtime-demo"
    ))

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

运行 `python piagent_runtime.py`。VeADK 依次查找 `PIAGENT_BINARY`、托管缓存；未找到时从 Pi Release 下载。下载只有在设置 `PIAGENT_BINARY_SHA256` 时才校验该摘要。离线与生产环境建议预先安装并指定可执行文件

| 环境变量 | 默认值 | 说明 |
| :- | :- | :- |
| `PIAGENT_BINARY` | 未设置 | 已安装的可执行文件 |
| `PIAGENT_INSTALL_DIR` | `~/.cache/veadk/piagent` | 下载与缓存目录 |
| `PIAGENT_BINARY_URL` | Release 地址 | 自定义下载地址 |
| `PIAGENT_BINARY_SHA256` | 未设置 | 下载归档的预期 SHA-256 |
| `PIAGENT_BINARY_VERSION` | `latest` | 自动下载的版本 |
| `PIAGENT_BINARY_REPO` | `earendil-works/pi` | Release 仓库 |
| `PIAGENT_BINARY_PLATFORM` | 当前系统与架构 | 支持 Linux/macOS 的 amd64、arm64，以及 Windows amd64 |
| `PIAGENT_AGENT_DIR` | 每次运行的临时目录 | 隔离配置与会话目录，不能指向真实 `~/.pi/agent` 或其子目录 |
| `PIAGENT_ALLOW_PARENT_PI_CODING_AGENT_DIR` | `false` | 未显式指定目录时，允许使用父进程的 `PI_CODING_AGENT_DIR`，仍需满足隔离约束 |
| `PIAGENT_WORKDIR` | 当前目录 | 任务工作目录 |
| `PIAGENT_TIMEOUT_SECONDS` | `600` | 单次执行超时秒数 |
| `PIAGENT_PROVIDER_ID` | `veadk` | Pi 中注册的提供商标识 |
| `PIAGENT_MODEL_API` | `openai-completions` | Pi 使用的模型 API 类型 |
| `PIAGENT_MODEL_API_KEY_ENV` | `VEADK_PI_MODEL_API_KEY` | 子进程接收模型 Key 的变量名称 |
| `PIAGENT_DISABLE_TOOLS` | `false` | 禁用工具的开关；桥接工具注册时可能重新启用工具，不应作为隔离保证 |
| `PIAGENT_DISABLE_BUILTIN_TOOLS` | `false` | 禁用 Pi 内置工具 |
| `PIAGENT_TOOL_ALLOWLIST` | 空 | 逗号分隔的内置工具允许列表 |
| `PIAGENT_EXCLUDE_TOOLS` | 空 | 逗号分隔的排除工具列表 |
| `PIAGENT_DISABLE_EXTENSION_DISCOVERY` | `true` | 禁用自动发现扩展 |
| `PIAGENT_ENABLE_EXTENSION_DISCOVERY` | 未设置 | 启用自动发现；存在对应 `DISABLE` 变量时后者优先 |
| `PIAGENT_DISABLE_SKILL_DISCOVERY` | `true` | 禁用自动发现技能，不影响显式传入的技能 |
| `PIAGENT_ENABLE_SKILL_DISCOVERY` | 未设置 | 启用自动发现；存在对应 `DISABLE` 变量时后者优先 |
| `PIAGENT_PROJECT_TRUST` | `deny` | 项目信任策略：`deny`、`approve` 或 Pi 默认的 `default` |

### 运行时兼容性

以下是切换到 `codex` 或 `piagent` 时需要检查的配置：

| 配置 | 外部运行时行为 |
| :- | :- |
| 自定义 `model`、`output_schema`、`planner`、智能体级 `code_executor` | 不支持，配置时会报错 |
| `include_contents="none"`、`enable_supervisor=True` | 不支持，配置时会报错 |
| `generate_content_config` | 只支持 `system_instruction`；其他显式字段会报错 |
| `model_name` 列表、`model_fallbacks` | 只使用主模型，忽略回退链 |
| `model_provider` | 不通过 LiteLLM 选择提供商，需使用兼容的模型端点 |
| `enable_responses`、`enable_responses_cache` | 不启用 ADK 的方舟 Responses 能力 |
| `model_extra_config` | Codex 会转发；PiAgent 不使用 |
| `knowledgebase`、`example_store` | 不自动提供相应内容；需要 ADK 或自行查询后放入指令 |
| 旧 `skills_mode`、`enable_skills_checklist` | 不提供旧工具集的完整行为；使用 ADK 或适配的原生技能入口 |
| `after_model_callback`、模型调用链路 | 按整轮结果处理，不能假定逐次模型调用都会产生回调和链路 |
| `RunConfig.max_llm_calls` | Codex 在调用前检查；PiAgent 在已完成调用后计数，可能超出一次才终止 |

## 智能体转移

`transfer_to_agent` 允许模型将当前任务转交给智能体树中的其他智能体，由目标继续执行并输出结果。ADK、Codex 与 PiAgent 均支持转交

| 目标 | 条件 |
| :- | :- |
| 子智能体 | 注册在 `sub_agents` 中，且不是 `single_turn` 或 `task` 模式 |
| 父智能体 | 存在父智能体且未设置 `disallow_transfer_to_parent` |
| 同级智能体 | 允许向同级转交，且目标可接收转交 |

```python transfer.py lines theme={null}
import asyncio
from veadk import Agent, Runner


writer = Agent(
    name="writer",
    description="Rewrite supplied text as a concise announcement.",
    instruction="Rewrite the user's text as an announcement without adding facts.",
)
agent = Agent(
    name="coordinator",
    instruction="Transfer announcement-writing tasks to writer.",
    sub_agents=[writer],
)

async def main():
    print(await Runner(agent=agent).run(
        messages='Write an announcement: the office closes at 17:00 on Friday.', session_id="runtime-demo"
    ))

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

运行 `python transfer.py` 后可得到公告文本。转交需要由 `Runner` 驱动，目标仍在同一次调用上下文中运行；它的 `output_key` 也会生效。需要固定顺序执行时，使用下节的顺序工作流

## 输出持久化

`output_key` 将最终回复保存到会话状态，ADK、Codex 和 PiAgent 均支持。它不是独立的磁盘持久化机制；跨进程保存取决于所选[会话存储](/productions/veadk/preview/zh/components/session/index)

```python pipeline.py lines theme={null}
import asyncio
from google.adk.agents import SequentialAgent
from veadk import Agent, Runner
from veadk.memory.short_term_memory import ShortTermMemory

planner = Agent(
    name="planner",
    instruction="Create an outline for the user's requested article.",
    output_key="plan",
)
writer = Agent(
    name="writer",
    instruction="Write the article using this outline: {plan}",
    output_key="draft",
)
pipeline = SequentialAgent(name="pipeline", sub_agents=[planner, writer])

async def main():
    runner = Runner(
        agent=pipeline, short_term_memory=ShortTermMemory(),
        app_name="writing_pipeline", user_id="demo-user",
    )
    print(await runner.run(
        messages="Write a short introduction to session state.", session_id="writing-demo"
    ))
    session = await runner.session_service.get_session(
        app_name="writing_pipeline", user_id="demo-user", session_id="writing-demo"
    )
    print(session.state["plan"])
    print(session.state["draft"])

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

运行 `python pipeline.py`。规划结果写入 `plan`，写作智能体通过 `{plan}` 读取，正文写入 `draft`。示例打印回复和两个状态值，以便检查传递结果

## 模型回调

回调可以使用同步或异步函数。ADK 按模型调用执行；外部运行时的前后回调围绕整轮执行，插件回调先于智能体回调。返回 `None` 表示继续默认处理

| 回调 | 行为 |
| :- | :- |
| `before_model_callback(callback_context, llm_request)` | 可修改输入内容、系统指令和可用工具；返回 `LlmResponse` 可直接提供结果并跳过执行 |
| `after_model_callback(callback_context, llm_response)` | 返回 `LlmResponse` 替换回复；外部运行时配置此回调后会缓冲并合并本轮最终文本 |
| `on_model_error_callback(callback_context, llm_request, error)` | 返回 `LlmResponse` 作为失败回复；返回 `None` 时异常继续传播 |

修改请求不意味着外部运行时支持全部 ADK 配置，仍需遵循兼容性表。例如不能通过回调假定获得 `output_schema` 支持

```python model_callback.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from google.adk.models.llm_response import LlmResponse
from google.genai import types

def fallback_on_error(callback_context, llm_request, error):
    return LlmResponse(content=types.Content(
        role="model",
        parts=[types.Part(text="The model is temporarily unavailable. Please retry later.")],
    ))

agent = Agent(name="assistant", on_model_error_callback=fallback_on_error)

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain session state.', session_id="runtime-demo"
    ))

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

运行 `python model_callback.py`：模型正常时输出回答，模型调用失败时返回预设文本。该回调不处理所有初始化、工具或业务错误

### 将 PDF 转为图片

`pdf_to_images_before_model_callback` 将输入中的 PDF 字节转为图片，适用于支持图片但不能直接读取 PDF 的模型。PDF 渲染依赖已包含在 VeADK 默认安装中

将可发送给模型服务的文件保存为当前目录的 `example.pdf`，选择支持图片输入的模型，然后运行以下脚本：

```python pdf_callback.py lines theme={null}
import asyncio
from pathlib import Path
from google.genai import types
from veadk import Agent, Runner
from veadk.utils.pdf_to_images import pdf_to_images_before_model_callback

async def main():
    agent = Agent(
        name="document_reader",
        before_model_callback=pdf_to_images_before_model_callback,
    )
    runner = Runner(agent=agent, app_name="pdf_demo", user_id="demo-user")
    await runner.session_service.create_session(
        app_name="pdf_demo", user_id="demo-user", session_id="pdf-session"
    )
    message = types.Content(role="user", parts=[
        types.Part(text="Summarize the document."),
        types.Part(inline_data=types.Blob(
            mime_type="application/pdf", data=Path("example.pdf").read_bytes()
        )),
    ])
    async for event in runner.run_async(
        user_id="demo-user", session_id="pdf-session", new_message=message
    ):
        if event.is_final_response() and event.content:
            print("".join(part.text or "" for part in event.content.parts or []))

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

默认最多处理每个 PDF 的前 10 页，渲染比例为 `2.0`。需要调整时，从同一模块导入 `make_pdf_to_images_callback`，通过 `max_pages` 和 `scale` 创建回调。更多页面或更高比例会增加图片数量、内存占用和模型输入用量

## 按工具选择执行位置

`RuntimeProvider` 决定单个工具调用如何执行，通常用于保留 ADK 流程而把部分工具交给自有服务。`DispatchRuntimeProvider` 调用你提供的派发函数；`LocalRuntimeProvider` 执行原有工具。需要自定义完整策略时，继承 `RuntimeProvider` 并实现 `execute(tool_call)`

### 使用示例

以下示例使用本地演示适配函数模拟服务结果，可验证派发后的库存为 `12`，原始工具返回的 `0` 不会再次执行。它不连接实际远端服务

```python dispatch.py lines theme={null}
import asyncio
from veadk import Agent, Runner
from veadk.runtime import DispatchRuntimeProvider, ToolCall

def lookup_stock(product: str) -> dict:
    """Look up stock for a product."""
    return {"product": product, "stock": 0, "source": "local"}

async def dispatch_task(tool_call: ToolCall):
    # Demonstration adapter: replace this body with your service client.
    return {
        "product": tool_call.arguments["product"],
        "stock": 12,
        "source": "dispatch-demo",
    }

agent = Agent(
    name="stock_assistant",
    instruction="Use lookup_stock to answer stock questions.",
    tools=[lookup_stock],
)
runtime_provider = DispatchRuntimeProvider(
    dispatch_task, dispatchable_tools={"lookup_stock"}
)

async def main():
    print(await Runner(agent=agent, plugins=[runtime_provider]).run(
        messages='How many notebooks are in stock?', session_id="runtime-demo"
    ))

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

接入真实服务时，将 `dispatch_task` 替换为已配置认证与超时的客户端调用，用 `tool_call.id` 关联请求。派发函数可以同步或异步，并应返回可作为工具结果的数据。服务接口由应用自行约定

也可将 `runtime_provider.before_tool_callback` 传给 `Agent.before_tool_callback`；不要同时注册两个入口，否则会重复拦截。MCP 工具仍使用自身连接执行，不受派发范围影响

### DispatchRuntimeProvider 参数

| 参数 | 类型 | 默认值 | 说明 |
| :- | :- | :- | :- |
| `dispatch_task` | `Callable[[ToolCall], Any]` | — | 必填。每次派发工具调用时执行的函数，接收 `ToolCall`，返回工具执行结果。可以是同步或异步函数 |
| `dispatchable_tools` | `Collection[str] \| None` | `("bash",)` | 需要派发到远端的非 MCP 工具名称集合。设为 `None` 时派发所有非 MCP 工具 |
| `local_runtime` | `RuntimeProvider \| None` | `None` | 用于执行未派发工具的本地 Provider。未设置时使用 `LocalRuntimeProvider` |
| `name` | `str` | `"veadk_dispatch_runtime_provider"` | 插件名称 |

### ToolCall 字段

| 字段/属性 | 类型 | 说明 |
| :- | :- | :- |
| `name` | `str` | 工具名称 |
| `arguments` | `dict[str, Any]` | 传递给工具的参数 |
| `tool` | `BaseTool` | ADK 工具对象，可用于在本地执行该工具 |
| `context` | `ToolContext` | ADK 工具上下文 |
| `id` | `str` | 属性。ADK 函数调用标识，不可用时为空字符串 |
| `session_id` | `str` | 属性。当前 VeADK 会话标识，不可用时为空字符串 |
| `RuntimeProvider(name=...)` 要求提供插件名。`LocalRuntimeProvider` 仅有可选 `name` 参数，默认 `"veadk_local_runtime_provider"` | | |

## 请求处理

`run_processor` 包裹 `Runner.run` 的整轮事件流，用于鉴权、计时、清理或事件转换。优先级依次为：本次 `Runner.run(run_processor=...)`、`Runner` 构造配置、根智能体配置，最后是默认的透传处理器。直接调用 `run_async` 不会自动使用这一包装

身份认证可使用 `veadk.integrations.ve_identity.AuthRequestProcessor`，并传给 `Agent(run_processor=...)`；所需身份配置见[入站认证](/productions/veadk/preview/zh/components/security/inbound)

### 自定义处理器

继承 `BaseRunProcessor` 并实现 `process_run(runner, message, **kwargs)`。下面的处理器转发事件，并在正常结束、失败或取消时记录耗时和关闭事件流

```python run_processor.py lines theme={null}
import asyncio
from veadk import Agent, Runner
import time
from contextlib import aclosing
from veadk.processors.base_run_processor import BaseRunProcessor

class TimingProcessor(BaseRunProcessor):
    def process_run(self, runner, message, **kwargs):
        def decorator(event_generator):
            async def wrapper():
                started = time.perf_counter()
                try:
                    async with aclosing(event_generator()) as events:
                        async for event in events:
                            yield event
                finally:
                    print(f"Run elapsed: {time.perf_counter() - started:.2f}s")
            return wrapper
        return decorator

agent = Agent(name="assistant", run_processor=TimingProcessor())

async def main():
    print(await Runner(agent=agent).run(
        messages='Explain what a runtime does.', session_id="runtime-demo"
    ))

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

运行 `python run_processor.py` 可看到回复及执行耗时。处理器若包含重试，应先确认任务是否已产生外部副作用，避免重复执行