Building FAQ Bots with QnA Maker

Creating a QnA Maker Knowledge Base

# Create QnA Maker resource
az cognitiveservices account create \
    --name myqnamaker \
    --resource-group myResourceGroup \
    --kind QnAMaker \
    --sku S0 \
    --location westus

# Create App Service for runtime
az appservice plan create \
    --name myqnaplan \
    --resource-group myResourceGroup \
    --sku S1

az webapp create \
    --name myqnaruntime \
    --resource-group myResourceGroup \
    --plan myqnaplan

Building Knowledge Base Programmatically

import requests
import json
import time

class QnAMakerClient:
    def __init__(self, subscription_key, endpoint, runtime_endpoint):
        self.subscription_key = subscription_key
        self.endpoint = endpoint
        self.runtime_endpoint = runtime_endpoint

    def create_knowledge_base(self, name, qna_pairs, urls=None, files=None):
        """Create a new knowledge base."""
        create_url = f"{self.endpoint}/qnamaker/v4.0/knowledgebases/create"

        body = {
            "name": name,
            "qnaList": qna_pairs,
            "urls": urls or [],
            "files": files or []
        }

        headers = {
            "Ocp-Apim-Subscription-Key": self.subscription_key,
            "Content-Type": "application/json"
        }

        response = requests.post(create_url, headers=headers, json=body)
        operation_id = response.headers.get("Location").split("/")[-1]

        # Wait for creation to complete
        return self._wait_for_operation(operation_id)

    def _wait_for_operation(self, operation_id):
        """Poll for operation completion."""
        status_url = f"{self.endpoint}/qnamaker/v4.0/operations/{operation_id}"
        headers = {"Ocp-Apim-Subscription-Key": self.subscription_key}

        while True:
            response = requests.get(status_url, headers=headers)
            result = response.json()

            if result["operationState"] == "Succeeded":
                return result["resourceLocation"].split("/")[-1]
            elif result["operationState"] == "Failed":
                raise Exception(f"Operation failed: {result}")

            time.sleep(5)

    def add_qna_pairs(self, kb_id, qna_pairs):
        """Add Q&A pairs to existing knowledge base."""
        update_url = f"{self.endpoint}/qnamaker/v4.0/knowledgebases/{kb_id}"

        body = {
            "add": {
                "qnaList": qna_pairs
            }
        }

        headers = {
            "Ocp-Apim-Subscription-Key": self.subscription_key,
            "Content-Type": "application/json"
        }

        response = requests.patch(update_url, headers=headers, json=body)
        return response.status_code == 204

    def publish(self, kb_id):
        """Publish the knowledge base."""
        publish_url = f"{self.endpoint}/qnamaker/v4.0/knowledgebases/{kb_id}"
        headers = {"Ocp-Apim-Subscription-Key": self.subscription_key}

        response = requests.post(publish_url, headers=headers)
        return response.status_code == 204

    def get_answer(self, kb_id, question, top=3):
        """Query the knowledge base for answers."""
        query_url = f"{self.runtime_endpoint}/qnamaker/knowledgebases/{kb_id}/generateAnswer"

        body = {
            "question": question,
            "top": top,
            "scoreThreshold": 30,
            "strictFilters": [],
            "context": None
        }

        headers = {
            "Authorization": f"EndpointKey {self.endpoint_key}",
            "Content-Type": "application/json"
        }

        response = requests.post(query_url, headers=headers, json=body)
        return response.json()


# Example usage
client = QnAMakerClient(
    subscription_key="your-subscription-key",
    endpoint="https://your-resource.cognitiveservices.azure.com",
    runtime_endpoint="https://your-runtime.azurewebsites.net"
)

# Define Q&A pairs
qna_pairs = [
    {
        "id": 1,
        "answer": "Our store hours are Monday to Friday, 9 AM to 6 PM, and Saturday 10 AM to 4 PM.",
        "source": "manual",
        "questions": [
            "What are your store hours?",
            "When are you open?",
            "What time do you open?",
            "What time do you close?"
        ],
        "metadata": [
            {"name": "category", "value": "general"}
        ]
    },
    {
        "id": 2,
        "answer": "You can return items within 30 days of purchase with a receipt. Items must be unused and in original packaging.",
        "source": "manual",
        "questions": [
            "What is your return policy?",
            "Can I return an item?",
            "How do I return something?",
            "What is the refund policy?"
        ],
        "metadata": [
            {"name": "category", "value": "returns"}
        ]
    },
    {
        "id": 3,
        "answer": "Standard shipping takes 5-7 business days. Express shipping takes 2-3 business days. Free shipping on orders over $50.",
        "source": "manual",
        "questions": [
            "How long does shipping take?",
            "What are the shipping options?",
            "Do you offer free shipping?",
            "How much is shipping?"
        ],
        "metadata": [
            {"name": "category", "value": "shipping"}
        ]
    }
]

# Create knowledge base
kb_id = client.create_knowledge_base("Customer Support FAQ", qna_pairs)
print(f"Knowledge base created: {kb_id}")

# Publish
client.publish(kb_id)
print("Knowledge base published")

# Query
result = client.get_answer(kb_id, "When are you open?")
print(f"Answer: {result['answers'][0]['answer']}")
print(f"Confidence: {result['answers'][0]['score']}")

Multi-Turn Conversations

# Define multi-turn Q&A with follow-up prompts
multi_turn_qna = [
    {
        "id": 10,
        "answer": "We offer several payment methods. Which would you like to know about?",
        "source": "manual",
        "questions": [
            "What payment methods do you accept?",
            "How can I pay?"
        ],
        "metadata": [],
        "context": {
            "isContextOnly": False,
            "prompts": [
                {
                    "displayOrder": 1,
                    "displayText": "Credit Cards",
                    "qnaId": 11
                },
                {
                    "displayOrder": 2,
                    "displayText": "PayPal",
                    "qnaId": 12
                },
                {
                    "displayOrder": 3,
                    "displayText": "Bank Transfer",
                    "qnaId": 13
                }
            ]
        }
    },
    {
        "id": 11,
        "answer": "We accept Visa, MasterCard, American Express, and Discover. All cards are processed securely.",
        "source": "manual",
        "questions": ["Credit cards", "Tell me about credit cards"],
        "metadata": [],
        "context": {
            "isContextOnly": True,
            "prompts": []
        }
    },
    {
        "id": 12,
        "answer": "PayPal is available at checkout. You can use your PayPal balance or linked cards.",
        "source": "manual",
        "questions": ["PayPal", "Tell me about PayPal"],
        "metadata": [],
        "context": {
            "isContextOnly": True,
            "prompts": []
        }
    },
    {
        "id": 13,
        "answer": "Bank transfers are available for orders over $500. Contact support for wire transfer details.",
        "source": "manual",
        "questions": ["Bank transfer", "Wire transfer"],
        "metadata": [],
        "context": {
            "isContextOnly": True,
            "prompts": []
        }
    }
]

# Query with context for multi-turn
def get_answer_with_context(kb_id, question, previous_qna_id=None, previous_question=None):
    """Query with conversation context."""
    body = {
        "question": question,
        "top": 3,
        "scoreThreshold": 30
    }

    if previous_qna_id:
        body["context"] = {
            "previousQnAId": previous_qna_id,
            "previousUserQuery": previous_question
        }

    # Make API call...
    return result

Integrating with Bot Framework

// qnaBot.js
const { ActivityHandler } = require('botbuilder');
const { QnAMaker, QnAMakerEndpoint } = require('botbuilder-ai');

class QnABot extends ActivityHandler {
    constructor() {
        super();

        const endpoint = {
            knowledgeBaseId: process.env.QNA_KB_ID,
            endpointKey: process.env.QNA_ENDPOINT_KEY,
            host: process.env.QNA_HOST
        };

        this.qnaMaker = new QnAMaker(endpoint, {
            scoreThreshold: 30,
            top: 3
        });

        this.onMessage(async (context, next) => {
            // Get QnA answer
            const qnaResults = await this.qnaMaker.getAnswers(context);

            if (qnaResults.length > 0) {
                const topAnswer = qnaResults[0];

                // Send answer
                await context.sendActivity(topAnswer.answer);

                // Handle follow-up prompts
                if (topAnswer.context && topAnswer.context.prompts.length > 0) {
                    await this.sendPrompts(context, topAnswer.context.prompts);
                }
            } else {
                await context.sendActivity(
                    "I couldn't find an answer to your question. " +
                    "Please try rephrasing or contact support."
                );
            }

            await next();
        });
    }

    async sendPrompts(context, prompts) {
        const { CardFactory } = require('botbuilder');

        const card = {
            type: 'AdaptiveCard',
            $schema: 'http://adaptivecards.io/schemas/adaptive-card.json',
            version: '1.3',
            body: [
                {
                    type: 'TextBlock',
                    text: 'You might also want to know:',
                    wrap: true
                }
            ],
            actions: prompts.map(prompt => ({
                type: 'Action.Submit',
                title: prompt.displayText,
                data: { question: prompt.displayText }
            }))
        };

        await context.sendActivity({
            attachments: [CardFactory.adaptiveCard(card)]
        });
    }
}

module.exports.QnABot = QnABot;

Active Learning

def train_with_feedback(client, kb_id, question, answer_id):
    """Submit user feedback for active learning."""
    train_url = f"{client.endpoint}/qnamaker/v4.0/knowledgebases/{kb_id}/train"

    body = {
        "feedbackRecords": [
            {
                "userId": "user123",
                "userQuestion": question,
                "qnaId": answer_id
            }
        ]
    }

    headers = {
        "Ocp-Apim-Subscription-Key": client.subscription_key,
        "Content-Type": "application/json"
    }

    response = requests.post(train_url, headers=headers, json=body)
    return response.status_code == 204


# Get active learning suggestions
def get_suggestions(client, kb_id):
    """Get suggested alternative questions."""
    suggestions_url = f"{client.endpoint}/qnamaker/v4.0/knowledgebases/{kb_id}/alterations"

    headers = {"Ocp-Apim-Subscription-Key": client.subscription_key}
    response = requests.get(suggestions_url, headers=headers)

    return response.json()


# Add alternative questions based on suggestions
def accept_suggestion(client, kb_id, qna_id, new_question):
    """Add suggested question to QnA pair."""
    update_url = f"{client.endpoint}/qnamaker/v4.0/knowledgebases/{kb_id}"

    body = {
        "update": {
            "qnaList": [
                {
                    "id": qna_id,
                    "questions": {
                        "add": [new_question]
                    }
                }
            ]
        }
    }

    headers = {
        "Ocp-Apim-Subscription-Key": client.subscription_key,
        "Content-Type": "application/json"
    }

    response = requests.patch(update_url, headers=headers, json=body)
    return response.status_code == 204

Import from URLs and Files

# Create KB from website FAQs
kb_with_urls = {
    "name": "Product FAQ",
    "qnaList": [],
    "urls": [
        "https://www.example.com/faq",
        "https://www.example.com/support/help"
    ],
    "files": []
}

# Create KB from documents
kb_with_files = {
    "name": "Documentation KB",
    "qnaList": [],
    "urls": [],
    "files": [
        {
            "fileName": "product-manual.pdf",
            "fileUri": "https://storage.blob.core.windows.net/docs/product-manual.pdf"
        },
        {
            "fileName": "faq.docx",
            "fileUri": "https://storage.blob.core.windows.net/docs/faq.docx"
        }
    ]
}

# Supported file types:
# - PDF
# - DOCX
# - TXT
# - TSV (tab-separated Q&A pairs)
# - XLSX (Excel with Question/Answer columns)

Analytics and Monitoring

from azure.monitor.query import LogsQueryClient
from azure.identity import DefaultAzureCredential

def get_qna_analytics(workspace_id, days=7):
    """Get QnA Maker usage analytics from Application Insights."""
    credential = DefaultAzureCredential()
    client = LogsQueryClient(credential)

    query = f"""
    customEvents
    | where timestamp > ago({days}d)
    | where name == "QnAMessage"
    | extend question = tostring(customDimensions.Question)
    | extend answer = tostring(customDimensions.Answer)
    | extend score = todouble(customDimensions.Score)
    | extend kbId = tostring(customDimensions.KnowledgeBaseId)
    | summarize
        TotalQueries = count(),
        AvgScore = avg(score),
        LowScoreCount = countif(score < 50)
        by bin(timestamp, 1d)
    | order by timestamp desc
    """

    response = client.query_workspace(workspace_id, query, timespan=None)
    return response.tables[0]


def get_unanswered_questions(workspace_id, days=7, threshold=30):
    """Find questions with low confidence scores."""
    credential = DefaultAzureCredential()
    client = LogsQueryClient(credential)

    query = f"""
    customEvents
    | where timestamp > ago({days}d)
    | where name == "QnAMessage"
    | extend question = tostring(customDimensions.Question)
    | extend score = todouble(customDimensions.Score)
    | where score < {threshold}
    | summarize Count = count() by question
    | order by Count desc
    | take 50
    """

    response = client.query_workspace(workspace_id, query, timespan=None)
    return response.tables[0]

Conclusion

QnA Maker simplifies building knowledge-based chatbots:

• Easy content import from URLs, documents, and manual entry
• Natural language understanding for question matching
• Multi-turn conversations with follow-up prompts
• Active learning to improve accuracy over time
• Bot Framework integration for multi-channel deployment

It’s the fastest path to creating FAQ bots and customer support assistants.