File size: 5,723 Bytes
c9f5b32 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
from google.adk.agents import Agent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.models import Gemini
from google.genai import types, Client
import logging
from typing import Optional
from functools import cached_property
logger = logging.getLogger(__name__)
# --- Custom Gemini Class for Explicit API Key ---
class CustomGemini(Gemini):
api_key: Optional[str] = None
@cached_property
def api_client(self) -> Client:
"""Overrides the default api_client to use the provided API key."""
return Client(
api_key=self.api_key,
http_options=types.HttpOptions(
headers=self._tracking_headers(),
retry_options=self.retry_options,
)
)
# --- Retry Configuration ---
retry_config = types.HttpRetryOptions(
attempts=5, # Maximum retry attempts
exp_base=7, # Delay multiplier
initial_delay=1,
http_status_codes=[429, 500, 503, 504], # Retry on these HTTP errors
)
def get_video_analysis_system(api_key: Optional[str] = None, model_name: str = "gemini-2.1-flash-lite"):
"""
Factory function to create a configured VideoAnalysisSystem.
"""
# Use the custom Gemini class with the provided API key
llm_model = CustomGemini(
model=model_name,
api_key=api_key,
retry_options=retry_config
)
# --- Individual Agents ---
# Context Summary Agent: Provides a neutral summary of the video content.
context_summary_agent = Agent(
name="ContextSummaryAgent",
model=llm_model,
instruction="""
<thinking>
1. Scan the video, caption, and transcript for key entities and events.
2. Identify the main narrative thread or objective of the content.
3. Synthesize the findings into a neutral, objective summary.
</thinking>
Output format: summary: text[1]{text}: "Summary text" """,
)
# Political Bias Agent: Identifies political leaning and quantifies bias.
political_bias_agent = Agent(
name="PoliticalBiasAgent",
model=llm_model,
instruction="""
<thinking>
1. Examine the language used for loaded terms or rhetorical devices.
2. Analyze the context of mentioned political figures or institutions.
3. Determine the leaning (Left/Right/Center) and evaluate the intensity of bias.
</thinking>
Output format: political_bias: details[1]{score,reasoning}: (Int),"Reasoning" """,
)
# Criticism Level Agent: Measures the degree of hostility or support in the tone.
criticism_level_agent = Agent(
name="CriticismLevelAgent",
model=llm_model,
instruction="""
<thinking>
1. Assess the emotional valence of the speaker and visual cues.
2. Identify instances of direct criticism, sarcasm, or praise.
3. Quantify the overall hostility level on a neutral-to-supportive scale.
</thinking>
Output format: criticism_level: details[1]{score,reasoning}: (Int),"Reasoning" """,
)
# Modalities Agent: Evaluates the consistency between video, audio, and text.
modalities_agent = Agent(
name="ModalitiesAgent",
model=llm_model,
instruction="""
<thinking>
1. Compare visual events with audio descriptions for sync issues or contradictions.
2. Verify if the user-provided caption accurately reflects the visual content.
3. Check the transcript against both the audio and the caption for discrepancies.
</thinking>
Output format:
video_audio_pairing: details[1]{score,reasoning}: (Int),"Reasoning"
video_caption_pairing: details[1]{score,reasoning}: (Int),"Reasoning"
audio_caption_pairing: details[1]{score,reasoning}: (Int),"Reasoning" """,
)
# Disinformation Agent: Analyzes potential manipulation and threat levels.
disinformation_agent = Agent(
name="DisinformationAgent",
model=llm_model,
instruction="""
<thinking>
1. Search for signs of technical manipulation (deepfakes, AI artifacts).
2. Analyze the intent behind potential misinformation (Political/Commercial).
3. Evaluate the risk level and the specific threat vector used.
</thinking>
Output format: disinformation_analysis: details[1]{level,intent,threat_vector}: (Int),(Intent),(Vector) """,
)
# Sentiment Bias Agent: Captures the overall emotional tone and inherent bias.
sentiment_bias_agent = Agent(
name="SentimentBiasAgent",
model=llm_model,
instruction="""
<thinking>
1. Aggregate the emotional signals from the entire video duration.
2. Identify recurring biased patterns or slanted perspectives.
3. Synthesize a comprehensive overview of the sentiment and bias.
</thinking>
Output format: sentiment_and_bias: text[1]{text}: "Synthesis text" """,
)
# --- Agent Grouping ---
# The ParallelAgent runs all its sub-agents simultaneously.
analysis_team = ParallelAgent(
name="AnalysisTeam",
sub_agents=[
context_summary_agent,
political_bias_agent,
criticism_level_agent,
modalities_agent,
disinformation_agent,
sentiment_bias_agent
],
)
# This SequentialAgent defines the high-level workflow.
system = SequentialAgent(
name="VideoAnalysisSystem",
sub_agents=[analysis_team],
)
return system
if __name__ == "__main__":
system = get_video_analysis_system()
print("Video Analysis Agent System created successfully.")
|