""" 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 os import gradio as gr from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware import uvicorn from typing import Dict, Any, List, Tuple, Optional import logging from pathlib import Path import asyncio from datetime import datetime import json from requests.adapters import HTTPAdapter, Retry from dataclasses import dataclass from agentic_system import AgenticSystem from orchestrator import AgentOrchestrator from team_management import TeamManager, TeamType from reasoning import ( UnifiedReasoningEngine, StrategyType, UnifiedResult ) from api.openai_compatible import OpenAICompatibleAPI from api.venture_api import VentureAPI # 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: # Update chat history self.chat_history = history # Process message response = await self._handle_message(message) # Update history if response: 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 an error. Please try again." 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 intent intent = await self._analyze_intent(message) intent_type = self._get_intent_type(intent) # Route to appropriate handler 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 handling: {str(e)}") return "I apologize, but I encountered an error processing your message. Please try again." def _get_intent_type(self, intent) -> str: """Safely extract intent type from various result formats.""" if isinstance(intent, dict): return intent.get("type", "general") 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 analysis = await self.orchestrator.reasoning_engine.reason( query=message, context={ "chat_history": self.chat_history, "active_objectives": self.active_objectives } ) return { "type": analysis.get("intent_type", "general"), "confidence": analysis.get("confidence", 0.5), "entities": analysis.get("entities", []), "action_required": analysis.get("action_required", False) } 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.""" try: # 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: response = await self._get_team_response(team_type, message) if response: responses.append(response) if not responses: return "I apologize, but I couldn't find a relevant answer to your query." # Combine and format responses return self._format_team_responses(responses) except Exception as e: logger.error(f"Error handling query: {str(e)}") return "I apologize, but I encountered an error processing your query. Please try again." async def _handle_objective(self, message: str) -> str: """Handle new objective creation.""" try: # Create new objective objective_id = await self.team_manager.create_objective(message) if not objective_id: return "I apologize, but I couldn't create the objective. Please try again." # Format and return response return self._format_objective_creation(objective_id) except Exception as e: logger.error(f"Error creating objective: {str(e)}") return "I apologize, but I encountered an error creating the objective. Please try again." async def _handle_status_request(self, message: str) -> str: """Handle status check requests.""" try: # 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) except Exception as e: logger.error(f"Error getting status: {str(e)}") return "I apologize, but I encountered an error getting the status. Please try again." async def _handle_general_chat(self, message: str) -> str: """Handle general chat interactions with error recovery.""" try: # Use reasoning engine for response generation response = await self.orchestrator.reasoning_engine.reason( query=message, context={ "chat_history": self.chat_history, "system_state": await self.agentic_system.get_system_status() } ) if not response or not response.get("response"): return "I apologize, but I couldn't generate a meaningful response. Please try again." return response["response"] except Exception as e: logger.error(f"Error in general chat: {str(e)}") return "I apologize, but I encountered an error processing your message. Please try again." async def _get_team_response(self, team_type: TeamType, query: str) -> Dict[str, Any]: """Get response from a specific team.""" try: team = self.team_manager.teams.get(team_type.value) if not team: return None # Get response from team's agents responses = [] for agent in team.agents: response = await agent.process_query(query) if response: responses.append(response) if not responses: return None # Return best response return self._combine_agent_responses(responses) except Exception as e: logger.error(f"Error getting team response: {str(e)}") return None def _combine_agent_responses(self, responses: List[Dict[str, Any]]) -> Dict[str, Any]: """Combine multiple agent responses into a coherent response.""" try: # Sort by confidence valid_responses = [ r for r in responses if r.get("success", False) and r.get("response") ] if not valid_responses: return None sorted_responses = sorted( valid_responses, key=lambda x: x.get("confidence", 0), reverse=True ) # Take the highest confidence response return sorted_responses[0] except Exception as e: logger.error(f"Error combining responses: {str(e)}") return None def _format_team_responses(self, responses: List[Dict[str, Any]]) -> str: """Format team responses into a readable message.""" try: if not responses: return "No team responses available." formatted = [] for resp in responses: if resp and resp.get("response"): team_name = resp.get("team_name", "Unknown Team") confidence = resp.get("confidence", 0) formatted.append( f"\n{team_name} (Confidence: {confidence:.2%}):\n{resp['response']}" ) if not formatted: return "No valid team responses available." return "\n".join(formatted) except Exception as e: logger.error(f"Error formatting responses: {str(e)}") return "Error formatting team responses." def _format_objective_creation(self, objective_id: str) -> str: """Format objective creation response.""" try: obj = self.active_objectives.get(objective_id) if not obj: return "Objective created but details not available." return "\n".join([ "New Objective Created:", f"Description: {obj['description']}", f"Status: {obj['status']}", f"Assigned Teams: {', '.join(t.value for t in obj['teams'])}" ]) except Exception as e: logger.error(f"Error formatting objective: {str(e)}") return "Error formatting objective details." def _format_status_response( self, system_status: Dict[str, Any], team_status: Dict[str, Any], objective_status: Dict[str, Any] ) -> str: """Format status response.""" try: # 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) except Exception as e: logger.error(f"Error formatting status: {str(e)}") return "Error formatting status information." class VentureUI: def __init__(self, app): self.app = app def create_interface(self): """Create the Gradio interface.""" with gr.Blocks( theme=gr.themes.Soft(), analytics_enabled=False, title="Advanced Agentic System" ) as interface: # Verify Gradio version gr.Markdown(f""" # Advanced Agentic System Chat Interface v{gr.__version__} 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, bubble_full_width=False, show_copy_button=True, render_markdown=True ) with gr.Row(): msg = gr.Textbox( label="Message", placeholder="Chat with the Agentic System...", lines=2, scale=9, autofocus=True, container=True ) submit = gr.Button( "Send", scale=1, variant="primary" ) with gr.Row(): clear = gr.ClearButton( [msg, chatbot], value="Clear Chat", variant="secondary", scale=1 ) retry = gr.Button( "Retry Last", variant="secondary", scale=1 ) 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, history_list = await self.app(message, history_list) # Update history if history is None: history = [] 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." if history is None: history = [] history.append((message, error_msg)) return "", history async def retry_last(history): if not history: return history last_user_msg = history[-1][0] history = history[:-1] # Remove last exchange return await respond(last_user_msg, history) msg.submit( respond, [msg, chatbot], [msg, chatbot], api_name="chat" ).then( lambda: gr.update(interactive=True), None, [msg, submit], queue=False ) submit.click( respond, [msg, chatbot], [msg, chatbot], api_name="submit" ).then( lambda: gr.update(interactive=True), None, [msg, submit], queue=False ) retry.click( retry_last, [chatbot], [chatbot], api_name="retry" ) # Event handlers for better UX msg.change(lambda x: gr.update(interactive=bool(x.strip())), [msg], [submit]) # Add example inputs gr.Examples( examples=[ "What can Team A (Coders) help me with?", "Create a new objective: Analyze market trends", "What's the status of all teams?", "Give me insights about recent developments" ], inputs=msg, label="Example Queries" ) return interface def create_chat_interface() -> gr.Blocks: """Create Gradio chat interface.""" chat = ChatInterface() ui = VentureUI(chat.process_message) return ui.create_interface() # Initialize FastAPI app = FastAPI( title="Advanced Agentic System", description="Venture Strategy Optimizer with OpenAI-compatible API", version="1.0.0" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include OpenAI-compatible routes from api.openai_compatible import OpenAICompatibleAPI reasoning_engine = UnifiedReasoningEngine() openai_api = OpenAICompatibleAPI(reasoning_engine) app.include_router(openai_api.router, tags=["OpenAI Compatible"]) # Original API routes @app.get("/api/health") async def health_check(): """Health check endpoint.""" return { "status": "healthy", "version": "1.0.0", "endpoints": { "openai_compatible": "/v1/chat/completions", "venture": "/api/venture", "ui": "/" } } @app.post("/api/reason") async def reason(query: str, context: Optional[Dict[str, Any]] = None): """Reasoning endpoint.""" try: result = await reasoning_engine.reason(query, context or {}) return result except Exception as e: logger.error(f"Reasoning error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/venture/analyze") async def analyze_venture( venture_type: str, description: str, metrics: Optional[Dict[str, Any]] = None ): """Venture analysis endpoint.""" try: result = await VentureAPI(reasoning_engine).analyze_venture( venture_type=venture_type, description=description, metrics=metrics or {} ) return result except Exception as e: logger.error(f"Analysis error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/venture/types") async def get_venture_types(): """Get available venture types.""" return VentureAPI(reasoning_engine).get_venture_types() # Create Gradio interface interface = create_chat_interface() # Mount Gradio app to FastAPI app = gr.mount_gradio_app(app, interface, path="/") if __name__ == "__main__": # Run with uvicorn when called directly uvicorn.run( "app:app", host="0.0.0.0", port=7860, reload=True, workers=4 )