import asyncio import logging from typing import Optional, List, AsyncGenerator, Any from core.config import config from core.exceptions import DependencyError, EngineError from .session import VoiceSession from .queue import get_queue from core.vision import VisionLoop # Native ADK Imports try: from google.adk.agents import LlmAgent, LiveRequestQueue from google.adk.runners import Runner, InMemorySessionService, RunConfig from google.adk.planners import BuiltInPlanner from google.adk.tools import FunctionTool from google.genai import types from google.genai.types import ThinkingConfig except ImportError as e: raise DependencyError(f"google-adk is required for Native Voice: {e}") logger = logging.getLogger("core.voice.client") # ============================================================================= # HYBRID ARCHITECTURE (2026) # ============================================================================= # VOICE ENGINE: gemini-2.5-flash-native-audio-latest (Live API / bidiGenerateContent) # VISION ENGINE: gemini-3-flash-preview (Standard API / generateContent) # ============================================================================= VOICE_MODEL = "gemini-2.5-flash-native-audio-latest" # Live API compatible VISION_MODEL = "gemini-3-flash-preview" # Agent Vision & Auto Browse def auto_browse_func(url: str, action: str) -> str: """ Launch a browser to see and interact with websites. Args: url: URL to visit. action: Action to perform. Returns: Result string. """ logger.info(f"[TOOL] auto_browse called: URL={url}, ACTION={action}") return f"Browsing {url} with action {action}. Visual context updated." class VisionAgent: """ Gemini 3 Flash Agent for Vision & Agentic Tasks. Uses standard generateContent API (not Live API). """ def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or config.get_api_key() self.model_id = VISION_MODEL self.runner: Optional[Runner] = None self.session_service: Optional[InMemorySessionService] = None # Tools for agentic tasks self.tools = [ FunctionTool(func=auto_browse_func) ] async def initialize(self): """Initialize the Vision Agent with Gemini 3 Flash.""" logger.info(f"Initializing Vision Agent ({self.model_id})...") # ThinkingConfig for reasoning thinking_planner = BuiltInPlanner( thinking_config=ThinkingConfig( include_thoughts=True, thinking_budget=1024 ) ) agent = LlmAgent( name="vision_agent", model=self.model_id, tools=self.tools, planner=thinking_planner, generate_content_config=types.GenerateContentConfig( temperature=0.7, safety_settings=[ types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_NONE") ] ) ) self.session_service = InMemorySessionService() self.runner = Runner( app_name="antigravity_vision", agent=agent, session_service=self.session_service ) logger.info(f"<< VISION AGENT READY ({self.model_id}) >>") async def process_task(self, task: str, user_id: str = "user", session_id: Optional[str] = None) -> str: """ Process an agentic task using Gemini 3 Flash. Uses standard run() method (not run_live). """ if not self.runner: await self.initialize() session_id = session_id or f"vision_{asyncio.get_event_loop().time()}" # Ensure session exists await self.session_service.create_session( app_name="antigravity_vision", user_id=user_id, session_id=session_id ) # Run task via standard API result_parts = [] async for event in self.runner.run( user_id=user_id, session_id=session_id, new_message=task ): if hasattr(event, 'content') and event.content: result_parts.append(str(event.content)) return "".join(result_parts) if result_parts else "Task completed." class VoiceClient: """ Hybrid Voice Client (2026). Architecture: - Voice Engine: gemini-2.5-flash-native-audio-latest (Live API for streaming) - Vision Engine: gemini-3-flash-preview (Standard API for agentic tasks) """ def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or config.get_api_key() self.voice_model = VOICE_MODEL self.vision_model = VISION_MODEL self.runner: Optional[Runner] = None self.current_session: Optional[VoiceSession] = None self.vision_loop = VisionLoop() self.vision_agent = VisionAgent(api_key) # Note: Tools are handled by VisionAgent for agentic tasks # Voice agent is lightweight for streaming async def start(self, system_instruction: Optional[str] = None): """ Ignite the Hybrid Voice Bridge. - Live API for voice streaming (2.5 Flash Native Audio) - Standard API for vision/browse tasks (3 Flash) """ logger.info(f"Igniting Hybrid Voice Bridge...") logger.info(f" Voice Engine: {self.voice_model}") logger.info(f" Vision Engine: {self.vision_model}") try: # 1. Initialize Vision Agent (Gemini 3 Flash) await self.vision_agent.initialize() # 2. Create Voice Agent (Live API compatible) voice_agent = LlmAgent( name="voice_agent", model=self.voice_model, # No tools on voice agent - delegate to vision agent generate_content_config=types.GenerateContentConfig( temperature=0.7, safety_settings=[ types.SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_NONE"), types.SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_NONE") ] ) ) logger.info(f"DEBUG: Voice Agent Model: {voice_agent.model}") # 3. Initialize Runner for Voice self.session_service = InMemorySessionService() self.runner = Runner( app_name="antigravity_voice", agent=voice_agent, session_service=self.session_service ) # 4. Create Session Tracker self.current_session = VoiceSession(self.voice_model) self.current_session.activate() await self.session_service.create_session( app_name="antigravity_voice", user_id="antigravity_user", session_id=self.current_session.session_id ) # 5. Initialize Queue live_queue = get_queue() # 6. Start Vision Loop self.vision_loop.start() logger.info("<< HYBRID VOICE-VISION BRIDGE ESTABLISHED >>") logger.info(f" Voice Stream: {self.voice_model} (Live API)") logger.info(f" Vision Tasks: {self.vision_model} (Standard API)") # 7. Run Live Voice Loop vision_task = asyncio.create_task(self._run_vision_cycle()) try: async for event in self.runner.run_live( live_request_queue=live_queue, user_id="antigravity_user", session_id=self.current_session.session_id ): # Handle events - delegate agentic tasks to vision agent await self._handle_event(event) finally: vision_task.cancel() except Exception as e: logger.error(f"Hybrid Voice Bridge Failure: {e}") if self.current_session: self.current_session.state = "ERROR" raise EngineError(f"Voice Bridge Crumbled: {e}") finally: self.vision_loop.stop() if self.current_session: self.current_session.end() async def _handle_event(self, event): """ Handle events from the Live Voice stream. Delegate agentic tasks (vision, browse) to Gemini 3 Flash. """ # Check if event requires agentic action if hasattr(event, 'requires_tool') and event.requires_tool: task_description = str(event.content) if hasattr(event, 'content') else "" if task_description: logger.info(f"Delegating to Vision Agent: {task_description[:50]}...") result = await self.vision_agent.process_task(task_description) logger.info(f"Vision Agent Result: {result[:100]}...") async def _run_vision_cycle(self): """Background task for Vision Loop ticks.""" while self.vision_loop.is_active: await self.vision_loop.process_frame() await asyncio.sleep(0.0625) # 16 FPS async def stop(self): """Stop the voice client.""" self.vision_loop.stop()