So greifen Sie mit einem Adk -Tool auf ein hochgeladenes Video/Artikel von Google ADK -Web zu ⇐ Python

[Guest]

Ich entwickle eine Anwendung mit ADK Web (insbesondere mit InMemoryService für Artefakte und Sitzungsverwaltung) und auf Schwierigkeiten auf den Zugriff auf Videodateien, die vom Benutzer hochgeladen wurden. Ich habe mehrere Ansätze versucht, aber bisher war keiner erfolgreich. Was ich ausprobiert habe: < /p>
Tool_Context.load_artifact (): Mein primärer Versuch war, < /p>
zu verwenden

Code: Select all

   `video_part = await tool_context.load_artifact(filename=video_filename)`
< /code>
Diese Methode scheint jedoch nicht die erwarteten Videodaten zurückzugeben.    Ich bin mir nicht sicher, ob load_artifact die korrekte Funktion zum Abrufen von benutzergeradenen Dateien ist oder ob es eine bestimmte Möglichkeit gibt, Videotateitypen damit zu verarbeiten. Zugriff auf die Videodatei oder rekonstruieren Sie sie. | < /p>
Forschung durchgeführt: Ich habe die offizielle ADK -Webdokumentation überprüft, insbesondere Abschnitte im Zusammenhang mit Artefaktmanagement, Sitzung und Datei -Uploads.     Trotz meiner Bemühungen habe ich keine klaren Beispiele oder spezifische Anleitungen zum programmgesteuerten Zugriff auf Dateien gefunden, die Benutzer über die ADK -Weboberfläche hochladen, wenn InMemoryService verwendet wird.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
)