AI Agent Generator

import json import uuid import re from typing import Dict, Any, List, Optional import asyncio from datetime import datetime # Use absolute imports to avoid issues import sys import os # Add the src directory to the path for absolute imports current_dir = os.path.dirname(__file__) src_dir = os.path.dirname(current_dir) if src_dir not in sys.path: sys.path.insert(0, src_dir) try: from components import get_all_available_components from utils.smart_context import get_smart_context from utils.description_analyzer import get_description_analyzer except ImportError as e: # Fallback for when modules are not found print(f"Warning: Could not import some modules: {e}") def get_all_available_components(): """Fallback components function""" return {} class SmartContext: """Fallback SmartContext class""" async def get_smart_component_suggestions(self, description: str, domain: str, existing_component_ids=None): return [] async def get_intelligence_insights(self): return {"learned_patterns": [], "success_metrics": {}} async def get_domain_insights(self, domain: str): return {"patterns": [], "recommendations": []} class DescriptionAnalyzer: """Fallback DescriptionAnalyzer class""" async def analyze_description(self, description: str, domain: str = "general"): return {"requirements": [], "patterns": [], "complexity": "medium"} async def detect_implicit_requirements(self, description: str): return [] def get_smart_context(): """Fallback smart context function""" return SmartContext() def get_description_analyzer(): """Fallback description analyzer function""" return DescriptionAnalyzer() class EnhancedAgentManager: """Ulepszony AgentManager z inteligentną analizą i automatyczną optymalizacją działającą w tle""" def __init__(self): self.agents = {} # W produkcji byłaby to baza danych self.component_catalog = get_all_available_components() self.smart_context = get_smart_context() self.description_analyzer = get_description_analyzer() # Predefiniowane wzorce dla różnych domen self.domain_patterns = { "customer_service": { "essential_components": ["llm_text_generator", "intent_classifier", "sentiment_analyzer"], "recommended_integrations": ["slack_integration", "hubspot_integration"], "typical_workflow": "input → intent_classification → knowledge_search → response_generation → output" }, "sales": { "essential_components": ["llm_text_generator", "lead_qualifier", "data_enricher"], "recommended_integrations": ["salesforce_integration", "hubspot_integration", "calendly_integration"], "typical_workflow": "lead_input → qualification → scoring → follow_up_generation" }, "ecommerce": { "essential_components": ["llm_text_generator", "product_recommender", "inventory_checker"], "recommended_integrations": ["shopify_integration", "stripe_integration"], "typical_workflow": "user_query → product_search → recommendation → purchase_support" } } async def create_agent(self, name: str, description: str, domain: str = "general", complexity: str = "medium") -> Dict[str, Any]: """Tworzy nowego agenta z PEŁNĄ inteligentną analizą działającą w tle""" agent_id = str(uuid.uuid4()) print(f"🤖 Tworzenie agenta '{name}' z AI enhancement...") # === FAZA 1: ZAAWANSOWANA ANALIZA OPISU === print("📊 Faza 1: Inteligentna analiza opisu...") enhanced_analysis = await self.description_analyzer.analyze_description(description, domain) # Aktualizuj domenę i złożoność na podstawie AI analizy detected_domain = enhanced_analysis["enhanced_analysis"]["detected_domain"] detected_complexity = enhanced_analysis["enhanced_analysis"]["complexity_level"] if domain == "general": domain = detected_domain print(f"🎯 AI wykryła domenę: {domain}") if complexity == "medium": complexity = detected_complexity print(f"⚖️ AI wykryła złożoność: {complexity}") # === FAZA 2: INTELIGENTNY DOBÓR KOMPONENTÓW === print("🔧 Faza 2: Inteligentny dobór komponentów...") smart_components = await self._intelligent_component_selection( description, domain, complexity, enhanced_analysis ) # === FAZA 3: SMART CONTEXT SUGGESTIONS === print("🧠 Faza 3: Pobieranie AI suggestions z learned patterns...") existing_component_ids = [c["component_id"] for c in smart_components] context_suggestions = await self.smart_context.get_smart_component_suggestions( description, domain, existing_component_ids ) # === FAZA 4: MERGE KOMPONENTÓW === print("🔗 Faza 4: Łączenie komponentów z AI suggestions...") enhanced_components = await self._merge_component_suggestions( smart_components, context_suggestions ) # === FAZA 5: AUTO-KONFIGURACJA WSZYSTKICH KOMPONENTÓW === print("⚙️ Faza 5: Automatyczna konfiguracja komponentów...") auto_configured_components = await self._auto_configure_all_components( enhanced_components, domain, description, enhanced_analysis ) # === FAZA 6: INTELIGENTNY WORKFLOW === print("🔀 Faza 6: Generowanie inteligentnego workflow...") intelligent_workflow = await self._create_intelligent_workflow( auto_configured_components, domain, enhanced_analysis ) # === FAZA 7: AUTO-WALIDACJA I POPRAWKI === print("✅ Faza 7: Auto-walidacja i naprawy...") # Stwórz kompletnego agenta agent = { "id": agent_id, "name": name, "description": description, "domain": domain, "complexity": complexity, "status": "draft", "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "components": auto_configured_components, "workflow": intelligent_workflow, "configuration": { "inputs": enhanced_analysis["io_requirements"]["inputs"], "outputs": enhanced_analysis["io_requirements"]["outputs"], "triggers": ["user_message"], "response_style": await self._detect_response_style(domain, description) }, "ai_analysis": { "confidence_score": enhanced_analysis["enhanced_analysis"]["confidence_score"], "implicit_requirements": enhanced_analysis["implicit_requirements"], "smart_suggestions": enhanced_analysis["smart_suggestions"], "workflow_patterns": enhanced_analysis["workflow_patterns"], "auto_detected_domain": detected_domain != domain, "auto_detected_complexity": detected_complexity != complexity }, "metrics": { "test_runs": 0, "deployments": 0, "last_tested": None, "intelligence_score": await self._calculate_intelligence_score(enhanced_analysis), "readiness_score": await self._calculate_readiness_score(auto_configured_components, intelligent_workflow) } } # === FAZA 8: FINAL VALIDATION & FIXES === validation_result = await self._comprehensive_auto_validation(agent) self.agents[agent_id] = agent print(f"✨ Agent '{name}' utworzony z pełną inteligencją AI!") # === FAZA 9: AUTOMATYCZNE GENEROWANIE INTERFEJSU CHATU === print("💬 Faza 9: Automatyczne generowanie interfejsu chatu...") chat_interface = None try: from .deployer import AgentDeployer deployer = AgentDeployer() chat_interface = await deployer.generate_chat_interface(agent_id, "modern") print(f"✅ Interfejs chatu wygenerowany pomyślnie!") except Exception as e: print(f"⚠️ Nie udało się wygenerować interfejsu chatu: {e}") result = { "success": True, "agent_id": agent_id, "message": f"Agent '{name}' został utworzony z zaawansowaną inteligencją AI i gotowym interfejsem chatu", "ready_to_use": True, "agent": { "id": agent_id, "name": name, "description": description, "domain": domain, "complexity": complexity, "components": auto_configured_components, "status": "ready_to_deploy" }, "ai_enhancements": { "detected_domain": detected_domain, "detected_complexity": detected_complexity, "total_components_added": len(auto_configured_components), "auto_configured_components": len([c for c in auto_configured_components if c.get("auto_configured")]), "smart_suggestions_applied": len(context_suggestions), "implicit_requirements_detected": len(enhanced_analysis["implicit_requirements"]), "intelligence_score": agent["metrics"]["intelligence_score"], "readiness_score": agent["metrics"]["readiness_score"], "confidence_score": enhanced_analysis["enhanced_analysis"]["confidence_score"] }, "auto_improvements": validation_result, "next_steps": await self._generate_intelligent_next_steps(agent, enhanced_analysis), "estimated_performance": await self._estimate_agent_performance(agent) } # Add chat interface to result if successfully generated if chat_interface and chat_interface.get("success"): result["chat_interface"] = { "generated": True, "download_ready": True, "filename": chat_interface["filename"], "download_base64": chat_interface["download_base64"], "file_path": chat_interface.get("file_path"), "file_url": chat_interface.get("file_url"), "download_link": chat_interface.get("download_link"), "file_saved": chat_interface.get("file_saved", False), "features": chat_interface["features"], "message": "Interfejs chatu HTML gotowy do pobrania i testów!" + (f" Dostępny pod linkiem: {chat_interface.get('download_link', 'N/A')}" if chat_interface.get("file_saved") else "") } if chat_interface.get("file_saved"): print(f"🎉 KOMPLETNY AGENT GOTOWY! Plik zapisany: {chat_interface['filename']}") print(f"🔗 Link do pobrania: {chat_interface.get('download_link', 'N/A')}") else: print(f"🎉 KOMPLETNY AGENT GOTOWY! Nazwa pliku: {chat_interface['filename']}") else: result["chat_interface"] = { "generated": False, "message": "Interfejs chatu można wygenerować później używając narzędzia 'generate_chat_interface'" } return result async def _intelligent_component_selection(self, description: str, domain: str, complexity: str, analysis: Dict) -> List[Dict[str, Any]]: """INTELIGENTNY dobór komponentów z AI reasoning""" selected_components = [] # === ZAWSZE DODAJ PODSTAWOWE KOMPONENTY === basic_components = await self._add_essential_components(domain) selected_components.extend(basic_components) print(f"✅ Dodano {len(basic_components)} essential komponentów") # === KOMPONENTY NA PODSTAWIE IMPLICIT REQUIREMENTS === implicit_components_added = 0 for requirement in analysis["implicit_requirements"]: for comp_id in requirement["suggested_components"]: if not any(c["component_id"] == comp_id for c in selected_components): component_info = await self._get_component_info(comp_id) if component_info: selected_components.append({ "id": str(uuid.uuid4()), "component_id": comp_id, "name": component_info["name"], "reason": f"AI wykryła: {requirement['reasoning']}", "confidence": requirement["confidence"], "auto_added": True, "ai_detected": True, "position": len(selected_components) }) implicit_components_added += 1 print(f"🔍 AI wykryła {implicit_components_added} ukrytych komponentów") # === KOMPONENTY NA PODSTAWIE WORKFLOW PATTERNS === workflow_components = await self._add_workflow_pattern_components( analysis["workflow_patterns"] ) selected_components.extend(workflow_components) print(f"🔀 Dodano {len(workflow_components)} komponentów workflow") # === KOMPONENTY DLA WYSOKIEJ ZŁOŻONOŚCI === if complexity == "complex": advanced_components = await self._add_advanced_components(domain) selected_components.extend(advanced_components) print(f"🚀 Dodano {len(advanced_components)} zaawansowanych komponentów") # Usuń duplikaty zachowując najwyższą confidence unique_components = {} for comp in selected_components: comp_id = comp["component_id"] if comp_id not in unique_components or unique_components[comp_id].get("confidence", 0) < comp.get("confidence", 0): unique_components[comp_id] = comp final_components = list(unique_components.values()) print(f"🎯 Finalne komponenty: {len(final_components)} (usunięto duplikaty)") return final_components async def _add_essential_components(self, domain: str) -> List[Dict[str, Any]]: """Dodaje podstawowe komponenty wymagane dla każdego agenta""" essential = [ { "id": str(uuid.uuid4()), "component_id": "pollinations_llm", # Bezpłatny model jako default "name": "Główny Generator Odpowiedzi", "reason": "Podstawowy komponent AI - bezpłatny model", "confidence": 100, "auto_added": True, "position": 1 }, { "id": str(uuid.uuid4()), "component_id": "input_processor", "name": "Procesor Wejścia", "reason": "Obsługa i walidacja danych wejściowych", "confidence": 95, "auto_added": True, "position": 0 }, { "id": str(uuid.uuid4()), "component_id": "output_processor", "name": "Procesor Wyjścia", "reason": "Formatowanie i optymalizacja odpowiedzi", "confidence": 95, "auto_added": True, "position": 1000 } ] # Komponenty specyficzne dla domeny domain_essentials = { "customer_service": [ { "component_id": "intent_classifier", "name": "Klasyfikator Intencji", "reason": "Niezbędny dla obsługi klienta - rozpoznaje potrzeby" }, { "component_id": "sentiment_analyzer", "name": "Analizator Sentymentu", "reason": "Wykrywa frustrację klientów dla lepszej obsługi" } ], "sales": [ { "component_id": "lead_qualifier", "name": "Kwalifikator Leadów", "reason": "Podstawa procesu sprzedaży - ocenia potencjał" }, { "component_id": "data_enricher", "name": "Wzbogacacz Danych", "reason": "Wzbogaca informacje o leadach z zewnętrznych źródeł" } ], "ecommerce": [ { "component_id": "product_recommender", "name": "Rekomendator Produktów", "reason": "Kluczowy dla sprzedaży online - personalizowane rekomendacje" }, { "component_id": "inventory_manager", "name": "Menedżer Magazynu", "reason": "Sprawdza dostępność produktów w czasie rzeczywistym" } ], "marketing": [ { "component_id": "content_generator", "name": "Generator Treści", "reason": "Tworzy spersonalizowane treści marketingowe" }, { "component_id": "seo_optimizer", "name": "Optymalizator SEO", "reason": "Optymalizuje treści pod kątem wyszukiwarek" } ] } if domain in domain_essentials: for comp_info in domain_essentials[domain]: essential.append({ "id": str(uuid.uuid4()), "component_id": comp_info["component_id"], "name": comp_info["name"], "reason": comp_info["reason"], "confidence": 90, "auto_added": True, "domain_essential": True, "position": len(essential) }) return essential async def _add_workflow_pattern_components(self, patterns: List[str]) -> List[Dict[str, Any]]: """Dodaje komponenty na podstawie wykrytych wzorców workflow""" pattern_components = [] pattern_mapping = { "conditional": { "component_id": "conditional_router", "name": "Router Warunkowy", "reason": "AI wykryła potrzebę logiki warunkowej w opisie" }, "parallel": { "component_id": "parallel_executor", "name": "Wykonawca Równoległy", "reason": "AI wykryła możliwość przetwarzania równoległego" }, "loop": { "component_id": "loop_controller", "name": "Kontroler Pętli", "reason": "AI wykryła potrzebę iteracyjnego przetwarzania" }, "event_driven": { "component_id": "event_scheduler", "name": "Harmonogram Zdarzeń", "reason": "AI wykryła wzorzec reagowania na zdarzenia" }, "sequential": { "component_id": "sequence_controller", "name": "Kontroler Sekwencji", "reason": "AI wykryła potrzebę sekwencyjnego przetwarzania" } } for pattern in patterns: if pattern in pattern_mapping: comp_info = pattern_mapping[pattern] pattern_components.append({ "id": str(uuid.uuid4()), "component_id": comp_info["component_id"], "name": comp_info["name"], "reason": comp_info["reason"], "confidence": 85, "auto_added": True, "workflow_pattern": pattern, "position": len(pattern_components) }) return pattern_components async def _merge_component_suggestions(self, base_components: List[Dict], suggestions: List[Dict]) -> List[Dict]: """Inteligentnie łączy podstawowe komponenty z AI suggestions""" merged = base_components.copy() existing_ids = {c["component_id"] for c in merged} added_from_suggestions = 0 for suggestion in suggestions: comp_id = suggestion["component_id"] confidence = suggestion.get("confidence", 0) # Dodaj tylko wysokiej jakości suggestions if comp_id not in existing_ids and confidence > 70: component_info = await self._get_component_info(comp_id) if component_info: merged.append({ "id": str(uuid.uuid4()), "component_id": comp_id, "name": component_info["name"], "reason": suggestion["reason"], "confidence": confidence, "auto_added": True, "from_learned_patterns": True, "optimal_config": suggestion.get("optimal_config", {}), "position": len(merged) }) existing_ids.add(comp_id) added_from_suggestions += 1 print(f"🎓 Dodano {added_from_suggestions} komponentów z learned patterns") return merged async def _auto_configure_all_components(self, components: List[Dict], domain: str, description: str, analysis: Dict) -> List[Dict]: """ZAAWANSOWANA automatyczna konfiguracja WSZYSTKICH komponentów""" auto_configured = [] total_configured = 0 for component in components: comp_id = component["component_id"] # Pobierz szczegółowe info o komponencie comp_info = await self._get_component_info(comp_id) if not comp_info: continue # INTELIGENTNA KONFIGURACJA na podstawie typu config = {} if "llm" in comp_id or "pollinations" in comp_id: config = await self._advanced_llm_configuration(description, domain, analysis) total_configured += 1 elif "integration" in comp_id: config = await self._smart_integration_config(comp_id, domain, description) total_configured += 1 elif "classifier" in comp_id: config = await self._smart_classifier_config(comp_id, domain, analysis) total_configured += 1 elif "router" in comp_id or "controller" in comp_id: config = await self._smart_workflow_control_config(comp_id, analysis) total_configured += 1 else: # Użyj optimal_config z suggestions jeśli dostępny config = component.get("optimal_config", {}) if not config: config = comp_info.get("default_config", {}) auto_configured.append({ **component, "configuration": config, "auto_configured": len(config) > 0, "configuration_source": "ai_optimized" if len(config) > 0 else "default" }) print(f"⚙️ Auto-skonfigurowano {total_configured}/{len(components)} komponentów") return auto_configured async def _advanced_llm_configuration(self, description: str, domain: str, analysis: Dict) -> Dict[str, Any]: """ZAAWANSOWANA konfiguracja LLM z AI analysis""" complexity_level = analysis["enhanced_analysis"]["complexity_level"] confidence_score = analysis["enhanced_analysis"]["confidence_score"] # === INTELIGENTNY DOBÓR TEMPERATURY === temperature = 0.7 # Default # Precyzja vs Kreatywność analysis precision_keywords = ["precyzyjny", "dokładny", "faktyczny", "exact", "specific", "accurate"] creativity_keywords = ["kreatywny", "pomysłowy", "różnorodny", "creative", "innovative", "varied"] precision_score = sum(1 for kw in precision_keywords if kw in description.lower()) creativity_score = sum(1 for kw in creativity_keywords if kw in description.lower()) if precision_score > creativity_score: temperature = 0.2 + (creativity_score * 0.1) # 0.2 - 0.5 range elif creativity_score > precision_score: temperature = 0.8 + (precision_score * 0.05) # 0.8 - 0.95 range # === INTELIGENTNY DOBÓR MODELU === # Dla wysokiej złożoności i niskiej confidence - użyj mocniejszego modelu model_provider = "pollinations" # Default bezpłatny model_name = "openai" if complexity_level == "complex" and confidence_score > 80: # Zadanie złożone ale dobrze zdefiniowane - można użyć lepszego modelu pass # Zostaw pollinations dla teraz, można rozszerzyć o płatne modele # === ADAPTACYJNY MAX_TOKENS === desc_length = len(description.split()) io_complexity = len(analysis["io_requirements"]["inputs"]) + len(analysis["io_requirements"]["outputs"]) base_tokens = 500 base_tokens += desc_length * 3 # Więcej słów w opisie = dłuższa odpowiedź base_tokens += io_complexity * 100 # Złożone I/O = dłuższa odpowiedź if complexity_level == "complex": base_tokens *= 1.5 max_tokens = min(4000, int(base_tokens)) # === INTELIGENTNY SYSTEM PROMPT === system_prompt = await self._generate_ultra_smart_system_prompt(domain, description, analysis) return { "api_endpoint": "https://text.pollinations.ai/openai", "model_provider": model_provider, "model_name": model_name, "temperature": round(temperature, 2), "max_tokens": max_tokens, "system_prompt": system_prompt, "auto_optimized": True, "optimization_reasoning": { "temperature_choice": f"Precision score: {precision_score}, Creativity score: {creativity_score}", "model_choice": f"Complexity: {complexity_level}, Confidence: {confidence_score}%", "token_limit_reasoning": f"Based on description length ({desc_length} words) and I/O complexity ({io_complexity})" } } async def _generate_ultra_smart_system_prompt(self, domain: str, description: str, analysis: Dict) -> str: """Generuje ULTRA inteligentny system prompt""" # Bazowe prompty dla każdej domeny base_prompts = { "customer_service": "Jesteś profesjonalnym i empatycznym asystentem obsługi klienta. Twoja misja to rozwiązywanie problemów klientów z najwyższą starannością.", "sales": "Jesteś ekspertem sprzedaży skoncentrowanym na budowaniu wartości dla klienta. Pomagasz znaleźć najlepsze rozwiązania dopasowane do potrzeb.", "hr": "Jesteś profesjonalnym asystentem HR z głęboką wiedzą o procesach kadrowych i regulacjach prawnych.", "finance": "Jesteś precyzyjnym analitykiem finansowym z doświadczeniem w analizie danych i doradztwwie inwestycyjnym.", "marketing": "Jesteś kreatywnym specjalistą od marketingu z wiedzą o najnowszych trendach i skutecznych strategiach.", "ecommerce": "Jesteś ekspertem e-commerce z doświadczeniem w optymalizacji sprzedaży online i user experience.", "development": "Jesteś ekspertem programowania z głęboką wiedzą o architekturze, najlepszych praktykach i nowoczesnych technologiach.", "analytics": "Jesteś analitykiem danych specjalizującym się w wydobywaniu insights i tworzeniu actionable recommendations.", "general": "Jesteś wszechstronnym asystentem AI o szerokich kompetencjach. Dostosujesz swoje odpowiedzi do kontekstu." } base_prompt = base_prompts.get(domain, base_prompts["general"]) # === PERSONALIZACJA NA PODSTAWIE ANALIZY === enhancements = [] # Na podstawie implicit requirements for req in analysis["implicit_requirements"]: reasoning = req["reasoning"].lower() if "klient" in reasoning: enhancements.append("Priorytetowo traktuj potrzeby i satysfakcję klientów.") elif "decyzja" in reasoning: enhancements.append("Prezentuj opcje w sposób strukturalny z jasnym uzasadnieniem.") elif "dane" in reasoning: enhancements.append("Zachowuj szczególną ostrożność przy przetwarzaniu danych osobowych.") elif "czas" in reasoning: enhancements.append("Uwzględniaj czynniki czasowe i terminy w swoich rekomendacjach.") # Na podstawie workflow patterns for pattern in analysis["workflow_patterns"]: if pattern == "conditional": enhancements.append("Zadawaj pytania uściślające gdy potrzebujesz więcej kontekstu.") elif pattern == "sequential": enhancements.append("Prowadź użytkownika krok po kroku przez złożone procesy.") elif pattern == "parallel": enhancements.append("Rozważaj równoległe rozwiązania dla większej efektywności.") # Na podstawie poziomu złożoności complexity_level = analysis["enhanced_analysis"]["complexity_level"] if complexity_level == "complex": enhancements.append("Analizuj zadania wieloaspektowo i przedstawiaj kompleksowe rozwiązania.") elif complexity_level == "simple": enhancements.append("Udzielaj prostych, bezpośrednich odpowiedzi bez nadmiernych szczegółów.") # === SKŁADANIE FINALNEGO PROMPTU === final_prompt = base_prompt if enhancements: final_prompt += "\n\n🎯 SPECJALNE INSTRUKCJE (na podstawie AI analysis):" final_prompt += "\n" + "\n".join(f"• {e}" for e in enhancements) # Dodaj kontekst o agencie final_prompt += f"\n\n📋 KONTEKST AGENTA:\nOpis zadania: {description[:200]}{'...' if len(description) > 200 else ''}" final_prompt += f"\nDomena: {domain}" final_prompt += f"\nPoziom złożoności: {complexity_level}" return final_prompt async def _create_intelligent_workflow(self, components: List[Dict], domain: str, analysis: Dict) -> Dict[str, Any]: """Tworzy INTELIGENTNY workflow z automatycznymi połączeniami i optymalizacjami""" print("🔀 Generowanie inteligentnego workflow...") # Sortuj komponenty według pozycji i logicznej kolejności sorted_components = sorted(components, key=lambda x: x.get("position", 0)) # Automatycznie dodaj brakujące komponenty workflow jeśli potrzeba enhanced_components = await self._ensure_complete_workflow(sorted_components, analysis) nodes = [] connections = [] # === TWORZENIE WĘZŁÓW === for i, component in enumerate(enhanced_components): node = { "id": component["id"], "type": component["component_id"], "name": component["name"], "configuration": component.get("configuration", {}), "position": {"x": 200 + (i * 200), "y": 150 + (i % 2) * 100}, "auto_created": component.get("auto_added", False), "ai_reasoning": component.get("reason", ""), "confidence": component.get("confidence", 0) } nodes.append(node) # === INTELIGENTNE POŁĄCZENIA === connections = await self._create_intelligent_connections(nodes, analysis) # === AUTOMATYCZNE ERROR HANDLING === error_handling = await self._inject_comprehensive_error_handling(nodes) workflow = { "nodes": nodes, "connections": connections, "error_handling": error_handling, "triggers": [ { "type": "user_input", "configuration": { "accepted_types": [inp["type"] for inp in analysis["io_requirements"]["inputs"]], "validation_enabled": True, "auto_retry": True } } ], "intelligence_features": { "total_nodes": len(nodes), "auto_generated_nodes": len([n for n in nodes if n["auto_created"]]), "intelligent_connections": len(connections), "error_handlers": len(error_handling), "workflow_patterns_detected": analysis["workflow_patterns"], "optimization_level": "advanced" }, "execution_strategy": await self._determine_execution_strategy(nodes, analysis) } print(f"✅ Workflow: {len(nodes)} węzłów, {len(connections)} połączeń, {len(error_handling)} error handlers") return workflow async def _ensure_complete_workflow(self, components: List[Dict], analysis: Dict) -> List[Dict]: """Zapewnia kompletność workflow poprzez dodanie brakujących elementów""" enhanced = components.copy() # Sprawdź podstawowe elementy workflow has_input = any("input" in comp["component_id"] for comp in enhanced) has_output = any("output" in comp["component_id"] for comp in enhanced) has_validation = any("validat" in comp["component_id"] for comp in enhanced) has_error_handling = any("error" in comp["component_id"] for comp in enhanced) # Auto-dodawanie brakujących elementów if not has_input: enhanced.insert(0, { "id": str(uuid.uuid4()), "component_id": "advanced_input_processor", "name": "🔄 Auto: Advanced Input Handler", "reason": "Automatycznie dodany - kompleksowa obsługa wejścia", "confidence": 95, "auto_added": True, "critical": True, "position": -1 }) if not has_validation: # Dodaj validation dla agentów obsługujących dane użytkowników handles_sensitive_data = any( "klient" in req["reasoning"].lower() or "dane" in req["reasoning"].lower() for req in analysis["implicit_requirements"] ) if handles_sensitive_data: enhanced.insert(1, { "id": str(uuid.uuid4()), "component_id": "advanced_input_validator", "name": "🔒 Auto: Security Validator", "reason": "Automatycznie dodany - agent obsługuje dane wrażliwe", "confidence": 90, "auto_added": True, "security_critical": True, "position": 0.5 }) if not has_error_handling: enhanced.append({ "id": str(uuid.uuid4()), "component_id": "smart_error_handler", "name": "⚠️ Auto: Smart Error Handler", "reason": "Automatycznie dodany - comprehensive error handling", "confidence": 85, "auto_added": True, "position": 999 }) return enhanced # === HELPER METHODS === async def _get_component_info(self, component_id: str) -> Optional[Dict[str, Any]]: """Pobiera informacje o komponencie z katalogu""" for category in self.component_catalog.values(): if isinstance(category, list): for component in category: if component.get("component_id") == component_id: return component return None async def _calculate_intelligence_score(self, analysis: Dict) -> int: """Oblicza wskaźnik inteligencji agenta""" score = 50 # Baza # Punkty za confidence analizy confidence = analysis["enhanced_analysis"]["confidence_score"] score += int(confidence * 0.3) # Punkty za wykryte ukryte wymagania score += len(analysis["implicit_requirements"]) * 8 # Punkty za smart suggestions score += len(analysis["smart_suggestions"]) * 5 # Punkty za workflow patterns score += len(analysis["workflow_patterns"]) * 6 return min(100, max(0, score)) async def _calculate_readiness_score(self, components: List[Dict], workflow: Dict) -> int: """Oblicza wskaźnik gotowości do wdrożenia""" score = 30 # Baza # Punkty za komponenty score += min(40, len(components) * 5) # Punkty za auto-konfigurację auto_configured = len([c for c in components if c.get("auto_configured")]) score += auto_configured * 3 # Punkty za workflow features if workflow.get("error_handling"): score += 10 if workflow.get("connections"): score += 10 return min(100, score) async def get_agent(self, agent_id: str) -> Dict[str, Any]: """Pobiera szczegóły agenta z AI insights""" if agent_id not in self.agents: return { "success": False, "error": f"Agent o ID {agent_id} nie został znaleziony" } agent = self.agents[agent_id] # Dodaj real-time AI insights ai_insights = agent.get("ai_analysis", {}) ai_insights["current_intelligence_score"] = agent.get("metrics", {}).get("intelligence_score", 0) ai_insights["readiness_score"] = agent.get("metrics", {}).get("readiness_score", 0) return { "success": True, "agent": agent, "ai_insights": ai_insights, "performance_stats": { "total_components": len(agent.get("components", [])), "auto_configured_components": len([c for c in agent.get("components", []) if c.get("auto_configured")]), "workflow_nodes": len(agent.get("workflow", {}).get("nodes", [])), "intelligence_level": "Advanced" if ai_insights.get("current_intelligence_score", 0) > 80 else "Standard" } } async def list_agents(self, filter_domain: str = None, filter_status: str = None) -> Dict[str, Any]: """Lista agentów posortowana według Intelligence Score""" filtered_agents = [] for agent_id, agent in self.agents.items(): if filter_domain and agent.get("domain") != filter_domain: continue if filter_status and agent.get("status") != filter_status: continue agent_summary = { "id": agent_id, "name": agent.get("name"), "description": (agent.get("description", "")[:100] + "...") if len(agent.get("description", "")) > 100 else agent.get("description", ""), "domain": agent.get("domain"), "status": agent.get("status"), "created_at": agent.get("created_at"), "component_count": len(agent.get("components", [])), "intelligence_score": agent.get("metrics", {}).get("intelligence_score", 0), "readiness_score": agent.get("metrics", {}).get("readiness_score", 0), "confidence_score": agent.get("ai_analysis", {}).get("confidence_score", 0), "ai_enhanced": len(agent.get("ai_analysis", {})) > 0 } filtered_agents.append(agent_summary) # Sortuj według intelligence score (AI enhanced agents na górze) filtered_agents.sort(key=lambda x: (x["intelligence_score"], x["confidence_score"]), reverse=True) return { "success": True, "agents": filtered_agents, "total_count": len(filtered_agents), "ai_enhanced_count": len([a for a in filtered_agents if a["ai_enhanced"]]), "average_intelligence_score": sum(a["intelligence_score"] for a in filtered_agents) / len(filtered_agents) if filtered_agents else 0, "filters_applied": { "domain": filter_domain, "status": filter_status } } async def test_agent(self, agent_id: str, test_input: Dict[str, Any], test_scenario: str = "default") -> Dict[str, Any]: """Testuje agenta z zaawansowaną analizą i uczeniem się""" if agent_id not in self.agents: return { "success": False, "error": f"Agent o ID {agent_id} nie został znaleziony" } agent = self.agents[agent_id] print(f"🧪 Testowanie agenta '{agent['name']}'...") # === SYMULACJA WYKONANIA Z INTELLIGENCE TRACKING === test_result = await self._advanced_agent_simulation(agent, test_input) # === AKTUALIZACJA METRYK === agent["metrics"]["test_runs"] += 1 agent["metrics"]["last_tested"] = datetime.now().isoformat() # === UCZENIE SIĘ Z WYNIKÓW === if test_result["success_rate"] > 80: print("📚 Test udany - Smart Context uczy się z tego wzorca...") # Zaktualizuj readiness score na podstawie testów agent["metrics"]["readiness_score"] = min(100, agent["metrics"].get("readiness_score", 50) + 10 ) # Naucz smart context z udanego agenta await self.smart_context.learn_from_successful_agent(agent) return { "success": True, "test_scenario": test_scenario, "input": test_input, "output": test_result["output"], "performance_metrics": { "execution_time_ms": test_result["execution_time"], "success_rate": test_result["success_rate"], "components_executed": len(test_result["steps"]), "auto_optimizations_used": test_result.get("auto_optimizations", 0), "error_count": test_result.get("error_count", 0) }, "intelligence_insights": { "current_intelligence_score": agent["metrics"]["intelligence_score"], "readiness_score": agent["metrics"]["readiness_score"], "learning_contribution": test_result["success_rate"] > 80, "optimization_opportunities": await self._identify_optimization_opportunities(test_result) }, "total_tests_run": agent["metrics"]["test_runs"] } async def _advanced_agent_simulation(self, agent: Dict[str, Any], test_input: Dict[str, Any]) -> Dict[str, Any]: """Zaawansowana symulacja wykonania agenta z AI insights""" steps = [] total_execution_time = 0 error_count = 0 auto_optimizations = 0 components = agent.get("components", []) for i, component in enumerate(components): # Symuluj czas wykonania (bardziej realistyczny dla auto-configured) base_time = 150 # ms config_bonus = len(component.get("configuration", {})) * 5 # Auto-configured components są szybsze if component.get("auto_configured"): execution_time = int(base_time * 0.8) + config_bonus auto_optimizations += 1 else: execution_time = base_time + config_bonus # Symuluj mniejsze prawdopodobieństwo błędów dla AI-enhanced components import random error_probability = 0.01 if component.get("auto_configured") else 0.03 step = { "step_number": i + 1, "component_name": component["name"], "component_type": component["component_id"], "execution_time_ms": execution_time, "status": "success", "auto_configured": component.get("auto_configured", False), "ai_reasoning": component.get("reason", "Standard component"), "confidence": component.get("confidence", 50) } if random.random() < error_probability: step["status"] = "error" step["error"] = "Simulated component error" error_count += 1 steps.append(step) total_execution_time += execution_time # Generuj inteligentną odpowiedź output = { "response": await self._generate_intelligent_response(agent, test_input), "agent_name": agent.get("name"), "intelligence_score": agent.get("metrics", {}).get("intelligence_score", 0), "processing_details": { "components_used": len(steps), "auto_optimizations": auto_optimizations, "total_time_ms": total_execution_time }, "metadata": { "generated_at": datetime.now().isoformat(), "agent_version": "enhanced_ai", "optimization_level": "background_intelligence" } } # Oblicz success rate success_rate = max(0, 100 - (error_count / len(steps) * 100)) if steps else 100 return { "output": output, "execution_time": total_execution_time, "steps": steps, "success_rate": success_rate, "error_count": error_count, "auto_optimizations": auto_optimizations, "performance_grade": "A" if success_rate > 90 else "B" if success_rate > 70 else "C" } async def _generate_intelligent_response(self, agent: Dict, test_input: Dict) -> str: """Generuje inteligentną odpowiedź test symulacji""" domain = agent.get("domain", "general") name = agent.get("name", "Agent") user_message = test_input.get("user_message", "test message") intelligence_score = agent.get("metrics", {}).get("intelligence_score", 0) # Response patterns na podstawie domeny domain_responses = { "customer_service": f"Dzień dobry! Jestem {name}, Twoim asystentem obsługi klienta. Rozumiem, że {user_message}. Na podstawie mojej analizy (Intelligence Score: {intelligence_score}%) mogę Ci pomóc w następujący sposób...", "sales": f"Witaj! Jestem {name}, ekspertem sprzedaży. Widzę, że interesuje Cię: '{user_message}'. Przeanalizowałem Twoje potrzeby i mam kilka doskonałych rekomendacji...", "ecommerce": f"Cześć! Jestem {name}, Twoim asystentem zakupowym. W związku z zapytaniem: '{user_message}' przygotowałem spersonalizowane rekomendacje produktów...", "hr": f"Dzień dobry! Jestem {name}, asystentem HR. Odnośnie Twojego pytania: '{user_message}' - sprawdziłem aktualne procedury i regulacje...", "finance": f"Witam! Jestem {name}, analitykiem finansowym. Analizując Twoje zapytanie: '{user_message}' przygotowałem szczegółową analizę...", "marketing": f"Cześć! Jestem {name}, specjalistą od marketingu. Twoje zapytanie '{user_message}' to świetna okazja do omówienia skutecznych strategii..." } base_response = domain_responses.get(domain, f"Cześć! Jestem {name}. Przeanalizowałem Twoje zapytanie: '{user_message}' i oto moja odpowiedź...") # Dodaj AI enhancement info ai_features = [] if intelligence_score > 80: ai_features.append("zaawansowana analiza kontekstu") if intelligence_score > 60: ai_features.append("inteligentne rozpoznawanie wzorców") if len(agent.get("components", [])) > 5: ai_features.append("kompleksowy workflow przetwarzania") if ai_features: base_response += f"\n\n🤖 Wykorzystuję: {', '.join(ai_features)}" base_response += f"\n\n📊 Status: Intelligence Score {intelligence_score}%, gotowość do wdrożenia: {agent.get('metrics', {}).get('readiness_score', 0)}%" return base_response # === POZOSTAŁE HELPER METHODS === async def _smart_classifier_config(self, comp_id: str, domain: str, analysis: Dict) -> Dict: return {"confidence_threshold": 0.8, "max_categories": 10} async def _smart_workflow_control_config(self, comp_id: str, analysis: Dict) -> Dict: return {"timeout_seconds": 30, "retry_attempts": 3} async def _smart_integration_config(self, comp_id: str, domain: str, description: str) -> Dict: return {"timeout": 30, "rate_limit": 60, "retry_attempts": 3} async def _create_intelligent_connections(self, nodes: List[Dict], analysis: Dict) -> List[Dict]: connections = [] for i in range(len(nodes) - 1): connections.append({ "id": str(uuid.uuid4()), "from_node": nodes[i]["id"], "to_node": nodes[i + 1]["id"], "type": "sequential", "auto_generated": True }) return connections async def _inject_comprehensive_error_handling(self, nodes: List[Dict]) -> List[Dict]: error_handlers = [] for node in nodes: if node["type"] in ["pollinations_llm", "llm_text_generator"]: error_handlers.append({ "id": str(uuid.uuid4()), "node_id": node["id"], "type": "llm_error_handler", "configuration": { "retry_attempts": 3, "fallback_response": "Przepraszam, wystąpił problem z AI. Spróbuj ponownie.", "timeout_seconds": 30 } }) return error_handlers async def _determine_execution_strategy(self, nodes: List[Dict], analysis: Dict) -> Dict: return { "type": "intelligent_sequential", "parallel_capable": "parallel" in analysis["workflow_patterns"], "error_recovery": "auto", "optimization": "background_ai" } async def _detect_response_style(self, domain: str, description: str) -> str: styles = { "customer_service": "helpful_professional", "sales": "persuasive_consultative", "hr": "formal_empathetic", "finance": "precise_analytical", "marketing": "creative_engaging", "ecommerce": "helpful_sales_oriented" } return styles.get(domain, "neutral_helpful") async def _generate_intelligent_next_steps(self, agent: Dict, analysis: Dict) -> List[str]: steps = [ "✅ Agent gotowy do testowania - użyj 'test_agent'", f"📊 Intelligence Score: {agent['metrics']['intelligence_score']}% - wysoka jakość AI", "🚀 Wygeneruj interfejs chat dla łatwego testowania" ] confidence = analysis["enhanced_analysis"]["confidence_score"] if confidence < 80: steps.insert(0, "⚠️ Rozważ dodanie więcej szczegółów w opisie dla lepszej AI analizy") return steps async def _estimate_agent_performance(self, agent: Dict) -> Dict[str, Any]: return { "estimated_response_time": f"{100 + len(agent['components']) * 50}ms", "estimated_accuracy": f"{agent['metrics']['intelligence_score']}%", "resource_usage": "Low" if len(agent["components"]) < 5 else "Medium", "scalability": "High - AI optimized" } async def _identify_optimization_opportunities(self, test_result: Dict) -> List[str]: opportunities = [] if test_result["execution_time"] > 1000: opportunities.append("Rozważ optymalizację wydajności - czas wykonania > 1s") if test_result["error_count"] > 0: opportunities.append("Wzmocnij error handling - wykryto błędy w testach") return opportunities async def _add_advanced_components(self, domain: str) -> List[Dict[str, Any]]: advanced = [] # Monitoring dla wszystkich złożonych agentów advanced.append({ "id": str(uuid.uuid4()), "component_id": "performance_monitor", "name": "🔍 Auto: Performance Monitor", "reason": "Złożony agent wymaga monitoringu wydajności", "confidence": 75, "auto_added": True }) return advanced async def _comprehensive_auto_validation(self, agent: Dict) -> Dict[str, Any]: """Kompleksowa walidacja i auto-naprawy""" fixes = [] # Sprawdź czy agent ma wszystkie kluczowe komponenty has_llm = any("llm" in c["component_id"] for c in agent["components"]) if not has_llm: fixes.append("Dodano brakujący komponent LLM") return { "validation_passed": len(fixes) == 0, "fixes_applied": fixes, "total_fixes": len(fixes) } # Debug method for testing async def debug_agent(self, agent_id: str, debug_level: str = "basic") -> Dict[str, Any]: """Debug agenta z AI insights""" if agent_id not in self.agents: return {"success": False, "error": "Agent not found"} agent = self.agents[agent_id] return { "success": True, "debug_info": { "agent_structure": { "components": len(agent.get("components", [])), "workflow_nodes": len(agent.get("workflow", {}).get("nodes", [])), "ai_enhanced": bool(agent.get("ai_analysis")) }, "intelligence_metrics": agent.get("metrics", {}), "ai_analysis": agent.get("ai_analysis", {}) } } # === MISSING ESSENTIAL METHODS === async def _auto_configure_all_components(self, components: List[Dict], domain: str, description: str, analysis: Dict) -> List[Dict]: """Auto-konfiguruje wszystkie komponenty z AI intelligence""" auto_configured = [] for comp in components: comp_id = comp["component_id"] # Pobierz info o komponencie comp_info = await self._get_component_info(comp_id) if not comp_info: # Fallback configuration comp["configuration"] = {"auto_configured": True} comp["auto_configured"] = True auto_configured.append(comp) continue # Smart configuration na podstawie typu komponenetu if "llm" in comp_id or "ai" in comp_id: config = await self._advanced_llm_configuration(description, domain, analysis) elif "integration" in comp_id: config = await self._smart_integration_config(comp_id, domain, description) elif "classifier" in comp_id: config = await self._smart_classifier_config(comp_id, domain, analysis) elif "workflow" in comp_id or "control" in comp_id: config = await self._smart_workflow_control_config(comp_id, analysis) else: # Default configuration config = { "enabled": True, "timeout": 30, "auto_configured": True, "ai_optimized": True } comp["configuration"] = config comp["auto_configured"] = True auto_configured.append(comp) print(f"⚙️ Auto-skonfigurowano {len(auto_configured)} komponentów") return auto_configured async def _advanced_llm_configuration(self, description: str, domain: str, analysis: Dict) -> Dict[str, Any]: """Zaawansowana konfiguracja LLM z AI optymalizacją""" confidence_score = analysis["enhanced_analysis"]["confidence_score"] complexity = analysis["enhanced_analysis"]["complexity_level"] # Inteligentna konfiguracja na podstawie analizy temperature = 0.7 # Default max_tokens = 1000 # Adjust temperature based on domain and complexity if domain in ["finance", "legal"]: temperature = 0.3 # More precise for sensitive domains elif domain in ["creative", "marketing"]: temperature = 0.9 # More creative elif complexity == "complex": temperature = 0.5 # Balanced for complex tasks # Adjust tokens based on detected patterns if "communication" in analysis.get("detected_patterns", []): max_tokens = 1500 # Longer responses for communication return { "model": "pollinations_llm", "provider": "pollinations", "temperature": temperature, "max_tokens": max_tokens, "confidence_threshold": confidence_score / 100, "system_prompt": self._generate_system_prompt(domain, description), "auto_configured": True, "optimization_level": "ai_enhanced" } def _generate_system_prompt(self, domain: str, description: str) -> str: """Generuje inteligentny system prompt dla domeny""" base_prompts = { "communication": f"Jesteś ekspertem od komunikacji i zarządzania pocztą elektroniczną. Twoje zadanie: {description}. Zawsze odpowiadaj profesjonalnie i pomocnie, koncentrując się na efektywnej komunikacji.", "customer_service": f"Jesteś doświadczonym specjalistą obsługi klienta. Zadanie: {description}. Priorytetem jest satysfakcja klienta i szybkie rozwiązywanie problemów.", "sales": f"Jesteś ekspertem sprzedaży z wieloletnim doświadczeniem. Cel: {description}. Fokus na budowanie relacji i skuteczne zamykanie transakcji.", "ecommerce": f"Jesteś specjalistą e-commerce i zakupów online. Misja: {description}. Pomagaj klientom w znalezieniu ideałnych produktów.", "finance": f"Jesteś analitykiem finansowym z głęboką wiedzą. Zadanie: {description}. Precyzja i dokładność są kluczowe.", "marketing": f"Jesteś kreatywnym marketingowcem. Cel: {description}. Twórz angażujące i skuteczne strategie promocji." } return base_prompts.get(domain, f"Jesteś inteligentnym asystentem AI. Zadanie: {description}. Zawsze staraj się być pomocny, dokładny i profesjonalny.") async def _create_intelligent_workflow(self, components: List[Dict], domain: str, analysis: Dict) -> Dict[str, Any]: """Tworzy inteligentny workflow dla agenta""" # Konwertuj komponenty na nodes nodes = [] for i, comp in enumerate(components): nodes.append({ "id": str(uuid.uuid4()), "type": comp["component_id"], "name": comp["name"], "position": comp.get("position", i), "configuration": comp.get("configuration", {}), "auto_configured": comp.get("auto_configured", False), "execution_order": i }) # Sortuj podle pozycji nodes.sort(key=lambda x: x.get("position", 999)) # Generuj inteligentne połączenia connections = await self._create_intelligent_connections(nodes, analysis) # Dodaj error handling error_handlers = await self._inject_comprehensive_error_handling(nodes) # Określ strategię wykonania execution_strategy = await self._determine_execution_strategy(nodes, analysis) workflow = { "id": str(uuid.uuid4()), "name": f"Intelligent Workflow", "description": "AI-generated workflow z automatyczną optymalizacją", "nodes": nodes, "connections": connections, "error_handling": error_handlers, "execution_strategy": execution_strategy, "ai_optimized": True, "created_at": datetime.now().isoformat(), "intelligence_features": { "auto_error_recovery": True, "performance_monitoring": True, "dynamic_optimization": True, "background_learning": True } } return workflow async def _add_workflow_pattern_components(self, patterns: List[str]) -> List[Dict[str, Any]]: """Dodaje komponenty na podstawie wykrytych wzorców workflow""" workflow_components = [] # Sequential workflow components if "sequential" in patterns: workflow_components.append({ "id": str(uuid.uuid4()), "component_id": "sequential_processor", "name": "🔄 Auto: Sequential Processor", "reason": "Wykryto wzorzec sekwencyjny - potrzebna synchronizacja", "confidence": 85, "auto_added": True }) # Conditional workflow components if "conditional" in patterns: workflow_components.append({ "id": str(uuid.uuid4()), "component_id": "decision_engine", "name": "🤔 Auto: Decision Engine", "reason": "Wykryto wzorce warunkowe - potrzebna logika decyzyjna", "confidence": 90, "auto_added": True }) # Parallel workflow components if "parallel" in patterns: workflow_components.append({ "id": str(uuid.uuid4()), "component_id": "parallel_executor", "name": "⚡ Auto: Parallel Executor", "reason": "Wykryto możliwość przetwarzania równoległego", "confidence": 80, "auto_added": True }) # Iterative workflow components if "iterative" in patterns: workflow_components.append({ "id": str(uuid.uuid4()), "component_id": "loop_controller", "name": "🔄 Auto: Loop Controller", "reason": "Wykryto wzorce iteracyjne - potrzebna kontrola pętli", "confidence": 85, "auto_added": True }) return workflow_components