Python용 Azure Communication JobRouter 패키지 클라이언트 라이브러리 - 버전 1.0.0
이 패키지에는 JobRouter용 Azure Communication Services 대한 Python SDK가 포함되어 있습니다. Azure Communication Services 대한 자세한 내용은 여기를 참조하세요.
고지 사항
Python 2.7에 대한 Azure SDK Python 패키지 지원은 2022년 1월 1일에 종료되었습니다. 자세한 내용과 질문은 을 참조하세요. https://github.com/Azure/azure-sdk-for-python/issues/20691
시작
필수 구성 요소
이 패키지를 사용하려면 Azure 구독 및 Communication Service 리소스 가 필요합니다.
- 이 패키지를 사용하려면 Python 3.7 이상이 필요합니다. 자세한 내용은 Python용 Azure SDK 버전 지원 정책 페이지를 참조하세요.
- 새 Communication Service를 만들려면 Azure Portal, Azure PowerShell
패키지 설치
pip를 사용하여 Python용 Azure Communication JobRouter 클라이언트 라이브러리를 설치합니다.
pip install azure-communication-jobrouter
주요 개념
작업
작업은 사용 가능한 작업자로 라우팅해야 하는 작업 단위를 나타냅니다. 이 예제의 실제 예는 콜 센터의 컨텍스트에서 들어오는 전화 또는 채팅일 수 있습니다.
작업자
작업자는 작업을 처리하는 데 사용할 수 있는 공급을 나타냅니다. 각 작업자는 작업을 받기 위해 큐 이상을 등록합니다. 실제 예제는 콜 센터에서 작업하는 에이전트일 수 있습니다.
큐
큐는 작업자가 제공되기를 기다리는 정렬된 작업 목록을 나타냅니다. 작업자는 큐에 등록하여 작업을 받습니다. 이 예제의 실제 예는 콜 센터의 통화 큐일 수 있습니다.
채널
채널은 일부 형식별 작업 그룹화입니다. 작업자가 작업을 받기 위해 등록하는 경우, 작업을 처리할 수 있는 채널과 각 채널에서 동시에 처리할 수 있는 양도 지정해야 합니다.
실제 예제는 콜 센터에 있거나 chats 있을 voice calls 수 있습니다.
제안
제품이 일치 항목을 결정할 때 특정 작업을 처리하기 위해 JobRouter에서 작업자로 확장되며, 이 알림은 일반적으로 EventGrid를 통해 전달됩니다. 작업자는 th JobRouter API를 사용하여 제안을 수락하거나 거절하거나 배포 정책에 구성된 라이브 시간에 따라 만료됩니다. 이에 대한 실제 예는 콜 센터에서 에이전트를 울리는 것일 수 있습니다.
배포 정책
배포 정책은 큐의 작업이 해당 큐에 등록된 작업자에게 배포되는 방식을 제어하는 구성 집합을 나타냅니다. 이 구성에는 제품이 만료되기 전에 유효한 기간과 여러 사용 가능한 경우 작업자가 선택되는 순서를 정의하는 배포 모드가 포함됩니다.
배포 모드
3가지 유형의 모드는 다음과 같습니다.
- 라운드 로빈: 작업자가
Id를 기준으로 정렬되고, 이전에 제안을 받은 작업자 다음 작업자가 선택됩니다. - 최장 유휴 상태: 가장 오랜 기간 작업을 진행하지 않은 작업자입니다.
- Best Worker: 식을 지정하여 2명의 작업자를 비교하여 선택할 작업자를 결정할 수 있습니다.
레이블
작업자, 작업 및 큐에 레이블을 연결할 수 있습니다. 이러한 쌍은 또는 numberboolean 데이터 형식일 수 있는 키 값 쌍입니다string.
이러한 실제 예는 특정 작업자 또는 팀 또는 지리적 위치의 기술 수준일 수 있습니다.
레이블 선택기
레이블 선택기는 큐에 서비스를 제공하는 작업자의 하위 집합을 대상으로 하기 위해 작업에 연결할 수 있습니다. 이에 대한 실제 예는 에이전트가 특정 제품에 대한 최소 수준의 지식을 가져야 한다는 들어오는 호출의 조건일 수 있습니다.
분류 정책
분류 정책을 사용하여 큐를 동적으로 선택하고, 작업 우선 순위를 결정하고, 규칙 엔진을 활용하여 작업자 레이블 선택기를 작업에 연결할 수 있습니다.
예외 정책
예외 정책에서는 트리거에 따라 작업의 동작을 제어하고 원하는 작업을 실행합니다. 예외 정책은 큐에서 작업의 동작을 제어할 수 있도록 큐에 연결됩니다.
예제
클라이언트 초기화
SMS 클라이언트를 초기화하기 위해 연결 문자열 사용하여 인스턴스화할 수 있습니다. 또는 DefaultAzureCredential을 사용하여 Active Directory 인증을 사용할 수도 있습니다.
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)
배포 정책
큐를 만들려면 배포 정책이 필요합니다.
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
)
큐
다음으로 큐를 만들 수 있습니다.
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
)
작업
이제 작업자 선택기를 사용하여 해당 큐에 직접 작업을 제출할 수 있습니다. 작업자는 레이블 Some-Skill 이 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
)
작업자
이제 레이블이 11인 해당 큐에서 작업을 받을 작업자를 Some-Skill 등록합니다.
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
)
제안
EventGrid 구독에서 RouterWorkerOfferIssued를 가져와야 합니다.
이벤트 처리기 역할을 하는 여러 가지 Azure 서비스가 있습니다. 이 시나리오에서는 이벤트 배달을 위해 Webhook를 가정합니다. Webhook 이벤트 배달에 대해 자세히 알아보기
이벤트가 이벤트 처리기에 전달되면 JSON 페이로드를 이벤트 목록으로 역직렬화할 수 있습니다.
# 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
그러나 몇 초 정도 기다린 다음 JobRouter API에 대해 작업자를 직접 쿼리하여 제안이 발급되었는지 확인할 수도 있습니다.
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}")
제품 수락
작업자가 제안을 받으면 수락 또는 거절이라는 두 가지 가능한 작업을 수행할 수 있습니다. 우리는 제안을 수락할 것입니다.
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}")
작업 완료
작업자가 작업을 완료하면 작업자는 작업을 로 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.")
작업 닫기
작업이 완료된 후 작업자는 작업을 닫기 전에 작업에 대한 래핑 작업을 수행하고 마지막으로 더 많은 들어오는 작업을 수락할 수 있는 용량을 해제할 수 있습니다.
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}")
문제 해결
문제가 발생했나요? 이 섹션에는 수행할 작업에 대한 세부 정보가 포함되어야 합니다.
다음 단계
추가 샘플 코드
이 라이브러리를 사용하는 방법에 대한 자세한 예제는 샘플 디렉터리를 참조하세요.
피드백 제공
버그가 발생하거나 제안이 있는 경우 프로젝트의 문제 섹션에 문제를 제출하세요.
참여
이 프로젝트에 대한 기여와 제안을 환영합니다. 대부분의 경우 기여하려면 권한을 부여하며 실제로 기여를 사용할 권한을 당사에 부여한다고 선언하는 CLA(기여자 라이선스 계약)에 동의해야 합니다. 자세한 내용은 cla.microsoft.com.
이 프로젝트에는 Microsoft Open Source Code of Conduct(Microsoft 오픈 소스 준수 사항)가 적용됩니다. 자세한 내용은 Code of Conduct FAQ(규정 FAQ)를 참조하세요. 또는 추가 질문이나 의견은 opencode@microsoft.com으로 문의하세요.
Azure SDK for Python