Conversational AI Flow Designer агент

class ConversationState:
    def __init__(self):
        self.current_intent = None
        self.entities = {}
        self.conversation_history = []
        self.user_context = {}
        self.flow_position = "start"
        self.confidence_threshold = 0.7

    def update_context(self, user_input, intent, entities):
        self.conversation_history.append({
            "user_input": user_input,
            "intent": intent,
            "entities": entities,
            "timestamp": datetime.now()
        })
        self.entities.update(entities)
        self.current_intent = intent

flows:
  booking_flow:
    entry_conditions:
      - intent: "book_appointment"
      - entities: ["service_type"]
    steps:
      - name: "collect_datetime"
        prompt: "When would you like to schedule your {service_type}?"
        validation: "datetime_validator"
        fallback: "datetime_fallback"
      - name: "confirm_booking"
        prompt: "Confirm {service_type} on {datetime}?"
        actions: ["create_booking", "send_confirmation"]

  fallback_flow:
    triggers: ["low_confidence", "unknown_intent"]
    strategy: "clarification_questions"

def slot_filling_handler(state, required_slots):
    missing_slots = []
    for slot in required_slots:
        if slot not in state.entities or not state.entities[slot]:
            missing_slots.append(slot)

    if missing_slots:
        next_slot = missing_slots[0]
        return generate_slot_prompt(next_slot, state)

    return proceed_to_next_step(state)

def generate_slot_prompt(slot_name, context):
    prompts = {
        "date": "What date works best for you?",
        "time": "What time would you prefer?",
        "email": "What's your email address?"
    }
    return prompts.get(slot_name, f"Please provide your {slot_name}")

class NLUPipeline:
    def __init__(self, confidence_threshold=0.7):
        self.threshold = confidence_threshold
        self.fallback_strategies = {
            "clarification": self.ask_clarification,
            "suggestion": self.suggest_alternatives,
            "escalation": self.escalate_to_human
        }

    def process_input(self, user_input, context):
        intent_result = self.classify_intent(user_input, context)
        entities = self.extract_entities(user_input, intent_result.intent)

        if intent_result.confidence < self.threshold:
            return self.handle_low_confidence(user_input, intent_result)

        return {
            "intent": intent_result.intent,
            "entities": entities,
            "confidence": intent_result.confidence,
            "next_action": self.determine_next_action(intent_result, entities)
        }

class ErrorRecovery:
    def __init__(self):
        self.max_retry_attempts = 3
        self.escalation_paths = ["clarify", "simplify", "menu", "human"]

    def handle_misunderstanding(self, state, attempt_count):
        if attempt_count >= self.max_retry_attempts:
            return self.escalate_to_menu()

        strategies = {
            1: "I didn't quite catch that. Could you rephrase?",
            2: "Let me try to help differently. Are you looking to: [options]?",
            3: "I'm having trouble understanding. Let me connect you with a human agent."
        }

        return strategies.get(attempt_count, strategies[3])

class ResponseGenerator:
    def __init__(self):
        self.templates = {
            "confirmation": [
                "Got it! {summary}. Is that correct?",
                "Let me confirm: {summary}. Does this look right?"
            ],
            "progress": [
                "Great! We've got {completed_steps}. Next, {next_step}.",
                "Perfect! Just need {remaining_items} and we'll be all set."
            ]
        }

    def generate_response(self, template_type, context):
        templates = self.templates.get(template_type, [])
        if not templates:
            return self.fallback_response()

        template = random.choice(templates)
        return template.format(**context)

{
  "response_type": "rich",
  "text": "Here are your options:",
  "components": [
    {
      "type": "quick_replies",
      "options": [
        {"title": "Schedule Appointment", "payload": "intent:book_appointment"},
        {"title": "Check Status", "payload": "intent:check_status"}
      ]
    },
    {
      "type": "card",
      "title": "Available Services",
      "image_url": "https://example.com/services.jpg",
      "actions": [{"type": "postback", "title": "Learn More"}]
    }
  ]
}

class ConversationAnalytics:
    def track_flow_metrics(self, conversation_id, metrics):
        return {
            "completion_rate": metrics.completed_flows / metrics.started_flows,
            "average_turns": metrics.total_turns / metrics.conversations,
            "fallback_rate": metrics.fallback_triggers / metrics.total_turns,
            "user_satisfaction": metrics.positive_feedback / metrics.total_feedback,
            "abandonment_points": self.identify_drop_off_points(conversation_id)
        }

    def optimize_flow(self, flow_data):
        bottlenecks = self.identify_bottlenecks(flow_data)
        return {
            "recommendations": self.generate_optimization_suggestions(bottlenecks),
            "a_b_test_candidates": self.suggest_test_variations(bottlenecks)
        }