上下文¶

什么是上下文¶

在智能体开发工具包（ADK）中，"上下文"指的是在特定操作期间可供你的智能体及其工具使用的关键信息包。可以将其视为有效处理当前任务或对话轮次所需的必要背景知识和资源。

智能体通常需要的不仅仅是最新的用户消息才能表现良好。上下文至关重要，因为它能够：

维持状态： 记住对话中多个步骤的详细信息（例如，用户偏好、之前的计算、购物车中的物品）。这主要通过会话状态管理。
传递数据： 在一个步骤（如 LLM 调用或工具执行）中发现或生成的信息与后续步骤共享。会话状态在这里也是关键。
访问服务： 与框架功能交互，如：
- 制品（Artifacts）存储： 保存或加载与会话相关的文件或数据块（如 PDF、图像、配置文件）。
- 记忆（Memory）： 从过去的交互或与用户相关的外部知识源中搜索相关信息。
- 认证： 请求并检索工具安全访问外部 API 所需的凭证。
身份和跟踪： 知道当前运行的是哪个智能体（agent.name）以及唯一标识当前请求 - 响应周期（invocation_id）以进行日志记录和调试。
工具特定操作： 启用工具内的专门操作，例如请求认证或搜索记忆（Memory），这些操作需要访问当前交互的详细信息。

保存单个完整的用户请求到最终响应周期（一次调用）所有信息的核心部分是InvocationContext。但是，你通常不会直接创建或管理此对象。ADK 框架在调用开始时创建它（例如，通过runner.run_async），并将相关上下文信息隐式传递给你的智能体代码、回调和工具。

PythonJava

# 概念伪代码：框架如何提供上下文（内部逻辑）

# runner = Runner(agent=my_root_agent, session_service=..., artifact_service=...)
# user_message = types.Content(...)
# session = session_service.get_session(...) # 或创建新会话

# --- 在 runner.run_async(...) 内部 ---
# 1. 框架为本次运行创建主上下文
# invocation_context = InvocationContext(
#     invocation_id="unique-id-for-this-run",
#     session=session,
#     user_content=user_message,
#     agent=my_root_agent, # 起始智能体
#     session_service=session_service,
#     artifact_service=artifact_service,
#     memory_service=memory_service,
#     # ... 其他必要字段 ...
# )
#
# 2. 框架调用智能体的 run 方法，隐式传递上下文
#    （智能体方法签名会接收它，例如 runAsyncImpl(InvocationContext invocationContext)）
# await my_root_agent.run_async(invocation_context)
#   --- 结束内部逻辑 ---
#
# 作为开发者，你只需在方法参数中使用框架提供的上下文对象。

/* 概念伪代码：框架如何提供上下文（内部逻辑） */
InMemoryRunner runner = new InMemoryRunner(agent);
Session session = runner
    .sessionService()
    .createSession(runner.appName(), USER_ID, initialState, SESSION_ID )
    .blockingGet();

try (Scanner scanner = new Scanner(System.in, StandardCharsets.UTF_8)) {
  while (true) {
    System.out.print("\nYou > ");
  }
  String userInput = scanner.nextLine();
  if ("quit".equalsIgnoreCase(userInput)) {
    break;
  }
  Content userMsg = Content.fromParts(Part.fromText(userInput));
  Flowable<Event> events = runner.runAsync(session.userId(), session.id(), userMsg);
  System.out.print("\nAgent > ");
  events.blockingForEach(event -> System.out.print(event.stringifyContent()));
}

不同类型的上下文¶

虽然InvocationContext作为全面的内部容器，但 ADK 提供了根据特定情况定制的专门上下文对象。这确保了你拥有适合手头任务的正确工具和权限，而无需在任何地方处理完整内部上下文的全部复杂性。以下是你将遇到的不同"风格"：

InvocationContext

使用场所： 在智能体的核心实现方法（_run_async_impl，_run_live_impl）中直接接收为ctx参数。
目的： 提供对当前调用整个状态的访问。这是最全面的上下文对象。
关键内容： 直接访问session（包括state和events）、当前agent实例、invocation_id、初始user_content、对已配置服务的引用（artifact_service、memory_service、session_service），以及与实时/流式模式相关的字段。
使用场景： 主要在智能体的核心逻辑需要直接访问整体会话或服务时使用，尽管状态和制品（Artifacts）交互通常委托给使用自己上下文的回调/工具。也用于控制调用本身（例如，设置ctx.end_invocation = True）。

PythonJava

# Pseudocode: Agent implementation receiving InvocationContext
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

class MyAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        # 直接访问示例
        agent_name = ctx.agent.name
        session_id = ctx.session.id
        print(f"Agent {agent_name} running in session {session_id} for invocation {ctx.invocation_id}")
        # ... 智能体逻辑使用 ctx ...
        yield # ... 事件 ...

// 伪代码：Agent 实现接收 InvocationContext
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.InvocationContext;

    LlmAgent root_agent =
        LlmAgent.builder()
            .model("gemini-***")
            .name("sample_agent")
            .description("Answers user questions.")
            .instruction(
                """
                provide instruction for the agent here.
                """
            )
            .tools(sampleTool)
            .outputKey("YOUR_KEY")
            .build();

    ConcurrentMap<String, Object> initialState = new ConcurrentHashMap<>();
    initialState.put("YOUR_KEY", "");

    InMemoryRunner runner = new InMemoryRunner(agent);
    Session session =
          runner
              .sessionService()
              .createSession(runner.appName(), USER_ID, initialState, SESSION_ID )
              .blockingGet();

   try (Scanner scanner = new Scanner(System.in, StandardCharsets.UTF_8)) {
        while (true) {
          System.out.print("\nYou > ");
          String userInput = scanner.nextLine();

          if ("quit".equalsIgnoreCase(userInput)) {
            break;
          }

          Content userMsg = Content.fromParts(Part.fromText(userInput));
          Flowable<Event> events = 
                  runner.runAsync(session.userId(), session.id(), userMsg);

          System.out.print("\nAgent > ");
          events.blockingForEach(event -> 
                  System.out.print(event.stringifyContent()));
      }

    protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
        // 直接访问示例
        String agentName = invocationContext.agent.name
        String sessionId = invocationContext.session.id
        String invocationId = invocationContext.invocationId
        System.out.println("Agent " + agentName + " running in session " + session_id + " for invocation " + invocationId)
        // ... 智能体逻辑使用 ctx ...
    }

ReadonlyContext

使用场所： 在只需要对基本信息进行只读访问且不允许修改的场景中提供（例如，InstructionProvider函数）。它也是其他上下文的基类。
目的： 提供基本上下文详细信息的安全、只读视图。
关键内容： invocation_id、agent_name以及当前state的只读视图。

PythonJava

# 伪代码：Instruction provider 接收 ReadonlyContext
from google.adk.agents import ReadonlyContext

def my_instruction_provider(context: ReadonlyContext) -> str:
    # 只读访问示例
    user_tier = context.state().get("user_tier", "standard") # 可以读取 state
    # context.state['new_key'] = 'value' # 这通常会报错或无效
    return f"Process the request for a {user_tier} user."

// 伪代码：Instruction provider 接收 ReadonlyContext
import com.google.adk.agents.ReadonlyContext;

public String myInstructionProvider(ReadonlyContext context){
    // 只读访问示例
    String userTier = context.state().get("user_tier", "standard");
    context.state().put('new_key', 'value'); // 这通常会报错
    return "Process the request for a " + userTier + " user.";
}

CallbackContext

使用场所： 作为callback_context传递给智能体生命周期回调（before_agent_callback，after_agent_callback）和模型交互回调（before_model_callback，after_model_callback）。
目的： 便于检查和修改状态、与制品（Artifacts）交互以及特别在回调中访问调用详情。
主要功能（添加到ReadonlyContext）：
- 可变state属性： 允许读取和写入会话状态。在此处所做的更改（callback_context.state['key'] = value）会被跟踪并与回调后框架生成的事件相关联。
- 制品（Artifacts）方法： 用于与配置的artifact_service交互的load_artifact(filename)和save_artifact(filename, part)方法。
- 直接user_content访问。

PythonJava

# 伪代码：Callback 接收 CallbackContext
from google.adk.agents import CallbackContext
from google.adk.models import LlmRequest
from google.genai import types
from typing import Optional

def my_before_model_cb(callback_context: CallbackContext, request: LlmRequest) -> Optional[types.Content]:
    # 读/写 state 示例
    call_count = callback_context.state.get("model_calls", 0)
    callback_context.state["model_calls"] = call_count + 1 # 修改 state

    # 可选：加载制品（Artifact）
    # config_part = callback_context.load_artifact("model_config.json")
    print(f"Preparing model call #{call_count + 1} for invocation {callback_context.invocation_id}")
    return None # 允许模型调用继续

// 伪代码：Callback 接收 CallbackContext
import com.google.adk.agents.CallbackContext;
import com.google.adk.models.LlmRequest;
import com.google.genai.types.Content;
import java.util.Optional;

public Maybe<LlmResponse> myBeforeModelCb(CallbackContext callbackContext, LlmRequest request){
    // 读/写 state 示例
    callCount = callbackContext.state().get("model_calls", 0);
    callbackContext.state().put("model_calls", callCount + 1); // 修改 state

    // 可选：加载制品（Artifact）
    // Maybe<Part> configPart = callbackContext.loadArtifact("model_config.json");
    System.out.println("Preparing model call " + (callCount + 1));
    return Maybe.empty(); // 允许模型调用继续
}

ToolContext

使用场所： 作为tool_context传递给支持FunctionTool的函数以及工具执行回调（before_tool_callback，after_tool_callback）。
目的： 提供CallbackContext所有功能，外加工具执行必需的专门方法，如处理认证、搜索记忆（Memory）和列出制品（Artifacts）。
主要功能（添加到CallbackContext）：
- 认证方法： request_credential(auth_config)触发认证流程，以及get_auth_response(auth_config)检索用户/系统提供的凭证。
- 制品（Artifacts）列表： list_artifacts()用于发现会话中可用的制品（Artifacts）。
- 记忆（Memory）搜索： search_memory(query)用于查询配置的memory_service。
- function_call_id属性： 标识触发此工具执行的 LLM 的特定函数调用，对于将认证请求或响应正确地链接回来至关重要。
- actions属性： 直接访问此步骤的EventActions对象，允许工具发出状态更改、认证请求等信号。

PythonJava

# 伪代码：工具函数接收 ToolContext
from google.adk.tools import ToolContext
from typing import Dict, Any

# 假设此函数被 FunctionTool 包装
def search_external_api(query: str, tool_context: ToolContext) -> Dict[str, Any]:
    api_key = tool_context.state.get("api_key")
    if not api_key:
        # 定义所需的认证配置
        # auth_config = AuthConfig(...)
        # tool_context.request_credential(auth_config) # 请求凭证
        # 使用 'actions' 属性来标记已发起认证请求
        # tool_context.actions.requested_auth_configs[tool_context.function_call_id] = auth_config
        return {"status": "Auth Required"}

    # 使用 API key...
    print(f"Tool executing for query '{query}' using API key. Invocation: {tool_context.invocation_id}")

    # 可选：搜索记忆或列出制品
    # relevant_docs = tool_context.search_memory(f"info related to {query}")
    # available_files = tool_context.list_artifacts()

    return {"result": f"Data for {query} fetched."}

// 伪代码：工具函数接收 ToolContext
import com.google.adk.tools.ToolContext;
import java.util.HashMap;
import java.util.Map;

// 假设此函数被 FunctionTool 包装
public Map<String, Object> searchExternalApi(String query, ToolContext toolContext){
    String apiKey = toolContext.state.get("api_key");
    if(apiKey.isEmpty()){
        // 定义所需的认证配置
        // authConfig = AuthConfig(...);
        // toolContext.requestCredential(authConfig); // 请求凭证
        // 使用 'actions' 属性来标记已发起认证请求
        ...
        return Map.of("status", "Auth Required");
    }

    // 使用 API key...
    System.out.println("Tool executing for query " + query + " using API key. ");

    // 可选：列出制品
    // Single<List<String>> availableFiles = toolContext.listArtifacts();

    return Map.of("result", "Data for " + query + " fetched");
}

理解这些不同的上下文对象以及何时使用它们是有效管理状态、访问服务以及控制 ADK 应用程序流程的关键。下一节将详细介绍使用这些上下文可以执行的常见任务。

使用上下文的常见任务¶

现在你已经了解了不同的上下文对象，让我们专注于在构建智能体和工具时如何使用它们来执行常见任务。

访问信息¶

你将经常需要读取存储在上下文中的信息。

读取会话状态： 访问之前步骤中保存的数据或用户/应用级别设置。在state属性上使用类似字典的访问。

PythonJava

# 伪代码：在 Tool 函数中
from google.adk.tools import ToolContext

def my_tool(tool_context: ToolContext, **kwargs):
    user_pref = tool_context.state.get("user_display_preference", "default_mode")
    api_endpoint = tool_context.state.get("app:api_endpoint") # Read app-level state

    if user_pref == "dark_mode":
        # ... apply dark mode logic ...
        pass
    print(f"Using API endpoint: {api_endpoint}")
    # ... rest of tool logic ...

# 伪代码：在 Callback 函数中
from google.adk.agents import CallbackContext

def my_callback(callback_context: CallbackContext, **kwargs):
    last_tool_result = callback_context.state.get("temp:last_api_result") # Read temporary state
    if last_tool_result:
        print(f"Found temporary result from last tool: {last_tool_result}")
    # ... callback logic ...

// 伪代码：在 Tool 函数中
import com.google.adk.tools.ToolContext;

public void myTool(ToolContext toolContext){
   String userPref = toolContext.state().get("user_display_preference");
   String apiEndpoint = toolContext.state().get("app:api_endpoint"); // Read app-level state
   if(userPref.equals("dark_mode")){
        // ... apply dark mode logic ...
        pass
    }
   System.out.println("Using API endpoint: " + api_endpoint);
   // ... rest of tool logic ...
}


// 伪代码：在 Callback 函数中
import com.google.adk.agents.CallbackContext;

    public void myCallback(CallbackContext callbackContext){
        String lastToolResult = (String) callbackContext.state().get("temp:last_api_result"); // Read temporary state
    }
    if(!(lastToolResult.isEmpty())){
        System.out.println("Found temporary result from last tool: " + lastToolResult);
    }
    // ... callback logic ...

获取当前标识符： 对于基于当前操作的日志记录或自定义逻辑很有用。

PythonJava

# 伪代码：在任何上下文中（ToolContext 示例）
from google.adk.tools import ToolContext

def log_tool_usage(tool_context: ToolContext, **kwargs):
    agent_name = tool_context.agent_nameSystem.out.println("Found temporary result from last tool: " + lastToolResult);
    inv_id = tool_context.invocation_id
    func_call_id = getattr(tool_context, 'function_call_id', 'N/A') # Specific to ToolContext

    print(f"Log: Invocation={inv_id}, Agent={agent_name}, FunctionCallID={func_call_id} - Tool Executed.")

// 伪代码：在任何上下文中（ToolContext 示例）
 import com.google.adk.tools.ToolContext;

 public void logToolUsage(ToolContext toolContext){
            String agentName = toolContext.agentName;
            String invId = toolContext.invocationId;
            String functionCallId = toolContext.functionCallId().get(); // Specific to ToolContext
            System.out.println("Log: Invocation= " + invId &+ " Agent= " + agentName);
        }

访问初始用户输入： 参考开始当前调用的消息。

PythonJava

# 伪代码：在 Callback 中
from google.adk.agents import CallbackContext

def check_initial_intent(callback_context: CallbackContext, **kwargs):
    initial_text = "N/A"
    if callback_context.user_content and callback_context.user_content.parts:
        initial_text = callback_context.user_content.parts[0].text or "Non-text input"

    print(f"This invocation started with user input: '{initial_text}'")

# 伪代码：在 Agent 的 _run_async_impl 中
# async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
#     if ctx.user_content and ctx.user_content.parts:
#         initial_text = ctx.user_content.parts[0].text
#         print(f"Agent logic remembering initial query: {initial_text}")
#     ...

// 伪代码：在 Callback 中
import com.google.adk.agents.CallbackContext;

public void checkInitialIntent(CallbackContext callbackContext){
    String initialText = "N/A";
    if((!(callbackContext.userContent().isEmpty())) && (!(callbackContext.userContent().parts.isEmpty()))){
        initialText = cbx.userContent().get().parts().get().get(0).text().get();
        ...
        System.out.println("This invocation started with user input: " + initialText)
    }
}

Managing Session State¶

状态对于记忆（Memory）和数据流至关重要。当你使用CallbackContext或ToolContext修改状态时，框架会自动跟踪并持久化这些更改。

工作原理： 写入callback_context.state['my_key'] = my_value或tool_context.state['my_key'] = my_value将此更改添加到与当前步骤事件关联的EventActions.state_delta中。然后SessionService在持久化事件时应用这些增量变化。

在工具之间传递数据：

PythonJava

# 伪代码：Tool 1 - Fetches user ID
from google.adk.tools import ToolContext
import uuid

def get_user_profile(tool_context: ToolContext) -> dict:
    user_id = str(uuid.uuid4()) # Simulate fetching ID
    # Save the ID to state for the next tool
    tool_context.state["temp:current_user_id"] = user_id
    return {"profile_status": "ID generated"}

# 伪代码：Tool 2 - Uses user ID from state
def get_user_orders(tool_context: ToolContext) -> dict:
    user_id = tool_context.state.get("temp:current_user_id")
    if not user_id:
        return {"error": "User ID not found in state"}

    print(f"Fetching orders for user ID: {user_id}")
    # ... logic to fetch orders using user_id ...
    return {"orders": ["order123", "order456"]}

// 伪代码：Tool 1 - Fetches user ID
import com.google.adk.tools.ToolContext;
import java.util.UUID;

public Map<String, String> getUserProfile(ToolContext toolContext){
    String userId = UUID.randomUUID().toString();
    // Save the ID to state for the next tool
    toolContext.state().put("temp:current_user_id", user_id);
    return Map.of("profile_status", "ID generated");
}

// 伪代码：Tool 2 - Uses user ID from state
public  Map<String, String> getUserOrders(ToolContext toolContext){
    String userId = toolContext.state().get("temp:current_user_id");
    if(userId.isEmpty()){
        return Map.of("error", "User ID not found in state");
    }
    System.out.println("Fetching orders for user id: " + userId);
     // ... logic to fetch orders using user_id ...
    return Map.of("orders", "order123");
}

更新用户偏好：

PythonJava

# 伪代码：Tool 或 Callback 识别偏好
from google.adk.tools import ToolContext # Or CallbackContext

def set_user_preference(tool_context: ToolContext, preference: str, value: str) -> dict:
    # Use 'user:' prefix for user-level state (if using a persistent SessionService)
    state_key = f"user:{preference}"
    tool_context.state[state_key] = value
    print(f"Set user preference '{preference}' to '{value}'")
    return {"status": "Preference updated"}

// 伪代码：Tool 或 Callback 识别偏好
import com.google.adk.tools.ToolContext; // Or CallbackContext

public Map<String, String> setUserPreference(ToolContext toolContext, String preference, String value){
    // Use 'user:' prefix for user-level state (if using a persistent SessionService)
    String stateKey = "user:" + preference;
    toolContext.state().put(stateKey, value);
    System.out.println("Set user preference '" + preference + "' to '" + value + "'");
    return Map.of("status", "Preference updated");
}

状态前缀： 虽然基本状态是会话特定的，但像app:和user:这样的前缀可以与持久SessionService实现（如DatabaseSessionService或VertexAiSessionService）一起使用，以指示更广泛的范围（应用范围或跨会话的用户范围）。temp:可以表示仅在当前调用中相关的数据。

使用制品（Artifacts）¶

使用制品（Artifacts）处理与会话相关的文件或大型数据块。常见用例：处理上传的文档。

文档摘要生成器示例流程：

引入引用（例如，在设置工具或回调中）： 将文档的路径或 URI保存为制品（Artifacts），而不是整个内容。

PythonJava

# 伪代码：在 callback 或 initial tool
from google.adk.agents import CallbackContext # Or ToolContext
from google.genai import types

def save_document_reference(context: CallbackContext, file_path: str) -> None:
    # Assume file_path is something like "gs://my-bucket/docs/report.pdf" or "/local/path/to/report.pdf"
    try:
        # Create a Part containing the path/URI text
        artifact_part = types.Part(text=file_path)
        version = context.save_artifact("document_to_summarize.txt", artifact_part)
        print(f"Saved document reference '{file_path}' as artifact version {version}")
        # Store the filename in state if needed by other tools
        context.state["temp:doc_artifact_name"] = "document_to_summarize.txt"
    except ValueError as e:
        print(f"Error saving artifact: {e}") # E.g., Artifact service not configured
    except Exception as e:
        print(f"Unexpected error saving artifact reference: {e}")

# Example usage:
# save_document_reference(callback_context, "gs://my-bucket/docs/report.pdf")

// 伪代码：在 callback 或 initial tool
import com.google.adk.agents.CallbackContext;
import com.google.genai.types.Content;
import com.google.genai.types.Part;


pubic void saveDocumentReference(CallbackContext context, String filePath){
    // Assume file_path is something like "gs://my-bucket/docs/report.pdf" or "/local/path/to/report.pdf"
    try{
        // Create a Part containing the path/URI text
        Part artifactPart = types.Part(filePath)
        Optional<Integer> version = context.saveArtifact("document_to_summarize.txt", artifactPart)
        System.out.println("Saved document reference" + filePath + " as artifact version " + version);
        // Store the filename in state if needed by other tools
        context.state().put("temp:doc_artifact_name", "document_to_summarize.txt");
    } catch(Exception e){
        System.out.println("Unexpected error saving artifact reference: " + e);
    }
}

// Example usage:
// saveDocumentReference(context, "gs://my-bucket/docs/report.pdf")

摘要工具： 加载制品（Artifacts）以获取路径/URI，使用适当的库读取实际文档内容，总结，并返回结果。

PythonJava

# 伪代码：在 Summarizer 工具函数中
from google.adk.tools import ToolContext
from google.genai import types
# Assume libraries like google.cloud.storage or built-in open are available
# Assume a 'summarize_text' function exists
# from my_summarizer_lib import summarize_text

def summarize_document_tool(tool_context: ToolContext) -> dict:
    artifact_name = tool_context.state.get("temp:doc_artifact_name")
    if not artifact_name:
        return {"error": "Document artifact name not found in state."}

    try:
        # 1. Load the artifact part containing the path/URI
        artifact_part = tool_context.load_artifact(artifact_name)
        if not artifact_part or not artifact_part.text:
            return {"error": f"Could not load artifact or artifact has no text path: {artifact_name}"}

        file_path = artifact_part.text
        print(f"Loaded document reference: {file_path}")

        # 2. Read the actual document content (outside ADK context)
        document_content = ""
        if file_path.startswith("gs://"):
            # Example: Use GCS client library to download/read
            # from google.cloud import storage
            # client = storage.Client()
            # blob = storage.Blob.from_string(file_path, client=client)
            # document_content = blob.download_as_text() # Or bytes depending on format
            pass # Replace with actual GCS reading logic
        elif file_path.startswith("/"):
             # Example: Use local file system
             with open(file_path, 'r', encoding='utf-8') as f:
                 document_content = f.read()
        else:
            return {"error": f"Unsupported file path scheme: {file_path}"}

        # 3. Summarize the content
        if not document_content:
             return {"error": "Failed to read document content."}

        # summary = summarize_text(document_content) # Call your summarization logic
        summary = f"Summary of content from {file_path}" # Placeholder

        return {"summary": summary}

    except ValueError as e:
         return {"error": f"Artifact service error: {e}"}
    except FileNotFoundError:
         return {"error": f"Local file not found: {file_path}"}
    # except Exception as e: # Catch specific exceptions for GCS etc.
    #      return {"error": f"Error reading document {file_path}: {e}"}

// 伪代码：在 Summarizer 工具函数中
import com.google.adk.tools.ToolContext;
import com.google.genai.types.Content;
import com.google.genai.types.Part;

public Map<String, String> summarizeDocumentTool(ToolContext toolContext){
    String artifactName = toolContext.state().get("temp:doc_artifact_name");
    if(artifactName.isEmpty()){
        return Map.of("error", "Document artifact name not found in state.");
    }
    try{
        // 1. Load the artifact part containing the path/URI
        Maybe<Part> artifactPart = toolContext.loadArtifact(artifactName);
        if((artifactPart == null) || (artifactPart.text().isEmpty())){
            return Map.of("error", "Could not load artifact or artifact has no text path: " + artifactName);
        }
        filePath = artifactPart.text();
        System.out.println("Loaded document reference: " + filePath);

        // 2. Read the actual document content (outside ADK context)
        String documentContent = "";
        if(filePath.startsWith("gs://")){
            // Example: Use GCS client library to download/read into documentContent
            pass; // Replace with actual GCS reading logic
        } else if(){
            // Example: Use local file system to download/read into documentContent
        } else{
            return Map.of("error", "Unsupported file path scheme: " + filePath); 
        }

        // 3. Summarize the content
        if(documentContent.isEmpty()){
            return Map.of("error", "Failed to read document content."); 
        }

        // summary = summarizeText(documentContent) // Call your summarization logic
        summary = "Summary of content from " + filePath; // Placeholder

        return Map.of("summary", summary);
    } catch(IllegalArgumentException e){
        return Map.of("error", "Artifact service error " + filePath + e);
    } catch(FileNotFoundException e){
        return Map.of("error", "Local file not found " + filePath + e);
    } catch(Exception e){
        return Map.of("error", "Error reading document " + filePath + e);
    }
}

Listing Artifacts: Discover what files are available.

PythonJava

# 伪代码：在 tool 函数中
from google.adk.tools import ToolContext

def check_available_docs(tool_context: ToolContext) -> dict:
    try:
        artifact_keys = tool_context.list_artifacts()
        print(f"Available artifacts: {artifact_keys}")
        return {"available_docs": artifact_keys}
    except ValueError as e:
        return {"error": f"Artifact service error: {e}"}

// 伪代码：在 tool 函数中
import com.google.adk.tools.ToolContext;

public Map<String, String> checkAvailableDocs(ToolContext toolContext){
    try{
        Single<List<String>> artifactKeys = toolContext.listArtifacts();
        System.out.println("Available artifacts" + artifactKeys.tostring());
        return Map.of("availableDocs", "artifactKeys");
    } catch(IllegalArgumentException e){
        return Map.of("error", "Artifact service error: " + e);
    }
}

Handling Tool Authentication¶

安全管理工具所需的 API 密钥或其他凭证。

# 伪代码：需要认证的工具
from google.adk.tools import ToolContext
from google.adk.auth import AuthConfig # 假设定义了适当的 AuthConfig

# 定义你需要的认证配置（例如，OAuth、API 密钥）
MY_API_AUTH_CONFIG = AuthConfig(...)
AUTH_STATE_KEY = "user:my_api_credential" # 存储检索到的凭证的键

def call_secure_api(tool_context: ToolContext, request_data: str) -> dict:
    # 1. 检查凭证是否已存在于状态中
    credential = tool_context.state.get(AUTH_STATE_KEY)

    if not credential:
        # 2. 如果不存在，请求它
        print("未找到凭证，正在请求...")
        try:
            tool_context.request_credential(MY_API_AUTH_CONFIG)
            # 框架处理生成事件。工具执行在此处停止本轮。
            return {"status": "需要认证。请提供凭证。"}
        except ValueError as e:
            return {"error": f"认证错误：{e}"} # 例如，缺少 function_call_id
        except Exception as e:
            return {"error": f"请求凭证失败：{e}"}

    # 3. 如果凭证存在（可能来自请求后的上一轮）
    #    或者如果这是外部认证流程完成后的后续调用
    try:
        # 如有需要，可选择重新验证/检索，或直接使用
        # 如果外部流程刚刚完成，这可能会检索凭证
        auth_credential_obj = tool_context.get_auth_response(MY_API_AUTH_CONFIG)
        api_key = auth_credential_obj.api_key # 或 access_token 等

        # 将其存回状态中以供会话内的未来调用使用
        tool_context.state[AUTH_STATE_KEY] = auth_credential_obj.model_dump() # 持久化检索到的凭证

        print(f"使用检索到的凭证调用 API，数据：{request_data}")
        # ... 使用 api_key 进行实际的 API 调用 ...
        api_result = f"{request_data}的 API 结果"

        return {"result": api_result}
    except Exception as e:
        # 处理检索/使用凭证时的错误
        print(f"使用凭证时出错：{e}")
        # 如果凭证无效，可能清除状态键？
        # tool_context.state[AUTH_STATE_KEY] = None
        return {"error": "使用凭证失败"}

记住：request_credential暂停工具并发出需要认证的信号。用户/系统提供凭证，在后续调用中，get_auth_response（或再次检查状态）允许工具继续。 框架隐式使用tool_context.function_call_id来链接请求和响应。

Leveraging Memory¶

访问来自过去或外部来源的相关信息。

# 伪代码：使用记忆（Memory）搜索的工具
from google.adk.tools import ToolContext

def find_related_info(tool_context: ToolContext, topic: str) -> dict:
    try:
        search_results = tool_context.search_memory(f"有关{topic}的信息")
        if search_results.results:
            print(f"为'{topic}'找到了{len(search_results.results)}条记忆（Memory）结果")
            # 处理 search_results.results（这些是 SearchMemoryResponseEntry）
            top_result_text = search_results.results[0].text
            return {"memory_snippet": top_result_text}
        else:
            return {"message": "未找到相关记忆（Memory）。"}
    except ValueError as e:
        return {"error": f"记忆服务（Memory Service）错误：{e}"} # 例如，服务未配置
    except Exception as e:
        return {"error": f"搜索记忆（Memory）时出现意外错误：{e}"}

Advanced: Direct `InvocationContext` Usage¶

虽然大多数交互通过CallbackContext或ToolContext进行，但有时智能体的核心逻辑（_run_async_impl/_run_live_impl）需要直接访问。

# Pseudocode: Inside agent's _run_async_impl
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

class MyControllingAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        # 示例：检查特定服务是否可用
        if not ctx.memory_service:
            print("本次调用没有可用的记忆服务（Memory Service）。")
            # 可能改变智能体行为

        # 示例：基于某些条件提前终止
        if ctx.session.state.get("critical_error_flag"):
            print("检测到严重错误，正在结束调用。")
            ctx.end_invocation = True # 向框架发出停止处理的信号
            yield Event(author=self.name, invocation_id=ctx.invocation_id, content="由于严重错误而停止。")
            return # 停止此智能体的执行

        # ... 正常的智能体处理 ...
        yield # ... 事件 ...

设置ctx.end_invocation = True是从智能体内部或其回调/工具（通过它们各自的上下文对象，这些对象也可以访问并修改底层InvocationContext的标志）优雅地停止整个请求 - 响应周期的方法。

关键要点和最佳实践¶

使用合适的上下文： 始终使用提供的最具体的上下文对象（工具/工具回调中的ToolContext，智能体/模型回调中的CallbackContext，适用情况下的ReadonlyContext）。仅在必要时直接在_run_async_impl / _run_live_impl中使用完整的InvocationContext（ctx）。
用于数据流的状态： context.state是在调用内部共享数据、记住偏好和管理对话记忆（Memory）的主要方式。使用持久存储时，要深思熟虑地使用前缀（app:、user:、temp:）。
用于文件的制品（Artifacts）： 使用context.save_artifact和context.load_artifact来管理文件引用（如路径或 URI）或更大的数据块。存储引用，按需加载内容。
跟踪更改： 通过上下文方法对状态或制品（Artifacts）所做的修改会自动链接到当前步骤的EventActions并由SessionService处理。
从简单开始： 首先专注于state和基本制品（Artifacts）用法。随着需求变得更加复杂，再探索认证、记忆（Memory）和高级InvocationContext字段（如用于实时流式处理的字段）。

通过理解并有效使用这些上下文对象，你可以使用 ADK 构建更复杂、状态化且功能强大的智能体。

上下文¶

什么是上下文¶

不同类型的上下文¶

使用上下文的常见任务¶

访问信息¶

Managing Session State¶

使用制品（Artifacts）¶

Handling Tool Authentication¶

Leveraging Memory¶

Advanced: Direct InvocationContext Usage¶

关键要点和最佳实践¶

Advanced: Direct `InvocationContext` Usage¶