制品¶

在 ADK 中，制品代表一种关键机制，用于管理与特定用户交互会话相关联或持久化存储在用户跨多个会话中的命名、版本化二进制数据。它们允许你的智能体和工具处理简单文本字符串之外的数据，实现涉及文件、图像、音频和其他二进制格式的更丰富交互。

什么是制品？¶

定义： 制品本质上是一段二进制数据（如文件内容），在特定范围内（会话或用户）由唯一的 filename 字符串标识。每次使用相同文件名保存制品时，都会创建一个新版本。

表示： 制品始终使用标准的 google.genai.types.Part 对象表示。核心数据通常存储在 Part 的 inline_data 属性中，该属性本身包含：

data：原始二进制内容，以 bytes 形式存在。
mime_type：指示数据类型的字符串（例如 'image/png'、'application/pdf'）。这对于后续正确解释数据至关重要。

# 示例：如何将制品表示为 types.Part
import google.genai.types as types

# 假设 'image_bytes' 包含 PNG 图像的二进制数据
image_bytes = b'\x89PNG\r\n\x1a\n...' # 实际图像字节的占位符

image_artifact = types.Part(
    inline_data=types.Blob(
        mime_type="image/png",
        data=image_bytes
    )
)

# 你也可以使用便捷构造函数：
# image_artifact_alt = types.Part.from_data(data=image_bytes, mime_type="image/png")

print(f"Artifact MIME Type: {image_artifact.inline_data.mime_type}")
print(f"Artifact Data (first 10 bytes): {image_artifact.inline_data.data[:10]}...")

持久化与管理： 制品不直接存储在智能体或会话状态中。它们的存储和检索由专用的制品服务（BaseArtifactService 的实现，定义在 google.adk.artifacts.base_artifact_service.py 中）管理。ADK 提供了像 InMemoryArtifactService（用于测试/临时存储，定义在 google.adk.artifacts.in_memory_artifact_service.py 中）和 GcsArtifactService（使用 Google Cloud Storage 进行持久存储，定义在 google.adk.artifacts.gcs_artifact_service.py 中）这样的实现。所选服务在保存数据时自动处理版本控制。

为什么使用制品？¶

虽然会话 state 适用于存储小块配置或会话上下文（如字符串、数字、布尔值或小型字典/列表），但制品设计用于涉及二进制或大数据的场景：

处理非文本数据： 轻松存储和检索图像、音频剪辑、视频片段、PDF、电子表格或与你的智能体功能相关的任何其他文件格式。
持久化大数据： 会话状态通常不适合存储大量数据。制品提供了一种专用机制，可以在不使会话状态混乱的情况下持久化更大的二进制数据。
用户文件管理： 提供用户上传文件（可以作为制品保存）以及检索或下载由智能体生成的文件（从制品加载）的功能。
共享输出： 使工具或智能体能够生成二进制输出（如 PDF 报告或生成的图像），可以通过 save_artifact 保存，并在应用程序的其他部分甚至在后续会话中（如果使用用户命名空间）访问。
缓存二进制数据： 将产生二进制数据的计算密集型操作结果（例如，渲染复杂图表图像）存储为制品，以避免在后续请求中重新生成它们。

本质上，每当你的智能体需要处理需要持久化、版本控制或共享的类似文件的二进制数据时，由 ArtifactService 管理的制品是 ADK 中的适当机制。

常见用例¶

制品提供了在 ADK 应用程序中处理二进制数据的灵活方式。

以下是它们证明有价值的一些典型场景：

生成的报告/文件：
- 工具或智能体生成报告（例如，PDF 分析、CSV 数据导出、图像图表）。
- 工具使用 tool_context.save_artifact("monthly_report_oct_2024.pdf", report_part) 存储生成的文件。
- 用户稍后可以要求智能体检索此报告，这可能涉及另一个工具使用 tool_context.load_artifact("monthly_report_oct_2024.pdf") 或使用 tool_context.list_artifacts() 列出可用报告。
处理用户上传：
- 用户通过前端界面上传文件（例如，用于分析的图像，用于总结的文档）。
- 应用程序后端接收文件，从其字节和 MIME 类型创建 types.Part，并使用 runner.session_service（或直接运行智能体之外的类似机制）或运行中的专用工具/回调通过 context.save_artifact 存储它，可能使用 user: 命名空间以便在会话之间持久化（例如，user:uploaded_image.jpg）。
- 然后可以提示智能体处理此上传的文件，使用 context.load_artifact("user:uploaded_image.jpg") 检索它。
存储中间二进制结果：
- 智能体执行复杂的多步骤流程，其中一个步骤生成中间二进制数据（例如，音频合成，模拟结果）。
- 使用 context.save_artifact 和临时或描述性名称（例如，"temp_audio_step1.wav"）保存此数据。
- 流程中的后续智能体或工具（可能在 SequentialAgent 中或稍后触发）可以使用 context.load_artifact 加载此中间制品以继续处理。
持久化用户数据：
- 存储不是简单键值状态的用户特定配置或数据。
- 智能体使用 context.save_artifact("user:profile_settings.json", settings_part) 或 context.save_artifact("user:avatar.png", avatar_part) 保存用户偏好或个人资料图片。
- 可以在该用户的任何未来会话中加载这些制品，以个性化其体验。
缓存生成的二进制内容：
- 智能体经常基于某些输入生成相同的二进制输出（例如，公司标志图像，标准音频问候）。
- 在生成之前，before_tool_callback 或 before_agent_callback 使用 context.load_artifact 检查制品是否存在。
- 如果存在，则使用缓存的制品，跳过生成步骤。
- 如果不存在，则生成内容，并在 after_tool_callback 或 after_agent_callback 中调用 context.save_artifact 将其缓存以供下次使用。

核心概念¶

理解制品涉及掌握几个关键组件：管理它们的服务、用于保存它们的数据结构，以及它们如何被识别和版本控制。

制品服务 (`BaseArtifactService`)¶

角色： 负责制品实际存储和检索逻辑的中央组件。它定义了制品如何以及在哪里持久化。
接口： 由抽象基类 BaseArtifactService（google.adk.artifacts.base_artifact_service.py）定义。任何具体实现都必须提供以下方法：
- save_artifact(...) -> int：存储制品数据并返回其分配的版本号。
- load_artifact(...) -> Optional[types.Part]：检索制品的特定版本（或最新版本）。
- list_artifact_keys(...) -> list[str]：列出给定范围内制品的唯一文件名。
- delete_artifact(...) -> None：删除制品（并可能删除所有版本，取决于实现）。
- list_versions(...) -> list[int]：列出特定制品文件名的所有可用版本号。
配置： 在初始化 Runner 时提供制品服务（例如 InMemoryArtifactService、GcsArtifactService）的实例。然后 Runner 通过 InvocationContext 将此服务提供给智能体和工具。

from google.adk.runners import Runner
from google.adk.artifacts import InMemoryArtifactService # 或 GcsArtifactService
from google.adk.agents import LlmAgent # 任何智能体
from google.adk.sessions import InMemorySessionService

# 示例：使用制品服务配置 Runner
my_agent = LlmAgent(name="artifact_user_agent", model="gemini-2.0-flash")
artifact_service = InMemoryArtifactService() # 选择一个实现
session_service = InMemorySessionService()

runner = Runner(
    agent=my_agent,
    app_name="my_artifact_app",
    session_service=session_service,
    artifact_service=artifact_service # 在此提供服务实例
)
# 现在，由此 runner 管理的运行中的上下文可以使用制品方法

制品数据 (`google.genai.types.Part`)¶

标准表示： 制品内容普遍使用 google.genai.types.Part 对象表示，这与 LLM 消息部分使用的结构相同。
关键属性 (inline_data)： 对于制品，最相关的属性是 inline_data，它是一个 google.genai.types.Blob 对象，包含：
- data (bytes)：制品的原始二进制内容。
- mime_type (str)：标准 MIME 类型字符串（例如 'application/pdf'、'image/png'、'audio/mpeg'），描述二进制数据的性质。这对于加载制品时的正确解释至关重要。
创建： 你通常使用其 from_data 类方法或通过直接构造带有 Blob 的对象来为制品创建 Part。

import google.genai.types as types

# 示例：从原始字节创建制品 Part
pdf_bytes = b'%PDF-1.4...' # 你的原始 PDF 数据
pdf_mime_type = "application/pdf"

# 使用构造函数
pdf_artifact = types.Part(
    inline_data=types.Blob(data=pdf_bytes, mime_type=pdf_mime_type)
)

# 使用便捷类方法（等效）
pdf_artifact_alt = types.Part.from_data(data=pdf_bytes, mime_type=pdf_mime_type)

print(f"Created artifact with MIME type: {pdf_artifact.inline_data.mime_type}")

文件名 (`str`)¶

标识符： 用于在特定命名空间内命名和检索制品的简单字符串（见下文）。
唯一性： 文件名在其范围内（会话或用户命名空间）必须唯一。
最佳实践： 使用描述性名称，可能包括文件扩展名（例如 "monthly_report.pdf"、"user_avatar.jpg"），尽管扩展名本身不决定行为 – mime_type 才是决定因素。

版本控制 (`int`)¶

自动版本控制： 制品服务自动处理版本控制。当你调用 save_artifact 时，服务会为该特定文件名和范围确定下一个可用的版本号（通常从 0 开始并递增）。
由 save_artifact 返回： save_artifact 方法返回分配给新保存制品的整数版本号。
检索：
load_artifact(..., version=None) （默认）：检索制品的最新可用版本。
load_artifact(..., version=N)：检索特定版本 N。
列出版本： 可以使用 list_versions 方法（在服务上，而不是上下文）查找制品的所有现有版本号。

命名空间（会话与用户）¶

概念： 制品可以限定在特定会话范围内，也可以更广泛地限定在应用程序内用户的所有会话中。这种限定由 filename 格式决定，并由 ArtifactService 内部处理。
默认（会话范围）： 如果你使用像 "report.pdf" 这样的普通文件名，制品将与特定的 app_name、user_id 和 session_id 关联。它只能在该确切的会话上下文中访问。
内部路径（示例）：app_name/user_id/session_id/report.pdf/<version> （在 GcsArtifactService._get_blob_name 和 InMemoryArtifactService._artifact_path 中可见）
用户范围（"user:" 前缀）： 如果你用 "user:" 前缀文件名，比如 "user:profile.png"，制品只与 app_name 和 user_id 关联。它可以从该用户在应用程序中的任何会话中访问或更新。
内部路径（示例）：app_name/user_id/user/user:profile.png/<version> （user: 前缀通常保留在最终路径段中以增加清晰度，如服务实现中所示）。
用例： 适用于属于用户本身的数据，与特定对话无关，如个人资料图片、用户偏好文件或长期报告。

# 说明命名空间差异的示例（概念性）

# 会话特定的制品文件名
session_report_filename = "summary.txt"

# 用户特定的制品文件名
user_config_filename = "user:settings.json"

# 保存 'summary.txt' 时，它与当前会话 ID 绑定。
# 保存 'user:settings.json' 时，它只与用户 ID 绑定。

这些核心概念共同提供了一个灵活的系统，用于在 ADK 框架内管理二进制数据。

与制品交互（通过上下文对象）¶

在智能体逻辑中（特别是在回调或工具中）与制品交互的主要方式是通过 CallbackContext 和 ToolContext 对象提供的方法。这些方法抽象了由 ArtifactService 管理的底层存储细节。

先决条件：配置 `ArtifactService`¶

在通过上下文对象使用任何制品方法之前，你必须在初始化 Runner 时提供 BaseArtifactService 实现（如 InMemoryArtifactService 或 GcsArtifactService）的实例。

from google.adk.runners import Runner
from google.adk.artifacts import InMemoryArtifactService # 或 GcsArtifactService
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService

# 你的智能体定义
agent = LlmAgent(name="my_agent", model="gemini-2.0-flash")

# 实例化所需的制品服务
artifact_service = InMemoryArtifactService()

# 将其提供给 Runner
runner = Runner(
    agent=agent,
    app_name="artifact_app",
    session_service=InMemorySessionService(),
    artifact_service=artifact_service # 服务必须在此处提供
)

如果在 InvocationContext 中没有配置 artifact_service（如果没有传递给 Runner，就会发生这种情况），在上下文对象上调用 save_artifact、load_artifact 或 list_artifacts 将引发 ValueError。

访问方法¶

制品交互方法直接在 CallbackContext（传递给智能体和模型回调）和 ToolContext（传递给工具回调）实例上可用。请记住，ToolContext 继承自 CallbackContext。

保存制品¶

方法：

context.save_artifact(filename: str, artifact: types.Part) -> int

可用上下文： CallbackContext、ToolContext。
动作：
1. 接受 filename 字符串（可能包括 "user:" 前缀用于用户范围）和包含制品数据的 types.Part 对象（通常在 artifact.inline_data 中）。
2. 将此信息传递给底层的 artifact_service.save_artifact。
3. 服务存储数据，为该文件名和范围分配下一个可用版本号。
4. 关键的是，上下文通过向当前事件的 actions.artifact_delta 字典（在 google.adk.events.event_actions.py 中定义）添加条目自动记录此操作。这个 delta 将 filename 映射到新分配的 version。
返回： 分配给保存制品的整数 version 号。
代码示例（在假设的工具或回调中）：

import google.genai.types as types
from google.adk.agents.callback_context import CallbackContext # Or ToolContext

async def save_generated_report(context: CallbackContext, report_bytes: bytes):
    """Saves generated PDF report bytes as an artifact."""
    report_artifact = types.Part.from_data(
        data=report_bytes,
        mime_type="application/pdf"
    )
    filename = "generated_report.pdf"

    try:
        version = context.save_artifact(filename=filename, artifact=report_artifact)
        print(f"Successfully saved artifact '{filename}' as version {version}.")
        # The event generated after this callback will contain:
        # event.actions.artifact_delta == {"generated_report.pdf": version}
    except ValueError as e:
        print(f"Error saving artifact: {e}. Is ArtifactService configured?")
    except Exception as e:
        # Handle potential storage errors (e.g., GCS permissions)
        print(f"An unexpected error occurred during artifact save: {e}")

# --- Example Usage Concept ---
# report_data = b'...' # Assume this holds the PDF bytes
# await save_generated_report(callback_context, report_data)

加载制品¶

方法：

context.load_artifact(filename: str, version: Optional[int] = None) -> Optional[types.Part]

可用上下文： CallbackContext、ToolContext。
动作：
1. 接受一个 filename 字符串（可能包括 "user:"）。
2. 可选地接受一个整数 version。如果 version 是 None（默认值），它将从服务请求最新版本。如果提供了特定的整数，它将请求该确切版本。
3. 调用底层的 artifact_service.load_artifact。
4. 服务尝试检索指定的制品。
返回： 如果找到，则返回包含制品数据的 types.Part 对象；如果制品（或指定版本）不存在，则返回 None。

代码示例（在假设的工具或回调中）：

import google.genai.types as types
from google.adk.agents.callback_context import CallbackContext # 或 ToolContext

async def process_latest_report(context: CallbackContext):
    """加载最新的报告制品并处理其数据。"""
    filename = "generated_report.pdf"
    try:
        # 加载最新版本
        report_artifact = context.load_artifact(filename=filename)

        if report_artifact and report_artifact.inline_data:
            print(f"成功加载最新制品 '{filename}'。")
            print(f"MIME 类型：{report_artifact.inline_data.mime_type}")
            # 处理 report_artifact.inline_data.data（字节）
            pdf_bytes = report_artifact.inline_data.data
            print(f"报告大小：{len(pdf_bytes)} 字节。")
            # ... 进一步处理 ...
        else:
            print(f"未找到制品 '{filename}'。")

        # 示例：加载特定版本（如果版本 0 存在）
        # specific_version_artifact = context.load_artifact(filename=filename, version=0)
        # if specific_version_artifact:
        #     print(f"已加载 '{filename}' 的版本 0。")

    except ValueError as e:
        print(f"加载制品时错误：{e}。ArtifactService 是否已配置？")
    except Exception as e:
        # 处理潜在的存储错误
        print(f"制品加载过程中发生意外错误：{e}")

# --- 使用概念示例 ---
# await process_latest_report(callback_context)

列出制品文件名（仅限工具上下文）¶

方法：

tool_context.list_artifacts() -> list[str]

可用上下文： 仅限 ToolContext。这个方法在基础 CallbackContext 上不可用。
动作： 调用底层的 artifact_service.list_artifact_keys 获取当前范围内可访问的所有唯一制品文件名列表（包括特定于会话的文件和带有 "user:" 前缀的用户范围文件）。
返回： 一个已排序的文件名 str 的 list。
代码示例（在工具函数中）：

from google.adk.tools.tool_context import ToolContext

def list_user_files(tool_context: ToolContext) -> str:
    """列出用户可用制品的工具。"""
    try:
        available_files = tool_context.list_artifacts()
        if not available_files:
            return "你没有已保存的制品。"
        else:
            # 为用户/LLM 格式化列表
            file_list_str = "\n".join([f"- {fname}" for fname in available_files])
            return f"以下是你可用的制品：\n{file_list_str}"
    except ValueError as e:
        print(f"列出制品时错误：{e}。ArtifactService 是否已配置？")
        return "错误：无法列出制品。"
    except Exception as e:
        print(f"制品列表过程中发生意外错误：{e}")
        return "错误：列出制品时发生意外错误。"

# 此函数通常会被包装在 FunctionTool 中
# from google.adk.tools import FunctionTool
# list_files_tool = FunctionTool(func=list_user_files)

这些上下文方法提供了一种便捷且一致的方式在 ADK 中管理二进制数据持久化，无论选择哪种后端存储实现（InMemoryArtifactService、GcsArtifactService 等）。

可用实现¶

ADK 提供了 BaseArtifactService 接口的具体实现，提供适用于不同开发阶段和部署需求的各种存储后端。这些实现根据 app_name、user_id、session_id 和 filename（包括 user: 命名空间前缀）处理制品数据的存储、版本控制和检索细节。

InMemoryArtifactService¶

源文件： google.adk.artifacts.in_memory_artifact_service.py
存储机制： 使用应用程序内存中保存的 Python 字典（self.artifacts) 存储制品。字典键表示制品路径（包含应用、用户、会话/用户范围和文件名），值是 types.Part 的列表，其中列表中的每个元素对应一个版本（索引 0 是版本 0，索引 1 是版本 1，等等）。
主要特性：
- 简单性： 除了核心 ADK 库之外，不需要外部设置或依赖项。
- 速度： 操作通常非常快，因为它们涉及内存字典查找和列表操作。
- 临时性： 当运行应用程序的 Python 进程终止时，所有存储的制品都将丢失。数据在应用程序重启之间不会持久保存。
用例：
- 适用于不需要持久性的本地开发和测试。
- 适合仅在应用程序单次运行中临时使用制品数据的短期演示或场景。
实例化：

from google.adk.artifacts import InMemoryArtifactService

# 简单实例化类
in_memory_service = InMemoryArtifactService()

# 然后将其传递给 Runner
# runner = Runner(..., artifact_service=in_memory_service)

GcsArtifactService¶

源文件： google.adk.artifacts.gcs_artifact_service.py
存储机制： 利用 Google Cloud Storage (GCS) 实现制品持久存储。制品的每个版本都作为指定 GCS 存储桶中的单独对象存储。
对象命名约定： 它使用分层路径结构构建 GCS 对象名（blob 名称），通常为：
- 会话范围：{app_name}/{user_id}/{session_id}/{filename}/{version}
- 用户范围：{app_name}/{user_id}/user/{filename}/{version}（注意：服务处理文件名中的 user: 前缀以确定路径结构）。
主要特性：
- 持久性： 存储在 GCS 中的制品在应用程序重启和部署之间持久保存。
- 可扩展性： 利用 Google Cloud Storage 的可扩展性和耐用性。
- 版本控制： 显式地将每个版本存储为不同的 GCS 对象。
- 需要配置： 需要配置目标 GCS bucket_name。
- 需要权限： 应用程序环境需要适当的凭证和 IAM 权限才能读取和写入指定的 GCS 存储桶。
用例：
- 需要持久制品存储的生产环境。
- 需要在不同应用程序实例或服务之间共享制品的场景（通过访问相同的 GCS 存储桶）。
- 需要长期存储和检索用户或会话数据的应用程序。
实例化：

from google.adk.artifacts import GcsArtifactService

# 指定 GCS 存储桶名称
gcs_bucket_name = "your-gcs-bucket-for-adk-artifacts" # 替换为你的存储桶名称

try:
    gcs_service = GcsArtifactService(bucket_name=gcs_bucket_name)
    print(f"GcsArtifactService 初始化，存储桶：{gcs_bucket_name}")
    # 确保你的环境有权限访问此存储桶。
    # 例如，通过应用程序默认凭证 (ADC)

    # 然后将其传递给 Runner
    # runner = Runner(..., artifact_service=gcs_service)

except Exception as e:
    # 捕获 GCS 客户端初始化期间的潜在错误（例如，身份验证问题）
    print(f"初始化 GcsArtifactService 时出错：{e}")
    # 适当处理错误 - 可能回退到 InMemory 或抛出异常

选择适当的 ArtifactService 实现取决于你的应用程序对数据持久性、可扩展性和操作环境的要求。

最佳实践¶

要有效且可维护地使用制品：

选择合适的服务： 对于快速原型设计、测试和不需要持久性的场景，使用 InMemoryArtifactService。对于需要数据持久性和可扩展性的生产环境，使用 GcsArtifactService（或为其他后端实现自己的 BaseArtifactService）。
有意义的文件名： 使用清晰、描述性的文件名。包括相关扩展名（.pdf、.png、.wav）有助于人类理解内容，尽管 mime_type 决定了程序化处理。为临时与持久制品名称建立约定。
指定正确的 MIME 类型： 为 save_artifact 创建 types.Part 时，始终提供准确的 mime_type。这对于稍后 load_artifact 的应用程序或工具正确解释 bytes 数据至关重要。尽可能使用标准 IANA MIME 类型。
理解版本控制： 记住，没有特定 version 参数的 load_artifact() 会检索最新版本。如果你的逻辑依赖于制品的特定历史版本，确保在加载时提供整数版本号。
有意识地使用命名空间（user:）： 只有当数据真正属于用户并且应该在其所有会话中都可访问时，才为文件名使用 "user:" 前缀。对于特定于单个对话或会话的数据，使用不带前缀的常规文件名。
错误处理：
- 在调用上下文方法（save_artifact、load_artifact、list_artifacts) 之前，始终检查 artifact_service 是否实际配置 - 如果服务为 None，它们将引发 ValueError。在 try...except ValueError 中包装调用。
- 检查 load_artifact 的返回值，因为如果制品或版本不存在，它将是 None。不要假设它总是返回 Part。
- 准备好处理底层存储服务的异常，特别是使用 GcsArtifactService 时（例如，权限问题的 google.api_core.exceptions.Forbidden，如果存储桶不存在的 NotFound，网络错误）。
大小考虑： 制品适用于典型的文件大小，但要注意极大文件可能的成本和性能影响，特别是使用云存储时。如果存储许多大型制品，InMemoryArtifactService 可能会消耗大量内存。评估非常大的数据是否可能通过直接 GCS 链接或其他专用存储解决方案更好地处理，而不是在内存中传递整个字节数组。
清理策略： 对于像 GcsArtifactService 这样的持久存储，制品会一直保留，直到显式删除。如果制品代表临时数据或有限的生命周期，实现清理策略。这可能涉及：
- 在存储桶上使用 GCS 生命周期策略。
- 构建利用 artifact_service.delete_artifact 方法的特定工具或管理功能（注意：为安全起见，删除不通过上下文对象公开）。
- 谨慎管理文件名，以允许基于模式的删除（如果需要）。

制品¶