AI-агент для колл-центру: цифровий співробітник Цифровий співробітник колл-центру - це не IVR і не чат-бот з кнопками. Це система, яка веде повноцінні неструктуровані розмови, звертається до даних клієнта у реальному часі, вирішує завдання самостійно і передає ті звернення, які вимагають людини. Різниця в архітектурі важлива: замість дерева рішень — LLM з інструментами та доступом до бізнес-систем. ### Архітектура production-grade AI агента```python
import asyncio import json from typing import Optional, Callable from openai import AsyncOpenAI from twilio.rest import Client as TwilioClient
class CallCenterAIAgent: """ Автономный агент колл-центра. Обрабатывает звонки через Twilio + OpenAI Realtime API. """
def __init__(self, openai_api_key: str,
twilio_account_sid: str,
twilio_auth_token: str,
system_prompt: str,
escalation_handler: Callable):
self.openai = AsyncOpenAI(api_key=openai_api_key)
self.twilio = TwilioClient(twilio_account_sid, twilio_auth_token)
self.system_prompt = system_prompt
self.escalation_handler = escalation_handler
# Инструменты агента
self.tools = self._define_tools()
def _define_tools(self) -> list[dict]:
"""Инструменты для работы агента во время звонка"""
return [
{
"type": "function",
"function": {
"name": "lookup_customer",
"description": "Поиск клиента в базе по номеру телефона или email",
"parameters": {
"type": "object",
"properties": {
"phone": {"type": "string", "description": "Номер телефона"},
"email": {"type": "string", "description": "Email адрес"}
}
}
}
},
{
"type": "function",
"function": {
"name": "get_order_status",
"description": "Статус заказа по его номеру",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string"}
},
"required": ["order_id"]
}
}
},
{
"type": "function",
"function": {
"name": "create_ticket",
"description": "Создать тикет поддержки для сложного обращения",
"parameters": {
"type": "object",
"properties": {
"summary": {"type": "string", "description": "Краткое описание проблемы"},
"priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]},
"category": {"type": "string"}
},
"required": ["summary", "priority"]
}
}
},
{
"type": "function",
"function": {
"name": "transfer_to_human",
"description": "Переключить звонок на живого оператора. Используй при: эскалации, жалобах, сложных технических вопросах, запросе клиента.",
"parameters": {
"type": "object",
"properties": {
"reason": {"type": "string", "description": "Причина переключения"},
"department": {
"type": "string",
"enum": ["general", "technical", "billing", "complaints"]
},
"context_summary": {"type": "string", "description": "Краткий контекст для оператора"}
},
"required": ["reason", "department"]
}
}
}
]
async def process_call_turn(self, conversation_history: list[dict],
user_message: str,
customer_context: dict) -> dict:
"""
Обработка одного хода разговора.
Возвращает: ответ агента, действия (tool calls), флаг эскалации.
"""
messages = [
{
"role": "system",
"content": self.system_prompt + f"\n\nКонтекст клиента: {json.dumps(customer_context, ensure_ascii=False)}"
}
] + conversation_history + [
{"role": "user", "content": user_message}
]
response = await self.openai.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=self.tools,
tool_choice="auto",
temperature=0.4,
)
message = response.choices[0].message
actions_taken = []
should_escalate = False
# Обрабатываем tool calls
if message.tool_calls:
tool_results = []
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
if func_name == "transfer_to_human":
should_escalate = True
await self.escalation_handler(
reason=func_args.get("reason"),
department=func_args.get("department"),
context=func_args.get("context_summary", "")
)
result = {"status": "transfer_initiated", "department": func_args.get("department")}
else:
result = await self._execute_tool(func_name, func_args)
actions_taken.append({"tool": func_name, "args": func_args, "result": result})
tool_results.append({
"tool_call_id": tool_call.id,
"role": "tool",
"content": json.dumps(result, ensure_ascii=False)
})
# Финальный ответ с результатами инструментов
if not should_escalate:
messages.append({"role": "assistant", "content": None, "tool_calls": message.tool_calls})
messages.extend(tool_results)
final_response = await self.openai.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.4,
)
agent_response = final_response.choices[0].message.content
else:
agent_response = "Позвольте переключить вас на специалиста. Один момент."
else:
agent_response = message.content
return {
"response": agent_response,
"actions": actions_taken,
"escalated": should_escalate,
}
async def _execute_tool(self, tool_name: str, args: dict) -> dict:
"""Диспетчер вызовов инструментов"""
# В production: реальные API-вызовы к CRM, ERP, базе знаний
tool_handlers = {
"lookup_customer": self._lookup_customer,
"get_order_status": self._get_order_status,
"create_ticket": self._create_ticket,
}
handler = tool_handlers.get(tool_name)
if handler:
return await handler(**args)
return {"error": f"Unknown tool: {tool_name}"}
async def _lookup_customer(self, phone: str = None, email: str = None) -> dict:
# Stub: replace with real CRM API call
return {"customer_id": "C001", "name": "Иван Петров", "tier": "gold", "open_tickets": 1}
async def _get_order_status(self, order_id: str) -> dict:
return {"order_id": order_id, "status": "в доставке", "eta": "завтра до 18:00"}
async def _create_ticket(self, summary: str, priority: str, category: str = "general") -> dict:
return {"ticket_id": "T-45231", "priority": priority, "status": "created"}
### Управління якістю та моніторингpython
class CallQualityMonitor:
"""Автоматический контроль качества AI-агента"""
def analyze_call(self, transcript: list[dict],
resolution: str,
duration_seconds: int) -> dict:
"""
Оценка качества завершённого звонка.
resolution: 'resolved' | 'escalated' | 'abandoned'
"""
turns = len([t for t in transcript if t['role'] == 'user'])
agent_responses = [t['content'] for t in transcript if t['role'] == 'assistant']
# Метрики разговора
avg_response_length = sum(len(r.split()) for r in agent_responses) / max(len(agent_responses), 1)
# Проверка на нарушения (отказ в обслуживании, раскрытие системного промпта)
red_flags = []
for response in agent_responses:
if any(phrase in response.lower() for phrase in [
'я не могу', 'извините, но я', 'как ai', 'as an ai'
]):
red_flags.append('possible_refusal')
# Оценка FCR (First Call Resolution)
fcr = resolution == 'resolved' and turns <= 8
return {
'resolution': resolution,
'duration_seconds': duration_seconds,
'turns': turns,
'avg_response_words': round(avg_response_length),
'first_call_resolution': fcr,
'escalated': resolution == 'escalated',
'red_flags': red_flags,
'quality_score': 100 - len(red_flags) * 20 - (0 if fcr else 15),
}