使用Google ADK Web开发时，如何访问用户上传的视频文件？

我正在使用ADK Web开发应用，采用InMemoryService进行工件和会话管理，当前在获取用户上传的视频文件时遇到阻碍，目标是在ADK Web应用内检索并处理这些视频。尝试以下两种方法均未成功：

• 使用tool_context.load_artifact()：执行代码video_part = await tool_context.load_artifact(filename=video_filename)，未返回预期视频数据，不确定该函数是否适用于获取用户上传文件，或处理视频文件是否存在特定方式；
• 将视频数据作为函数参数传递：尝试过Dict类型、Part类型、base64字符串，但均无法成功访问或重建视频文件。

我已查阅官方ADK Web文档中关于工件管理、会话处理和文件上传的内容，但未找到使用InMemoryService时通过编程方式访问用户上传文件的清晰示例或具体指导。

import base64
import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent
from google.cloud import videointelligence
import os
import shutil
from google.adk.artifacts import InMemoryArtifactService
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.adk.runners import Runner
from google.adk.tools import FunctionTool
from google.genai import types
from typing import Optional, Dict, Any
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def save_video_to_current_dir(filename: str) -> str:
    """
    Save the video from ADK web UI to current directory.
    
    Args:
        filename: The filename of the video from ADK web UI
        
    Returns:
        The path where the video was saved
    """
    try:
        # Get the current directory
        current_dir = os.getcwd()
        
        # Create a new filename with timestamp
        timestamp = datetime.datetime.now(ZoneInfo('UTC')).strftime('%Y%m%d_%H%M%S')
        new_filename = f"input_video_{timestamp}.mp4"
        
        # Full path for the new file
        output_path = os.path.join(current_dir, new_filename)
        
        # Copy the file from ADK web UI location to current directory
        shutil.copy2(filename, output_path)
        
        logger.info(f"Successfully saved video to: {output_path}")
        return output_path
        
    except Exception as e:
        logger.error(f"Error saving video: {str(e)}")
        raise

async def process_video_input(video_data: Dict[str, Any], tool_context: ToolContext) -> Dict[str, Any]:
    """
    Process video input from ADK web UI.
    
    Args:
        video_data: Dictionary containing video data from the UI with a 'video' key containing the video filename
        tool_context: The context object provided by the ADK framework
        
    Returns:
        A dictionary containing:
        - status: A string indicating the status of video processing
        - video: The processed video part (if successful)
    """
    try:
        logger.info("Received video input from ADK web UI")
        logger.info(f"Video data received: {video_data}")
        
        # Get the video filename from the input data
        if 'video' not in video_data:
            raise ValueError("No video data found in input")
            
        video_filename = video_data['video']
        logger.info(f"Processing video with filename: {video_filename}")
        
        # Load the video content using tool_context
        video_part = await tool_context.load_artifact(filename=video_filename)
        if not video_part:
            raise ValueError(f"Could not load video content for {video_filename}")
            
        logger.info(f"Successfully loaded video content")
        
        # Save as artifact with a new name
        timestamp = datetime.datetime.now(ZoneInfo('UTC')).strftime('%Y%m%d_%H%M%S')
        artifact_name = f"input_video_{timestamp}.mp4"
        
        version = await tool_context.save_artifact(
            filename=artifact_name,
            artifact=video_part
        )
        
        logger.info(f"Successfully saved video as artifact: {artifact_name} (version: {version})")
        
        return {
            "status": f"Video successfully processed and saved as {artifact_name}",
            "video": video_part
        }
        
    except Exception as e:
        logger.error(f"Error processing video: {str(e)}")
        return {
            "status": f"Error processing video: {str(e)}",
            "video": None
        }

def create_troll_video(filename: str, tool_context: ToolContext) -> Optional[str]:
    """
    Creates a troll video from the input video.
    
    Args:
        filename: The filename of the input video artifact
        tool_context: The context object provided by the ADK framework
        
    Returns:
        A string indicating the status of video creation
    """
    try:
        # Load the input video artifact using the tool context's load_artifact method
        input_video = tool_context.load_artifact(filename=filename)
        
        if not input_video:
            return f"Could not find video artifact: {filename}"
            
        # Process the video (implement your video processing logic here)
        # For now, we'll just return a success message
        return f"Successfully processed video {filename}"
        
    except Exception as e:
        logger.error(f"Error creating troll video: {str(e)}")
        return f"Error creating troll video: {str(e)}"

# Initialize services
artifact_service = InMemoryArtifactService()
session_service = InMemorySessionService()

# Create the agent with video processing capabilities
root_agent = Agent(
    name="troll_generator",
    model="gemini-2.0-flash",
    description=(
        "This agent creates troll videos based on user queries. "
        "It accepts video input from the ADK web UI and processes it to create entertaining content."
    ),
    instruction=(
        "You are a video creation agent that can create troll videos based on user queries. "
        "You can accept video input from the ADK web UI and process it to create entertaining content. "
        "Before creating a troll video, you need to process the video input first. "
        "When a user uploads a video, it will be available as a video part in the input. "
        "You should process this video part and save it as an artifact before proceeding with video creation. "
        "After processing, you should return both the status message and the processed video back to the user."
    ),
    tools=[
        FunctionTool(process_video_input)
    ]
)

# Initialize the runner
runner = Runner(
    agent=root_agent,
    app_name="video_agent_app",
    session_service=session_service,
    artifact_service=artifact_service
)

内容的提问来源于stack exchange，提问作者EL BEAST Z