Advanced Chatbot Architectures: Beyond Simple Q&A

User Message
    ↓
[Message Preprocessing]
    ├── Language detection
    ├── Input sanitization
    └── Query rewriting (for search)
    ↓
[Retrieval Pipeline (RAG)]
    ├── Embed user query
    ├── Vector search (knowledge base)
    ├── Keyword search (BM25 hybrid)
    └── Rerank retrieved chunks
    ↓
[Context Assembly]
    ├── System prompt
    ├── Retrieved documents
    ├── Conversation history
    ├── User profile
    └── Tool definitions
    ↓
[LLM Generation]
    ├── Streaming response
    ├── Tool calls (if needed)
    └── Follow-up suggestions
    ↓
[Post-processing]
    ├── Citation extraction
    ├── Response validation
    └── Guardrails check
    ↓
Response to User

from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_anthropic import ChatAnthropic
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory

class RAGChatbot:
    def __init__(self, knowledge_base_index: str, provider: str = "openai"):
        # Initialize LLM - choose based on your needs
        if provider == "openai":
            self.llm = ChatOpenAI(
                model="gpt-5.1",  # Latest GPT with Instant/Thinking modes
                temperature=0.7,
                streaming=True
            )
        elif provider == "anthropic":
            self.llm = ChatAnthropic(
                model="claude-sonnet-4-5-20251022",  # Best for coding
                temperature=0.7,
                streaming=True
            )
        elif provider == "google":
            self.llm = ChatGoogleGenerativeAI(
                model="gemini-3-pro",  # #1 on LMArena
                temperature=0.7,
                streaming=True
            )

        # Initialize embeddings and vector store
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vectorstore = PineconeVectorStore.from_existing_index(
            index_name=knowledge_base_index,
            embedding=self.embeddings
        )

        # Conversation memory (last 10 turns)
        self.memory = ConversationBufferWindowMemory(
            k=10,
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )

        # Build retrieval chain
        self.chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(
                search_type="mmr",  # Maximal Marginal Relevance
                search_kwargs={"k": 5, "fetch_k": 20}
            ),
            memory=self.memory,
            return_source_documents=True,
            verbose=True
        )

    async def chat(self, user_message: str) -> dict:
        """Process user message and return response with sources."""
        result = await self.chain.ainvoke({"question": user_message})

        return {
            "answer": result["answer"],
            "sources": [
                {
                    "title": doc.metadata.get("title"),
                    "url": doc.metadata.get("url"),
                    "snippet": doc.page_content[:200]
                }
                for doc in result["source_documents"]
            ]
        }

from langchain_community.document_loaders import WebBaseLoader
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pinecone import Pinecone

class KnowledgeBaseBuilder:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

        # Initialize Pinecone (2025 SDK)
        self.pc = Pinecone(api_key="...")

    def ingest_documents(self, sources: list[dict]):
        """Ingest documents into vector store."""
        all_chunks = []

        for source in sources:
            # Load document
            if source["type"] == "web":
                loader = WebBaseLoader(source["url"])
            elif source["type"] == "pdf":
                loader = PDFLoader(source["path"])

            documents = loader.load()

            # Split into chunks
            chunks = self.text_splitter.split_documents(documents)

            # Add metadata
            for chunk in chunks:
                chunk.metadata["source"] = source["name"]
                chunk.metadata["url"] = source.get("url")

            all_chunks.extend(chunks)

        # Create vector store
        Pinecone.from_documents(
            documents=all_chunks,
            embedding=self.embeddings,
            index_name="chatbot-knowledge-base"
        )

        return len(all_chunks)

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, vectorstore, documents):
        self.vectorstore = vectorstore
        self.documents = documents

        # Build BM25 index
        tokenized_docs = [doc.page_content.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)

    def retrieve(self, query: str, k: int = 5) -> list:
        """Hybrid retrieval combining vector and keyword search."""
        # Vector search
        vector_results = self.vectorstore.similarity_search_with_score(query, k=k*2)

        # BM25 keyword search
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_top_k = np.argsort(bm25_scores)[-k*2:][::-1]

        # Combine and rerank using Reciprocal Rank Fusion
        doc_scores = {}

        for rank, (doc, score) in enumerate(vector_results):
            doc_id = doc.metadata.get("id")
            doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank + 60)

        for rank, idx in enumerate(bm25_top_k):
            doc_id = self.documents[idx].metadata.get("id")
            doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank + 60)

        # Sort by combined score
        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
        return [self.get_doc_by_id(doc_id) for doc_id, _ in sorted_docs[:k]]

User: "What's the pricing for the enterprise plan?"
Bot: "Enterprise is $99/seat/month with volume discounts..."

User: "What about for 500 users?"
Bot: ❌ "I'm not sure what you're referring to..."  # Bad - lost context
Bot: ✅ "For 500 users on Enterprise, that would be $89/seat..."  # Good - retained context

class ChatbotMemory:
    """Multi-tier memory system for LLM chatbot."""

    def __init__(self, user_id: str):
        self.user_id = user_id

        # Tier 1: Working Memory (current conversation)
        self.working_memory = []

        # Tier 2: Short-term Memory (recent conversations, summarized)
        self.short_term = self.load_recent_summaries(user_id)

        # Tier 3: Long-term Memory (vector store of all interactions)
        self.long_term = self.init_user_memory_store(user_id)

        # Tier 4: User Profile (structured data)
        self.user_profile = self.load_user_profile(user_id)

    def add_turn(self, role: str, content: str):
        """Add a conversation turn to working memory."""
        self.working_memory.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now()
        })

        # If working memory gets too long, compress
        if len(self.working_memory) > 20:
            self.compress_working_memory()

    def compress_working_memory(self):
        """Summarize older turns to save context space."""
        older_turns = self.working_memory[:-10]
        recent_turns = self.working_memory[-10:]

        # Use LLM to summarize older conversation
        summary = self.llm.summarize(older_turns)

        self.working_memory = [
            {"role": "system", "content": f"Previous conversation summary: {summary}"}
        ] + recent_turns

    def build_context(self) -> list:
        """Build full context for LLM call."""
        context = []

        # Add user profile context
        context.append({
            "role": "system",
            "content": f"User profile: {self.user_profile.to_string()}"
        })

        # Add relevant long-term memories
        relevant_memories = self.long_term.search(
            self.working_memory[-1]["content"],
            k=3
        )
        if relevant_memories:
            context.append({
                "role": "system",
                "content": f"Relevant past interactions: {relevant_memories}"
            })

        # Add working memory
        context.extend(self.working_memory)

        return context

class ConversationMemoryStore:
    """Vector store for conversation history."""

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            collection_name=f"user_{user_id}_memory",
            embedding_function=self.embeddings
        )

    def store_conversation(self, conversation: list, summary: str):
        """Store a completed conversation."""
        # Create a document from the conversation
        doc = Document(
            page_content=summary,
            metadata={
                "user_id": self.user_id,
                "timestamp": datetime.now().isoformat(),
                "turn_count": len(conversation),
                "topics": self.extract_topics(conversation)
            }
        )
        self.vectorstore.add_documents([doc])

    def search(self, query: str, k: int = 3) -> list:
        """Find relevant past conversations."""
        results = self.vectorstore.similarity_search(
            query,
            k=k,
            filter={"user_id": self.user_id}
        )
        return results

class FollowUpHandler:
    """Handle follow-up questions with context resolution."""

    def __init__(self, llm):
        self.llm = llm
        self.entity_tracker = {}  # Track mentioned entities

    def resolve_references(self, query: str, conversation_history: list) -> str:
        """Rewrite query to be standalone by resolving references."""

        # If query seems complete, return as-is
        if self.is_standalone_query(query):
            return query

        # Use LLM to rewrite with context
        prompt = f"""Given this conversation history and follow-up question,
rewrite the question to be standalone (include all necessary context).

Conversation:
{self.format_history(conversation_history)}

Follow-up question: {query}

Standalone question:"""

        resolved = self.llm.invoke(prompt)
        return resolved.content

    def is_standalone_query(self, query: str) -> bool:
        """Check if query needs context resolution."""
        # Pronouns and references that typically need resolution
        context_markers = [
            "it", "that", "this", "those", "these",
            "the same", "another", "more", "less",
            "previous", "last", "earlier",
            "he", "she", "they", "them"
        ]
        query_lower = query.lower()
        return not any(marker in query_lower for marker in context_markers)

    def extract_entities(self, message: str) -> dict:
        """Extract and track entities from messages."""
        # Use NER or LLM to extract entities
        entities = self.llm.invoke(
            f"Extract key entities (products, prices, dates, names) from: {message}"
        )
        return entities

class ConversationThreadManager:
    """Manage multiple conversation threads/topics."""

    def __init__(self):
        self.threads = {}  # topic -> conversation history
        self.current_thread = "general"

    def detect_topic_change(self, message: str, current_context: list) -> str:
        """Detect if user is switching topics."""

        prompt = f"""Analyze if this message continues the current topic or starts a new one.

Current topic context: {current_context[-3:] if current_context else 'None'}
New message: {message}

Response format:
- CONTINUE: if staying on current topic
- NEW_TOPIC: <topic_name> if switching topics
- RETURN: <topic_name> if returning to a previous topic"""

        result = self.llm.invoke(prompt)
        return self.parse_topic_result(result.content)

    def handle_topic_switch(self, new_topic: str, old_topic: str):
        """Handle switching between conversation threads."""

        # Save current thread state
        self.threads[old_topic] = {
            "history": self.current_history.copy(),
            "entities": self.entity_tracker.copy(),
            "last_active": datetime.now()
        }

        # Load or create new thread
        if new_topic in self.threads:
            # Returning to previous topic
            thread = self.threads[new_topic]
            self.current_history = thread["history"]
            self.entity_tracker = thread["entities"]
        else:
            # New topic
            self.current_history = []
            self.entity_tracker = {}

        self.current_thread = new_topic

Pattern 1: Direct Reference
User: "Tell me about product X"
User: "How much does IT cost?" → "How much does product X cost?"

Pattern 2: Implicit Reference
User: "I'm looking for a laptop for video editing"
User: "What about RAM?" → "What RAM is recommended for video editing laptops?"

Pattern 3: Comparative Reference
User: "What's the price of Plan A?"
User: "And Plan B?" → "What's the price of Plan B?"
User: "Which is better?" → "Which is better, Plan A or Plan B?"

Pattern 4: Topic Return
User: "Help me with billing" → [billing thread]
User: "Actually, quick question about shipping" → [shipping thread]
User: "OK back to my billing issue" → [resume billing thread]

class ProactiveEngine:
    """Engine for proactive chatbot engagement."""

    def __init__(self, user_context: dict):
        self.user = user_context
        self.triggers = self.load_triggers()

    def check_proactive_triggers(self, event: dict) -> Optional[str]:
        """Check if any proactive message should be sent."""

        # Browsing behavior triggers
        if event["type"] == "page_view":
            if event["page"] == "pricing" and event["view_count"] >= 3:
                return self.generate_pricing_help()

            if event["page"] == "checkout" and event["time_on_page"] > 60:
                return self.generate_checkout_assistance()

        # User state triggers
        if event["type"] == "cart_abandonment":
            return self.generate_cart_recovery()

        # Subscription triggers
        if event["type"] == "subscription_expiring":
            days_left = event["days_until_expiry"]
            if days_left <= 7:
                return self.generate_renewal_reminder(days_left)

        # Usage triggers
        if event["type"] == "feature_struggle":
            return self.generate_feature_help(event["feature"])

        return None

    def generate_pricing_help(self) -> str:
        return """I noticed you're checking out our pricing options.
Would you like me to help you find the right plan for your needs?
I can also explain any features or answer questions about billing."""

    def generate_checkout_assistance(self) -> str:
        return """I see you're on the checkout page. Having any trouble?
I can help with:
• Payment options
• Discount codes
• Order questions"""

class ReactiveHandler:
    """Handle reactive chatbot behaviors."""

    def analyze_user_state(self, conversation: list) -> dict:
        """Analyze conversation for user sentiment and intent."""

        recent_messages = conversation[-5:]

        # Detect frustration signals
        frustration_indicators = [
            "not working", "still broken", "again", "already told you",
            "doesn't help", "useless", "frustrated", "annoyed"
        ]

        # Detect confusion signals
        confusion_indicators = [
            "don't understand", "what do you mean", "confused",
            "unclear", "lost", "?"  # Multiple question marks
        ]

        # Detect urgency signals
        urgency_indicators = [
            "asap", "urgent", "immediately", "right now",
            "deadline", "emergency", "critical"
        ]

        user_text = " ".join([m["content"] for m in recent_messages if m["role"] == "user"])

        return {
            "frustrated": any(ind in user_text.lower() for ind in frustration_indicators),
            "confused": any(ind in user_text.lower() for ind in confusion_indicators),
            "urgent": any(ind in user_text.lower() for ind in urgency_indicators),
            "repeated_question": self.detect_repetition(recent_messages)
        }

    def adapt_response(self, response: str, user_state: dict) -> str:
        """Adapt response based on user state."""

        if user_state["frustrated"]:
            # Lead with empathy, be concise, offer escalation
            return f"""I understand this has been frustrating, and I apologize.

{response}

Would you prefer to speak with a human agent? I can connect you right away."""

        if user_state["confused"]:
            # Simplify, offer step-by-step
            return f"""Let me explain this more clearly:

{self.simplify_response(response)}

Would a step-by-step walkthrough help?"""

        if user_state["urgent"]:
            # Be direct, prioritize action
            return f"""I understand this is urgent. Here's the fastest path:

{self.prioritize_actions(response)}"""

        return response

class FollowUpSuggester:
    """Generate contextual follow-up suggestions."""

    def generate_suggestions(self, query: str, response: str, context: dict) -> list:
        """Generate relevant follow-up questions for the user."""

        prompt = f"""Based on this conversation, suggest 2-3 natural follow-up questions
the user might want to ask next.

User asked: {query}
Bot responded: {response}
User profile: {context.get('user_type', 'general')}

Generate follow-ups that:
1. Dive deeper into the topic
2. Address related concerns
3. Help the user take next steps

Format: Return as a JSON array of strings."""

        suggestions = self.llm.invoke(prompt)
        return json.loads(suggestions.content)

# Example output:
# User: "What's your return policy?"
# Bot: "You can return items within 30 days..."
# Suggestions:
# - "How do I start a return?"
# - "What items can't be returned?"
# - "How long until I get my refund?"

class ConversationFlowManager:
    """Manage conversation flow and prevent dead ends."""

    def ensure_conversation_continues(self, response: str, context: dict) -> str:
        """Ensure response doesn't create a dead end."""

        # Check if response is a dead end
        if self.is_dead_end(response):
            # Add a conversation continuer
            continuer = self.generate_continuer(context)
            response = f"{response}\n\n{continuer}"

        return response

    def is_dead_end(self, response: str) -> bool:
        """Check if response might end conversation prematurely."""
        dead_end_patterns = [
            r"^(ok|okay|sure|got it|done|alright)\.?$",
            r"^(yes|no)\.?$",
            r"hope (this|that) helps",
        ]
        return any(re.match(p, response.lower().strip()) for p in dead_end_patterns)

    def generate_continuer(self, context: dict) -> str:
        """Generate a conversation continuer."""
        continuers = [
            "Is there anything else you'd like to know?",
            "Do you have any other questions about this?",
            "Would you like me to explain anything in more detail?",
            "Can I help you with anything else today?"
        ]

        # Context-aware continuers
        if context.get("topic") == "troubleshooting":
            return "Did that solve your issue, or should we try something else?"
        if context.get("topic") == "purchase":
            return "Ready to proceed, or would you like to explore other options?"

        return random.choice(continuers)

from openai import OpenAI

class ToolEnabledChatbot:
    def __init__(self):
        self.client = OpenAI()
        self.tools = self.define_tools()

    def define_tools(self):
        return [
            {
                "type": "function",
                "function": {
                    "name": "search_knowledge_base",
                    "description": "Search the knowledge base for information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Search query"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "check_order_status",
                    "description": "Check the status of a customer order",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "order_id": {"type": "string", "description": "Order ID"}
                        },
                        "required": ["order_id"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "create_support_ticket",
                    "description": "Create a support ticket for complex issues",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "title": {"type": "string"},
                            "description": {"type": "string"},
                            "priority": {"type": "string", "enum": ["low", "medium", "high"]}
                        },
                        "required": ["title", "description"]
                    }
                }
            }
        ]

    async def chat(self, messages: list) -> str:
        """Process chat with tool calling."""
        response = self.client.chat.completions.create(
            model="gpt-5.1",  # Latest GPT with superior tool use
            messages=messages,
            tools=self.tools,
            tool_choice="auto",
            parallel_tool_calls=True  # Enable parallel execution
        )

        message = response.choices[0].message

        # Handle tool calls
        if message.tool_calls:
            # Execute each tool call
            tool_results = []
            for tool_call in message.tool_calls:
                result = await self.execute_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                tool_results.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "content": json.dumps(result)
                })

            # Get final response with tool results
            messages.append(message)
            messages.extend(tool_results)

            final_response = self.client.chat.completions.create(
                model="gpt-5.1",
                messages=messages
            )
            return final_response.choices[0].message.content

        return message.content

    async def execute_tool(self, name: str, args: dict) -> dict:
        """Execute a tool and return results."""
        if name == "search_knowledge_base":
            return await self.search_kb(args["query"])
        elif name == "check_order_status":
            return await self.get_order(args["order_id"])
        elif name == "create_support_ticket":
            return await self.create_ticket(args)

async def stream_response(self, messages: list):
    """Stream chatbot response for real-time display."""
    response = self.client.chat.completions.create(
        model="gpt-5.1",
        messages=messages,
        stream=True
    )

    full_response = ""
    async for chunk in response:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_response += token
            yield token  # Yield each token as it arrives

    # Store complete response in memory
    self.memory.add_turn("assistant", full_response)

User Input
    ↓
[Input Processing]
    ├── Speech-to-text (if voice)
    ├── Language detection
    └── Input normalization
    ↓
[Context Assembly]
    ├── Conversation history
    ├── User profile
    ├── Session state
    └── Retrieved knowledge
    ↓
[Intent & Entity Recognition]
    ↓
[Dialog Management]
    ├── State machine (structured flows)
    └── LLM reasoning (open-ended)
    ↓
[Action Execution]
    ├── API calls
    ├── Database operations
    └── External integrations
    ↓
[Response Generation]
    ↓
[Output Processing]
    ├── Text-to-speech (if voice)
    ├── Formatting
    └── Personalization
    ↓
Response to User

User: "I want to cancel my subscription"
     ↓
[FSM: Cancellation Flow triggered]
     ↓
FSM: "I can help with that. First, can you confirm your account email?"
User: "Wait, actually what happens to my data if I cancel?"
     ↓
[LLM: Off-script question detected]
     ↓
LLM: [Retrieves data policy, generates response]
     ↓
[FSM: Resume cancellation flow]
FSM: "Great question - your data is retained for 30 days... Now, your account email?"

Turn 1: User asks about pricing
Turn 2: Bot explains tiers
Turn 3: User asks "what about enterprise?" ← Needs Turn 1-2 context

[Full history] → [LLM summary] → Compressed context
"User is evaluating enterprise pricing, concerned about per-seat costs,
has team of ~50, interested in SSO features"

{
  "product": "Enterprise Plan",
  "team_size": 50,
  "concerns": ["price", "SSO"],
  "timeline": "Q1 decision"
}

{
  "user_id": "u_12345",
  "name": "Sarah",
  "company": "Acme Corp",
  "role": "VP Engineering",
  "history": {
    "support_tickets": 3,
    "nps_score": 8,
    "last_interaction": "2024-11-15",
    "common_topics": ["billing", "API"]
  },
  "preferences": {
    "communication_style": "direct",
    "technical_level": "high"
  }
}

{
  "workflow": "subscription_upgrade",
  "current_step": "payment_confirmation",
  "collected_data": {
    "new_plan": "enterprise",
    "billing_cycle": "annual",
    "payment_method": "invoice"
  },
  "pending_actions": ["generate_contract", "notify_sales"],
  "can_resume": true
}

Intents:
- billing_inquiry
- technical_support
- account_management
- sales_inquiry
- feedback
- other

Input: "My card was charged twice"
→ billing_inquiry (0.94)

System: Classify the user's intent and extract key entities.
User: "I need to add my coworker John to our account but I'm not sure if we have seats available"

Response: {
  "primary_intent": "add_team_member",
  "secondary_intent": "check_seat_availability",
  "entities": {
    "action": "add",
    "target_user": "John",
    "relationship": "coworker"
  },
  "uncertainty": "seat_availability"
}

workflow: subscription_cancellation
triggers:
  - intent: cancel_subscription
  - keywords: ["cancel", "unsubscribe", "end subscription"]

steps:
  - id: confirm_identity
    action: verify_account
    required_fields: [email, last_4_cc]
    on_failure: escalate_to_human

  - id: understand_reason
    action: collect_feedback
    options:
      - too_expensive
      - not_using
      - missing_features
      - switching_competitor
      - other

  - id: retention_offer
    condition: "reason in ['too_expensive', 'switching_competitor']"
    action: present_offer
    offers:
      - discount_20_percent
      - pause_subscription
      - downgrade_plan

  - id: process_cancellation
    condition: "offer_accepted == false"
    action: cancel_subscription
    side_effects:
      - send_confirmation_email
      - update_crm
      - trigger_winback_sequence

  - id: confirm_cancellation
    action: summarize_and_confirm

class WorkflowEngine:
    def __init__(self, workflow_definition):
        self.definition = workflow_definition
        self.current_step = None
        self.collected_data = {}

    def process_input(self, user_input, context):
        # Determine if input advances workflow
        if self.current_step:
            result = self.process_step_input(user_input)
            if result.complete:
                return self.advance_to_next_step()
            else:
                return result.follow_up_prompt
        else:
            # Start workflow
            return self.start_workflow()

    def advance_to_next_step(self):
        next_step = self.get_next_step()
        if next_step:
            self.current_step = next_step
            return self.execute_step(next_step)
        else:
            return self.complete_workflow()

FSM: "What's your account email?"
User: "Actually, how much would it cost to upgrade instead of canceling?"

→ Detect tangent (upgrade inquiry)
→ Pause cancellation workflow
→ Address upgrade question
→ Offer to resume: "Would you like to continue with the cancellation, or explore the upgrade?"

template: subscription_confirmed
text: "Great news, {{name}}! Your {{plan}} subscription is now active.
       Your next billing date is {{next_billing_date}}."

Generate a friendly confirmation that the user's subscription is active.
Include: their name (Sarah), plan (Enterprise), and next billing date (Jan 15).
Tone: Professional but warm. Mention the key benefits they now have access to.

Novice: "To access your API key, go to Settings (the gear icon in the top right),
        then click on 'API Access'. You'll see your key there!"

Expert: "API key: Settings → API Access"

Frustrated user: Lead with empathy, be concise, offer escalation
Curious user: Provide detail, suggest related topics
Rushed user: Get to the point, offer async follow-up

Returning user: "Welcome back, Sarah! How can I help today?"
VIP customer: Route to senior support, proactive offers
At-risk user: Extra care, retention focus

[Chatbot Core]
      ↓
[API Gateway]
      ├── CRM (Salesforce, HubSpot)
      ├── Billing (Stripe, Zuora)
      ├── Support (Zendesk, Intercom)
      ├── Product (internal APIs)
      └── External (shipping, payments)

class ActionExecutor:
    async def execute(self, action, params, context):
        # Validate action is permitted
        if not self.authorize(action, context.user):
            return ActionResult(success=False, error="Not authorized")

        # Execute with timeout
        try:
            result = await asyncio.wait_for(
                self.dispatch(action, params),
                timeout=10.0
            )
            return ActionResult(success=True, data=result)
        except TimeoutError:
            return ActionResult(success=False, error="Action timed out")
        except Exception as e:
            return ActionResult(success=False, error=str(e))

"I want to make sure you get the best help with this. I'm connecting you
with Sarah from our support team. I've shared our conversation so you
won't need to repeat yourself. Sarah will be with you in about 2 minutes."

Production conversations
        ↓
Quality filter (successful resolutions)
        ↓
Human review and correction
        ↓
Training data for models
        ↓
Improved chatbot
        ↓
Better conversations

[User views pricing page for 3rd time]
→ Chatbot: "I noticed you're checking out our pricing.
   Have questions I can help answer?"

[User's subscription renewing in 7 days]
→ Chatbot: "Quick heads up - your subscription renews next week.
   Everything look good, or would you like to make changes?"

personality:
  name: "Alex"
  traits:
    - helpful
    - knowledgeable
    - slightly playful
    - never condescending

  style_guide:
    greeting: "Hey there! 👋"
    apology: "I'm sorry about that - let me help fix it."
    confusion: "Hmm, I want to make sure I understand..."
    success: "Awesome! That's all set."

  boundaries:
    - Never comment on competitors
    - No political or controversial topics
    - Escalate safety concerns immediately