Spaces:
Runtime error
Runtime error
File size: 3,934 Bytes
dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 3921722 dcb2a99 |
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 |
"""
Advanced Agentic System Interface
-------------------------------
Provides an interface to interact with the autonomous agent system
using local LLM for improved performance.
"""
import gradio as gr
import asyncio
from typing import Dict, Any, List
import json
from datetime import datetime
import logging
import os
import socket
import requests
from requests.adapters import HTTPAdapter, Retry
from agentic_system import AgenticSystem
from team_management import TeamManager
from orchestrator import AgentOrchestrator
from reasoning.unified_engine import UnifiedReasoningEngine
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Configure network settings
TIMEOUT = int(os.getenv('REQUESTS_TIMEOUT', '30'))
RETRIES = 3
# Configure session with retries
session = requests.Session()
retries = Retry(total=RETRIES,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
session.mount('http://', HTTPAdapter(max_retries=retries))
def check_network():
"""Check network connectivity."""
try:
# Try to resolve huggingface.co
socket.gethostbyname('huggingface.co')
# Try to connect to the API
response = session.get('https://huggingface.co/api/health',
timeout=TIMEOUT)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Network check failed: {e}")
return False
class AgentInterface:
"""Interface for the agentic system."""
def __init__(self):
"""Initialize the interface components."""
# Check network connectivity
if not check_network():
logger.warning("Network connectivity issues detected")
self.orchestrator = AgentOrchestrator()
self.reasoning_engine = UnifiedReasoningEngine(
min_confidence=0.7,
parallel_threshold=3,
learning_rate=0.1
)
async def process_query(self, message: str) -> str:
"""Process user query through the reasoning system."""
try:
# Prepare context
context = {
'timestamp': datetime.now().isoformat(),
'objective': 'Provide helpful and accurate responses',
'mode': 'analytical'
}
# Get response from reasoning engine
result = await self.reasoning_engine.reason(
query=message,
context=context
)
return result.get('answer', 'No response generated')
except Exception as e:
logger.error(f"Error processing query: {e}")
return f"An error occurred: {str(e)}"
# Initialize interface
interface = AgentInterface()
# Create Gradio interface
with gr.Blocks(title="Advanced Reasoning System") as demo:
gr.Markdown("# Advanced Reasoning System")
with gr.Row():
with gr.Column():
input_text = gr.Textbox(
label="Your Query",
placeholder="Enter your query here...",
lines=3
)
submit_btn = gr.Button("Submit")
with gr.Column():
output_text = gr.Textbox(
label="Response",
lines=5
)
submit_btn.click(
fn=interface.process_query,
inputs=input_text,
outputs=output_text
)
# Start the server with health check endpoint
app = gr.mount_gradio_app(None, demo, path="/")
@app.get("/health")
async def health_check():
"""Health check endpoint."""
if check_network():
return {"status": "healthy"}
return {"status": "degraded", "reason": "network_issues"}
# Launch the interface
if __name__ == "__main__":
demo.launch()
|