Development of an Autonomous AI Customer Communication System
An autonomous communication system manages the complete customer dialogue cycle: initiates communication based on triggers, maintains multi-turn conversations with context preservation, adapts tone to the customer, executes actions (booking, payment, order changes) within the dialogue, and correctly escalates to an operator when needed.
Key difference from FAQ bots: the system understands intent, manages dialogue state, remembers customer interaction history, and is capable of executing multi-step scenarios.
Dialogue System Architecture
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import TypedDict, Annotated, Optional
import operator
class ConversationState(TypedDict):
# Identification
session_id: str
customer_id: str
channel: str # "whatsapp", "telegram", "web_chat", "sms"
# Dialogue history
messages: Annotated[list, operator.add]
# Customer profile
customer_profile: Optional[dict]
customer_tier: str # "standard", "premium", "vip"
preferred_language: str # "ru", "en"
# Dialogue state
current_intent: Optional[str]
dialog_context: dict # Accumulated data from current scenario
pending_confirmations: list[dict] # Awaiting customer confirmation
# Executed actions
completed_actions: Annotated[list, operator.add]
# Escalation management
escalated: bool
escalation_reason: Optional[str]
escalation_priority: Optional[str]
Customer Context Manager
class CustomerContextManager:
"""Loads and updates customer profile"""
async def load_context(self, customer_id: str) -> dict:
profile_task = crm.get_customer(customer_id)
orders_task = orders_db.get_recent(customer_id, limit=10)
preferences_task = preference_store.get(customer_id)
loyalty_task = loyalty_service.get_status(customer_id)
results = await asyncio.gather(
profile_task, orders_task, preferences_task, loyalty_task,
return_exceptions=True,
)
return {
"profile": results[0] if not isinstance(results[0], Exception) else {},
"recent_orders": results[1] if not isinstance(results[1], Exception) else [],
"preferences": results[2] if not isinstance(results[2], Exception) else {},
"loyalty": results[3] if not isinstance(results[3], Exception) else {},
}
def build_system_context(self, customer_context: dict) -> str:
profile = customer_context.get("profile", {})
loyalty = customer_context.get("loyalty", {})
context_parts = [
f"Customer: {profile.get('name', 'customer')}",
f"Status: {loyalty.get('tier', 'standard')}",
f"Language: {profile.get('preferred_language', 'ru')}",
]
orders = customer_context.get("recent_orders", [])
if orders:
last_order = orders[0]
context_parts.append(
f"Last order: #{last_order['id']} from {last_order['date']}, "
f"status: {last_order['status']}"
)
return "\n".join(context_parts)
Intent Detection and Scenario Management System
from pydantic import BaseModel
from typing import Literal
class IntentDetection(BaseModel):
intent: Literal[
"order_status", "order_change", "order_cancel",
"delivery_issue", "return_request", "payment_issue",
"product_question", "complaint", "compliment",
"account_management", "general_question", "farewell",
]
confidence: float
entities: dict # Extracted entities (order number, date, etc.)
requires_action: bool # Need to execute action in system
needs_clarification: bool # Need additional data from customer
SYSTEM_PROMPT_TEMPLATE = """You are an AI customer service assistant for company "{company_name}".
Customer information:
{customer_context}
Communication rules:
- Address customer by name
- Tone: {tone} (depends on customer status)
- Language: {language}
- Never promise what you cannot deliver
- On errors — acknowledge and offer solution
- Do not disclose internal systems and databases
Available actions:
- Order status
- Change delivery address (if order not transferred to courier)
- Initiate return
- Reschedule delivery date
- Transfer to operator
If you don't know the answer — be honest and offer to transfer to specialist."""
def build_system_prompt(state: ConversationState) -> str:
tone_map = {
"standard": "professional, friendly",
"premium": "personal, attentive",
"vip": "exclusive, highly personalized",
}
return SYSTEM_PROMPT_TEMPLATE.format(
company_name="RetailCo",
customer_context=state.get("customer_profile", {}).get("context", ""),
tone=tone_map.get(state["customer_tier"], "professional"),
language=state["preferred_language"],
)
Dialogue Action Execution
async def execute_dialog_action(action_name: str, params: dict, state: ConversationState) -> dict:
"""Executes action and returns result for dialogue inclusion"""
action_handlers = {
"get_order_status": lambda p: orders_api.get_status(p["order_id"]),
"change_delivery_address": lambda p: orders_api.update_address(
p["order_id"], p["new_address"]
),
"initiate_return": lambda p: returns_service.create_request(
order_id=p["order_id"],
reason=p["reason"],
customer_id=state["customer_id"],
),
"reschedule_delivery": lambda p: delivery_api.reschedule(
p["order_id"], p["new_date"]
),
}
handler = action_handlers.get(action_name)
if not handler:
return {"success": False, "error": f"Unknown action: {action_name}"}
try:
result = await handler(params)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
Confirmation Management (Confirmation Flow)
def needs_confirmation(action_name: str) -> bool:
"""Actions requiring customer confirmation"""
return action_name in {"cancel_order", "initiate_return", "change_payment_method"}
async def handle_pending_confirmation(state: ConversationState) -> ConversationState:
"""Handles customer response to confirmation request"""
if not state["pending_confirmations"]:
return state
last_message = state["messages"][-1].content.lower()
confirmation_words = {"да", "yes", "подтверждаю", "согласен", "ок", "ok"}
rejection_words = {"нет", "no", "отмена", "отказываюсь", "стоп"}
if any(word in last_message for word in confirmation_words):
# Execute pending action
pending = state["pending_confirmations"][0]
result = await execute_dialog_action(pending["action"], pending["params"], state)
return {
**state,
"pending_confirmations": state["pending_confirmations"][1:],
"completed_actions": [{"action": pending["action"], "result": result}],
}
elif any(word in last_message for word in rejection_words):
return {
**state,
"pending_confirmations": [],
"messages": [AIMessage("OK, action cancelled. How else can I help?")],
}
# Ambiguous response — ask again
return {
**state,
"messages": [AIMessage("Please answer 'yes' to confirm or 'no' to cancel.")],
}
Trigger-Based Communication
class OutboundCommunicationEngine:
"""Initiates outbound communication based on business triggers"""
TRIGGER_TEMPLATES = {
"order_shipped": {
"message": "Your order #{order_id} has shipped! Tracking: {tracking_url}. Expected delivery: {eta}.",
"channel_priority": ["sms", "whatsapp", "email"],
},
"delivery_delay": {
"message": "We're notifying you of a delivery delay for order #{order_id}. New date: {new_eta}. Sorry for the inconvenience.",
"channel_priority": ["whatsapp", "telegram", "sms"],
},
"return_approved": {
"message": "Your return for order #{order_id} has been approved. Funds will return within {refund_days} days.",
"channel_priority": ["email", "whatsapp"],
},
}
async def send_trigger_message(self, customer_id: str, trigger: str, params: dict):
template_config = self.TRIGGER_TEMPLATES.get(trigger)
if not template_config:
return
# Personalize message via LLM
customer = await crm.get_customer(customer_id)
base_message = template_config["message"].format(**params)
if customer.get("tier") in ("premium", "vip"):
# VIP gets more personalized message
personalized = await personalize_message(base_message, customer)
else:
personalized = base_message
channel = await self.get_preferred_channel(customer_id, template_config["channel_priority"])
await channel_dispatcher.send(customer_id, channel, personalized)
Practical Case: Telecom Operator, 18,000 Dialogues/Day
Company: Regional telecom with 850,000 subscribers, 120-person contact center.
Implemented scenarios:
- Balance and traffic checking
- Service activation/deactivation
- Number unlock (verification by code word)
- Technical problem diagnostics (by algorithm)
- Quality complaint handling (auto ticket creation + compensation per rules)
- Payment receipt via dialogue (redirect to payment form)
Results:
- Autonomous dialogue resolution: 67%
- Average response time: 4.5 min → 8 seconds
- CSAT (Customer Satisfaction Score): 3.8/5.0 → 4.2/5.0 (unexpected increase due to speed)
- Operators focused on complaints and complex cases
- NPS: +7 points per quarter
Challenges: Tone adjustment for angry customers (4 weeks of iterations). VIP clients expect "live" communication — added option for immediate human transfer.
Timeline
- Architecture and basic dialogue engine: 2–3 weeks
- Scenario implementation (each ~3–5 days): from 3 weeks
- Communication channel integration: 1–2 weeks
- CRM/ERP/API integration: 2–3 weeks
- Training, testing, launch: 2–3 weeks
- Total: 10–14 weeks