制品

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

什么是Artifacts？

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

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

data: 原始二进制内容，格式为bytes。
mime_type: 表示数据类型的字符串（例如'image/png'、'application/pdf'）。这对于后续正确解析数据至关重要。

# Example of how an artifact might be represented as a types.Part
import google.genai.types as types

# Assume 'image_bytes' contains the binary data of a PNG image
image_bytes = b'\x89PNG\r\n\x1a\n...' # Placeholder for actual image bytes

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

# You can also use the convenience constructor:
# 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中）等实现。所选服务会在保存数据时自动处理版本控制。

为什么使用Artifacts？

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

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

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

常见使用场景

Artifacts为您的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:uploaded_image.jpg），则可能使用user:命名空间。
- 然后可以提示智能体处理这个上传的文件，使用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 # Or GcsArtifactService
from google.adk.agents import LlmAgent # Any agent
from google.adk.sessions import InMemorySessionService

# Example: Configuring the Runner with an Artifact Service
my_agent = LlmAgent(name="artifact_user_agent", model="gemini-2.0-flash")
artifact_service = InMemoryArtifactService() # Choose an implementation
session_service = InMemorySessionService()

runner = Runner(
    agent=my_agent,
    app_name="my_artifact_app",
    session_service=session_service,
    artifact_service=artifact_service # Provide the service instance here
)
# Now, contexts within runs managed by this runner can use artifact methods

工件数据 (`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')，用于描述二进制数据的性质。这对于正确加载工件时的解释至关重要。
创建：通常您可以使用Part的from_data类方法或直接通过Blob构造来为工件创建一个Part。

import google.genai.types as types

# Example: Creating an artifact Part from raw bytes
pdf_bytes = b'%PDF-1.4...' # Your raw PDF data
pdf_mime_type = "application/pdf"

# Using the constructor
pdf_artifact = types.Part(
    inline_data=types.Blob(data=pdf_bytes, mime_type=pdf_mime_type)
)

# Using the convenience class method (equivalent)
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/（如GcsArtifactService._get_blob_name和InMemoryArtifactService._artifact_path中所示）
用户作用域（"user:"前缀）：如果文件名以"user:"为前缀，例如"user:profile.png"，则该工件仅与app_name和user_id相关联。它可以从该用户在应用内的任何会话中访问或更新。
内部路径（示例）：app_name/user_id/user/user:profile.png/（在服务实现中可以看到，user:前缀通常会保留在最终路径段中以保持清晰）。
使用场景：适用于属于用户自身、与特定对话无关的数据，例如个人头像、用户偏好设置文件或长期报告。

# Example illustrating namespace difference (conceptual)

# Session-specific artifact filename
session_report_filename = "summary.txt"

# User-specific artifact filename
user_config_filename = "user:settings.json"

# When saving 'summary.txt', it's tied to the current session ID.
# When saving 'user:settings.json', it's tied only to the user ID.

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

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

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

前提条件：配置 `ArtifactService`

在使用上下文对象调用任何智能体构件方法之前，必须在初始化Runner时提供一个BaseArtifactService实现类的实例（例如InMemoryArtifactService或GcsArtifactService）。

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

# Your agent definition
agent = LlmAgent(name="my_agent", model="gemini-2.0-flash")

# Instantiate the desired artifact service
artifact_service = InMemoryArtifactService()

# Provide it to the Runner
runner = Runner(
    agent=agent,
    app_name="artifact_app",
    session_service=InMemorySessionService(),
    artifact_service=artifact_service # Service must be provided here
)

如果在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中）添加条目来自动记录此操作。这个增量将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。

Code Example (within a hypothetical tool or callback):

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

async def process_latest_report(context: CallbackContext):
    """Loads the latest report artifact and processes its data."""
    filename = "generated_report.pdf"
    try:
        # Load the latest version
        report_artifact = context.load_artifact(filename=filename)

        if report_artifact and report_artifact.inline_data:
            print(f"Successfully loaded latest artifact '{filename}'.")
            print(f"MIME Type: {report_artifact.inline_data.mime_type}")
            # Process the report_artifact.inline_data.data (bytes)
            pdf_bytes = report_artifact.inline_data.data
            print(f"Report size: {len(pdf_bytes)} bytes.")
            # ... further processing ...
        else:
            print(f"Artifact '{filename}' not found.")

        # Example: Load a specific version (if version 0 exists)
        # specific_version_artifact = context.load_artifact(filename=filename, version=0)
        # if specific_version_artifact:
        #     print(f"Loaded version 0 of '{filename}'.")

    except ValueError as e:
        print(f"Error loading artifact: {e}. Is ArtifactService configured?")
    except Exception as e:
        # Handle potential storage errors
        print(f"An unexpected error occurred during artifact load: {e}")

# --- Example Usage Concept ---
# await process_latest_report(callback_context)

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

方法：

tool_context.list_artifacts() -> list[str]

可用上下文： 仅限ToolContext。此方法不适用于基础CallbackContext。
操作：调用底层的artifact_service.list_artifact_keys来获取当前作用域内可访问的所有唯一工件文件名列表（包括特定会话文件和以"user:"为前缀的用户作用域文件）。
返回： 一个排序后的list，包含str类型的文件名。
代码示例（在工具函数内部）：

from google.adk.tools.tool_context import ToolContext

def list_user_files(tool_context: ToolContext) -> str:
    """Tool to list available artifacts for the user."""
    try:
        available_files = tool_context.list_artifacts()
        if not available_files:
            return "You have no saved artifacts."
        else:
            # Format the list for the user/LLM
            file_list_str = "\n".join([f"- {fname}" for fname in available_files])
            return f"Here are your available artifacts:\n{file_list_str}"
    except ValueError as e:
        print(f"Error listing artifacts: {e}. Is ArtifactService configured?")
        return "Error: Could not list artifacts."
    except Exception as e:
        print(f"An unexpected error occurred during artifact list: {e}")
        return "Error: An unexpected error occurred while listing artifacts."

# This function would typically be wrapped in a 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，依此类推）。
Key Features:
- 简单性：除了核心ADK库外，不需要任何外部设置或依赖。
- 速度：操作通常非常快，因为它们涉及内存中的字典查找和列表操作。
- 临时性: 当运行应用的Python进程终止时，所有存储的工件都将丢失。数据在应用重启之间不会持久保存。
Use Cases:
- 非常适合不需要持久化的本地开发和测试场景。
- 适用于短期演示场景，或当工件数据仅在应用程序单次运行期间作为临时数据存在的情况。
实例化：

from google.adk.artifacts import InMemoryArtifactService

# Simply instantiate the class
in_memory_service = InMemoryArtifactService()

# Then pass it to the Runner
# runner = Runner(..., artifact_service=in_memory_service)

GcsArtifactService

源文件: google.adk.artifacts.gcs_artifact_service.py
存储机制： 利用Google Cloud Storage (GCS)实现持久化工件存储。每个工件版本都作为独立对象存储在指定的GCS存储桶中。
Object Naming Convention: It constructs GCS object names (blob names) using a hierarchical path structure, typically:
- 会话作用域: {app_name}/{user_id}/{session_id}/{filename}/{version}
- 用户作用域: {app_name}/{user_id}/user/{filename}/{version} (注意: 服务通过处理文件名中的user:前缀来确定路径结构).
Key Features:
- 持久性：存储在GCS中的工件在应用重启和部署后仍然保留。
- 可扩展性：利用Google Cloud Storage的可扩展性和持久性。
- 版本控制：明确将每个版本存储为独立的GCS对象。
- 需要配置：需要使用目标GCS bucket_name进行配置。
- 所需权限：应用程序环境需要具备适当的凭证和IAM权限，才能对指定的GCS存储桶进行读写操作。
Use Cases:
- 需要持久化存储产物的生产环境。
- 需要跨不同应用实例或服务共享产物的场景（通过访问同一个GCS存储桶）。
- 需要长期存储和检索用户或会话数据的应用程序。
实例化：

from google.adk.artifacts import GcsArtifactService

# Specify the GCS bucket name
gcs_bucket_name = "your-gcs-bucket-for-adk-artifacts" # Replace with your bucket name

try:
    gcs_service = GcsArtifactService(bucket_name=gcs_bucket_name)
    print(f"GcsArtifactService initialized for bucket: {gcs_bucket_name}")
    # Ensure your environment has credentials to access this bucket.
    # e.g., via Application Default Credentials (ADC)

    # Then pass it to the Runner
    # runner = Runner(..., artifact_service=gcs_service)

except Exception as e:
    # Catch potential errors during GCS client initialization (e.g., auth issues)
    print(f"Error initializing GcsArtifactService: {e}")
    # Handle the error appropriately - maybe fall back to InMemory or raise

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

最佳实践

为了有效且可持续地使用构件：

选择合适的服务： 使用InMemoryArtifactService进行快速原型设计、测试以及不需要持久化的场景。对于需要数据持久化和可扩展性的生产环境，请使用GcsArtifactService（或为其他后端实现自己的BaseArtifactService）。
有意义的文件名： 使用清晰、描述性的文件名。包含相关扩展名（.pdf、.png、.wav）有助于人类理解内容，尽管mime_type决定了程序处理方式。为临时与持久工件名称建立命名规范。
指定正确的MIME类型: 在创建types.Part用于save_artifact时，始终提供准确的mime_type。这对于后续需要load_artifact来正确解释bytes数据的应用程序或工具至关重要。尽可能使用标准的IANA MIME类型。
理解版本控制：请记住，不带特定version参数的load_artifact()会获取最新版本。如果您的逻辑依赖于某个特定的历史版本工件，请确保在加载时提供整数版本号。
谨慎使用命名空间(user:)：仅当数据确实属于用户且需要在所有会话中共享时，才在文件名前添加"user:"前缀。对于特定于单次对话或会话的数据，请使用不带前缀的常规文件名。
Error Handling:
- 在调用上下文方法（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链接或其他专用存储解决方案来处理超大数据，而非在内存中传递整个字节数组。
Cleanup Strategy: For persistent storage like GcsArtifactService, artifacts remain until explicitly deleted. If artifacts represent temporary data or have a limited lifespan, implement a strategy for cleanup. This might involve:
- 在存储桶上使用GCS生命周期策略。
- 构建特定的工具或管理功能，利用artifact_service.delete_artifact方法（注意：出于安全考虑，删除操作不通过上下文对象暴露）。
- 仔细管理文件名，以便在需要时支持基于模式的删除。

制品