Compartir a través de

Biblioteca cliente del paquete JobRouter de Azure Communication para Python, versión 1.0.0

  • Artículo

Este paquete contiene un SDK de Python para Azure Communication Services para JobRouter. Obtenga más información sobre Azure Communication Services aquí.

Código | fuentePaquete (Pypi) | Documentación del producto

Declinación de responsabilidades

Los paquetes de Python del SDK de Azure admiten Python 2.7 finalizó el 01 de enero de 2022. Para más información y preguntas, consulte https://github.com/Azure/azure-sdk-for-python/issues/20691.

Introducción

Requisitos previos

Necesita una suscripción de Azure y un recurso de Communication Service para usar este paquete.

Instalar el paquete

Instale la biblioteca cliente de JobRouter de Azure Communication para Python con pip:

pip install azure-communication-jobrouter

Conceptos clave

Trabajo

Un trabajo representa la unidad de trabajo que se debe enrutar a un rol de trabajo disponible. Un ejemplo real de esto puede ser una llamada entrante o chat en el contexto de un centro de llamadas.

Trabajo

Un trabajador representa la fuente disponible para controlar un trabajo. Cada trabajador se registra con o más colas para recibir trabajos. Un ejemplo real de esto puede ser un agente que trabaja en un centro de llamadas.

Cola

Una cola representa una lista ordenada de trabajos que esperan ser atendidos por un trabajador. Los trabajadores se registrarán con una cola para recibir trabajo de él. Un ejemplo real de esto puede ser una cola de llamadas en un centro de llamadas.

Canal

Un canal representa una agrupación de trabajos por algún tipo. Cuando un trabajador se registra para recibir trabajo, también debe especificar para qué canales puede administrar el trabajo, y cuánto de cada uno puede administrar de forma simultánea. Un ejemplo real de esto puede ser voice calls o chats en un centro de llamadas.

Oferta

JobRouter amplía una oferta a un trabajador para controlar un trabajo determinado cuando determina una coincidencia, esta notificación se entrega normalmente a través de EventGrid. El trabajo puede aceptar o rechazar la oferta mediante la API jobRouter, o expirará según el período de vida configurado en la directiva de distribución. Un ejemplo real de esto puede ser la llamada de un agente en un centro de llamadas.

Directiva de distribución

Una directiva de distribución representa un conjunto de configuración que rige cómo se distribuyen los trabajos de una cola a los trabajos registrados con esa cola. Esta configuración incluye cuánto tiempo es válida una oferta antes de que expire y el modo de distribución, que definen el orden en el que se seleccionan los trabajos cuando hay varios disponibles.

Modo de distribución

Los tres tipos de modos son

  • Round Robin: los trabajadores se ordenan por Id y se selecciona el siguiente trabajador después del anterior que obtuvo una oferta.
  • Máxima inactividad: el trabajador que más tiempo ha estado sin trabajar en un trabajo.
  • Mejor trabajo: puede especificar una expresión para comparar dos trabajos para determinar cuál elegir.

Etiquetas

Puede adjuntar etiquetas a trabajos, trabajos y colas. Estos son pares clave-valor que pueden ser de tipos de stringdatos o numberboolean . Un ejemplo real de esto puede ser el nivel de aptitud de un trabajador determinado o el equipo o la ubicación geográfica.

Selectores de etiquetas

Los selectores de etiquetas se pueden asociar a un trabajo para tener como destino un subconjunto de trabajos que sirven a la cola. Un ejemplo real de esto puede ser una condición en una llamada entrante que el agente debe tener un nivel mínimo de conocimiento de un producto determinado.

Directiva de clasificación

Se puede usar una directiva de clasificación para seleccionar dinámicamente una cola, determinar la prioridad del trabajo y adjuntar selectores de etiquetas de trabajo a un trabajo aprovechando un motor de reglas.

Directiva de excepciones

Una directiva de excepciones controla el comportamiento de un trabajo en función de un desencadenador y ejecuta una acción deseada. La directiva de excepciones se asocia a una cola para que pueda controlar el comportamiento de los trabajos de esa cola.

Ejemplos

Inicialización del cliente

Para inicializar el cliente SMS, se puede usar el cadena de conexión para crear instancias. Como alternativa, también puede usar la autenticación de Active Directory mediante DefaultAzureCredential.

from azure.communication.jobrouter import (
    JobRouterClient,
    JobRouterAdministrationClient
)

connection_string = "endpoint=ENDPOINT;accessKey=KEY"
router_client = JobRouterClient.from_connection_string(conn_str = connection_string)
router_admin_client = JobRouterAdministrationClient.from_connection_string(conn_str = connection_string)

Directiva de distribución

Para poder crear una cola, necesitamos una directiva de distribución.

from azure.communication.jobrouter.models import (
    LongestIdleMode,
    DistributionPolicy
)

distribution_policy: DistributionPolicy = DistributionPolicy(
    offer_expires_after_seconds = 24 * 60 * 60,
    mode = LongestIdleMode(
        min_concurrent_offers = 1,
        max_concurrent_offers = 1
    )
)

distribution_policy: DistributionPolicy = router_admin_client.upsert_distribution_policy(
    "distribution-policy-1",
    distribution_policy
)

Cola

A continuación, podemos crear la cola.

from azure.communication.jobrouter.models import (
    RouterQueue
)

queue: RouterQueue = RouterQueue(
    distribution_policy_id = "distribution-policy-1"
)

queue: RouterQueue = router_admin_client.upsert_queue(
    "queue-1",
    queue
)

Trabajo

Ahora se puede enviar un trabajo directamente a esa cola, con un selector de trabajo que exige que el trabajo tenga la etiqueta Some-Skill mayor que 10.

from azure.communication.jobrouter.models import (
    RouterJob,
    RouterWorkerSelector,
    LabelOperator
)

router_job: RouterJob = RouterJob(
    channel_id = "my-channel",
    queue_id = "queue-1",
    channel_reference = "12345",
    priority = 1,
    requested_worker_selectors = [
        RouterWorkerSelector(key = "Some-Skill", label_operator = LabelOperator.EQUAL, value = 10)
    ]
)

job: RouterJob = router_client.upsert_job(
    "jobId-1",
    router_job
)

Trabajo

Ahora, registramos un trabajo para recibir trabajo de esa cola, con una etiqueta igual Some-Skill a 11.

from azure.communication.jobrouter.models import (
    RouterWorker,
    RouterChannel
)

router_worker: RouterWorker = RouterWorker(
    capacity = 1,
    queues = [
        "queue-1"
    ],
    labels = {
        "Some-Skill": 11
    },
    channels = [
        RouterChannel(channel_id = "my-channel", capacity_cost_per_job = 1)
    ],
    available_for_offers = True
)

worker = router_client.upsert_worker(
    "worker-1",
    router_worker
)

Oferta

Se debe obtener un valor RouterWorkerOfferIssued de la suscripción de EventGrid.

Hay varios servicios de Azure diferentes que actúan como controlador de eventos. En este escenario, vamos a asumir webhooks para la entrega de eventos. Más información sobre la entrega de eventos de Webhook

Una vez que los eventos se entregan al controlador de eventos, podemos deserializar la carga JSON en una lista de eventos.

# Parse the JSON payload into a list of events
from azure.eventgrid import EventGridEvent
import json

## deserialize payload into a list of typed Events
events = [EventGridEvent.from_json(json.loads(msg)) for msg in payload]

offer_id = ""
for event in events:
    if event.event_type == "Microsoft.Communication.RouterWorkerOfferIssued":
        offer_id = event.data.offer_id
    else:
        continue

Pero también se podría esperar unos segundos y consultar el trabajo directamente en la API de JobRouter para ver si se le ha emitido una oferta.

from azure.communication.jobrouter.models import (
    RouterWorker,
)

router_worker: RouterWorker = router_client.get_worker(worker_id = "worker-1")

for offer in router_worker.offers:
    print(f"Worker {router_worker.id} has an active offer for job {offer.job_id}")

Aceptar una oferta

Una vez que un trabajador recibe una oferta, puede realizar dos acciones posibles: aceptar o rechazar. Aceptaremos la oferta.

from azure.communication.jobrouter.models import (
    RouterJobOffer,
    AcceptJobOfferResult,
    RouterJobStatus
)

# fetching the offer id
job_offer: RouterJobOffer = [offer for offer in router_worker.offers if offer.job_id == "jobId-1"][0]
offer_id = job_offer.offer_id

# accepting the offer sent to `worker-1`
accept_job_offer_result: AcceptJobOfferResult = router_client.accept_job_offer(
    worker_id = "worker-1",
    offer_id = offer_id
)

print(f"Offer: {job_offer.offer_id} sent to worker: {router_worker.id} has been accepted")
print(f"Job has been assigned to worker: {router_worker.id} with assignment: {accept_job_offer_result.assignment_id}")

# verify job assignment is populated when querying job
updated_job = router_client.get_job(job_id = "jobId-1")
print(f"Job assignment has been successful: {updated_job.job_status == RouterJobStatus.Assigned and accept_job_offer_result.assignment_id in updated_job.assignments}")

Finalización de un trabajo

Una vez que el trabajo se realiza con el trabajo, el trabajo tiene que marcar el trabajo como completed.

import datetime
from azure.communication.jobrouter.models import (
    CompleteJobOptions
)
complete_job_result = router_client.complete_job(
    "jobId-1",
    accept_job_offer_result.assignment_id,
    CompleteJobOptions(
        note = f"Job has been completed by {router_worker.id} at {datetime.datetime.utcnow()}"
    )
)

print(f"Job has been successfully completed.")

Cierre de un trabajo

Una vez completado un trabajo, el trabajo puede realizar acciones de resumen en el trabajo antes de cerrar el trabajo y finalmente liberar su capacidad para aceptar más trabajos entrantes.

from azure.communication.jobrouter.models import (
    RouterJob,
    RouterJobStatus,
    CloseJobOptions,
)

close_job_result = router_client.close_job(
    "jobId-1",
    accept_job_offer_result.assignment_id,
    CloseJobOptions(
        note = f"Job has been closed by {router_worker.id} at {datetime.datetime.utcnow()}"
    )
)

print(f"Job has been successfully closed.")

update_job: RouterJob = router_client.get_job(job_id = "jobId-1")
print(f"Updated job status: {update_job.job_status == RouterJobStatus.CLOSED}")

import time
from datetime import datetime, timedelta
from azure.communication.jobrouter.models import (
    RouterJob,
    RouterJobStatus,
    CloseJobOptions,
)

close_job_in_future_result = router_client.close_job(
    "jobId-1",
    accept_job_offer_result.assignment_id,
    CloseJobOptions(
        note = f"Job has been closed by {router_worker.id} at {datetime.utcnow()}",
        close_at = datetime.utcnow() + timedelta(seconds = 2)
    )
)

print(f"Job has been marked to close")
time.sleep(secs = 2)
update_job: RouterJob = router_client.get_job(job_id = "jobId-1")
print(f"Updated job status: {update_job.job_status == RouterJobStatus.CLOSED}")

Solución de problemas

¿Algún problema? Esta sección debe contener detalles sobre qué hacer allí.

Pasos siguientes

Más código de ejemplo

Eche un vistazo al directorio de ejemplos para obtener ejemplos detallados de cómo usar esta biblioteca.

Envío de comentarios

Si encuentra algún error o tiene sugerencias, envíe un problema en la sección Problemas del proyecto.

Contribuciones

Este proyecto agradece las contribuciones y sugerencias. La mayoría de las contribuciones requieren que acepte un Contrato de licencia para el colaborador (CLA) que declara que tiene el derecho a concedernos y nos concede los derechos para usar su contribución. Para obtener más información, visite cla.microsoft.com.

Este proyecto ha adoptado el Código de conducta de Microsoft Open Source. Para más información, consulte las preguntas más frecuentes del código de conducta o póngase en contacto con opencode@microsoft.com si tiene cualquier otra pregunta o comentario.