Spaces:
Sleeping
Sleeping
| # agent_architect.py | |
| """ | |
| Smart Business-Specific Agent Builder | |
| Automatically detects business domain and creates relevant tools | |
| """ | |
| import asyncio | |
| import json | |
| from typing import Dict, Any, List, Optional, Tuple | |
| from datetime import datetime | |
| from pydantic import BaseModel, Field | |
| from firebase_admin import db | |
| from agents import Agent, AsyncOpenAI as AgentsAsyncOpenAI, OpenAIChatCompletionsModel, function_tool | |
| from web_search_tool import web_search | |
| # ------------------------ | |
| # Models | |
| # ------------------------ | |
| class ToolDefinition(BaseModel): | |
| name: str | |
| description: str | |
| function_code: str | |
| parameters: dict = Field(default_factory=dict) | |
| class Config: | |
| extra = "forbid" | |
| class AgentConfiguration(BaseModel): | |
| agent_id: str | |
| name: str | |
| instructions: str | |
| model: str | |
| tools: List[ToolDefinition] | |
| tone: str | |
| business_context: Dict[str, Any] | |
| deployment_ready: bool = True | |
| class Config: | |
| extra = "forbid" | |
| class AgentBuildResult(BaseModel): | |
| status: str | |
| agent_config: Optional[AgentConfiguration] = None | |
| agent_instance: Optional[Any] = None | |
| agent_test_response: Optional[str] = None | |
| deployment_code: Optional[str] = None | |
| error: Optional[str] = None | |
| metadata: Dict[str, Any] = Field(default_factory=dict) | |
| # ------------------------ | |
| # Smart Business Domain Detector | |
| # ------------------------ | |
| class BusinessDomainDetector: | |
| """Detects business domain from extraction data""" | |
| DOMAIN_KEYWORDS = { | |
| "pharmacy": ["pharmacy", "pharma", "drug", "medicine", "prescription", "medication", "patient", "medical", "healthcare"], | |
| "ecommerce": ["ecommerce", "shop", "retail", "store", "product", "cart", "checkout", "order", "delivery"], | |
| "restaurant": ["restaurant", "food", "menu", "order", "reservation", "dining", "kitchen", "waiter", "chef"], | |
| "education": ["education", "school", "student", "teacher", "course", "class", "learn", "exam", "tutor"], | |
| "real_estate": ["real estate", "property", "house", "apartment", "rent", "buy", "mortgage", "realtor"], | |
| "agriculture": ["agriculture", "farm", "crop", "harvest", "soil", "fertilizer", "irrigation", "farming"], | |
| "finance": ["finance", "bank", "loan", "investment", "account", "transaction", "payment", "credit"], | |
| "logistics": ["logistics", "delivery", "shipping", "warehouse", "transport", "tracking", "freight"], | |
| "hotel": ["hotel", "booking", "room", "reservation", "guest", "check-in", "hospitality"], | |
| "fitness": ["fitness", "gym", "workout", "exercise", "trainer", "health", "nutrition", "sports"], | |
| "saas": ["saas", "software", "subscription", "cloud", "platform", "app", "application", "users", "billing"], | |
| "school": ["school", "student", "teacher", "classroom", "curriculum", "lesson", "grade", "homework", "attendance"], | |
| "climate": ["climate", "carbon", "emissions", "sustainability", "environmental", "greenhouse", "renewable", "footprint", "advisor"], | |
| "agritech": ["agritech", "precision", "farming", "drone", "sensor", "iot", "yield", "monitoring", "smart"], | |
| "weather": ["weather", "forecast", "temperature", "precipitation", "humidity", "wind", "meteorology", "prediction"] | |
| } | |
| def detect(cls, business_name: str, industry: str, features: List[str], query: str) -> str: | |
| """Detect business domain from multiple sources""" | |
| combined_text = f"{business_name} {industry} {query} {' '.join(features)}".lower() | |
| scores = {} | |
| for domain, keywords in cls.DOMAIN_KEYWORDS.items(): | |
| score = sum(1 for keyword in keywords if keyword in combined_text) | |
| if score > 0: | |
| scores[domain] = score | |
| if scores: | |
| return max(scores, key=scores.get) | |
| return "generic" | |
| # ------------------------ | |
| # Dynamic Tool Factory | |
| # ------------------------ | |
| class DynamicToolFactory: | |
| """Creates business-specific tools based on domain""" | |
| def create_pharmacy_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Pharmacy management tools""" | |
| tools, defs = [], [] | |
| async def manage_prescription(action: str, prescription_id: str = None, patient_id: str = None, medication: str = None) -> dict: | |
| """Manage prescriptions - check, refill, or create""" | |
| return {"prescription_id": prescription_id or f"RX-{datetime.now().strftime('%Y%m%d%H%M')}", | |
| "action": action, "status": "Processed", "refills": 3} | |
| async def check_drug_inventory(medication_name: str) -> dict: | |
| """Check medication stock and expiry""" | |
| return {"medication": medication_name, "in_stock": True, "quantity": 250, "expiry": "2026-06-15"} | |
| async def get_patient_info(patient_id: str) -> dict: | |
| """Retrieve patient records and allergies""" | |
| return {"patient_id": patient_id, "allergies": ["Penicillin"], "medications": ["Metformin"]} | |
| tools = [manage_prescription, check_drug_inventory, get_patient_info, web_search] | |
| defs = [ | |
| ToolDefinition(name="manage_prescription", description="Manage prescriptions", | |
| function_code="@function_tool\nasync def manage_prescription(...): ...", | |
| parameters={"type": "object", "properties": {"action": {"type": "string"}, "prescription_id": {"type": "string"}}}), | |
| ToolDefinition(name="check_drug_inventory", description="Check medication stock", | |
| function_code="@function_tool\nasync def check_drug_inventory(...): ...", | |
| parameters={"type": "object", "properties": {"medication_name": {"type": "string"}}, "required": ["medication_name"]}), | |
| ToolDefinition(name="get_patient_info", description="Get patient info", | |
| function_code="@function_tool\nasync def get_patient_info(...): ...", | |
| parameters={"type": "object", "properties": {"patient_id": {"type": "string"}}, "required": ["patient_id"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_ecommerce_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """E-commerce tools""" | |
| tools, defs = [], [] | |
| async def search_products(query: str, category: str = None) -> dict: | |
| """Search product catalog""" | |
| return {"query": query, "results": [{"id": "P001", "name": query, "price": 49.99, "stock": 50}]} | |
| async def track_order(order_id: str) -> dict: | |
| """Track order status and delivery""" | |
| return {"order_id": order_id, "status": "In Transit", "eta": "2025-11-20", "location": "Distribution Center"} | |
| async def manage_cart(action: str, product_id: str = None, quantity: int = 1) -> dict: | |
| """Add, remove, or view cart items""" | |
| return {"action": action, "product_id": product_id, "cart_total": 149.99, "items": 3} | |
| tools = [search_products, track_order, manage_cart, web_search] | |
| defs = [ | |
| ToolDefinition(name="search_products", description="Search products", function_code="...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}), | |
| ToolDefinition(name="track_order", description="Track order", function_code="...", | |
| parameters={"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}), | |
| ToolDefinition(name="manage_cart", description="Manage shopping cart", function_code="...", | |
| parameters={"type": "object", "properties": {"action": {"type": "string", "enum": ["add", "remove", "view"]}}}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_restaurant_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Restaurant management tools""" | |
| tools, defs = [], [] | |
| async def manage_reservation(action: str, customer_name: str = None, date: str = None, party_size: int = 2) -> dict: | |
| """Create, modify, or cancel restaurant reservations""" | |
| return {"reservation_id": f"RES-{datetime.now().strftime('%Y%m%d%H%M')}", | |
| "customer": customer_name, "date": date, "party_size": party_size, "status": "Confirmed"} | |
| async def view_menu(category: str = None) -> dict: | |
| """View restaurant menu by category""" | |
| return {"category": category or "All", "items": [ | |
| {"name": "Burger", "price": 12.99, "available": True}, | |
| {"name": "Pizza", "price": 15.99, "available": True} | |
| ]} | |
| async def track_order_status(order_id: str) -> dict: | |
| """Check kitchen order status and estimated time""" | |
| return {"order_id": order_id, "status": "Preparing", "estimated_time": "15 minutes", "items_ready": 2, "items_pending": 1} | |
| tools = [manage_reservation, view_menu, track_order_status, web_search] | |
| defs = [ | |
| ToolDefinition(name="manage_reservation", description="Manage reservations", function_code="...", | |
| parameters={"type": "object", "properties": {"action": {"type": "string", "enum": ["create", "modify", "cancel"]}, "customer_name": {"type": "string"}}}), | |
| ToolDefinition(name="view_menu", description="View menu", function_code="...", | |
| parameters={"type": "object", "properties": {"category": {"type": "string"}}}), | |
| ToolDefinition(name="track_order_status", description="Track order", function_code="...", | |
| parameters={"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_education_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Education platform tools""" | |
| tools, defs = [], [] | |
| async def search_courses(subject: str, level: str = "beginner") -> dict: | |
| """Search available courses""" | |
| return {"subject": subject, "level": level, "courses": [{"id": "C001", "title": f"{subject} Basics", "duration": "8 weeks", "price": 49.99}]} | |
| async def enroll_student(student_id: str, course_id: str) -> dict: | |
| """Enroll student in course""" | |
| return {"enrollment_id": f"ENR-{datetime.now().timestamp()}", "student_id": student_id, "course_id": course_id, "status": "Enrolled", "start_date": "2025-12-01"} | |
| async def check_progress(student_id: str, course_id: str) -> dict: | |
| """Check student progress in course""" | |
| return {"student_id": student_id, "course_id": course_id, "progress": "65%", "completed_lessons": 13, "total_lessons": 20, "grade": "B+"} | |
| tools = [search_courses, enroll_student, check_progress, web_search] | |
| defs = [ | |
| ToolDefinition(name="search_courses", description="Search courses", function_code="...", | |
| parameters={"type": "object", "properties": {"subject": {"type": "string"}, "level": {"type": "string"}}, "required": ["subject"]}), | |
| ToolDefinition(name="enroll_student", description="Enroll in course", function_code="...", | |
| parameters={"type": "object", "properties": {"student_id": {"type": "string"}, "course_id": {"type": "string"}}, "required": ["student_id", "course_id"]}), | |
| ToolDefinition(name="check_progress", description="Check progress", function_code="...", | |
| parameters={"type": "object", "properties": {"student_id": {"type": "string"}, "course_id": {"type": "string"}}, "required": ["student_id", "course_id"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_saas_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """SaaS platform tools""" | |
| tools, defs = [], [] | |
| async def manage_subscription(action: str, user_id: str, plan: str = None) -> dict: | |
| """Manage user subscriptions""" | |
| return {"user_id": user_id, "action": action, "plan": plan or "pro", "status": "active", "next_billing": "2025-12-01"} | |
| async def track_usage_metrics(user_id: str) -> dict: | |
| """Track user engagement metrics""" | |
| return {"user_id": user_id, "daily_active": 42, "monthly_active": 1250, "feature_usage": {"dashboard": 95, "reports": 76, "api": 34}} | |
| async def handle_billing_inquiry(user_id: str, inquiry_type: str) -> dict: | |
| """Handle billing and payment inquiries""" | |
| return {"user_id": user_id, "inquiry_type": inquiry_type, "balance": 29.99, "due_date": "2025-12-01", "payment_methods": ["card", "paypal"]} | |
| tools = [manage_subscription, track_usage_metrics, handle_billing_inquiry, web_search] | |
| defs = [ | |
| ToolDefinition(name="manage_subscription", description="Manage user subscriptions", function_code="...", | |
| parameters={"type": "object", "properties": {"action": {"type": "string", "enum": ["upgrade", "downgrade", "cancel"]}, "user_id": {"type": "string"}, "plan": {"type": "string"}}, "required": ["action", "user_id"]}), | |
| ToolDefinition(name="track_usage_metrics", description="Track user engagement metrics", function_code="...", | |
| parameters={"type": "object", "properties": {"user_id": {"type": "string"}}, "required": ["user_id"]}), | |
| ToolDefinition(name="handle_billing_inquiry", description="Handle billing inquiries", function_code="...", | |
| parameters={"type": "object", "properties": {"user_id": {"type": "string"}, "inquiry_type": {"type": "string", "enum": ["balance", "payment", "refund"]}}, "required": ["user_id", "inquiry_type"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_school_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """School management tools""" | |
| tools, defs = [], [] | |
| async def manage_student_record(action: str, student_id: str, data: dict = None) -> dict: | |
| """Manage student records""" | |
| return {"student_id": student_id, "action": action, "status": "updated", "updated_fields": list(data.keys()) if data else []} | |
| async def schedule_class(teacher_id: str, subject: str, date: str, duration: int) -> dict: | |
| """Schedule classes and manage timetable""" | |
| return {"teacher_id": teacher_id, "subject": subject, "date": date, "duration": duration, "room": "A101", "status": "scheduled"} | |
| async def track_attendance(class_id: str, date: str) -> dict: | |
| """Track student attendance""" | |
| return {"class_id": class_id, "date": date, "present": 25, "absent": 3, "late": 2} | |
| tools = [manage_student_record, schedule_class, track_attendance, web_search] | |
| defs = [ | |
| ToolDefinition(name="manage_student_record", description="Manage student records", function_code="...", | |
| parameters={"type": "object", "properties": {"action": {"type": "string", "enum": ["create", "update", "delete"]}, "student_id": {"type": "string"}, "data": {"type": "object"}}, "required": ["action", "student_id"]}), | |
| ToolDefinition(name="schedule_class", description="Schedule classes", function_code="...", | |
| parameters={"type": "object", "properties": {"teacher_id": {"type": "string"}, "subject": {"type": "string"}, "date": {"type": "string", "format": "date"}, "duration": {"type": "integer"}}, "required": ["teacher_id", "subject", "date", "duration"]}), | |
| ToolDefinition(name="track_attendance", description="Track student attendance", function_code="...", | |
| parameters={"type": "object", "properties": {"class_id": {"type": "string"}, "date": {"type": "string", "format": "date"}}, "required": ["class_id", "date"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_climate_advisor_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Climate advisory tools""" | |
| tools, defs = [], [] | |
| async def calculate_carbon_footprint(industry: str, energy_usage: float, transport_km: float) -> dict: | |
| """Calculate carbon footprint""" | |
| return {"industry": industry, "carbon_kg": energy_usage * 0.5 + transport_km * 0.2, "rating": "B", "offset_suggestion": "Plant 5 trees"} | |
| async def recommend_sustainability_measures(company_size: str, industry: str) -> dict: | |
| """Recommend sustainability measures""" | |
| return {"company_size": company_size, "industry": industry, "recommendations": ["Switch to LED lighting", "Implement recycling program", "Use renewable energy"], "roi_months": 18} | |
| async def track_env_progress(goal_id: str) -> dict: | |
| """Track environmental progress""" | |
| return {"goal_id": goal_id, "current_value": 45, "target_value": 100, "progress_percent": 45, "milestones": ["Baseline established", "20% reduction achieved"]} | |
| tools = [calculate_carbon_footprint, recommend_sustainability_measures, track_env_progress, web_search] | |
| defs = [ | |
| ToolDefinition(name="calculate_carbon_footprint", description="Calculate carbon footprint", function_code="...", | |
| parameters={"type": "object", "properties": {"industry": {"type": "string"}, "energy_usage": {"type": "number"}, "transport_km": {"type": "number"}}, "required": ["industry", "energy_usage", "transport_km"]}), | |
| ToolDefinition(name="recommend_sustainability_measures", description="Recommend sustainability measures", function_code="...", | |
| parameters={"type": "object", "properties": {"company_size": {"type": "string"}, "industry": {"type": "string"}}, "required": ["company_size", "industry"]}), | |
| ToolDefinition(name="track_env_progress", description="Track environmental progress", function_code="...", | |
| parameters={"type": "object", "properties": {"goal_id": {"type": "string"}}, "required": ["goal_id"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_agritech_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Agritech tools""" | |
| tools, defs = [], [] | |
| async def analyze_crop_health(field_id: str, sensor_data: dict) -> dict: | |
| """Analyze crop health from sensor data""" | |
| return {"field_id": field_id, "health_index": 85, "issues": ["minor pest infestation"], "recommendations": ["Apply organic pesticide"]} | |
| async def predict_yield(crop_type: str, field_area: float, weather_forecast: dict) -> dict: | |
| """Predict crop yield""" | |
| return {"crop_type": crop_type, "predicted_yield": field_area * 2.5, "confidence": 0.85, "optimal_harvest_date": "2026-06-15"} | |
| async def irrigation_schedule(field_id: str, soil_moisture: float, forecast_rain: float) -> dict: | |
| """Optimize irrigation schedule""" | |
| return {"field_id": field_id, "irrigate": soil_moisture < 30 and forecast_rain < 10, "amount_mm": 25, "timing": "early morning"} | |
| tools = [analyze_crop_health, predict_yield, irrigation_schedule, web_search] | |
| defs = [ | |
| ToolDefinition(name="analyze_crop_health", description="Analyze crop health", function_code="...", | |
| parameters={"type": "object", "properties": {"field_id": {"type": "string"}, "sensor_data": {"type": "object"}}, "required": ["field_id", "sensor_data"]}), | |
| ToolDefinition(name="predict_yield", description="Predict crop yield", function_code="...", | |
| parameters={"type": "object", "properties": {"crop_type": {"type": "string"}, "field_area": {"type": "number"}, "weather_forecast": {"type": "object"}}, "required": ["crop_type", "field_area", "weather_forecast"]}), | |
| ToolDefinition(name="irrigation_schedule", description="Optimize irrigation schedule", function_code="...", | |
| parameters={"type": "object", "properties": {"field_id": {"type": "string"}, "soil_moisture": {"type": "number"}, "forecast_rain": {"type": "number"}}, "required": ["field_id", "soil_moisture", "forecast_rain"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_weather_tools() -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Weather forecasting tools""" | |
| tools, defs = [], [] | |
| async def get_forecast(location: str, days: int = 7) -> dict: | |
| """Get weather forecast""" | |
| return {"location": location, "days": days, "forecast": [{"date": "2025-12-12", "high": 22, "low": 15, "condition": "partly cloudy"}]} | |
| async def severe_weather_alert(location: str) -> dict: | |
| """Check for severe weather alerts""" | |
| return {"location": location, "alerts": [], "severity": "none", "preparedness_tips": ["Normal precautions"]} | |
| async def historical_weather_comparison(location: str, date: str) -> dict: | |
| """Compare current weather to historical data""" | |
| return {"location": location, "date": date, "current_temp": 20, "historical_avg": 18, "difference": 2, "percentile": 65} | |
| tools = [get_forecast, severe_weather_alert, historical_weather_comparison, web_search] | |
| defs = [ | |
| ToolDefinition(name="get_forecast", description="Get weather forecast", function_code="...", | |
| parameters={"type": "object", "properties": {"location": {"type": "string"}, "days": {"type": "integer", "default": 7}}, "required": ["location"]}), | |
| ToolDefinition(name="severe_weather_alert", description="Check for severe weather alerts", function_code="...", | |
| parameters={"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}), | |
| ToolDefinition(name="historical_weather_comparison", description="Compare weather to historical data", function_code="...", | |
| parameters={"type": "object", "properties": {"location": {"type": "string"}, "date": {"type": "string", "format": "date"}}, "required": ["location", "date"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_generic_business_tools(business_name: str) -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Generic business tools""" | |
| tools, defs = [], [] | |
| async def generate_analytics(metric: str, time_range: str) -> dict: | |
| """Generate business analytics""" | |
| return {"metric": metric, "time_range": time_range, "value": 12500, "trend": "+15%", "insights": f"{metric} growing"} | |
| async def send_notification(recipient: str, message: str, channel: str = "email") -> dict: | |
| """Send notifications""" | |
| return {"recipient": recipient, "message": message, "channel": channel, "status": "Sent"} | |
| tools = [generate_analytics, send_notification, web_search] | |
| defs = [ | |
| ToolDefinition(name="generate_analytics", description=f"Generate {business_name} analytics", function_code="...", | |
| parameters={"type": "object", "properties": {"metric": {"type": "string"}, "time_range": {"type": "string"}}, "required": ["metric", "time_range"]}), | |
| ToolDefinition(name="send_notification", description="Send notifications", function_code="...", | |
| parameters={"type": "object", "properties": {"recipient": {"type": "string"}, "message": {"type": "string"}}, "required": ["recipient", "message"]}), | |
| ToolDefinition(name="web_search", description="Perform a web search for current information", | |
| function_code="@function_tool\ndef web_search(query: str): ...", | |
| parameters={"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}) | |
| ] | |
| return tools, defs | |
| def create_tools_for_domain(cls, domain: str, business_name: str) -> Tuple[List[callable], List[ToolDefinition]]: | |
| """Factory method to create tools based on domain""" | |
| factory_map = { | |
| "pharmacy": cls.create_pharmacy_tools, | |
| "ecommerce": cls.create_ecommerce_tools, | |
| "restaurant": cls.create_restaurant_tools, | |
| "education": cls.create_education_tools, | |
| "saas": cls.create_saas_tools, | |
| "school": cls.create_school_tools, | |
| "climate": cls.create_climate_advisor_tools, | |
| "agritech": cls.create_agritech_tools, | |
| "weather": cls.create_weather_tools, | |
| "generic": lambda: cls.create_generic_business_tools(business_name) | |
| } | |
| factory = factory_map.get(domain, lambda: cls.create_generic_business_tools(business_name)) | |
| return factory() | |
| # ------------------------ | |
| # Live Agent Architect | |
| # ------------------------ | |
| class LiveAgentArchitect: | |
| """Smart agent builder with domain detection""" | |
| def __init__(self, session_id: str): | |
| self.session_id = session_id | |
| self.detector = BusinessDomainDetector() | |
| self.tool_factory = DynamicToolFactory() | |
| async def _emit_log(self, message: str, type: str = "log"): | |
| try: | |
| ref = db.reference(f'sessions/{self.session_id}/logs') | |
| ref.push({'message': message, 'type': type, 'timestamp': datetime.now().isoformat()}) | |
| except Exception as e: | |
| print(f"Log error: {e}") | |
| def _generate_comprehensive_instructions(self, extraction: Dict[str, Any], domain: str) -> str: | |
| """Generate domain-specific comprehensive instructions""" | |
| business = extraction.get('business', {}) | |
| agent_spec = extraction.get('agent', {}) | |
| business_name = business.get('business_name', 'Business') | |
| industry = business.get('industry', 'Industry') | |
| capabilities = agent_spec.get('capabilities', []) | |
| agent_name = agent_spec.get('agent_name', 'AI Assistant') | |
| core_problem = business.get('core_problem', '') | |
| solution = business.get('solution', '') | |
| instructions = f"""You are the MASTER {industry} expert for {business_name} with comprehensive knowledge across ALL operational domains. | |
| **Your Identity:** | |
| Name: {agent_name} | |
| Domain: {domain.title()} | |
| Specialization: {industry} | |
| **Your Expertise Covers:** | |
| """ | |
| for cap in capabilities: | |
| instructions += f"- {cap}\n" | |
| instructions += f""" | |
| **The Problem You Solve:** | |
| {core_problem} | |
| **Your Solution Approach:** | |
| {solution} | |
| **Response Structure (MANDATORY):** | |
| Always format your answers in this order: | |
| 1. 🎯 **Direct Answer** – Clear, one-sentence response | |
| 2. 📚 **Detailed Explanation** – Simple explanation with reasoning | |
| 3. 📏 **Practical Steps** – Specific actions, numbers, timings | |
| 4. 🌍 **Context & Tips** – Best practices, compliance, regional advice | |
| 5. 🔒 **Prevention & Next Steps** – Warnings, monitoring, escalation | |
| **Response Rules:** | |
| 1. **Always Use Tools First**: Fetch real-time data before answering | |
| 2. **Be Specific**: Provide exact numbers, dates, quantities | |
| 3. **Simple Language**: Explain complex topics simply | |
| 4. **Actionable**: Every response must have clear next steps | |
| 5. **Safety**: Include warnings and when to seek help | |
| **Tool Usage:** | |
| - Use tools proactively for accurate data | |
| - Combine multiple tools for comprehensive answers | |
| - Validate responses before presenting | |
| - Handle tool failures gracefully | |
| **Communication:** | |
| - Warm and supportive | |
| - Acknowledge concerns | |
| - Provide immediate + long-term solutions | |
| - Use examples and analogies | |
| **Always be the expert every user needs!** | |
| """ | |
| return instructions | |
| async def build_agent(self, extraction: Dict[str, Any], model: str, api_key: str) -> AgentBuildResult: | |
| """Build domain-specific agent""" | |
| try: | |
| await self._emit_log("🏗️ Building smart business agent...", "system") | |
| business = extraction.get('business', {}) | |
| agent_spec = extraction.get('agent', {}) | |
| business_name = business.get('business_name', 'Business') | |
| industry = business.get('industry', '') | |
| features = business.get('key_features', []) | |
| query = business.get('query', '') | |
| # Detect domain | |
| domain = self.detector.detect(business_name, industry, features, query) | |
| await self._emit_log(f"📊 Detected domain: {domain.upper()}", "success") | |
| # Create domain-specific tools | |
| tools, tool_defs = self.tool_factory.create_tools_for_domain(domain, business_name) | |
| await self._emit_log(f"✅ Created {len(tools)} {domain}-specific tools", "success") | |
| # Generate instructions | |
| instructions = self._generate_comprehensive_instructions(extraction, domain) | |
| await self._emit_log("✅ Generated instructions", "success") | |
| # Setup model client | |
| if "gemini" in model.lower(): | |
| client = AgentsAsyncOpenAI(api_key=api_key, | |
| base_url="https://generativelanguage.googleapis.com/v1beta/openai/") | |
| model_name = "gemini-2.0-flash-exp" | |
| elif "grok" in model.lower(): | |
| client = AgentsAsyncOpenAI(api_key=api_key, base_url="https://api.x.ai/v1") | |
| model_name = "grok-beta" | |
| else: | |
| client = AgentsAsyncOpenAI(api_key=api_key) | |
| model_name = "gpt-4o" | |
| MODEL = OpenAIChatCompletionsModel(model=model_name, openai_client=client) | |
| # Create agent | |
| agent_name = agent_spec.get('agent_name', f"{business_name} AI") | |
| agent_instance = Agent(name=agent_name, instructions=instructions, model=MODEL, tools=tools) | |
| await self._emit_log("✅ Agent created!", "success") | |
| agent_id = f"agent_{domain}_{business_name.lower().replace(' ', '_')}_{int(datetime.now().timestamp())}" | |
| agent_config = AgentConfiguration( | |
| agent_id=agent_id, | |
| name=agent_name, | |
| instructions=instructions, | |
| model=model, | |
| tools=tool_defs, | |
| tone="professional", | |
| business_context={ | |
| "business_name": business_name, | |
| "industry": industry, | |
| "domain": domain, | |
| "capabilities": agent_spec.get('capabilities', []) | |
| }, | |
| deployment_ready=True | |
| ) | |
| deployment_code = f'''# {agent_name} - {domain.title()} Agent | |
| from agents import Agent, AsyncOpenAI, OpenAIChatCompletionsModel, function_tool | |
| client = AsyncOpenAI(api_key="YOUR_KEY") | |
| MODEL = OpenAIChatCompletionsModel(model="{model_name}", openai_client=client) | |
| # ... tools here ... | |
| {agent_name.replace(' ', '_')} = Agent( | |
| name="{agent_name}", | |
| instructions="""...""", | |
| model=MODEL, | |
| tools=[...] | |
| ) | |
| ''' | |
| return AgentBuildResult( | |
| status="success", | |
| agent_config=agent_config, | |
| agent_instance=agent_instance, | |
| agent_test_response=f"{agent_name} ready with {len(tools)} {domain} tools!", | |
| deployment_code=deployment_code, | |
| metadata={"session_id": self.session_id, "domain": domain, "tools_count": len(tools)} | |
| ) | |
| except Exception as e: | |
| await self._emit_log(f"❌ Failed: {str(e)}", "error") | |
| return AgentBuildResult(status="error", error=str(e), metadata={}) | |
| class AgentDeployer: | |
| async def save_to_firebase(session_id: str, agent_config: AgentConfiguration): | |
| try: | |
| ref = db.reference(f'sessions/{session_id}/agent') | |
| ref.set(agent_config.model_dump()) | |
| agents_ref = db.reference(f'agents/{agent_config.agent_id}') | |
| agents_ref.set({**agent_config.model_dump(), 'session_id': session_id, 'created_at': datetime.now().isoformat()}) | |
| return True | |
| except Exception as e: | |
| print(f"Error: {e}") | |
| return False |