""" Advanced Agentic System Interface ------------------------------- Provides a chat interface to interact with the autonomous agent teams: - Team A: Coders (App/Software Developers) - Team B: Business (Entrepreneurs) - Team C: Research (Deep Online Research) - Team D: Crypto & Sports Trading """ import gradio as gr import asyncio from typing import Dict, Any, List, Tuple import json from datetime import datetime import logging import os import socket import requests from requests.adapters import HTTPAdapter, Retry from urllib3.util.retry import Retry import time from agentic_system import AgenticSystem from team_management import TeamManager, TeamType, TeamObjective from orchestrator import AgentOrchestrator from reasoning import UnifiedReasoningEngine as ReasoningEngine # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Configure network settings TIMEOUT = int(os.getenv('REQUESTS_TIMEOUT', '30')) MAX_RETRIES = 5 RETRY_BACKOFF = 1 def setup_requests_session(): """Configure requests session with retries.""" session = requests.Session() retry_strategy = Retry( total=MAX_RETRIES, backoff_factor=RETRY_BACKOFF, status_forcelist=[408, 429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("https://", adapter) session.mount("http://", adapter) return session def check_network(max_attempts=3): """Check network connectivity with retries.""" session = setup_requests_session() for attempt in range(max_attempts): try: # Try multiple DNS servers for dns in ['8.8.8.8', '8.8.4.4', '1.1.1.1']: try: socket.gethostbyname('huggingface.co') break except socket.gaierror: continue # Test connection to Hugging Face response = session.get('https://huggingface.co/api/health', timeout=TIMEOUT) if response.status_code == 200: return True except (requests.RequestException, socket.gaierror) as e: logger.warning(f"Network check attempt {attempt + 1} failed: {e}") if attempt < max_attempts - 1: time.sleep(RETRY_BACKOFF * (attempt + 1)) continue logger.error("Network connectivity check failed after all attempts") return False class ChatInterface: def __init__(self): # Check network connectivity if not check_network(): logger.warning("Network connectivity issues detected - continuing with degraded functionality") # Initialize core components with consistent configuration config = { "min_confidence": 0.7, "parallel_threshold": 3, "learning_rate": 0.1, "strategy_weights": { "LOCAL_LLM": 0.8, "CHAIN_OF_THOUGHT": 0.6, "TREE_OF_THOUGHTS": 0.5, "META_LEARNING": 0.4 } } self.orchestrator = AgentOrchestrator(config) self.agentic_system = AgenticSystem(config) self.team_manager = TeamManager(self.orchestrator) self.chat_history = [] self.active_objectives = {} # Set up network session self.session = setup_requests_session() # Initialize teams asyncio.run(self.team_manager.initialize_team_agents()) async def process_message( self, message: str, history: List[List[str]] ) -> Tuple[str, List[List[str]]]: """Process incoming chat message.""" try: # Validate input if not message or not message.strip(): return "Please provide a message to process.", history # Check network before processing if not check_network(): return "Network connectivity issues detected. Some features may be limited.", history # Process the message response = await self._handle_message(message) # Update chat history if isinstance(response, tuple): response = response[0] # Extract message if tuple returned history.append([message, response]) return response, history except Exception as e: logger.error(f"Error processing message: {str(e)}") error_msg = "I apologize, but I encountered a temporary error. Please try again or rephrase your request." history.append([message, error_msg]) return error_msg, history async def _handle_message(self, message: str) -> str: """Handle message processing with error recovery.""" try: # Analyze message intent intent = await self._analyze_intent(message) intent_type = self._get_intent_type(intent) # Process based on intent if intent_type == "query": return await self._handle_query(message) elif intent_type == "objective": return await self._handle_objective(message) elif intent_type == "status": return await self._handle_status_request(message) else: return await self._handle_general_chat(message) except Exception as e: logger.error(f"Error in message handler: {str(e)}") raise def _get_intent_type(self, intent) -> str: """Safely extract intent type from various result formats.""" try: if hasattr(intent, "type"): return intent.type elif hasattr(intent, "intent_type"): return intent.intent_type elif isinstance(intent, dict): return intent.get("type") or intent.get("intent_type", "general") else: return "general" except Exception as e: logger.error(f"Error getting intent type: {str(e)}") return "general" async def _analyze_intent(self, message: str) -> Dict[str, Any]: """Analyze user message intent with error handling.""" try: # Use reasoning engine to analyze intent result = await self.orchestrator.reasoning_engine.reason( query=message, context={ "chat_history": self.chat_history, "active_objectives": self.active_objectives } ) # Handle different result types if hasattr(result, "to_dict"): return result.to_dict() elif hasattr(result, "__dict__"): return result.__dict__ elif isinstance(result, dict): return result else: return {"type": "general", "confidence": 0.5} except Exception as e: logger.error(f"Error analyzing intent: {str(e)}") return {"type": "general", "confidence": 0.5} async def _handle_query(self, message: str) -> str: """Handle information queries.""" # Get relevant teams for the query recommended_teams = await self.team_manager.get_team_recommendations(message) # Get responses from relevant teams responses = [] for team_type in recommended_teams: team_response = await self._get_team_response(team_type, message) responses.append(team_response) # Combine and format responses combined_response = self._format_team_responses(responses) return combined_response async def _handle_objective(self, message: str) -> str: """Handle new objective creation.""" # Analyze objective requirements analysis = await self.orchestrator.reasoning_engine.reason( query=f"Analyze objective requirements: {message}", context={"teams": self.team_manager.teams} ) # Determine required teams required_teams = [ TeamType[team.upper()] for team in analysis.get("required_teams", []) ] # Create cross-team objective objective_id = await self.team_manager.create_cross_team_objective( objective=message, required_teams=required_teams ) self.active_objectives[objective_id] = { "description": message, "teams": required_teams, "status": "initiated", "created_at": datetime.now() } return self._format_objective_creation(objective_id) async def _handle_status_request(self, message: str) -> str: """Handle status check requests.""" # Get system status system_status = await self.agentic_system.get_system_status() # Get team status team_status = {} for team_id, team in self.team_manager.teams.items(): team_status[team.name] = await self.team_manager.monitor_objective_progress(team_id) # Get objective status objective_status = {} for obj_id, obj in self.active_objectives.items(): objective_status[obj_id] = await self.team_manager.monitor_objective_progress(obj_id) return self._format_status_response(system_status, team_status, objective_status) async def _handle_general_chat(self, message: str) -> str: """Handle general chat interactions with error recovery.""" try: # Get responses from all teams responses = [] for team_type in self.team_manager.teams: try: response = await self._get_team_response(team_type, message) if response: responses.append(response) except Exception as e: logger.error(f"Error getting response from team {team_type}: {str(e)}") continue if not responses: return "I'm having trouble processing that request. Could you please rephrase it?" # Combine responses return self._format_team_responses(responses) except Exception as e: logger.error(f"Error in general chat handler: {str(e)}") return "I apologize, but I'm having trouble understanding that. Could you try asking in a different way?" async def _get_team_response(self, team_type: TeamType, query: str) -> Dict[str, Any]: """Get response from a specific team.""" team_id = next( (tid for tid, team in self.team_manager.teams.items() if team.type == team_type), None ) if not team_id: return { "team": team_type.value, "response": "Team not available", "confidence": 0.0 } # Get team agents team_agents = self.team_manager.agents[team_id] # Aggregate responses from team agents responses = [] for agent in team_agents.values(): agent_response = await agent.process_query(query) responses.append(agent_response) # Combine responses combined_response = self._combine_agent_responses(responses) return { "team": team_type.value, "response": combined_response, "confidence": sum(r.get("confidence", 0) for r in responses) / len(responses) } def _combine_agent_responses(self, responses: List[Dict[str, Any]]) -> str: """Combine multiple agent responses into a coherent response.""" # Sort by confidence valid_responses = [ r for r in responses if r.get("success", False) and r.get("response") ] if not valid_responses: return "No valid response available" sorted_responses = sorted( valid_responses, key=lambda x: x.get("confidence", 0), reverse=True ) # Take the highest confidence response best_response = sorted_responses[0] return best_response.get("response", "No response available") def _format_team_responses(self, responses: List[Dict[str, Any]]) -> str: """Format team responses into a readable message.""" formatted = [] for response in responses: if response.get("confidence", 0) > 0.3: # Confidence threshold formatted.append( f"Team {response['team'].title()}:\n" f"{response['response']}\n" ) if not formatted: return "No team was able to provide a confident response." return "\n".join(formatted) def _format_objective_creation(self, objective_id: str) -> str: """Format objective creation response.""" objective = self.active_objectives[objective_id] return ( f"Objective created successfully!\n\n" f"Objective ID: {objective_id}\n" f"Description: {objective['description']}\n" f"Assigned Teams: {', '.join(t.value for t in objective['teams'])}\n" f"Status: {objective['status']}\n" f"Created: {objective['created_at'].strftime('%Y-%m-%d %H:%M:%S')}" ) def _format_status_response( self, system_status: Dict[str, Any], team_status: Dict[str, Any], objective_status: Dict[str, Any] ) -> str: """Format status response.""" # Format system status status = [ "System Status:", f"- State: {system_status['state']}", f"- Active Agents: {system_status['agent_count']}", f"- Active Tasks: {system_status['active_tasks']}", "\nTeam Status:" ] # Add team status for team_name, team_info in team_status.items(): status.extend([ f"\n{team_name}:", f"- Active Agents: {team_info['active_agents']}", f"- Completion Rate: {team_info['completion_rate']:.2%}", f"- Collaboration Score: {team_info['collaboration_score']:.2f}" ]) # Add objective status if objective_status: status.append("\nActive Objectives:") for obj_id, obj_info in objective_status.items(): obj = self.active_objectives[obj_id] status.extend([ f"\n{obj['description']}:", f"- Status: {obj['status']}", f"- Teams: {', '.join(t.value for t in obj['teams'])}", f"- Progress: {sum(t['completion_rate'] for t in obj_info.values())/len(obj_info):.2%}" ]) return "\n".join(status) class VentureUI: def __init__(self, app): self.app = app def create_interface(self): with gr.Blocks(theme=gr.themes.Soft()) as interface: gr.Markdown(""" # Advanced Agentic System Chat Interface Chat with our autonomous agent teams: - Team A: Coders (App/Software Developers) - Team B: Business (Entrepreneurs) - Team C: Research (Deep Online Research) - Team D: Crypto & Sports Trading You can: 1. Ask questions 2. Create new objectives 3. Check status of teams and objectives 4. Get insights and recommendations """) chatbot = gr.Chatbot(label="Chat History", height=400) with gr.Row(): msg = gr.Textbox( label="Message", placeholder="Chat with the Agentic System...", lines=2, scale=9 ) submit = gr.Button("Send", scale=1) clear = gr.ClearButton([msg, chatbot], value="Clear") async def respond(message, history): try: # Convert history to the format expected by process_message history_list = [[x, y] for x, y in history] if history else [] response, _ = await self.app(message, history_list) history.append((message, response)) return "", history except Exception as e: logger.error(f"Error in chat response: {str(e)}") error_msg = "I apologize, but I encountered an error. Please try again." history.append((message, error_msg)) return "", history submit.click( respond, [msg, chatbot], [msg, chatbot], queue=False ).then( lambda: gr.update(interactive=True), None, [submit], queue=False ) msg.submit( respond, [msg, chatbot], [msg, chatbot], queue=False ).then( lambda: gr.update(interactive=True), None, [submit], queue=False ) return interface def create_chat_interface() -> gr.Blocks: """Create Gradio chat interface.""" chat = ChatInterface() ui = VentureUI(chat.process_message) return ui.create_interface() # Create and launch the interface interface = create_chat_interface() if __name__ == "__main__": interface.launch( server_name="0.0.0.0", server_port=7860, share=False )