Conversational AI: Building Intelligent Chatbots
self, members_added: list[ChannelAccount], turn_context: TurnContext ): """Welcome new users to the conversation.""" for member in members_added: # Don't greet the bot itself if member.id != turn_context.activity.recipient.id: welcome_message = ( "Hello! 👋 I'm your Customer Support Assistant. I can help you with:\n\n" "- Order status and tracking\n" "- Account information\n" "- Product inquiries\n" "- Returns and refunds\n" "- Technical support\n\n" "How can I assist you today?" ) await turn_context.send_activity(MessageFactory.text(welcome_message)) logger.info(f"Welcomed new member: {member.id}")
async def on_members_removed_activity( self, members_removed: list[ChannelAccount], turn_context: TurnContext ): """Log when members leave the conversation.""" for member in members_removed: if member.id != turn_context.activity.recipient.id: logger.info(f"Member left conversation: {member.id}")
async def on_message_reaction_activity(self, turn_context: TurnContext): """Handle reactions to bot messages (Teams, Slack).""" for reaction in turn_context.activity.reactions_added or []: logger.info(f"Reaction added: {reaction.type}") # Track positive/negative reactions for satisfaction metrics if reaction.type == "like": logger.info("Positive reaction received") elif reaction.type == "angry" or reaction.type == "sad": logger.warning("Negative reaction received - may need escalation")
async def on_event_activity(self, turn_context: TurnContext): """Handle custom events from channels.""" event_name = turn_context.activity.name logger.info(f"Event received: {event_name}")
if event_name == "webchat/join": await turn_context.send_activity("Thanks for joining our web chat!") elif event_name == "tokens/response": # OAuth token response logger.info("OAuth token response received")
async def on_turn_error(self, turn_context: TurnContext, error: Exception): """Global error handler for unhandled exceptions.""" logger.error(f"Bot error: {str(error)}", exc_info=error)
# Send friendly error message to user error_message = ( "I'm sorry, I encountered an error processing your request. " "Please try again or contact support if the issue persists." ) await turn_context.send_activity(MessageFactory.text(error_message))
# Clear conversation state to prevent stuck dialogs await self.conversation_state.delete(turn_context)
Bot Application Server (aiohttp)
Host the bot with an async HTTP server:
from aiohttp import web
from aiohttp.web import Request, Response
from botbuilder.core import BotFrameworkAdapter, BotFrameworkAdapterSettings
from botbuilder.schema import Activity
import sys
import traceback
## Bot Framework Adapter configuration
APP_ID = os.environ.get("MicrosoftAppId", "")
APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
settings = BotFrameworkAdapterSettings(APP_ID, APP_PASSWORD)
adapter = BotFrameworkAdapter(settings)
## Create bot instance
bot = CustomerSupportBot()
## Error handler for adapter
async def on_error(context: TurnContext, error: Exception):
```text
print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
traceback.print_exc()
await context.send_activity("The bot encountered an error or bug.")
adapter.on_turn_error = on_error
HTTP endpoint for Bot Framework messages
Figure: Copilot Studio – topic flow designer with test chat panel.
async def messages(req: Request) -> Response:
"""
Main bot endpoint receiving activities from Bot Framework Service.
"""
## Verify Bot Framework signature
if "application/json" in req.headers["Content-Type"]:
body = await req.json()
else:
return Response(status=415)
## Process activity
activity = Activity().deserialize(body)
auth_header = req.headers["Authorization"] if "Authorization" in req.headers else ""
try:
response = await adapter.process_activity(activity, auth_header, bot.on_turn)
if response:
return Response(status=response.status, body=response.body)
return Response(status=201)
except Exception as exception:
raise exception
Create web application
app = web.Application() app.router.add_post("/api/messages", messages)
Start server
Figure: Configuration and management dashboard with status overview.
if name == "main":
try:
web.run_app(app, host="0.0.0.0", port=3978)
except Exception as error:
raise error
## Dialog Management
Dialog management is the core of multi-turn conversations, enabling the bot to guide users through complex workflows by maintaining conversation context across multiple message exchanges.
### Waterfall Dialogs (Sequential Step-by-Step Flows)
Waterfall dialogs execute a series of sequential steps where each step prompts the user, processes the response, and advances to the next step. This pattern is ideal for linear workflows with predictable sequences (booking, forms, troubleshooting wizards).
```python
from botbuilder.dialogs import (
```text
WaterfallDialog,
WaterfallStepContext,
DialogTurnResult,
ComponentDialog,
DialogSet,
DialogTurnStatus```
)
from botbuilder.dialogs.prompts import (
```text
TextPrompt,
ChoicePrompt,
ConfirmPrompt,
NumberPrompt,
DateTimePrompt,
PromptOptions,
PromptValidatorContext```
)
from botbuilder.core import MessageFactory, StatePropertyAccessor
from botbuilder.schema import InputHints, Activity
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class FlightBookingDialog(ComponentDialog):
```python
"""
Multi-turn dialog for flight booking with validation and confirmation.
Demonstrates waterfall pattern with prompts, validation, and state management.
"""
def __init__(self, dialog_id: str = None):
super().__init__(dialog_id or FlightBookingDialog.__name__)
# Add prompts with custom validators
self.add_dialog(
TextPrompt("DestinationPrompt", FlightBookingDialog.destination_validator)
)
self.add_dialog(
DateTimePrompt("DatePrompt", FlightBookingDialog.date_validator)
)
self.add_dialog(
NumberPrompt("PassengersPrompt", FlightBookingDialog.passengers_validator)
)
self.add_dialog(
ChoicePrompt("ClassPrompt")
)
self.add_dialog(
ConfirmPrompt("ConfirmPrompt")
)
# Define waterfall steps
self.add_dialog(
WaterfallDialog(
"BookingWaterfall",
[
self.destination_step,
self.date_step,
self.passengers_step,
self.class_step,
self.confirm_step,
self.final_step
]
)
)
self.initial_dialog_id = "BookingWaterfall"
async def destination_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""First step: Ask for destination."""
prompt_message = MessageFactory.text(
"Where would you like to travel? (e.g., New York, London, Tokyo)",
"Where would you like to travel?",
InputHints.expecting_input
)
return await step_context.prompt("DestinationPrompt", PromptOptions(prompt=prompt_message))
async def date_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Second step: Store destination and ask for travel date."""
step_context.values["destination"] = step_context.result
prompt_message = MessageFactory.text(
f"When would you like to travel to {step_context.result}? (e.g., tomorrow, next Monday, 2025-07-15)",
f"When would you like to travel to {step_context.result}?",
InputHints.expecting_input
)
return await step_context.prompt("DatePrompt", PromptOptions(prompt=prompt_message))
async def passengers_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Third step: Store date and ask for number of passengers."""
# DateTimePrompt returns a list of DateTimeResolution
date_resolutions = step_context.result
if date_resolutions:
step_context.values["date"] = date_resolutions[0].value
prompt_message = MessageFactory.text(
"How many passengers? (1-9)",
"How many passengers?",
InputHints.expecting_input
)
return await step_context.prompt("PassengersPrompt", PromptOptions(prompt=prompt_message))
async def class_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Fourth step: Store passengers and ask for cabin class."""
step_context.values["passengers"] = step_context.result
prompt_message = MessageFactory.text(
"Which cabin class would you prefer?",
"Which cabin class?",
InputHints.expecting_input
)
return await step_context.prompt(
"ClassPrompt",
PromptOptions(
prompt=prompt_message,
choices=["Economy", "Premium Economy", "Business", "First Class"]
)
)
async def confirm_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Fifth step: Store class and present confirmation summary."""
step_context.values["class"] = step_context.result.value
# Build confirmation message
destination = step_context.values["destination"]
date = step_context.values["date"]
passengers = step_context.values["passengers"]
cabin_class = step_context.values["class"]
confirmation_message = (
f"**Flight Booking Summary:**\n\n"
f"- **Destination:** {destination}\n"
f"- **Date:** {date}\n"
f"- **Passengers:** {passengers}\n"
f"- **Class:** {cabin_class}\n\n"
f"Would you like to proceed with this booking?"
)
prompt_message = MessageFactory.text(
confirmation_message,
confirmation_message,
InputHints.expecting_input
)
return await step_context.prompt("ConfirmPrompt", PromptOptions(prompt=prompt_message))
async def final_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Final step: Process confirmation and complete booking."""
if step_context.result: # User confirmed
booking_data = {
"destination": step_context.values["destination"],
"date": step_context.values["date"],
"passengers": step_context.values["passengers"],
"class": step_context.values["class"]
}
logger.info(f"Booking confirmed: {booking_data}")
# Simulate booking API call
confirmation_number = f"BK{datetime.now().strftime('%Y%m%d%H%M%S')}"
success_message = (
f"✅ **Booking Confirmed!**\n\n"
f"Confirmation Number: **{confirmation_number}**\n\n"
f"You'll receive a confirmation email shortly. "
f"Thank you for booking with us!"
)
await step_context.context.send_activity(MessageFactory.text(success_message))
# Return booking data to parent dialog/bot
return await step_context.end_dialog(booking_data)
else: # User cancelled
logger.info("Booking cancelled by user")
await step_context.context.send_activity(
MessageFactory.text("Booking cancelled. Is there anything else I can help you with?")
)
return await step_context.end_dialog()
## Custom validators
@staticmethod
async def destination_validator(prompt_context: PromptValidatorContext) -> bool:
"""Validate destination is not empty and has reasonable length."""
if not prompt_context.recognized.succeeded:
await prompt_context.context.send_activity(
"Please enter a valid destination."
)
return False
destination = prompt_context.recognized.value.strip()
if len(destination) < 2:
await prompt_context.context.send_activity(
"Destination must be at least 2 characters."
)
return False
if len(destination) > 50:
await prompt_context.context.send_activity(
"Destination is too long. Please use a shorter name."
)
return False
return True
@staticmethod
async def date_validator(prompt_context: PromptValidatorContext) -> bool:
"""Validate travel date is in the future."""
if not prompt_context.recognized.succeeded:
await prompt_context.context.send_activity(
"Please enter a valid date (e.g., tomorrow, next week, 2025-07-15)."
)
return False
# DateTimePrompt returns list of DateTimeResolution
date_resolutions = prompt_context.recognized.value
if not date_resolutions:
return False
# Parse the date
try:
date_str = date_resolutions[0].value
travel_date = datetime.fromisoformat(date_str)
# Check if date is in the future
if travel_date.date() < datetime.now().date():
await prompt_context.context.send_activity(
"Travel date must be in the future. Please choose a later date."
)
return False
# Check if date is not too far in the future (e.g., within 1 year)
if travel_date.date() > (datetime.now() + timedelta(days=365)).date():
await prompt_context.context.send_activity(
"We can only book flights within the next year. Please choose an earlier date."
)
return False
return True
except Exception as e:
logger.error(f"Date validation error: {e}")
return False
@staticmethod
async def passengers_validator(prompt_context: PromptValidatorContext) -> bool:
"""Validate passenger count is between 1 and 9."""
if not prompt_context.recognized.succeeded:
await prompt_context.context.send_activity(
"Please enter a valid number of passengers (1-9)."
)
return False
passengers = prompt_context.recognized.value
if passengers < 1 or passengers > 9:
await prompt_context.context.send_activity(
"Number of passengers must be between 1 and 9."
)
return False
return True
## Component Dialogs (Reusable Dialog Components)
Component dialogs encapsulate complex dialog logic into reusable components that can be composed into larger workflows:
```python
class MainDialog(ComponentDialog):
```python
"""
Main dialog orchestrating multiple sub-dialogs.
Demonstrates dialog composition and routing.
"""
def __init__(self):
super().__init__(MainDialog.__name__)
# Add child dialogs
self.add_dialog(FlightBookingDialog())
self.add_dialog(OrderStatusDialog())
self.add_dialog(AccountInfoDialog())
self.add_dialog(TextPrompt("TextPrompt"))
# Main waterfall
self.add_dialog(
WaterfallDialog(
"MainWaterfall",
[
self.intro_step,
self.route_step,
self.final_step
]
)
)
self.initial_dialog_id = "MainWaterfall"
async def intro_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Present main menu options."""
message = (
"What would you like to do today?\n\n"
"1. Book a flight\n"
"2. Check order status\n"
"3. View account information\n"
"4. Speak to an agent\n\n"
"Please type a number or describe your request."
)
return await step_context.prompt(
"TextPrompt",
PromptOptions(prompt=MessageFactory.text(message))
)
async def route_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Route to appropriate sub-dialog based on user choice."""
user_choice = step_context.result.lower().strip()
if "1" in user_choice or "book" in user_choice or "flight" in user_choice:
return await step_context.begin_dialog(FlightBookingDialog.__name__)
elif "2" in user_choice or "order" in user_choice or "status" in user_choice:
return await step_context.begin_dialog(OrderStatusDialog.__name__)
elif "3" in user_choice or "account" in user_choice or "information" in user_choice:
return await step_context.begin_dialog(AccountInfoDialog.__name__)
elif "4" in user_choice or "agent" in user_choice or "human" in user_choice:
await step_context.context.send_activity(
"Connecting you to a live agent. Please wait..."
)
# Trigger handoff to human agent
return await step_context.end_dialog({"handoff": True})
else:
await step_context.context.send_activity(
"I didn't understand that option. Let's try again."
)
return await step_context.replace_dialog(self.id)
async def final_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Complete the main dialog."""
result = step_context.result
if result and isinstance(result, dict) and result.get("handoff"):
# Handoff to agent was triggered
return await step_context.end_dialog(result)
# Ask if user needs anything else
message = "Is there anything else I can help you with?"
await step_context.context.send_activity(MessageFactory.text(message))
# Restart main dialog
return await step_context.replace_dialog(self.id)
## Natural Language Understanding
### Conversational Language Understanding (CLU) Integration
Conversational Language Understanding (CLU) is the successor to LUIS, providing advanced natural language understanding with improved accuracy (95%+), better entity extraction, and streamlined project management through Azure AI Language Service.
#### CLU Project Setup
```bash
## Create Azure AI Language Service resource
az cognitiveservices account create \
--name CustomerSupportLanguage \
--resource-group ConversationalAI-RG \
--kind TextAnalytics \
--sku S \
--location eastus \
--yes
## Get endpoint and key
ENDPOINT=$(az cognitiveservices account show \
--name CustomerSupportLanguage \
--resource-group ConversationalAI-RG \
--query properties.endpoint --output tsv)
KEY=$(az cognitiveservices account keys list \
--name CustomerSupportLanguage \
--resource-group ConversationalAI-RG \
--query key1 --output tsv)
echo "CLU_ENDPOINT=$ENDPOINT"
echo "CLU_KEY=$KEY"
CLU Recognizer Implementation
Figure: Configuration and management dashboard with status overview.
from azure.core.credentials import AzureKeyCredential
from azure.ai.language.conversations import ConversationAnalysisClient
from botbuilder.core import TurnContext, RecognizerResult
import os
import logging
logger = logging.getLogger(__name__)
class CLURecognizer:
```python
"""
Conversational Language Understanding recognizer for intent and entity extraction.
Processes user utterances to determine intent and extract structured information.
"""
def __init__(
self,
endpoint: str = None,
key: str = None,
project_name: str = "CustomerSupportBot",
deployment_name: str = "production"
):
self.endpoint = endpoint or os.environ.get("CLU_ENDPOINT")
self.key = key or os.environ.get("CLU_KEY")
self.project_name = project_name
self.deployment_name = deployment_name
# Initialize CLU client
credential = AzureKeyCredential(self.key)
self.client = ConversationAnalysisClient(self.endpoint, credential)
async def recognize(self, turn_context: TurnContext) -> RecognizerResult:
"""
Analyze user utterance and return recognized intent and entities.
"""
utterance = turn_context.activity.text
try:
# Call CLU analysis
result = self.client.analyze_conversation(
task={
"kind": "Conversation",
"analysisInput": {
"conversationItem": {
"participantId": "user",
"id": "1",
"modality": "text",
"language": "en",
"text": utterance
}
},
"parameters": {
"projectName": self.project_name,
"deploymentName": self.deployment_name,
"verbose": True
}
}
)
# Extract prediction
prediction = result["result"]["prediction"]
top_intent = prediction["topIntent"]
confidence = prediction["intents"][0]["confidenceScore"]
entities = prediction.get("entities", [])
logger.info(
f"CLU Result - Intent: {top_intent} (confidence: {confidence:.2f}), "
f"Entities: {len(entities)}"
)
# Convert to RecognizerResult format
recognizer_result = RecognizerResult(
text=utterance,
intents={
intent["category"]: {"score": intent["confidenceScore"]}
for intent in prediction["intents"]
},
entities=self._extract_entities(entities)
)
# Add top intent
recognizer_result.properties["topIntent"] = top_intent
recognizer_result.properties["topIntentScore"] = confidence
return recognizer_result
except Exception as e:
logger.error(f"CLU recognition error: {e}")
# Return empty result on error
return RecognizerResult(
text=utterance,
intents={"None": {"score": 1.0}},
entities={}
)
def _extract_entities(self, entities: list) -> dict:
"""
Convert CLU entities to RecognizerResult format.
"""
extracted = {}
for entity in entities:
category = entity["category"]
text = entity["text"]
confidence = entity["confidenceScore"]
if category not in extracted:
extracted[category] = []
extracted[category].append({
"text": text,
"confidence": confidence,
"offset": entity.get("offset"),
"length": entity.get("length")
})
return extracted
Usage in bot logic
Figure: Copilot Studio – topic flow designer with test chat panel.
class IntentBot(ActivityHandler):
"""
Bot that routes to dialogs based on CLU intent recognition.
"""
def __init__(self, clu_recognizer: CLURecognizer, dialog_set: DialogSet):
self.clu_recognizer = clu_recognizer
self.dialog_set = dialog_set
async def on_message_activity(self, turn_context: TurnContext):
"""Process message with CLU intent routing."""
# Recognize intent
recognizer_result = await self.clu_recognizer.recognize(turn_context)
top_intent = recognizer_result.properties.get("topIntent")
confidence = recognizer_result.properties.get("topIntentScore", 0.0)
logger.info(f"Top intent: {top_intent} (confidence: {confidence:.2f})")
# Route based on intent with confidence threshold
if confidence < 0.6:
# Low confidence - ask for clarification
await turn_context.send_activity(
"I'm not sure I understood that. Could you rephrase or choose from:\n"
"- Book a flight\n"
"- Check order status\n"
"- Account information\n"
"- Speak to an agent"
)
return
# High confidence - route to appropriate dialog
if top_intent == "BookFlight":
# Extract entities for pre-filling dialog
entities = recognizer_result.entities
destination = entities.get("Destination", [{}])[0].get("text")
date = entities.get("Date", [{}])[0].get("text")
logger.info(f"Booking flight - Destination: {destination}, Date: {date}")
# Begin booking dialog with pre-filled data
dialog_context = await self.dialog_set.create_context(turn_context)
initial_data = {"destination": destination, "date": date}
await dialog_context.begin_dialog(FlightBookingDialog.__name__, initial_data)
elif top_intent == "OrderStatus":
order_number = entities.get("OrderNumber", [{}])[0].get("text")
await self._check_order_status(turn_context, order_number)
elif top_intent == "AccountInfo":
await dialog_context.begin_dialog(AccountInfoDialog.__name__)
elif top_intent == "EscalateToAgent":
await self._escalate_to_agent(turn_context)
elif top_intent == "Greeting":
await turn_context.send_activity(
"Hello! How can I assist you today? I can help with:\n"
"- Flight bookings\n"
"- Order status\n"
"- Account information"
)
elif top_intent == "Cancel":
await turn_context.send_activity("Request cancelled. What else can I help with?")
else:
# Fallback
await turn_context.send_activity(
f"I recognized you want to '{top_intent}', but I'm not sure how to help with that yet. "
"Could you try rephrasing?"
)
async def _check_order_status(self, turn_context: TurnContext, order_number: str):
"""Check order status from backend system."""
if not order_number:
await turn_context.send_activity("What's your order number?")
return
# Simulate API call
logger.info(f"Checking order status for: {order_number}")
await turn_context.send_activity(
f"Your order {order_number} is currently:\n"
f"**Status:** Shipped\n"
f"**Expected Delivery:** Tomorrow by 5 PM\n"
f"**Tracking:** 1Z999AA10123456784"
)
async def _escalate_to_agent(self, turn_context: TurnContext):
"""Escalate to human agent."""
await turn_context.send_activity(
"Connecting you to a live agent. Please wait..."
)
# Trigger handoff event to live agent platform (Dynamics Omnichannel, etc.)
logger.warning(f"Escalation triggered for user: {turn_context.activity.from_property.id}")
## Training CLU Models
CLU models are trained through Azure AI Language Studio with labeled utterances:
**Example Training Data (JSON):**
```json
{
"projectKind": "Conversation",
"intents": [
```json
{
"category": "BookFlight",
"examples": [
"Book a flight to New York",
"I want to fly to London next week",
"Can you help me book a trip to Paris?",
"I need a flight to Tokyo on July 15th"
]
},
{
"category": "OrderStatus",
"examples": [
"Where is my order?",
"Check order status for 12345",
"Track my shipment",
"What's the status of order BK20250616?"
]
},
{
"category": "EscalateToAgent",
"examples": [
"I need to speak to a person",
"Connect me to an agent",
"This isn't working, I need human help",
"Can I talk to someone?"
]
}```
],
"entities": [
```json
{
"category": "Destination",
"examples": [
{"text": "New York", "offset": 16, "length": 8},
{"text": "London", "offset": 15, "length": 6}
]
},
{
"category": "OrderNumber",
"examples": [
{"text": "12345", "offset": 24, "length": 5},
{"text": "BK20250616", "offset": 23, "length": 10}
]
}```
]
}
Best Practices for CLU Training:
- Diverse utterances: Include 15-50 examples per intent with varied phrasing, synonyms, sentence structures
- Balanced intents: Similar number of examples per intent to avoid bias
- Entity labeling: Mark all entity mentions in training data for accurate extraction
- Test set: Reserve 20% of data for testing to measure true accuracy
- Active learning: Periodically review low-confidence predictions and add to training data
- Multilingual: Train with utterances in all target languages (CLU supports 90+ languages)
State Management
State management enables bots to maintain context across multiple conversation turns by persisting user preferences, conversation data, and dialog state. Bot Framework SDK provides three state scopes: User State (persists across all conversations with a user), Conversation State (persists within a single conversation), and Private Conversation State (persists within a single conversation for a specific user in group chats).
Storage Backends
from botbuilder.core import (
```text
ConversationState,
UserState,
MemoryStorage,
BotState```
)
from botbuilder.azure import CosmosDbPartitionedStorage, BlobStorage
import os
## Memory Storage (development only - data lost on restart)
memory_storage = MemoryStorage()
## Azure Blob Storage (production)
blob_storage = BlobStorage(
```text
connection_string=os.environ["AZURE_STORAGE_CONNECTION_STRING"],
container_name="bot-state"```
)
## Azure Cosmos DB (production - recommended for high-scale)
cosmos_config = {
```text
"endpoint": os.environ["COSMOS_ENDPOINT"],
"master_key": os.environ["COSMOS_KEY"],
"database": "botdb",
"container": "botstate"```
}
cosmos_storage = CosmosDbPartitionedStorage(cosmos_config)
## Use Cosmos DB for production
storage = cosmos_storage
State Configuration and Accessors
Figure: Site permissions – groups, external sharing, and access request settings.
from botbuilder.core import (
```text
ConversationState,
UserState,
TurnContext,
StatePropertyAccessor```
)
from typing import Dict, Any
class BotStateManager:
```python
"""
Centralized state management for bot.
Provides strongly-typed accessors for user and conversation data.
"""
def __init__(self, storage):
# Create state management objects
self.user_state = UserState(storage)
self.conversation_state = ConversationState(storage)
# User state accessors (persist across all conversations)
self.user_profile_accessor = self.user_state.create_property("UserProfile")
self.user_preferences_accessor = self.user_state.create_property("UserPreferences")
self.authentication_data_accessor = self.user_state.create_property("AuthData")
# Conversation state accessors (persist within current conversation)
self.conversation_data_accessor = self.conversation_state.create_property("ConversationData")
self.dialog_state_accessor = self.conversation_state.create_property("DialogState")
async def get_user_profile(self, turn_context: TurnContext) -> Dict[str, Any]:
"""Get or create user profile."""
return await self.user_profile_accessor.get(
turn_context,
lambda: {
"name": None,
"email": None,
"phone": None,
"preferred_language": "en",
"created_at": None,
"total_interactions": 0
}
)
async def set_user_profile(self, turn_context: TurnContext, profile: Dict[str, Any]):
"""Update user profile."""
await self.user_profile_accessor.set(turn_context, profile)
async def get_user_preferences(self, turn_context: TurnContext) -> Dict[str, Any]:
"""Get user preferences for personalization."""
return await self.user_preferences_accessor.get(
turn_context,
lambda: {
"notifications_enabled": True,
"preferred_cabin_class": "Economy",
"home_airport": None,
"newsletter_subscribed": False
}
)
async def get_conversation_data(self, turn_context: TurnContext) -> Dict[str, Any]:
"""Get conversation-specific data (current session only)."""
return await self.conversation_data_accessor.get(
turn_context,
lambda: {
"started_at": None,
"turn_count": 0,
"intents_recognized": [],
"escalation_requested": False,
"sentiment_scores": []
}
)
async def increment_turn_count(self, turn_context: TurnContext):
"""Track conversation turns for analytics."""
conversation_data = await self.get_conversation_data(turn_context)
conversation_data["turn_count"] += 1
await self.conversation_data_accessor.set(turn_context, conversation_data)
async def save_all_changes(self, turn_context: TurnContext):
"""Persist all state changes."""
await self.user_state.save_changes(turn_context)
await self.conversation_state.save_changes(turn_context)
async def clear_conversation_state(self, turn_context: TurnContext):
"""Clear conversation state (e.g., on error or reset)."""
await self.conversation_state.clear_state(turn_context)
await self.conversation_state.save_changes(turn_context)
Usage in bot
Figure: Copilot Studio – topic flow designer with test chat panel.
class StatefulBot(ActivityHandler):
"""
Bot with comprehensive state management.
"""
def __init__(self, state_manager: BotStateManager):
self.state_manager = state_manager
async def on_message_activity(self, turn_context: TurnContext):
"""Handle message with state tracking."""
# Load state
user_profile = await self.state_manager.get_user_profile(turn_context)
conversation_data = await self.state_manager.get_conversation_data(turn_context)
# Increment turn count
await self.state_manager.increment_turn_count(turn_context)
# Update user profile on first interaction
if user_profile["name"] is None:
user_profile["name"] = turn_context.activity.from_property.name
user_profile["created_at"] = turn_context.activity.timestamp.isoformat()
await self.state_manager.set_user_profile(turn_context, user_profile)
# Increment total interactions
user_profile["total_interactions"] += 1
await self.state_manager.set_user_profile(turn_context, user_profile)
# Personalized greeting
name = user_profile.get("name", "there")
turn_count = conversation_data["turn_count"]
response = f"Hello {name}! This is turn {turn_count} of our conversation. "
response += f"You've interacted with me {user_profile['total_interactions']} times total."
await turn_context.send_activity(response)
# Save all state changes
await self.state_manager.save_all_changes(turn_context)
async def on_turn_error(self, turn_context: TurnContext, error: Exception):
"""Clear state on error to prevent stuck dialogs."""
logger.error(f"Bot error: {error}")
await turn_context.send_activity("Sorry, an error occurred. Starting fresh...")
# Clear conversation state but keep user state
await self.state_manager.clear_conversation_state(turn_context)
## Dialog State Management
Dialog state is automatically managed by the dialog system but requires explicit configuration:
```python
from botbuilder.dialogs import DialogSet, DialogTurnStatus
class DialogBot(ActivityHandler):
```python
"""
Bot with dialog state management.
"""
def __init__(self, conversation_state: ConversationState, main_dialog: Dialog):
self.conversation_state = conversation_state
self.dialog_state_accessor = conversation_state.create_property("DialogState")
self.dialog_set = DialogSet(self.dialog_state_accessor)
self.dialog_set.add(main_dialog)
self.main_dialog_id = main_dialog.id
async def on_message_activity(self, turn_context: TurnContext):
"""Route messages through dialog system."""
# Create dialog context
dialog_context = await self.dialog_set.create_context(turn_context)
# Continue existing dialog or start new one
results = await dialog_context.continue_dialog()
if results.status == DialogTurnStatus.Empty:
# No active dialog - start main dialog
await dialog_context.begin_dialog(self.main_dialog_id)
# Save state after each turn
await self.conversation_state.save_changes(turn_context)
## Authentication and Security
### OAuth 2.0 Authentication Flow
Authenticate users with Azure AD for accessing protected resources (Microsoft Graph, backend APIs):
```python
from botbuilder.core import TurnContext, MessageFactory
from botbuilder.schema import Activity, ActivityTypes, TokenResponse
from botbuilder.dialogs import (
```text
WaterfallDialog,
WaterfallStepContext,
DialogTurnResult,
ComponentDialog```
)
from botbuilder.dialogs.prompts import OAuthPrompt, OAuthPromptSettings
import aiohttp
import logging
logger = logging.getLogger(__name__)
class AuthenticationDialog(ComponentDialog):
```python
"""
OAuth 2.0 authentication dialog for Azure AD sign-in.
Enables bot to access user data from Microsoft Graph with user consent.
"""
def __init__(self, connection_name: str):
super().__init__(AuthenticationDialog.__name__)
self.connection_name = connection_name
# OAuth prompt settings
oauth_settings = OAuthPromptSettings(
connection_name=connection_name,
text="Please sign in to continue",
title="Sign In",
timeout=300000 # 5 minutes
)
self.add_dialog(OAuthPrompt("OAuthPrompt", oauth_settings))
self.add_dialog(
WaterfallDialog(
"AuthWaterfall",
[
self.prompt_step,
self.login_step,
self.display_token_step
]
)
)
self.initial_dialog_id = "AuthWaterfall"
async def prompt_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Prompt user to sign in."""
return await step_context.begin_dialog("OAuthPrompt")
async def login_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Process sign-in token."""
# Get token response
token_response: TokenResponse = step_context.result
if token_response and token_response.token:
logger.info(f"User authenticated: {token_response.connection_name}")
# Store token in user state for subsequent API calls
user_state = step_context.context.turn_state.get("user_state")
if user_state:
auth_data = {"token": token_response.token, "provider": token_response.connection_name}
await user_state.set(step_context.context, auth_data)
# Get user profile from Microsoft Graph
user_profile = await self._get_user_profile(token_response.token)
if user_profile:
step_context.values["user_profile"] = user_profile
return await step_context.next(user_profile)
else:
await step_context.context.send_activity("Failed to retrieve user profile.")
return await step_context.end_dialog()
else:
await step_context.context.send_activity("Authentication failed. Please try again.")
return await step_context.end_dialog()
async def display_token_step(self, step_context: WaterfallStepContext) -> DialogTurnResult:
"""Display authenticated user information."""
user_profile = step_context.values.get("user_profile")
if user_profile:
message = (
f"Welcome, {user_profile.get('displayName', 'User')}!\n\n"
f"Email: {user_profile.get('mail', user_profile.get('userPrincipalName', 'N/A'))}\n"
f"Job Title: {user_profile.get('jobTitle', 'N/A')}"
)
await step_context.context.send_activity(MessageFactory.text(message))
return await step_context.end_dialog(user_profile)
async def _get_user_profile(self, token: str) -> dict:
"""Call Microsoft Graph to get user profile."""
graph_url = "https://graph.microsoft.com/v1.0/me"
headers = {"Authorization": f"Bearer {token}"}
try:
async with aiohttp.ClientSession() as session:
async with session.get(graph_url, headers=headers) as response:
if response.status == 200:
return await response.json()
else:
logger.error(f"Graph API error: {response.status}")
return None
except Exception as e:
logger.error(f"Graph API call failed: {e}")
return None
Sign-out handler
async def sign_out_user(turn_context: TurnContext, connection_name: str):
"""
Sign out user from OAuth provider.
"""
try:
adapter = turn_context.adapter
await adapter.sign_out_user(turn_context, connection_name)
await turn_context.send_activity("You have been signed out.")
logger.info(f"User signed out: {turn_context.activity.from_property.id}")
except Exception as e:
logger.error(f"Sign-out error: {e}")
await turn_context.send_activity("Error signing out. Please try again.")
## Azure AD Application Registration
```bash
## Register Azure AD app for bot authentication
az ad app create \
--display-name "CustomerSupportBot" \
--sign-in-audience AzureADMyOrg \
--web-redirect-uris "https://token.botframework.com/.auth/web/redirect"
## Get application ID
APP_ID=$(az ad app list --display-name "CustomerSupportBot" --query [0].appId -o tsv)
## Create client secret
SECRET=$(az ad app credential reset --id $APP_ID --query password -o tsv)
echo "APP_ID=$APP_ID"
echo "APP_SECRET=$SECRET"
## Configure API permissions (Microsoft Graph)
az ad app permission add \
--id $APP_ID \
--api 00000003-0000-0000-c000-000000000000 \
--api-permissions e1fe6dd8-ba31-4d61-89e7-88639da4683d=Scope # User.Read
## Grant admin consent
az ad app permission admin-consent --id $APP_ID
Multi-Channel Deployment
Azure Bot Service supports 10+ channels out-of-the-box with unified bot code. Each channel has specific capabilities (rich UI, file uploads, voice) that require channel-specific handling.
Supported Channels
- Microsoft Teams: Enterprise collaboration (chat, channels, meetings, tabs, messaging extensions, adaptive cards)
- Slack: Team communication (slash commands, interactive messages, home tab, workflows)
- Facebook Messenger: Consumer messaging (quick replies, templates, persistent menu)
- Telegram: Secure messaging (inline keyboards, bot commands, file sharing)
- Web Chat: Website embedding (customizable UI, speech, video)
- SMS (Twilio): Text messaging (simple text, media messages)
- WhatsApp Business: Consumer messaging (message templates, media, location)
- Email: Asynchronous communication (attachments, HTML formatting)
- Direct Line: Custom applications (REST/WebSocket API, full control)
- Alexa/Cortana: Voice assistants (speech synthesis, skill integration)
Channel-Specific Handling
from botbuilder.core import TurnContext, MessageFactory, CardFactory
from botbuilder.schema import Activity, Attachment, HeroCard, CardAction, ActionTypes
class MultiChannelBot(ActivityHandler):
```python
"""
Bot with channel-specific features and adaptive responses.
"""
async def on_message_activity(self, turn_context: TurnContext):
"""Handle messages with channel-specific logic."""
channel_id = turn_context.activity.channel_id
logger.info(f"Message received from channel: {channel_id}")
# Route to channel-specific handlers
if channel_id == "msteams":
await self._handle_teams_message(turn_context)
elif channel_id == "slack":
await self._handle_slack_message(turn_context)
elif channel_id == "facebook":
await self._handle_facebook_message(turn_context)
elif channel_id == "sms":
await self._handle_sms_message(turn_context)
elif channel_id == "webchat":
await self._handle_webchat_message(turn_context)
else:
# Generic handler for other channels
await self._handle_generic_message(turn_context)
async def _handle_teams_message(self, turn_context: TurnContext):
"""Microsoft Teams-specific features."""
# Teams supports rich adaptive cards
card = self._create_adaptive_card()
message = MessageFactory.attachment(card)
await turn_context.send_activity(message)
# Teams also supports @mentions, file uploads, meeting events
if turn_context.activity.attachments:
for attachment in turn_context.activity.attachments:
if attachment.content_type == "application/vnd.microsoft.teams.file.download.info":
logger.info(f"File uploaded: {attachment.name}")
async def _handle_slack_message(self, turn_context: TurnContext):
"""Slack-specific features (blocks, interactive components)."""
# Slack uses Block Kit for rich UI
message = MessageFactory.text("Welcome to our support bot!")
message.channel_data = {
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*How can we help you today?*"
}
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "Book Flight"},
"value": "book_flight",
"action_id": "book_flight"
},
{
"type": "button",
"text": {"type": "plain_text", "text": "Order Status"},
"value": "order_status",
"action_id": "order_status"
}
]
}
]
}
await turn_context.send_activity(message)
async def _handle_sms_message(self, turn_context: TurnContext):
"""SMS-specific handling (plain text only, character limits)."""
# SMS has 160 character limit per message
response = "Your order #12345 shipped today. Track: bit.ly/track12345"
await turn_context.send_activity(MessageFactory.text(response))
async def _handle_webchat_message(self, turn_context: TurnContext):
"""Web chat with custom styling and suggested actions."""
message = MessageFactory.text("How can I assist you?")
message.suggested_actions = {
"actions": [
{"type": "imBack", "title": "📦 Track Order", "value": "track order"},
{"type": "imBack", "title": "✈️ Book Flight", "value": "book flight"},
{"type": "imBack", "title": "👤 Account Info", "value": "account info"},
{"type": "imBack", "title": "🤝 Talk to Agent", "value": "talk to agent"}
]
}
await turn_context.send_activity(message)
async def _handle_generic_message(self, turn_context: TurnContext):
"""Fallback handler for channels without specific features."""
await turn_context.send_activity(
MessageFactory.text(f"You said: {turn_context.activity.text}")
)
def _create_adaptive_card(self) -> Attachment:
"""Create adaptive card for rich UI (Teams, Outlook, etc.)."""
return CardFactory.adaptive_card({
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.4",
"body": [
{
"type": "TextBlock",
"text": "Customer Support",
"weight": "Bolder",
"size": "Large"
},
{
"type": "TextBlock",
"text": "How can we help you today?",
"wrap": True
}
],
"actions": [
{
"type": "Action.Submit",
"title": "Book Flight",
"data": {"action": "book_flight"}
},
{
"type": "Action.Submit",
"title": "Order Status",
"data": {"action": "order_status"}
}
]
})
Channel Configuration
## Enable Microsoft Teams channel
az bot msteams create \
--resource-group ConversationalAI-RG \
--name CustomerSupportBot \
--enable-calling false \
--calling-web-hook ""
## Enable Slack channel (requires Slack App credentials)
az bot slack create \
--resource-group ConversationalAI-RG \
--name CustomerSupportBot \
--client-id "<slack-client-id>" \
--client-secret "<slack-client-secret>" \
--verification-token "<slack-verification-token>"
## Enable SMS channel via Twilio
az bot sms create \
--resource-group ConversationalAI-RG \
--name CustomerSupportBot \
--account-sid "<twilio-account-sid>" \
--auth-token "<twilio-auth-token>" \
--phone "<twilio-phone-number>"
## Enable Web Chat (no additional config needed - embed token in website)
az bot webchat show \
--resource-group ConversationalAI-RG \
--name CustomerSupportBot
```text
## Adaptive Cards
Adaptive Cards are platform-agnostic UI cards that render natively across channels (Teams, Outlook, web chat). They support rich formatting, input controls, actions, and data binding for interactive user experiences.
### Comprehensive Adaptive Card Example
```python
from botbuilder.core import CardFactory, MessageFactory
from botbuilder.schema import Attachment
import json
def create_flight_booking_card(destination: str, date: str) -> Attachment:
```sql
"""
Create rich adaptive card for flight booking with input fields and actions.
"""
card_json = {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.4",
"body": [
{
"type": "Container",
"style": "emphasis",
"items": [
{
"type": "ColumnSet",
"columns": [
{
"type": "Column",
"width": "auto",
"items": [
{
"type": "Image",
"url": "https://mycompany.azurewebsites.net/flight-icon.png",
"size": "Medium"
}
]
},
{
"type": "Column",
"width": "stretch",
"items": [
{
"type": "TextBlock",
"text": "Flight Booking",
"weight": "Bolder",
"size": "Large"
},
{
"type": "TextBlock",
"text": f"Destination: {destination}",
"spacing": "None",
"isSubtle": True
}
]
}
]
}
]
},
{
"type": "Container",
"items": [
{
"type": "TextBlock",
"text": "Complete your booking details:",
"wrap": True,
"separator": True
},
{
"type": "Input.Text",
"id": "passengerName",
"label": "Passenger Name",
"": "John Doe",
"isRequired": True,
"errorMessage": "Name is required"
},
{
"type": "Input.Date",
"id": "departureDate",
"label": "Departure Date",
"value": date,
"isRequired": True
},
{
"type": "Input.Number",
"id": "passengers",
"label": "Number of Passengers",
"min": 1,
"max": 9,
"value": 1,
"isRequired": True
},
{
"type": "Input.ChoiceSet",
"id": "cabinClass",
"label": "Cabin Class",
"style": "compact",
"value": "economy",
"choices": [
{"title": "Economy", "value": "economy"},
{"title": "Premium Economy", "value": "premium"},
{"title": "Business", "value": "business"},
{"title": "First Class", "value": "first"}
]
},
{
"type": "Input.Toggle",
"id": "needsHotel",
"title": "Also book hotel?",
"value": "false"
}
]
},
{
"type": "Container",
"items": [
{
"type": "FactSet",
"facts": [
{"title": "Base Fare:", "value": "$450.00"},
{"title": "Taxes & Fees:", "value": "$85.50"},
{"title": "**Total:**", "value": "**$535.50**"}
]
}
]
}
],
"actions": [
{
"type": "Action.Submit",
"title": "Confirm Booking",
"style": "positive",
"data": {
"action": "confirmBooking",
"destination": destination
}
},
{
"type": "Action.Submit",
"title": "Modify Search",
"data": {
"action": "modifySearch"
}
},
{
"type": "Action.Submit",
"title": "Cancel",
"style": "destructive",
"data": {
"action": "cancel"
}
}
]
}
return CardFactory.adaptive_card(card_json)
Send adaptive card
async def send_booking_card(turn_context: TurnContext, destination: str, date: str):
"""Send flight booking adaptive card."""
card = create_flight_booking_card(destination, date)
message = MessageFactory.attachment(card)
await turn_context.send_activity(message)
Handle adaptive card submissions
Figure: Teams adaptive card – card designer with preview panel.
async def on_teams_card_action_invoke(turn_context: TurnContext):
"""Process adaptive card action submissions."""
card_data = turn_context.activity.value
action = card_data.get("action")
if action == "confirmBooking":
passenger_name = card_data.get("passengerName")
departure_date = card_data.get("departureDate")
passengers = card_data.get("passengers")
cabin_class = card_data.get("cabinClass")
needs_hotel = card_data.get("needsHotel") == "true"
logger.info(
f"Booking submitted: {passenger_name}, {departure_date}, "
f"{passengers} passengers, {cabin_class} class, hotel: {needs_hotel}"
)
# Process booking
confirmation_card = create_confirmation_card(passenger_name, departure_date)
return await turn_context.send_activity(MessageFactory.attachment(confirmation_card))
elif action == "modifySearch":
await turn_context.send_activity("Let's modify your search...")
elif action == "cancel":
await turn_context.send_activity("Booking cancelled.")
def create_confirmation_card(passenger_name: str, date: str) -> Attachment:
"""Create booking confirmation card with success message."""
card_json = {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.4",
"body": [
{
"type": "Container",
"style": "good",
"items": [
{
"type": "TextBlock",
"text": "✅ Booking Confirmed!",
"weight": "Bolder",
"size": "Large",
"color": "Good"
}
]
},
{
"type": "TextBlock",
"text": f"Thank you, {passenger_name}! Your flight is confirmed for {date}.",
"wrap": True
},
{
"type": "FactSet",
"facts": [
{"title": "Confirmation Number:", "value": "BK20250616001"},
{"title": "Status:", "value": "Confirmed"},
{"title": "Email Sent:", "value": "Yes"}
]
}
]
}
return CardFactory.adaptive_card(card_json)
## Proactive Messaging
Proactive messages allow bots to initiate conversations with users (order updates, reminders, alerts) without waiting for user input:
```python
from botbuilder.core import BotAdapter, TurnContext
from botbuilder.schema import ConversationReference, Activity, ConversationAccount, ChannelAccount
import asyncio
import logging
logger = logging.getLogger(__name__)
class ProactiveMessageManager:
```python
"""
Manage proactive messaging with conversation reference storage.
"""
def __init__(self, adapter: BotAdapter, app_id: str):
self.adapter = adapter
self.app_id = app_id
self.conversation_references = {} # Store in database for production
def store_conversation_reference(self, turn_context: TurnContext):
"""Store conversation reference for later proactive messaging."""
reference = TurnContext.get_conversation_reference(turn_context.activity)
user_id = turn_context.activity.from_property.id
self.conversation_references[user_id] = reference
logger.info(f"Stored conversation reference for user: {user_id}")
async def send_proactive_message(
self,
user_id: str,
message_text: str,
card: Attachment = None
):
"""
Send proactive message to user.
"""
reference = self.conversation_references.get(user_id)
if not reference:
logger.warning(f"No conversation reference found for user: {user_id}")
return False
async def send_message(turn_context: TurnContext):
if card:
await turn_context.send_activity(MessageFactory.attachment(card))
else:
await turn_context.send_activity(MessageFactory.text(message_text))
logger.info(f"Proactive message sent to user: {user_id}")
try:
await self.adapter.continue_conversation(
reference,
send_message,
self.app_id
)
return True
except Exception as e:
logger.error(f"Failed to send proactive message: {e}")
return False
async def send_bulk_proactive_messages(self, user_ids: list, message_text: str):
"""Send proactive messages to multiple users (e.g., announcements)."""
tasks = [
self.send_proactive_message(user_id, message_text)
for user_id in user_ids
]
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = sum(1 for r in results if r is True)
logger.info(f"Sent proactive messages: {successful}/{len(user_ids)} successful")
return successful
Usage in bot
Figure: Copilot Studio – topic flow designer with test chat panel.
class ProactiveBot(ActivityHandler):
"""
Bot with proactive messaging capability.
"""
def __init__(self, adapter: BotAdapter, app_id: str):
self.proactive_manager = ProactiveMessageManager(adapter, app_id)
async def on_message_activity(self, turn_context: TurnContext):
"""Store conversation reference on every interaction."""
self.proactive_manager.store_conversation_reference(turn_context)
# Handle message
await turn_context.send_activity("Message received. I'll notify you of updates!")
async def send_order_shipped_notification(self, user_id: str, order_number: str):
"""Example: Send order shipped notification."""
message = (
f"📦 **Order Shipped!**\n\n"
f"Your order {order_number} has been shipped and is on its way!\n"
f"Expected delivery: Tomorrow by 5 PM"
)
await self.proactive_manager.send_proactive_message(user_id, message)
## Testing and Debugging
### Bot Framework Emulator
The Bot Framework Emulator is a desktop tool for testing bots locally:
```bash
## Download Bot Framework Emulator
## https://github.com/Microsoft/BotFramework-Emulator/releases
## Start bot locally
python app.py
## In Emulator:
## 1. Click "Open Bot"
## 2. Enter bot URL: http://localhost:3978/api/messages
## 3. Leave Microsoft App ID and Password empty for local testing
## 4. For production testing, enter bot credentials
```text
## Comprehensive Unit Testing
```python
import pytest
from botbuilder.testing import DialogTestClient
from botbuilder.core import TurnContext, ConversationState, MemoryStorage
from botbuilder.dialogs import DialogSet
from unittest.mock import Mock, AsyncMock, patch
import json
@pytest.fixture
def dialog_test_client():
```text
"""Create dialog test client fixture."""
storage = MemoryStorage()
conversation_state = ConversationState(storage)
dialog = FlightBookingDialog()
return DialogTestClient(
"test",
dialog,
initial_dialog_options=None,
middlewares=[]
)
@pytest.mark.asyncio async def test_booking_dialog_full_flow(dialog_test_client):
"""Test complete booking dialog flow."""
## Start dialog
reply = await dialog_test_client.send_activity("book flight")
assert "Where would you like to travel?" in reply.text
## Enter destination
reply = await dialog_test_client.send_activity("New York")
assert "When would you like to travel" in reply.text
## Enter date
reply = await dialog_test_client.send_activity("tomorrow")
assert "How many passengers?" in reply.text
## Enter passengers
reply = await dialog_test_client.send_activity("2")
assert "cabin class" in reply.text.lower()
## Select class
reply = await dialog_test_client.send_activity("Business")
assert "confirm" in reply.text.lower()
assert "New York" in reply.text
## Confirm
reply = await dialog_test_client.send_activity("yes")
assert "Booking Confirmed" in reply.text or "confirmed" in reply.text.lower()
@pytest.mark.asyncio async def test_booking_dialog_cancellation(dialog_test_client):
"""Test booking cancellation flow."""
await dialog_test_client.send_activity("book flight")
await dialog_test_client.send_activity("London")
await dialog_test_client.send_activity("next week")
await dialog_test_client.send_activity("1")
await dialog_test_client.send_activity("Economy")
## Cancel booking
reply = await dialog_test_client.send_activity("no")
assert "cancelled" in reply.text.lower()
@pytest.mark.asyncio async def test_clu_recognizer():
"""Test CLU intent recognition."""
with patch("azure.ai.language.conversations.ConversationAnalysisClient") as mock_client:
# Mock CLU response
mock_client.return_value.analyze_conversation.return_value = {
"result": {
"prediction": {
"topIntent": "BookFlight",
"intents": [{"category": "BookFlight", "confidenceScore": 0.95}],
"entities": [
{"category": "Destination", "text": "Paris", "confidenceScore": 0.92}
]
}
}
}
recognizer = CLURecognizer(
endpoint="https://test.cognitiveservices.azure.com",
key="test-key"
)
# Create mock turn context
activity = Activity(text="Book a flight to Paris", type="message")
turn_context = TurnContext(Mock(), activity)
result = await recognizer.recognize(turn_context)
assert result.properties["topIntent"] == "BookFlight"
assert result.properties["topIntentScore"] == 0.95
assert "Destination" in result.entities
assert result.entities["Destination"][0]["text"] == "Paris"
@pytest.mark.asyncio async def test_state_management():
"""Test user and conversation state persistence."""
storage = MemoryStorage()
state_manager = BotStateManager(storage)
## Create mock turn context
activity = Activity(
type="message",
text="test",
from_property=ChannelAccount(id="user123", name="Test User")
)
turn_context = TurnContext(Mock(), activity)
## Get initial user profile
profile = await state_manager.get_user_profile(turn_context)
assert profile["total_interactions"] == 0
## Update profile
profile["name"] = "Test User"
profile["total_interactions"] = 5
await state_manager.set_user_profile(turn_context, profile)
await state_manager.save_all_changes(turn_context)
## Verify persistence
profile2 = await state_manager.get_user_profile(turn_context)
assert profile2["name"] == "Test User"
assert profile2["total_interactions"] == 5
def test_adaptive_card_creation():
"""Test adaptive card JSON generation."""
card = create_flight_booking_card("Tokyo", "2025-07-15")
assert card.content_type == "application/vnd.microsoft.card.adaptive"
card_json = card.content
assert card_json["type"] == "AdaptiveCard"
assert "Tokyo" in json.dumps(card_json)
assert len(card_json["actions"]) >= 2 # At least Confirm and Cancel
Integration test
Figure: Test Studio – recorded test cases, assertions, and execution results.
@pytest.mark.integration @pytest.mark.asyncio async def test_end_to_end_conversation():
"""End-to-end conversation test with CLU and dialogs."""
## Setup
storage = MemoryStorage()
conversation_state = ConversationState(storage)
clu_recognizer = CLURecognizer(
endpoint=os.environ["CLU_ENDPOINT"],
key=os.environ["CLU_KEY"]
)
bot = IntentBot(clu_recognizer, DialogSet())
## Simulate conversation
activity = Activity(
type="message",
text="I want to book a flight to London",
from_property=ChannelAccount(id="user123")
)
turn_context = TurnContext(Mock(), activity)
await bot.on_message_activity(turn_context)
## Verify dialog started (in real test, check turn_context calls)
## This is a simplified example
## Production Monitoring and Operations
### Application Insights Integration
```python
from botbuilder.applicationinsights import ApplicationInsightsTelemetryClient
from botbuilder.core import BotTelemetryClient, TurnContext
from applicationinsights import TelemetryClient
import logging
import time
class BotMonitoring:
```python
"""
Comprehensive monitoring for conversational AI bots.
"""
def __init__(self, instrumentation_key: str):
self.telemetry_client = TelemetryClient(instrumentation_key)
self.logger = logging.getLogger(__name__)
def track_conversation_turn(
self,
turn_context: TurnContext,
intent: str,
confidence: float,
response_time_ms: float
):
"""Track conversation metrics."""
properties = {
"userId": turn_context.activity.from_property.id,
"channelId": turn_context.activity.channel_id,
"intent": intent,
"confidence": confidence,
"conversationId": turn_context.activity.conversation.id
}
measurements = {
"responseTime": response_time_ms,
"messageLength": len(turn_context.activity.text or "")
}
self.telemetry_client.track_event(
"ConversationTurn",
properties,
measurements
)
def track_dialog_completion(
self,
dialog_id: str,
turn_count: int,
success: bool,
duration_seconds: float
):
"""Track dialog completion metrics."""
properties = {
"dialogId": dialog_id,
"success": str(success)
}
measurements = {
"turnCount": turn_count,
"duration": duration_seconds
}
self.telemetry_client.track_event(
"DialogCompleted",
properties,
measurements
)
def track_escalation(self, turn_context: TurnContext, reason: str):
"""Track escalations to human agents."""
properties = {
"userId": turn_context.activity.from_property.id,
"reason": reason,
"channelId": turn_context.activity.channel_id
}
self.telemetry_client.track_event("BotEscalation", properties)
self.logger.warning(f"Escalation: {reason} for user {turn_context.activity.from_property.id}")
def track_error(self, error: Exception, turn_context: TurnContext):
"""Track bot errors."""
properties = {
"userId": turn_context.activity.from_property.id if turn_context else "unknown",
"errorType": type(error).__name__
}
self.telemetry_client.track_exception(error, properties=properties)
self.logger.error(f"Bot error: {error}", exc_info=error)
Middleware for automatic monitoring
Figure: Azure Monitor Logs – KQL query results with time-series visualization.
class TelemetryMiddleware:
"""
Middleware to automatically track all conversation metrics.
"""
def __init__(self, monitoring: BotMonitoring):
self.monitoring = monitoring
async def on_turn(self, turn_context: TurnContext, next_fn):
"""Track metrics for each turn."""
start_time = time.time()
try:
# Process turn
await next_fn()
# Calculate response time
response_time = (time.time() - start_time) * 1000
# Track metrics (intent will be tracked separately by CLU recognizer)
self.monitoring.track_conversation_turn(
turn_context,
intent="Unknown", # Will be enriched by recognizer
confidence=0.0,
response_time_ms=response_time
)
except Exception as error:
self.monitoring.track_error(error, turn_context)
raise
## Key Performance Indicators (KPIs)
| Metric | Target | Alert Threshold | Business Impact | Measurement Method |
|--------|--------|-----------------|-----------------|-------------------|
| **Conversation Success Rate** | 75%+ | <65% | User satisfaction, deflection rate | (Completed dialogs / Total conversations) × 100 |
| **Intent Recognition Accuracy** | 90%+ | <85% | Correct routing, user experience | CLU confidence scores, manual validation |
| **Average Response Time** | <2s | >5s | User experience, engagement | Time from user message to bot response |
| **Deflection Rate** | 60%+ | <50% | Cost savings, agent workload | (Resolved by bot / Total inquiries) × 100 |
| **Escalation Rate** | <25% | >35% | Bot effectiveness, complexity handling | (Escalated to agent / Total conversations) × 100 |
| **User Satisfaction (CSAT)** | 75%+ | <65% | Product quality, retention | Post-conversation surveys, sentiment analysis |
| **Active Users (DAU/MAU)** | Growth trend | 20% decline | Bot adoption, value proposition | Unique users per day/month |
| **Average Conversation Turns** | 3-7 | <2 or >10 | Dialog efficiency, complexity | Turns per completed conversation |
| **Entity Extraction Accuracy** | 85%+ | <75% | Data quality, downstream processing | Manual validation of extracted entities |
**Monitoring Queries (Azure Monitor/KQL):**
```kusto
// Conversation success rate by channel
customEvents
| where name == "DialogCompleted"
| extend Success = tobool(customDimensions.success)
| summarize
```text
TotalConversations = count(),
SuccessfulConversations = countif(Success == true),
SuccessRate = round(100.0 * countif(Success == true) / count(), 2)```
by tostring(customDimensions.channelId)
| project channelId = customDimensions_channelId, TotalConversations, SuccessfulConversations, SuccessRate
// Average response time by intent
customEvents
| where name == "ConversationTurn"
| extend ResponseTime = todouble(customMeasurements.responseTime)
| extend Intent = tostring(customDimensions.intent)
| summarize
```text
AvgResponseTime = round(avg(ResponseTime), 2),
P95ResponseTime = round(percentile(ResponseTime, 95), 2),
Count = count()```
by Intent
| order by AvgResponseTime desc
// Escalation rate trends
customEvents
| where name == "BotEscalation"
| summarize EscalationCount = count() by bin(timestamp, 1h)
| join kind=leftouter (
```text
customEvents
| where name == "ConversationTurn"
| summarize TotalTurns = count() by bin(timestamp, 1h)```
) on timestamp
| project timestamp, EscalationRate = round(100.0 * EscalationCount / TotalTurns, 2)
| render timechart
// Low confidence intents requiring review
customEvents
| where name == "ConversationTurn"
| extend Confidence = todouble(customDimensions.confidence)
| where Confidence < 0.7
| extend Intent = tostring(customDimensions.intent)
| summarize Count = count() by Intent
| order by Count desc
| take 10
> **Architecture Overview:** ## Conversational AI Maturity Model
## Architecture Decision and Tradeoffs
When designing AI/ML solutions with Azure AI Services, consider these key architectural trade-offs:
| Approach | Best For | Tradeoff |
|----------|----------|----------|
| Managed / platform service | Rapid delivery, reduced ops burden | Less customisation, potential vendor lock-in |
| Custom / self-hosted | Full control, advanced tuning | Higher operational overhead and cost |
> **Recommendation:** Start with the managed approach for most workloads and move to custom only when specific requirements demand it.
## Security and Governance Considerations
- **Least Privilege:** Grant only the permissions required for each role
- **Secret Management:** Store credentials in Azure Key Vault or equivalent; never hard-code secrets
- **Audit Logging:** Enable diagnostic and activity logs for compliance and forensic analysis
- **Data Protection:** Encrypt data at rest and in transit; classify data with sensitivity labels where applicable
## Cost and Performance Notes
- **Primary Cost Drivers:** Compute tier, storage volume, and network egress
- **Optimization Levers:** Right-size resources, use reserved instances or savings plans, and review Azure Advisor recommendations regularly
- **Performance Baseline:** Define SLAs, latency targets, and throughput thresholds before going live
- **Scaling Strategy:** Use auto-scale rules and monitor utilisation to balance cost and responsiveness
## Validation and Versioning
- **Last Validated:** April 2026
- **Tested With:** Current generally-available Azure AI Services APIs and SDKs
- **Known Constraints:** Check regional availability and service limits before production deployment
## Official Microsoft References
- [Microsoft Learn – Azure AI Services](https://learn.microsoft.com)
- [Azure AI Services Documentation](https://learn.microsoft.com)
- [Azure Architecture Center](https://learn.microsoft.com/azure/architecture/)
## Public Examples from Official Sources
- [Microsoft official samples on GitHub](https://github.com/Azure-Samples)
- [Microsoft Learn training modules](https://learn.microsoft.com/training/)
Discussion