Как получить доступ к загруженному видео/статье из Google Adk Web с помощью инструмента ADK ⇐ Python

[Anonymous]

Я разрабатываю приложение с использованием ADK Web (в частности, с InmemoryService для артефакта и управления сеансами), и сталкиваюсь с трудностями с доступом к видеофайлам, загруженным пользователем.
Моя цель - получить и обработать эти загруженные видеофайлы в моем веб -приложении ADK. Я пытался сделать несколько подходов, но до сих пор ни один не был успешным. Что я пробовал: < /p>
tool_context.load_artifact (): моя основная попытка была использовать < /p>
`video_part = await tool_context.load_artifact(filename=video_filename)`
< /code>
Однако этот метод не возвращает ожидаемые видеоданные. Я не уверен, что load_artifact является правильной функцией для извлечения пользовательских файлов, или есть ли определенный способ обработки типов видео файлов. Успешный доступ или реконструировать видеофайл. | < /p>
Проводятся исследования: я рассмотрел официальную веб -документацию ADK, особенно разделы, связанные с управлением артефактом, обработкой сессий и загрузки файлов. Несмотря на мои усилия, я не нашел никаких четких примеров или конкретных рекомендаций о том, как программный доход от файлов, которые пользователи загружают через веб -интерфейс ADK, когда используется 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
)


Подробнее здесь: https://stackoverflow.com/questions/796 ... g-adk-tool