Python 运行时参考文档

您可以在 Python 工具和回调中使用 Python 代码。本参考文档介绍了 Python 运行时、导入选项和类、全局变量以及可在代码中使用的函数。

Python 环境

Python 工具和回调在安全的沙盒环境中运行。 此环境运行的是 Python 3.12。

导入

您只能导入以下模块:

类

AsyncTools

用于异步调用工具的别名类。 如需了解详情,请参阅 async_tools 全局变量。

Blob

内嵌字节数据。

属性:

属性 说明
display_name: Optional[str] Blob 的显示名称。用于提供标签或文件名以区分 blob。
data: Optional[bytes] Base64 编码的字节。
raw_data: Optional[bytes] 原始解码字节。
mime_type: Optional[str] 来源数据的 IANA 标准 MIME 类型。

方法:

方法 说明
transcript() -> Optional[str] 返回 Blob 数据的缓存转写(如果有)。这仅适用于音频 blob。
from_json(data: str) -> Blob 一个类方法,用于从 JSON 字符串创建 Blob 实例,并将 mime_type 设置为“application/json”。

样本:

# Create a blob from raw bytes
blob = Blob(mime_type='text/plain')
blob.raw_data = b'hello world'

# Create a blob from a JSON string
Blob.from_json(data='{"key": "value"}')

CallbackContext

回调期间可用的会话信息。

属性:

属性 说明
user_content: Optional[Content] 用户最近一次输入的内容。
invocation_id: str 特定回调调用的唯一标识符,对调试很有用。
agent_name: str 与当前回调关联的代理的显示名称。
session_id: str 当前正在进行的会话的唯一会话标识符。
variables: dict[str, Any] 包含在设计时定义或在运行时注入的变量键值对的字典。这是回调执行时变量的当前状态。
state: dict[str, Any] 与 variables 属性相同。
events: list[Event] 会话事件。

方法:

方法 说明
get_variable(key: str, default: Any) -> Any 从状态中获取变量。如果变量不存在,则返回默认值。
set_variable(key: str, value: Any) -> None 在状态中设置变量。
remove_variable(key: str) -> None 从状态中移除变量。
get_last_user_input() -> list[Part] 获取最后一批用户事件中所有部分的列表。
get_last_agent_output() -> list[Part] 获取代理事件最后一个分块中所有部分的列表。
parts() -> list[Part] 会话记录中记录的所有部分的列表。

Content

来自用户或代理的消息内容。

属性:

属性 说明
parts: Optional[list[Part]] 构成单条消息的部分的列表。每个部分可能具有不同的 IANA MIME 类型。
role: Optional[str] 内容作者的角色。'user' 或 'agent'。

方法:

方法 说明
is_user() -> bool 如果角色为 'user',则返回 True。
is_model() -> bool 如果角色为 'model',则返回 True。

Event

会话中事件的表示形式。

属性:

属性 说明
id: str 事件标识符。
invocation_id: str 事件调用标识符。
author: str “user”或代理的名称,表示谁参加了会话。
timestamp: int 事件时间戳。
content: Content 与此事件相关联的内容。
actions: EventActions 智能体执行的操作。
long_running_tool_ids: set[str] 长时间运行的函数调用的标识符集。
partial: bool True,用于处理 LLM 流式回答中的不完整分块。
turn_complete: bool 如果当前回合已完成,则为 True。
error_code: str 错误代码。
error_message: str 错误消息。
interrupted: bool 如果回合被中断,则为 True。
branch: str 事件的分支。

格式类似于 agent_1.agent_2.agent_3,其中 agent_1 是 agent_2 的父级,agent_2 是 agent_3 的父级。

当多个分代理不应看到其对等代理的对话记录时,可以使用分支。
grounding_metadata: Any 事件的接地元数据。

方法:

方法 说明
is_user() -> bool 如果事件的作者是“user”,则为 True。
is_agent(agent_name: Optional[str] = None) -> bool 如果事件的作者是客服人员,则为 True。如果提供了 agent_name,则会检查作者是否与该特定代理匹配。
has_error() -> bool True(如果事件有关联的错误代码)。
parts() -> list[Part] 一种便捷方法,用于从事件的 content 中获取 Part 对象列表。如果没有任何内容或部分,则返回空列表。

EventActions

事件发生的操作。

属性:

属性 说明
skip_summarization: bool 如果为 True,则不会调用模型来总结函数响应。仅用于 function_response 事件。
state_delta: dict[str,Any] 此事件导致的变量更改。
artifact_delta: dict[str,Any] 此事件对制品所做的更改。键是文件名,值是版本。
transfer_to_agent: str 如果设置了此属性,事件会转移到指定的代理。
escalate: bool 客服人员正在将问题上报给更高级别的客服人员。
requested_auth_configs: dict[str,dict[str,Any]] 工具回答请求的身份验证配置。

此字段仅由指示工具请求身份验证凭据的工具回答事件设置。

键:函数调用标识符。因为一个函数响应事件可能包含多个函数响应,这些响应对应于多个函数调用。每个函数调用都可能请求不同的身份验证配置。此标识符用于标识函数调用。

值:请求的身份验证配置。
end_invocation: bool 智能体循环中断。

ExternalResponse

表示来自 Python 环境外部的响应,例如工具调用或 HTTP 请求。

属性:

属性 说明
text: str 响应正文(字符串)。
status_code: int HTTP 状态代码。
reason: str 抛出错误的原因。如果未发生错误,则为空。
ok: bool 如果 status_code 小于 400,则为 True,否则为 False。

方法:

方法 说明
json() -> Any 解析文本属性 JSON 并返回结果。如果解析失败,则会抛出错误。
raise_for_status() 如果响应不正常 (ok == False),则会引发 StatusError。

FunctionCall

表示函数调用。

属性:

属性 说明
id: Optional[str] 函数调用的唯一标识符。
args: Optional[dict[str,Any]] 以 JSON 对象格式表示的函数参数和值。
name: Optional[str] 函数名称。

FunctionDeclaration

从模型返回的预测 FunctionCall,其中包含表示 FunctionDeclaration name 属性的字符串,以及参数及其值。

属性:

属性 说明
name: Optional[str] 函数名称。

FunctionResponse

FunctionCall 的结果,其中包含表示 FunctionDeclaration name 属性的字符串和包含函数调用的任何输出的结构化 JSON 对象。它用作模型的上下文。

属性:

属性 说明
id: Optional[str] 相应函数调用的标识符。
name: Optional[str] 函数名称。
response: Optional[dict[str,Any]] 以 JSON 对象格式表示的函数响应。使用“output”键指定函数输出,使用“error”键指定错误详情(如有)。如果未指定“output”和“error”键,则整个“response”都会被视为函数输出。

GenerateContentConfig

可选模型配置参数。

属性:

属性 说明
system_instruction: Optional[Content] 有关引导模型获得更好性能的说明。例如,“回答尽可能简明扼要”或“请勿在回答中使用技术词汇”。
tools: Optional[list[ToolDeclaration]] 模型可以执行的可用工具列表。
excluded_tools: Optional[list[str]] 模型将忽略的工具名称列表。这会替换 tools。

方法:

方法 说明
hide_tool(tool_name: str) 将 tool_name 添加到 excluded_tools 列表中。

HttpMethod

表示 HTTP 方法的字符串枚举。 可能的值包括:

  • GET
  • POST
  • PUT
  • DELETE
  • PATCH
  • HEAD
  • OPTIONS

LlmRequest

用于表示对 LLM 的请求的数据模型。

属性:

属性 说明
model: Optional[str] 模型名称
contents: List[Content] 发送到模型的内容列表。
config: Optional[GeneralContentConfig] 模型配置参数。

LlmResponse

用于表示 LLM 回答的数据模型。

属性:

属性 说明
content: Content 模型响应的第一个 Content。
partial: Optional[bool] 指明内容是否表示不完整的模型回答。智能体将在发出只答完一部分的回答后继续处理。

方法:

方法 说明
from_parts(parts: list[Part]) -> LlmResponse 从模型返回 LlmResponse 的类方法。

样本:

response = LlmResponse.from_parts(
  parts=[
    Part.from_text(text="hello world")
  ]
)

Part

包含媒体内容的数据类型。 Part 中应仅设置一个字段,用于表示所传达内容的具体类型。在同一 Part 实例中使用多个字段会被视为无效。

属性:

属性 说明
function_call: Optional[FunctionCall] 从模型返回的预测 FunctionCall,其中包含表示函数名称的字符串和包含参数及其值的结构化 JSON 对象。
function_response: Optional[FunctionResponse] FunctionCall 的结果输出,其中包含表示函数名称的字符串和包含函数调用的任何输出的结构化 JSON 对象。它用作模型的上下文。
text: Optional[str] 消息文字。
inline_data: Optional[Blob] 内嵌字节数据。

方法:

方法 说明
text_or_transcript() -> Optional[str] 返回文本(如果有),否则返回内嵌数据的转写内容。
has_function_call(name) -> bool 如果该部分包含特定函数调用,则返回 True。
has_function_response(name) -> bool 如果该部分包含特定函数响应,则返回 True。
from_text(text: str) -> Part 用于创建文本 Part 的类方法。
from_function_call(name: str, args: dict[str, Any]) -> Part 用于创建函数调用 Part 的类方法。
from_function_response(name: str, response: dict[str, Any]) -> Part 用于创建函数响应 Part 的类方法。
from_inline_data(data: bytes, mime_type: str) -> Part 用于创建内嵌数据 Part 的类方法。
from_json(data: str) -> Part 用于创建 JSON 内嵌数据 Part 的类方法。
from_agent_transfer(agent: str) -> Part 用于创建 Part 以转移到其他代理的类方法。
from_end_session(*, reason: str, escalated: bool = False) -> Part 用于创建 Part 以终止会话的类方法。
from_customized_response(*, content: str, disable_barge_in: bool = False, enable_dtmf: bool = False, dtmf_finish_digit = str: '#', dtmf_endpointing_timeout: int = 3) -> Part 类方法,用于创建具有自定义行为(例如,停用打断、启用 DTMF 输入等)的回答 Part。

样本:

text_part = ces_public.Part.from_text(text="Hello from the user!")

tool_part = ces_public.Part.from_function_call(
  name="get_weather",
  args={"location": "Mountain View"}
)

Requests

用于发出 HTTP 请求的别名类。 如需了解详情,请参阅 ces_requests 全局变量。

方法:

  • get(url, params=None, **kwargs)
  • post(url, data=None, json=None, **kwargs)
  • put(url, data=None, json=None, **kwargs)
  • delete(url, **kwargs)
  • patch(url, data=None, json=None, **kwargs)
  • head(url, **kwargs)
  • options(url, **kwargs)

StatusError

用于引发状态代码的错误。

属性:

属性 说明
status_code: int 与此错误关联的 HTTP 状态代码。
reason: str 抛出错误的原因。

Tool

表示具有名称和说明的工具。

属性:

属性 说明
name: str 工具的名称。
description: str 对工具功能的说明。

ToolContext

派生自 CallbackContext。 执行工具时可用的会话信息。

属性:

属性 说明
function_call_id: str 当前正在执行的工具调用的函数调用标识符。

Tools

用于同步调用工具的别名类。 如需了解详情,请参阅工具全局变量。

ToolDeclaration

可向模型显示的工具架构。

属性:

属性 说明
function_declarations: Optional[list[FunctionDeclaration]] 工具支持的函数声明列表。

函数

get_variable

此函数使用提供的键从会话状态中检索值。 它可用作 context.state.get(key) 或 context.variables.get(key) 的快捷方式。

示例代码:

def get_a_value() -> int:
  # Retrieve the value of 'my_key' from the state
  my_value = get_variable('my_key')
  return my_value + 5

remove_variable

此函数会从会话状态中移除键值对。 这是 del context.state[key] 的快捷方式。

示例代码:

def remove_a_value() -> None:
  # Delete 'my_key' from the state
  remove_variable('my_key')

set_variable

此函数用于在会话状态中为给定键设置值。 如果该键已存在,系统会更新其值。 这是context.state[key] = value的快捷方式。

示例代码:

def set_a_value() -> None:
  # Set the value of 'my_key' to 10
  set_variable('my_key', 10)

全局变量

_session

您可以访问对话的当前渠道和模态,其方式与在模块化组件中的操作类似。使用名为 _session 的预定义变量访问这些值。

Python 工具示例(使用全局 context 变量):

def check_order_status(order_id: str) -> str:
  # Resolve modality and channel from the session state
  session_vars = context.variables.get('_session', {})
  modality = session_vars.get('modality', 'TEXT')
  channel = session_vars.get('channel', 'API')

  status = get_raw_order_status(order_id)  # Internal API lookup

  if modality == 'VOICE':
    # Keep response concise and remove markdown/links for natural TTS playback
    return "Your order has shipped and is estimated to arrive tomorrow."
  else:
    # Rich text with markdown links for Chat/Web interfaces
    return f"Your order shipped [Track it](https://shipper.com/{order_id})."

回调示例(使用 CallbackContext):

def before_agent_callback(callback_context: CallbackContext) -> None:
  session_vars = callback_context.get_variable('_session', {})
  modality = session_vars.get('modality', 'TEXT')
  channel = session_vars.get('channel', 'API')
  callback_context.set_variable('active_modality', modality)

async_tools

AsyncTools 的实例,可让您对工具进行异步调用。

样本:

response_future = async_tools.<TOOL_DISPLAY_NAME>(<ARGS_AS_DICT>)
# ... misc work
response = response_future() # poll for response

# Check if the tool call was successful
try:
  response.raise_for_status()
except StatusError:
  print(f"Request failed with status {response.status_code}")

# Convert the response to json
data = response.json()

ces_requests

Requests 的实例。 Requests 允许您使用与热门 Python 请求模块类似的语法进行 HTTP 调用。

样本:

# Make a GET request
response = ces_requests.get('https://api.example.com/data')

# Check if the request was successful
try:
  response.raise_for_status()
except StatusError:
  print(f"Request failed with status {response.status_code}")

# Convert the response to json
data = response.json()

tools

Tools 的实例,可让您对工具进行同步调用。

样本:

response = tools.<TOOL_DISPLAY_NAME>(<ARGS_AS_DICT>)

# Check if the tool call was successful
try:
  response.raise_for_status()
except StatusError:
  print(f"Request failed with status {response.status_code}")

# Convert the response to json
data = response.json()

context

在代码中,您可以访问 ADK 上下文,该上下文可用于读取和写入多种类型的 ADK 情境数据。您可以在代码中使用名为 context 的全局变量。

context.state 和 context.variables 对象可以互换。 为了与 ADK 代码兼容,系统支持 state 对象,但新代码应使用 variables 对象。