> 継続的なプロセス改善 > カスタマーサービス > Genesys Cloud CX > データテンプレート

カスタマーサービスのデータテンプレート

Genesys Cloud CX

カスタマーサービスの属性(19) カスタマーサービスのアクティビティ(13) データ抽出ガイド(3)

カスタマーサービスのデータテンプレート

このテンプレートでは、カスタマーサービスのプロセスを分析、改善するために必要なデータの収集方法を確認できます。推奨する属性や記録すべき主要なアクティビティに加え、ソースシステムからデータを抽出する手順も紹介します。プロセスマイニングに必要なイベントログの作成にお役立てください。

収集を推奨する属性
記録すべき主なアクティビティ
データ抽出の手順

イベントログを初めて作成する方はこちら：プロセスマイニング用イベントログの作成方法.

テンプレートをダウンロード

カスタマーサービスの属性

カスタマーサービスを詳しく分析するため、イベントログに含めることを推奨するデータ項目です。

5 必須 6 推奨 8 任意

	名前	説明
	アクティビティ名 ActivityName	サービスリクエストの過程で、特定の時点に発生したイベントまたはタスクの名前です。
説明この属性は、カスタマーサービスのプロセスにおける具体的なステップや状態の変化を示します。各アクティビティは、「担当者が応対を受け入れ」や「後処理コードを割り当て」など、個別のイベントを表します。アクティビティの順序や頻度の分析は、プロセスマイニングの基本です。プロセスの流れを可視化し、逸脱やボトルネック、非効率なステップを特定できます。重要な理由アクティビティはプロセスマップの基盤となり、実際のプロセスの流れを設計上の流れと比較して可視化、分析できます。入手先通常は、Genesys Cloud CXのシステムイベント、担当者の状態、特定の監査証跡の記録を、標準化されたアクティビティ名に対応付けて設定します。例応対開始担当者が応対を受け入れ通話後作業を終了サービスリクエストを再開
	サービスリクエスト ServiceRequest	顧客との応対に関するすべてのアクティビティを関連付ける主な識別子です。
説明サービスリクエストは、チケットまたはケースとも呼ばれ、1件の顧客からの問い合わせや問題を識別する主な情報です。最初の問い合わせから最終的な解決まで、関連するすべてのイベントを1つのプロセスインスタンスにまとめます。サービスリクエスト単位で分析すると、顧客対応の全体像と社内の処理状況を一貫して確認できます。解決までの合計時間や初回問い合わせ解決率など、重要な指標を計算する基盤となります。重要な理由すべてのプロセスステップを結び付ける重要なケースIDです。顧客との応対を開始から終了まで一貫して分析できます。入手先概念上のケース識別子です。Genesysでは通常、会話IDに対応しますが、導入方法によってはカスタム識別子を使う場合があります。例 SR-20240521-00123SR-20240521-00124SR-20240522-00001
	開始時刻 EventTime	アクティビティまたはイベントが開始した日時を示すタイムスタンプです。
説明この属性には、各アクティビティが発生した正確な日時が記録されます。イベントを正しい順序に並べ、時間に基づく分析を行うために欠かせません。開始時刻を使ってプロセスの流れを時系列に並べ、アクティビティ間の所要時間、待ち時間、サイクルタイムを計算します。信頼できるプロセス分析とパフォーマンスのモニタリングには、正確なタイムスタンプが必要です。重要な理由このタイムスタンプでアクティビティを時系列に並べることで、プロセスの流れや所要時間、ボトルネックを正確に分析できます。入手先 Genesys Cloud CXのイベントログや応対の詳細から取得します。通常、記録された各システムイベントまたはユーザーイベントに関連付けられています。例 2024-05-21T10:00:15Z2024-05-21T10:02:30Z2024-05-21T10:15:00Z
	ソースシステム SourceSystem	データの抽出元となるシステムです。
説明この属性は、イベントデータの発生元を示します。ここではGenesys Cloud CXです。複数のシステムが連携する環境では、データの来歴の確認、問題の調査、データのコンテキストの把握に欠かせません。正しいデータソースに基づいて分析できるほか、複数のシステムのデータを1つのプロセスビューにまとめる場合に、データの絞り込みや分類にも役立ちます。重要な理由データの発生元を示します。データガバナンスや検証、複数システムのデータ統合に欠かせない情報です。入手先通常は、レコードの発生元を示すため、データの抽出と変換の際に固定値として追加します。例 Genesys Cloud CXGenesysCloudCX_US1Genesys
	最終データ更新 LastDataUpdate	直近のデータ更新日時を示すタイムスタンプです。
説明この属性は、データセットがソースシステムから最後に更新された日時を示します。分析対象のデータがどの程度新しいかを把握するための情報であり、適切なタイミングで的確な業務判断を行ううえで重要です。ダッシュボードやレポートでは、この情報を目立つ位置に表示し、データの更新状況を確認してインサイトの妥当性を判断できるようにします。重要な理由データの鮮度を示し、最新の情報に基づく分析と意思決定を支援します。入手先この値は、データを抽出する時点、またはプロセスマイニングツールに読み込む時点で生成され、データセットに記録されます。例 2024-05-23T04:00:00Z2024-05-24T04:00:00Z
	エージェントID AgentId	インタラクションまたはアクティビティを担当したエージェントを一意に識別するIDです。
説明エージェントIDは、各カスタマーサービス担当者を一意に識別するキーです。エージェントのパフォーマンス、業務量の配分、効率を分析する際に欠かせない属性です。プロセスマップを絞り込み、特定のエージェントやチームの対応状況を確認できます。また、解決時間、再作業率、標準手順の遵守状況などの指標を比較できます。「エージェントのパフォーマンスと効率」ダッシュボードの基礎となる属性です。重要な理由プロセスのアクティビティを担当者に結び付け、個人のパフォーマンスや業務量、チームの効率を分析できます。入手先 Genesys Cloud CXの会話詳細レコードで確認できます。インタラクションの特定のセグメントを担当したユーザーに関連付けられています。例 a1b2c3d4-e5f6-7890-1234-567890abcdeff0e9d8c7-b6a5-4321-fedc-ba0987654321
	エージェント名 AgentName	インタラクションまたはアクティビティを担当したエージェントの氏名です。
説明エージェント名は、エージェントIDに対応する、担当者を人が読みやすい形で示す名前です。ダッシュボードやレポートが分かりやすくなり、管理者やチームリーダーが内容を把握しやすくなります。エージェントごとの分析で広く使われ、システムIDを照合しなくても、パフォーマンスの確認や研修ニーズの特定、業務量の公平な配分に役立ちます。重要な理由エージェント名を分かりやすく表示することで、技術的なIDを使わずにパフォーマンスを分析し、結果を共有できます。入手先 Genesys Cloud CXのUsersまたはDirectoryサービスでAgentIdを検索して取得します。例 John SmithJane DoePeter Jones
	キュー名 QueueName	インタラクションの振り分け先となったキューの名前です。
説明この属性は、エージェントに割り当てられる前にインタラクションが待機していたキューを示します。キュー名ごとのデータ分析は、「サービスキューのボトルネック検出」ダッシュボードに欠かせません。スキルグループやサービスラインごとの業務量の把握、キューごとの待ち時間の測定、人員不足や過負荷が続くキューの特定に役立ちます。リソース配分を見直し、顧客の待ち時間を短縮するための重要な情報です。重要な理由サービスリクエストが割り当てを待っている場所を示し、ボトルネックの特定や業務量の配分状況の分析に役立ちます。入手先 Genesys Cloud CXの会話詳細レコードで確認できます。1つのインタラクションが複数のキューを経由する場合もあります。例第1階層サポート－音声請求に関する問い合わせ－チャットテクニカルサポート－メール
	コミュニケーションチャネル MediaType	音声、チャット、メールなど、インタラクションに使用されたコミュニケーションチャネルです。
説明この属性は、顧客とエージェントがやり取りした手段を示します。主なチャネルには、音声、チャット、メール、ソーシャルメディアがあります。「コミュニケーションチャネル別パフォーマンス」ダッシュボードのように、チャネルごとの実績を分析すると、チャネル別の件数や解決時間、顧客満足度を把握できます。チャネル戦略やリソース配分を見直すうえで重要な情報です。重要な理由コミュニケーションチャネルごとにプロセスを分けて分析すると、チャネル別のパフォーマンスや顧客の行動、必要なリソースを把握できます。入手先 Genesys Cloud CXの会話詳細レコードに標準で含まれるフィールドです。例音声チャットメールメッセージ
	対応結果コード WrapUpCode	インタラクションの終了時にエージェントが割り当てるコードです。対応結果や問い合わせ内容を分類します。
説明対応結果コードは、顧客とのインタラクションの内容や解決状況を分類するためにエージェントが選択するラベルです。顧客がサポートに連絡した理由を、構造化されたデータとして記録できます。対応結果コードを分析すると、よくある問題の種類や解決状況、特定の問い合わせの頻度を把握できます。根本原因の分析やサービス需要の傾向を理解するうえで役立つデータです。重要な理由インタラクションの結果を分類し、よくある問題や解決の有効性、問い合わせ理由を分析するための構造化データを提供します。入手先 Genesys Cloud CXの会話詳細レコードで確認できます。エージェント参加者のセッション詳細に記録されています。例パスワードのリセット請求に関する異議申し立てを解決製品情報の問い合わせ第2階層サポートにエスカレーション
	終了時刻 EventEndTime	アクティビティまたはイベントが終了した日時を示すタイムスタンプです。
説明この属性には、アクティビティが終了した正確な時刻が記録されます。開始時刻と組み合わせることで、個々のアクティビティにかかった時間を正確に算出できます。アクティビティごとの所要時間を分析すると、プロセスのどのステップに最も時間がかかっているかが分かり、非効率な箇所や改善の余地を特定できます。たとえば、「対応中の保留」が長引いている時間や、「通話後の作業」にかかる時間が長いことを把握できます。重要な理由個々のアクティビティの所要時間を正確に算出できます。時間のかかるステップやパフォーマンス上のボトルネックを特定するうえで欠かせません。入手先イベントログまたはGenesys Cloud CXのインタラクション詳細で確認できます。後続イベントの開始時刻から算出することもできます。例 2024-05-21T10:02:30Z2024-05-21T10:15:00Z2024-05-21T10:18:45Z
	SLA目標解決時間 SlaTargetResolutionTime	サービスリクエストの解決について、契約で合意した目標時間です。
説明この属性は、サービスレベル契約（SLA）に基づき、サービスリクエストの解決に許容される最大時間を定義します。実際の解決時間を測る際の基準となります。「SLAコンプライアンス概要」ダッシュボードや「SLA遵守率」の重要業績評価指標（KPI）に必要なデータです。顧客との約束に対する組織の実績を確認し、違反が発生する前にその可能性を特定できます。重要な理由 SLA遵守状況を測る基準となり、サービスのパフォーマンスと契約上の義務を確認するための重要な指標です。入手先会話のカスタム属性として保存するか、キュー、顧客の種類、リクエストの種類に関するルールに基づいて算出できます。例 86400144003600
	SLA遵守状況 IsSlaCompliant	サービスリクエストがSLAの目標時間内に解決されたかどうかを示すフラグです。
説明このブール型属性は、サービスリクエストが解決時間の目標を達成したかどうかを示します。「ServiceResolutionTime」と「SlaTargetResolutionTime」を比較して算出します。このフラグを使うと、「SLAコンプライアンス概要」ダッシュボードで分析や可視化を簡単に行えます。また、「SLA遵守率」の重要業績評価指標（KPI）の算出にも使います。遵守したケースと未達のケースをすばやく絞り込み、SLA違反に共通するパターンを特定できます。重要な理由各ケースのSLA遵守または未達を明確に示し、SLAパフォーマンスの分析や未達の根本原因の特定を容易にします。入手先算出フィールド：「ServiceResolutionTime <= SlaTargetResolutionTime」の場合はTrue、それ以外はFalseです。例 truefalse
	サービスリクエストの種類 ServiceRequestType	「問い合わせ」「苦情」「技術的な問題」など、サービスリクエストの分類です。
説明この属性は、内容や目的に応じてサービスリクエストを分類します。リクエストの種類ごとに分析を分け、対応状況を把握できます。たとえば、「苦情」と「一般的な問い合わせ」では、プロセスの流れや解決時間が大きく異なる場合があります。「社内エスカレーション分析」などのダッシュボードで、特定の種類のリクエストがエスカレーションされやすいかを確認する際に役立ちます。重要な理由リクエストの種類ごとにプロセスを分けて対応状況を比較し、種類に特有のボトルネックや非効率な箇所を特定できます。入手先この情報は、IVRでの選択、顧客がウェブフォームで選んだ項目、またはエージェントによる割り当てを通じて記録される場合があります。Genesysでは、参加者属性または対応結果コードとして保存できます。例請求に関する問い合わせテクニカルサポートアカウント管理製品に関する苦情
	サービス解決時間 ServiceResolutionTime	最初の顧客とのやり取りの開始から、最終的な解決までにかかった合計時間です。
説明この指標は、顧客が連絡を開始してから問題が解決済みになるまでの、サービスリクエスト全体の所要時間を測定します。プロセス全体の効率と顧客体験を評価する重要業績評価指標（KPI）です。算出属性であり、「サービスリクエスト解決時間」ダッシュボードと「平均サービス解決時間」の重要業績評価指標（KPI）で中心となる指標です。分布を分析すると、長期化しているケースや全体に共通する遅延を特定できます。重要な理由プロセス全体の効率と顧客体験への影響を測る重要な重要業績評価指標（KPI）です。入手先算出フィールド：最終解決アクティビティのタイムスタンプから、最初の顧客連絡アクティビティのタイムスタンプを差し引きます。例 90010800172800
	会話ID ConversationId	Genesysが会話全体に割り当てる一意の識別子です。
説明会話IDは、Genesys Cloud CXで1件の顧客との会話に関連するすべてのインタラクション、セグメント、参加者をまとめる主要な技術キーです。概念上の「サービスリクエスト」をケースIDとして使う一方、Genesys APIから関連データを取得する際の基盤となるキーが会話IDです。データの抽出やデータセット間の結合、プロセスデータの技術的な検証に欠かせません。重要な理由 Genesysで使われる主要な技術キーです。データの抽出やトラブルシューティング、ソースシステムとの照合に欠かせません。入手先 Genesys Cloud CXの分析APIと会話APIで主要なフィールドとして使われます。例 d8a7c6b5-e4f3-2109-8765-fedcba098765c7b6a5d4-f3e2-1098-7654-edcbaf987654
	再オープン IsReopened	解決済みのサービスリクエストが後から再オープンされたかどうかを示すフラグです。
説明このブール型属性は、解決済みまたはクローズ済みになった後、再び対応中になったサービスリクエストを示します。再オープンされたケースは、最初の解決策が有効でなかった、または不十分だった可能性があり、顧客の不満や追加作業につながります。この属性は、「サービスリクエスト再オープンの傾向」ダッシュボードと「サービスリクエスト再オープン率」の重要業績評価指標（KPI）に使われます。こうしたケースを分析すると、解決が不十分だった根本原因を特定できます。重要な理由解決プロセスの問題を明らかにし、解決策の品質に関する課題や、再作業、顧客体験の低下につながる要因を示します。入手先「サービスリクエスト解決」アクティビティの後にケースを再開するアクティビティを検出して算出するフィールドです。例 truefalse
	初回対応で解決 IsFirstContactResolution	サービスリクエストが1回のインタラクションで解決されたかどうかを示すフラグです。
説明このブール型属性は、顧客への追加確認や社内での引き継ぎを必要とせず、最初のインタラクションで解決したケースを示します。「true」は、理想的で効率的な解決を意味します。この属性は、「初回対応解決率」の重要業績評価指標（KPI）と対応するダッシュボードの基礎となります。初回の連絡で解決できなかったケースの特徴を分析すると、エージェントの研修、ナレッジベースの改善、プロセスの見直しにつながる可能性があります。重要な理由初回の対応で問題を解決することは、効率と顧客満足度を高め、良好なサービス体験につながります。入手先ケースのイベント順序を分析して算出するフィールドです。転送や再オープンなど、特定のアクティビティを挟まずに解決したケースを初回対応解決（FCR）とします。例 truefalse
	顧客満足度スコア（CSAT） CustomerSatisfactionScore	インタラクション後のアンケートで顧客が回答した満足度スコアです。
説明顧客満足度スコア（CSAT）は、受けたサービスに対する顧客の評価を直接示します。通常、インタラクションの終了後に送信するアンケートで収集します。CSATスコアとプロセスデータを合わせて分析すると、プロセスの違いや解決時間、特定のエージェントが顧客満足度に与える影響を把握できます。プロセス全体の成果を評価するための重要な指標です。重要な理由顧客満足度を直接測定できるため、プロセスのパフォーマンスと顧客の評価との相関を分析できます。入手先通常、Genesys Cloud CXのQuality Managementモジュール、またはGenesysと連携したサードパーティのアンケートツールから取得します。例 5413

必須推奨任意

カスタマーサービスのアクティビティ

プロセスを正確に把握するため、イベントログに記録すべき主な手順と節目です。

5 推奨 8 任意

	アクティビティ	説明
	後処理コードを割り当て	担当者が、あらかじめ定義された後処理コード（処理結果コード）を応対に割り当てます。「解決済み」や「エスカレーション」など、サービスリクエストの結果を明示的に分類します。
重要な理由後処理コードは、リクエストの業務上の結果を判断する主な情報源です。解決率の計算や、分析対象となるケースの分類に欠かせません。入手先会話の詳細レコードにある明示的なイベントで、通常は担当者の参加者のセッションデータに記録されます。wrapUpコードとタイムスタンプが保存されます。記録担当者が応対の後処理コードを選択したときに記録されます。イベントタイプ explicit
	応対を切断	すべての参加者が切断され、会話が終了したことを示すアクティビティです。サービスリクエストへの応対が技術的に終了した時点として扱われることがよくあります。
重要な理由応対自体の明確な終了イベントです。応対時間の合計を計算するうえで重要で、リクエストの終了を示す指標として使われることもよくあります。入手先会話の詳細レコードから取得します。通常はconversationEndタイムスタンプに記録される、会話オブジェクト自体の終了時刻に対応します。記録すべての参加者が切断され、会話が終了したときに記録されます。イベントタイプ explicit
	応対を転送	担当者が応対を別のキューまたは担当者に転送したことを示します。担当者がすぐに切断するブラインド転送と、転送先に事前に相談するコンサルテーション転送があります。
重要な理由このアクティビティは、社内エスカレーション率のKPIを測定するうえで欠かせません。転送率が高い場合、トレーニングの必要性、振り分けの誤り、プロセスの複雑さが考えられます。入手先会話の詳細レコードで、最初の担当者が対応した後に新しいACD参加者または担当者の参加者が会話に追加された場合に特定できます。多くの場合、「transfer」イベントがきっかけとなります。記録参加者のセッションデータにある転送イベントから特定します。イベントタイプ explicit
	応対開始	このアクティビティは、着信電話、チャット、メールなど、顧客との応対の開始を示します。Genesys Cloud CXでは、新しい会話オブジェクトがシステムに作成されると、このイベントが明示的に記録されます。
重要な理由サービスリクエストプロセスの主な開始イベントです。解決までの合計時間を計算し、顧客からの問い合わせの傾向を把握するうえで欠かせません。入手先会話の詳細レコードから取得するイベントです。通常はconversationStartタイムスタンプに記録される、会話オブジェクト自体の開始時刻に対応します。記録システムで新しい会話が開始されたときに記録されます。イベントタイプ explicit
	担当者が応対を受け入れ	担当者が提示された応対を受け入れ、顧客と接続した時点を示します。サービスリクエストへの直接対応が始まる重要な節目です。
重要な理由このアクティビティは、初回問い合わせ解決率と担当者の対応時間を測定するうえで重要です。担当者がリクエストへの対応を開始したことを示します。入手先会話の詳細レコードから確認できます。担当者の参加者状態が「connected」に変わり、対応するタイムスタンプが記録されます。記録担当者の参加者状態が「connected」に変わったときに記録されます。イベントタイプ explicit
	サービスリクエストを再開	解決済みとされた直後に、顧客が同じ問題についてサービスセンターへ再度問い合わせた状況を示します。明示的なイベントではなく、業務ルールに基づいて算出します。
重要な理由再開されたリクエストを追跡すると、サービスリクエスト再開率を測定できます。最初の解決が不十分だったことを示し、顧客満足度や業務効率に影響します。入手先以前の応対が解決した後、あらかじめ定めた期間内に、同じ顧客（顧客ID）から関連する問題（サービスリクエストの種類など）について新たな「応対開始」イベントが発生した場合に推定します。記録同じ顧客と問題に関する新しい応対を、最近終了した応対に関連付けて算出します。イベントタイプ calculated
	会話をキューに振り分け	新しい応対が、対応可能な担当者を待つために特定のキューへ入れられた時点を示します。Genesysの振り分けエンジン（ACD）が明示的に記録するイベントです。
重要な理由このアクティビティを追跡すると、キューの待ち時間を測定し、振り分けプロセスのボトルネックを特定できます。このイベントから担当者の割り当てまでに時間がかかる場合、人員配置や振り分けルールに問題がある可能性があります。入手先会話の詳細レコードから取得します。会話内のACD参加者の「purpose」と「state」を確認し、メトリクス配列にあるキューの「enterTime」でこのイベントを特定します。記録 ACD振り分けエンジンが会話をキューに入れたときに記録されます。イベントタイプ explicit
	応対中に保留を開始	応対中に担当者が顧客を保留にしたときに記録されるアクティビティです。会話内の担当者の参加者状態が明示的に変化します。
重要な理由保留の頻度と時間を分析すると、担当者が情報を探したり、他の人に相談したりすることが多いなど、プロセスの非効率な箇所を明らかにできます。入手先会話の詳細レコードにある担当者のセッションデータから取得します。状態が「held」に変わり、タイムスタンプが記録されます。記録担当者の参加者状態が「held」に変わったときに記録されます。イベントタイプ explicit
	応対中の保留を解除	担当者が顧客の保留を解除し、会話を再開したときに発生します。保留時間の終了を示すイベントで、状態の変化として記録されます。
重要な理由「応対中に保留を開始」と組み合わせることで、保留時間の合計を正確に計算できます。これは、対応時間全体や顧客体験を分析するうえで重要な要素です。入手先担当者の状態が「held」から「connected」に戻ったときのセッションデータから算出します。記録担当者の参加者状態が「held」から「connected」に変わったときに記録されます。イベントタイプ explicit
	担当者に応対を提示	システムが特定の担当者に応対を提示したときに発生するイベントです。担当者が会話を受け入れるか拒否するまで、担当者のステータスは「alerting」に変わります。
重要な理由このアクティビティから、担当者の応答性や振り分けアルゴリズムの有効性を分析できます。この時点以降に遅延がある場合、担当者が速やかに応対を受け入れていない可能性があります。入手先会話の詳細レコードにある担当者の参加者データから確認できます。担当者のセッションには、「alerting」状態のメトリクスと対応するタイムスタンプが記録されます。記録振り分けエンジンが新しい応対を担当者に通知したときに記録されます。イベントタイプ explicit
	通話後作業を終了	担当者の通話後作業が終了したことを示します。この時点で担当者は次の応対に対応できる状態になります。
重要な理由このアクティビティを「通話後作業を開始」と組み合わせると、後処理にかかった時間を正確に把握できます。担当者の生産性やプロセスの業務負荷の分析に役立ちます。入手先担当者の状態が「acw」から「idle」などの対応可能な状態に変わったときに推定します。この状態変化のタイムスタンプを使用します。記録担当者の状態が「acw」から対応可能な状態に変わったことから推定します。イベントタイプ inferred
	通話後作業を開始	通話後作業（ACW）の開始を示すアクティビティです。担当者は顧客との接続を終了し、メモの記録やシステムの更新などを行う専用の状態に移ります。
重要な理由 ACWの時間を測定すると、担当者の効率や対応時間全体を把握できます。ACWが長い場合、応対後のプロセスが非効率である可能性があります。入手先担当者の参加者のセッションデータから取得します。顧客側の参加者が切断された後、担当者の状態が「acw」に変わったときに記録されます。記録担当者の参加者状態が「acw」に変わったときに記録されます。イベントタイプ explicit
	顧客満足度アンケートを送信	応対の終了後に顧客満足度（CSAT）アンケートを送信したときに発生するアクティビティです。多くの場合、あらかじめ定めたルールに基づいてシステムが自動で送信します。
重要な理由このアクティビティは、CSATアンケート送信の適時性に関するKPIを測定するために必要です。顧客の記憶が新しいうちにアンケートを送ることで、正確なフィードバックを得やすくなります。入手先通常、この情報はGenesys Cloud CXのアンケートまたは品質管理データから取得でき、元の会話IDに関連付けられます。記録特定の会話に対するアンケートを配信したときに、アンケートモジュールが記録します。イベントタイプ explicit

推奨任意

データ抽出ガイド

Genesys Cloud CXからデータを取得する方法

ステップ

Analytics Workspaceに移動：Genesys Cloud CXにログインします。メニューからAdminを開き、Analyticsの下にあるWorkspaceを選択します。
新しいビューを作成：プラスアイコンをクリックして新しいタブを追加し、「新しいビューを作成」を選択します。ビューの種類として「インタラクション」を選ぶと、インタラクションビューの編集画面が開きます。
期間を指定：左側のフィルターパネルで、抽出するデータの日付と時刻の範囲を設定します。まずは1週間など短い期間を指定し、手順を確認することをおすすめします。
エクスポートする列を選択：データ準備における重要な手順です。「列」ボタンをクリックし、使用可能な項目から次の列を選んでビューに追加します。これらの項目はイベントログの作成に必要です。
- Conversation ID
- Conversation Start
- Conversation End
- Media Type
- Participants（エージェント、キュー、顧客のデータが入れ子になった複合項目です）
- Metrics.tAnswered
- Metrics.tACW
- Metrics.tHeld
- Surveys
追加のフィルターを適用：必要に応じてフィルターパネルでデータを絞り込みます。Media Type（例：音声、チャット、メール）、キュー、Direction（受信／発信）、その他の会話属性でフィルターを設定し、対象のサービスリクエストプロセスに絞ることができます。
ビューを保存して実行：ビューに「Process Mining Service Request Export」など内容が分かる名前を付けて保存します。設定した条件に一致するデータが読み込まれます。
エクスポートを開始：ビューにデータが読み込まれたら、通常はダウンロードアイコンで表示される「エクスポート」ボタンをクリックします。「CSVとしてエクスポート」または「JSONとしてエクスポート」を選択します。初期処理にはCSVが適しています。
エクスポートしたファイルをダウンロード：エクスポート処理はバックグラウンドで実行されます。Performance > My Exportsに移動すると進捗を確認できます。完了したら、ファイルをローカルシステムにダウンロードします。
生のデータを変換：エクスポートファイルは会話ごとに1行で、参加者とセグメントのデータが1つの列に入れ子になっています。QueryセクションにあるPythonスクリプトなどを実行し、このデータを展開する必要があります。スクリプトは生のエクスポートデータを解析し、「エージェントがインタラクションを受諾」や「インタラクションを保留」など、アクティビティごとに1行のファイルを作成します。
独自のロジックを実装：変換処理に、業務に応じたロジックを組み込みます。たとえば「サービスリクエストの再開」アクティビティは、定めた期間内に同じ顧客から発生した後続のインタラクションを分析して算出する必要があります。
イベントログを完成：変換スクリプトで、ServiceRequest、ActivityName、EventTimeなど、ProcessMindで必要な形式の列を含むCSVファイルを出力します。すべてのタイムスタンプの形式が統一されていることを確認します。
ProcessMindにアップロード：変換済みの最終CSVファイルを、プロセスマイニングツールにアップロードできます。

設定

権限：Analytics Workspaceへのアクセスとインタラクションビューの作成には、Genesys Cloud CXのユーザーロールにAnalytics > Conversation > View権限が必要です。
期間：初回の抽出では、代表的な期間として3～6か月を指定します。取扱量の多いコンタクトセンターでは、システム上限への到達を避け、ファイルサイズを管理するため、週単位や月単位など、短い期間に分けてエクスポートすることをおすすめします。
主なフィルター：
- Media Type：voice、chat、emailなど、カスタマーサービスのプロセスに関連するチャネルに絞り込みます。
- キュー：特定のチームがプロセスを担当している場合は、該当するキューIDで絞り込み、対象プロセスのデータだけを抽出します。
- Direction：通常は、顧客からのサービスリクエストを記録するため、inboundのインタラクションに絞り込みます。
エクスポートする列：Participants列には、エージェントとキューのイベントの詳細な時系列が含まれるため、欠かせません。この列がないと、個々のアクティビティを抽出できません。Conversation ID、Conversation Start、Conversation Endも含めてください。
パフォーマンスに関する注意点：列数が多い場合や期間が長い場合、エクスポートにかなり時間がかかることがあります。処理が失敗する、またはタイムアウトする場合は、期間を短くして再試行してください。Genesys Cloudでは、1回のエクスポートで扱える会話数は約100万件までです。

クエリの例 python

import pandas as pd
import json
from datetime import datetime, timezone

class=class="hl-string">"hl-comment"># This script transforms a Genesys Cloud Conversation Detail Export into a process mining event log.
class=class="hl-string">"hl-comment"># It assumes the raw data is in a CSV file named class="hl-string">'genesys_export.csv'.

def transform_genesys_export(file_path):
    class=class="hl-string">"hl-comment"># Load the raw export file
    try:
        df = pd.read_csv(file_path)
    except FileNotFoundError:
        print(fclass="hl-string">"Error: The file {file_path} was not found.")
        return None

    event_log = []
    source_system = class="hl-string">'Genesys Cloud CX'
    last_data_update = datetime.now(timezone.utc).isoformat()

    class=class="hl-string">"hl-comment"># Rename columns for easier access
    df.rename(columns={
        class="hl-string">'Conversation ID': class="hl-string">'conversationId',
        class="hl-string">'Conversation Start': class="hl-string">'conversationStart',
        class="hl-string">'Conversation End': class="hl-string">'conversationEnd',
        class="hl-string">'Media Type': class="hl-string">'mediaType'
    }, inplace=True)

    for index, row in df.iterrows():
        service_request_id = row[class="hl-string">'conversationId']
        media_type = row[class="hl-string">'mediaType']
        
        class=class="hl-string">"hl-comment"># Activity: Interaction Started
        event_log.append({
            class="hl-string">'ServiceRequest': service_request_id,
            class="hl-string">'ActivityName': class="hl-string">'Interaction Started',
            class="hl-string">'EventTime': row[class="hl-string">'conversationStart'],
            class="hl-string">'EventEndTime': None,
            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
        })

        class=class="hl-string">"hl-comment"># The class="hl-string">'Participants' column contains a JSON string that needs to be parsed
        participants = json.loads(row[class="hl-string">'Participants'])

        for p in participants:
            participant_id = p.get(class="hl-string">'participantId')
            user_id = p.get(class="hl-string">'userId')
            queue_name = p.get(class="hl-string">'queueName')
            purpose = p.get(class="hl-string">'purpose')

            for s in p.get(class="hl-string">'sessions', []):
                wrap_up_code = s.get(class="hl-string">'wrapUpCode')
                for seg in s.get(class="hl-string">'segments', []):
                    seg_start = seg.get(class="hl-string">'segmentStart')
                    seg_end = seg.get(class="hl-string">'segmentEnd')
                    seg_type = seg.get(class="hl-string">'segmentType')

                    activity_name = None
                    if purpose == class="hl-string">'acd' and seg_type == class="hl-string">'interact':
                        activity_name = class="hl-string">'Conversation Routed to Queue'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'alert':
                        activity_name = class="hl-string">'Agent Offered Interaction'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'interacting':
                        activity_name = class="hl-string">'Agent Accepted Interaction'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'hold':
                        activity_name = class="hl-string">'Hold Placed on Interaction'
                        class=class="hl-string">"hl-comment"># Also create the Hold Removed event
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': class="hl-string">'Hold Removed from Interaction',
                            class="hl-string">'EventTime': seg_end,
                            class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': None,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'acw':
                        activity_name = class="hl-string">'After-Call Work Started'
                        class=class="hl-string">"hl-comment"># Also create ACW Ended and Wrap-Up Assigned events
                        if wrap_up_code:
                            event_log.append({
                                class="hl-string">'ServiceRequest': service_request_id,
                                class="hl-string">'ActivityName': class="hl-string">'Wrap-Up Code Assigned',
                                class="hl-string">'EventTime': seg_end, class=class="hl-string">"hl-comment"># Assume code assigned at end of ACW
                                class="hl-string">'EventEndTime': None,
                                class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': wrap_up_code,
                                class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                            })
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': class="hl-string">'After-Call Work Ended',
                            class="hl-string">'EventTime': seg_end,
                            class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': wrap_up_code,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                    
                    if seg.get(class="hl-string">'disconnectType') == class="hl-string">'transfer':
                        activity_name = class="hl-string">'Interaction Transferred'

                    if activity_name:
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': activity_name,
                            class="hl-string">'EventTime': seg_start,
                            class="hl-string">'EventEndTime': seg_end,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': None,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
        
        class=class="hl-string">"hl-comment"># Activity: Interaction Disconnected
        event_log.append({
            class="hl-string">'ServiceRequest': service_request_id,
            class="hl-string">'ActivityName': class="hl-string">'Interaction Disconnected',
            class="hl-string">'EventTime': row[class="hl-string">'conversationEnd'],
            class="hl-string">'EventEndTime': None,
            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
        })

        class=class="hl-string">"hl-comment"># Activity: Customer Satisfaction Survey Sent (from surveys column)
        if pd.notna(row.get(class="hl-string">'Surveys')):
            surveys = json.loads(row[class="hl-string">'Surveys'])
            for survey in surveys:
                if survey.get(class="hl-string">'surveyStatus') == class="hl-string">'Sent':
                    class=class="hl-string">"hl-comment"># Survey sent time is not always directly available, using conversation end as an approximation
                    event_log.append({
                        class="hl-string">'ServiceRequest': service_request_id,
                        class="hl-string">'ActivityName': class="hl-string">'Customer Satisfaction Survey Sent',
                        class="hl-string">'EventTime': survey.get(class="hl-string">'surveySentDate', row[class="hl-string">'conversationEnd']),
                        class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
                        class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

    class=class="hl-string">"hl-comment"># NOTE: class="hl-string">'Service Request Re-Opened' requires custom cross-conversation logic.
    class=class="hl-string">"hl-comment"># This logic would need to be implemented here, analyzing the entire dataset
    class=class="hl-string">"hl-comment"># to find subsequent interactions from the same customer within a specific timeframe.

    final_df = pd.DataFrame(event_log)
    class=class="hl-string">"hl-comment"># Sort by case and event time to create a chronological log
    final_df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime'], inplace=True)
    
    return final_df

if __name__ == class="hl-string">'__main__':
    class=class="hl-string">"hl-comment"># Specify the path to your downloaded Genesys export CSV file
    input_file = class="hl-string">'genesys_export.csv'
    output_file = class="hl-string">'genesys_event_log.csv'
    
    event_log_df = transform_genesys_export(input_file)
    
    if event_log_df is not None:
        event_log_df.to_csv(output_file, index=False)
        print(fclass="hl-string">"Transformation complete. Event log saved to {output_file}")

ステップ

前提条件とセットアップ：Python 3.8以降をインストールし、pandasとpurecloud-platform-client-v2ライブラリを用意します。次のpipコマンドでインストールできます：pip install pandas purecloud-platform-client-v2。
Genesys Cloud OAuthクライアントを作成：Genesys Cloud CXの管理者アカウントでログインします。Integrations > OAuthに移動し、付与タイプに「Client Credentials」を指定して新しいクライアントを作成します。analytics:conversationDetail:view権限を含むロールを割り当てます。
スクリプトの認証情報を設定：前の手順で取得したClient IDとClient Secretを保存します。以下のPythonスクリプトで、CLIENT_ID、CLIENT_SECRET、GENESYS_CLOUD_REGIONのプレースホルダーを実際の認証情報に置き換えます。
抽出期間を指定：スクリプト内のquery_body変数を探し、intervalの値を変更して抽出する日付と時刻の範囲を指定します。テストでは、まず1日など短い期間を指定することをおすすめします。
Pythonスクリプトを実行：ターミナルでpython your_script_name.pyを実行します。スクリプトがGenesys Cloud APIで認証し、非同期の分析ジョブを送信して、ジョブIDを出力します。
ジョブの実行状況を確認：スクリプトは15秒ごとにAPIを自動で確認し、ジョブの状態を調べます。コンソールに状況が表示されます。要求したデータ量によっては、完了まで数分以上かかることがあります。
データを処理、変換：ジョブが完了すると、スクリプトがJSONデータをダウンロードします。各会話レコードの参加者、セッション、セグメントのデータを解析し、主なアクティビティと属性を特定して構造化します。
算出するイベントを処理：スクリプトには「サービスリクエストの再開」イベントを特定する後処理が含まれています。必要に応じてロジックを変更し、前のインタラクションが解決してから指定期間内に同じ顧客から新たなインタラクションがあったかを確認できます。
イベントログを作成：変換したデータをpandasのDataFrameにまとめます。プロセスマイニング用のイベントログと同じ構造です。
最終CSVファイルをエクスポート：スクリプトは最終的なDataFrameをgenesys_event_log.csvという名前のCSVファイルに保存します。このファイルはProcessMindに直接アップロードできる形式です。

設定

API認証情報：認証にはOAuth Client IDとClient Secretが必要です。これらはGenesys Cloud CXの管理画面で作成します。
API権限：関連付けられたOAuthクライアントのロールには、必要な会話データにアクセスするためのanalytics:conversationDetail:view権限が必要です。
抽出期間：APIクエリ本文のintervalパラメーターで抽出期間を指定します。大規模な環境では、APIのタイムアウトやメモリ使用量を抑えるため、1週間ずつなど、短い期間に分けて順番に抽出することをおすすめします。
データフィルター：APIクエリ本文にconversationFilters、segmentFilters、evaluationFiltersを追加して、抽出対象を絞り込めます。たとえば、特定のqueueIdやmediaTypeでフィルターを設定し、カスタマーサービスプロセスの一部を対象にできます。
再開の判定ロジック：「サービスリクエストの再開」イベントを特定する期間は、スクリプト内で設定できます。初期値は72時間ですが、組織の業務ルールに合わせて調整してください。
環境：スクリプトの実行には、必要なSDKとライブラリを備えたローカルのPython環境が必要です。Genesys Cloud CXのサーバーやデータベースに直接アクセスする必要はありません。

クエリの例 python

import PureCloudPlatformClientV2
import pandas as pd
import requests
import json
import time
import os
from datetime import datetime, timedelta

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your Genesys Cloud credentials and region
CLIENT_ID = class="hl-string">"[Your_Client_Id]"
CLIENT_SECRET = class="hl-string">"[Your_Client_Secret]"
GENESYS_CLOUD_REGION = class="hl-string">"[Your_Genesys_Cloud_Region]" class=class="hl-string">"hl-comment"># e.g., mypurecloud.com, mypurecloud.de

class=class="hl-string">"hl-comment"># Set the API environment
PureCloudPlatformClientV2.configuration.host = fclass="hl-string">'https://api.{GENESYS_CLOUD_REGION}'

class=class="hl-string">"hl-comment"># --- Authentication ---
api_client = PureCloudPlatformClientV2.api_client.ApiClient().get_client_credentials_token(CLIENT_ID, CLIENT_SECRET)
analytics_api = PureCloudPlatformClientV2.AnalyticsApi(api_client)


def parse_conversation(conv_detail):
    class="hl-string">"""Parses a single conversation detail object into multiple event log rows."""
    events = []
    conversation_id = conv_detail.get(class="hl-string">'conversationId')
    source_system = class="hl-string">'Genesys Cloud CX'
    last_data_update = datetime.utcnow().isoformat() + class="hl-string">'Z'

    class=class="hl-string">"hl-comment"># Event: Interaction Started
    events.append({
        class="hl-string">'ServiceRequest': conversation_id,
        class="hl-string">'ActivityName': class="hl-string">'Interaction Started',
        class="hl-string">'EventTime': conv_detail.get(class="hl-string">'conversationStart'),
        class="hl-string">'EventEndTime': None,
        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
        class="hl-string">'MediaType': next((p.get(class="hl-string">'sessions')[0].get(class="hl-string">'mediaType') for p in conv_detail.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None),
        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
    })

    for p in conv_detail.get(class="hl-string">'participants', []):
        agent_id = p.get(class="hl-string">'userId')
        agent_name = p.get(class="hl-string">'participantName')

        for s in p.get(class="hl-string">'sessions', []):
            media_type = s.get(class="hl-string">'mediaType')

            for seg in s.get(class="hl-string">'segments', []):
                seg_start = seg.get(class="hl-string">'segmentStart')
                seg_end = seg.get(class="hl-string">'segmentEnd')

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'alert':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Agent Offered Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'interact' and p.get(class="hl-string">'purpose') == class="hl-string">'agent':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Agent Accepted Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'hold':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Hold Placed on Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
                    if seg_end:
                        events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Hold Removed from Interaction', class="hl-string">'EventTime': seg_end, class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'acw':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'After-Call Work Started', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p.get(class="hl-string">'wrapUp', {}).get(class="hl-string">'code'), class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
                    if seg_end:
                         events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'After-Call Work Ended', class="hl-string">'EventTime': seg_end, class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p.get(class="hl-string">'wrapUp', {}).get(class="hl-string">'code'), class="hl-string">'MediaType': media_type,
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                
                if seg.get(class="hl-string">'segmentType') == class="hl-string">'transmitting' and seg.get(class="hl-string">'disconnectType') == class="hl-string">'transfer':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Interaction Transferred', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
            
            if s.get(class="hl-string">'flow', {}).get(class="hl-string">'outcomeValue') == class="hl-string">'SURVEY_OFFERED':
                events.append({
                    class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Customer Satisfaction Survey Sent', class="hl-string">'EventTime': s.get(class="hl-string">'flow', {}).get(class="hl-string">'flowEnd'), class="hl-string">'EventEndTime': None,
                    class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                    class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                })

        if p.get(class="hl-string">'purpose') == class="hl-string">'acd':
             for s in p.get(class="hl-string">'sessions', []):
                for seg in s.get(class="hl-string">'segments', []):
                    if seg.get(class="hl-string">'segmentType') == class="hl-string">'interact':
                         events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Conversation Routed to Queue', class="hl-string">'EventTime': seg.get(class="hl-string">'segmentStart'), class="hl-string">'EventEndTime': seg.get(class="hl-string">'segmentEnd'),
                            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': s.get(class="hl-string">'mediaType'),
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })

        if p.get(class="hl-string">'wrapUp'):
            events.append({
                class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Wrap-Up Code Assigned', class="hl-string">'EventTime': p[class="hl-string">'wrapUp'].get(class="hl-string">'wrapUpEnd'), class="hl-string">'EventEndTime': None,
                class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p[class="hl-string">'wrapUp'].get(class="hl-string">'code'), class="hl-string">'MediaType': s.get(class="hl-string">'mediaType'),
                class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
            })

    class=class="hl-string">"hl-comment"># Event: Interaction Disconnected
    events.append({
        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Interaction Disconnected', class="hl-string">'EventTime': conv_detail.get(class="hl-string">'conversationEnd'), class="hl-string">'EventEndTime': None,
        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
        class="hl-string">'MediaType': next((p.get(class="hl-string">'sessions')[0].get(class="hl-string">'mediaType') for p in conv_detail.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None),
        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
    })
    
    return events


def main():
    class=class="hl-string">"hl-comment"># --- Define Analytics Query ---
    class=class="hl-string">"hl-comment"># Define your time interval here. Format: YYYY-MM-DDTHH:MM:SS.sssZ/YYYY-MM-DDTHH:MM:SS.sssZ
    interval = class="hl-string">"2023-10-01T00:00:00.000Z/2023-10-02T00:00:00.000Z"
    
    query_body = {
        class="hl-string">"interval": interval,
        class="hl-string">"paging": {
            class="hl-string">"pageSize": 100,
            class="hl-string">"pageNumber": 1
        }
    }

    class=class="hl-string">"hl-comment"># --- Submit and Monitor Job ---
    try:
        print(fclass="hl-string">"Submitting analytics job for interval: {interval}")
        api_response = analytics_api.post_analytics_conversations_details_jobs(query_body)
        job_id = api_response.job_id
        print(fclass="hl-string">"Job submitted successfully. Job ID: {job_id}")

        while True:
            job_status = analytics_api.get_analytics_conversations_details_job(job_id)
            print(fclass="hl-string">"Job status: {job_status.state}. Progress: {job_status.percent_complete}%")
            if job_status.state == class="hl-string">'FULFILLED':
                print(class="hl-string">"Job completed successfully.")
                break
            elif job_status.state in [class="hl-string">'FAILED', class="hl-string">'CANCELLED']:
                print(fclass="hl-string">"Job failed or was cancelled. State: {job_status.state}, Error: {job_status.error}")
                return
            time.sleep(15)

        class=class="hl-string">"hl-comment"># --- Fetch and Process Results ---
        all_events = []
        results_response = analytics_api.get_analytics_conversations_details_job_results(job_id)
        
        if not results_response.conversations:
            print(class="hl-string">"No conversations found for the given interval.")
            return
            
        print(fclass="hl-string">"Processing {len(results_response.conversations)} conversations...")
        for conv in results_response.conversations:
            all_events.extend(parse_conversation(conv))

        df = pd.DataFrame(all_events)
        df[class="hl-string">'EventTime'] = pd.to_datetime(df[class="hl-string">'EventTime'], errors=class="hl-string">'coerce')
        df = df.dropna(subset=[class="hl-string">'EventTime'])
        df = df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime']).reset_index(drop=True)

        class=class="hl-string">"hl-comment"># --- Post-Processing for class="hl-string">'Service Request Re-Opened' ---
        class=class="hl-string">"hl-comment"># This logic is an example. Adjust the window and criteria to fit your business rules.
        print(class="hl-string">"Applying post-processing for 'Re-Opened' events...")
        reopen_events = []
        df_resolved = df[df[class="hl-string">'WrapUpCode'].str.contains(class="hl-string">'Resolved', na=False)].copy()
        df_resolved = df_resolved.sort_values(class="hl-string">'EventTime').drop_duplicates(class="hl-string">'ServiceRequest', keep=class="hl-string">'last')
        reopen_window = timedelta(hours=72)

        for index, row in df_resolved.iterrows():
            customer_id = next((p.get(class="hl-string">'participantName') for c in results_response.conversations if c.get(class="hl-string">'conversationId') == row[class="hl-string">'ServiceRequest'] for p in c.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None)
            if not customer_id: continue
            
            resolved_time = row[class="hl-string">'EventTime']
            future_interactions = df[(df[class="hl-string">'EventTime'] > resolved_time) & (df[class="hl-string">'EventTime'] <= resolved_time + reopen_window)]
            
            for _, future_row in future_interactions.iterrows():
                future_customer_id = next((p.get(class="hl-string">'participantName') for c in results_response.conversations if c.get(class="hl-string">'conversationId') == future_row[class="hl-string">'ServiceRequest'] for p in c.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None)
                if customer_id == future_customer_id and future_row[class="hl-string">'ActivityName'] == class="hl-string">'Interaction Started':
                    reopen_events.append({
                        class="hl-string">'ServiceRequest': row[class="hl-string">'ServiceRequest'], class="hl-string">'ActivityName': class="hl-string">'Service Request Re-Opened', class="hl-string">'EventTime': future_row[class="hl-string">'EventTime'],
                        class="hl-string">'SourceSystem': class="hl-string">'Genesys Cloud CX', class="hl-string">'LastDataUpdate': datetime.utcnow().isoformat() + class="hl-string">'Z'
                    })
                    break

        if reopen_events:
            df_reopen = pd.DataFrame(reopen_events)
            df = pd.concat([df, df_reopen], ignore_index=True)
            df = df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime']).reset_index(drop=True)
        
        class=class="hl-string">"hl-comment"># --- Save to CSV ---
        output_filename = class="hl-string">'genesys_event_log.csv'
        df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%S.%fZ')
        print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

    except PureCloudPlatformClientV2.ApiException as e:
        print(fclass="hl-string">"Error calling Genesys Cloud API: {e}")
    except Exception as e:
        print(fclass="hl-string">"An unexpected error occurred: {e}")

if __name__ == class="hl-string">'__main__':
    main()

ステップ

Genesys Cloud CXの組織で必要なEventBridge連携機能、適切な管理者権限、会話イベントへのアクセス権限、およびイベントの受信権限を持つAWSアカウントがあることを確認します。対象のAWSリージョン、EventBridgeイベントバス、組織、部門、キュー、メディアに関するフィルターをシステム管理者と確認してください。
Genesys Cloud CXで、現在の管理画面と組織の連携文書に従ってAWS EventBridge連携を設定します。詳細なインタラクションのライフサイクル監視に必要なAWSアカウント、リージョン、イベントバス、会話イベントのカテゴリを選択します。Genesys Cloudのメニュー名や利用可能なイベントスキーマは、組織やリリースによって異なる場合があります。組織固有の手順や設定には[Configure based on your system]を使用してください。
選択したイベントバスでGenesys Cloudのイベントを受信できるよう、AWS EventBridgeを設定します。サポートされている場合は、最小権限の適用、暗号化、保持期間、監視、デッドレター先を設定します。必要なイベントを、データレイク、データウェアハウス、ストリーム処理基盤、またはオブジェクトストレージの受け入れ先で使用するコンシューマーに転送するルールを作成します。
元のタイムスタンプや識別子を変更せずに、生のEventBridgeペイロードを取得します。ペイロード、取り込み時刻、イベントタイプ、ソース、詳細タイプ、配信メタデータを[Your raw event storage]に保存します。Genesys Cloudのスキーマ変更時にマッピングを再検証できるよう、生のペイロードを保持してください。
生のペイロードを読み込み、会話識別子、参加者とセッションの詳細、キュー情報、エージェント情報、メディアタイプ、後処理データ、イベント時刻を特定する正規化処理をPythonで作成します。明示的に記録されたアクティビティごとに1行を出力してください。組織で受信したイベントサンプルに基づき、実際のフィールドパスを使用します。明示的なイベントや状態遷移で示されていないアクティビティを推測しないでください。
Genesys Cloudの会話や参加者の明示的な状態変更を、必要なActivityNameの値にマッピングします。会話作成イベントがある場合は「インタラクション開始」を出力します。ルーティングイベントがある場合は「キューにルーティング」を、エージェントの状態が呼び出し中の場合は「エージェントにインタラクションを提示」を出力します。エージェントが受諾または接続した場合は「エージェントがインタラクションを受諾」を出力します。対応する保留状態の遷移には「インタラクションを保留」と「インタラクションの保留を解除」を、明示的な転送イベントには「インタラクションを転送」を出力します。明示的なACWの遷移には「通話後作業を開始」と「通話後作業を終了」を、後処理コードの割り当てがある場合は「後処理コードを割り当て」を出力します。会話または該当する最終参加者セッションが切断された場合は「インタラクション切断」を出力します。
「サービスリクエストの再開」と「顧客満足度アンケート送信」は別々に扱います。EventBridgeの会話イベントに、これらのアクティビティが直接含まれるとは限りません。組織に明示的なアンケートイベントソースがある場合は、そのイベントを受信したときに限り含めます。「サービスリクエストの再開」は、文書化した業務ルールに基づく別の処理で算出し、結果の行をイベントログに追加します。ProcessMindがこのアクティビティを算出、推測することはありません。
業務上のマッピングで選定した安定したサービスリクエスト識別子または会話識別子をServiceRequestに設定します。ActivityNameには必要なアクティビティ名を1つだけ、EventTimeにはアクティビティの開始時刻を設定します。利用可能な場合はEventEndTimeも設定し、SourceSystemにはGenesys Cloud CXとAWS EventBridge、LastDataUpdateには抽出の更新時刻を設定します。任意の属性は、値がある場合に入力し、ない場合はNULLのままにします。
プロバイダーのイベント識別子がある場合はそれを使い、ない場合はServiceRequest、ActivityName、EventTime、参加者またはエージェント識別子、ソースイベント識別子を含む、文書化された複合キーでイベントを重複排除します。EventBridgeの再試行や順序どおりに届かないイベントも考慮してください。別々に発生した保留、転送、提示などの正当な繰り返しアクティビティは重複として除外しないでください。
正規化した出力を、生のペイロード数、イベント時刻、会話識別子、キューとエージェントの詳細、アクティビティ一覧全体と照合して検証します。必須列のServiceRequest、ActivityName、EventTime、SourceSystem、LastDataUpdateと、利用可能な推奨列を含むフラットファイルをエクスポートするか、テーブルに読み込みます。タイムスタンプのタイムゾーンと形式を統一し、ProcessMindで受け入れられることを確認してから、サービスリクエストプロセスの入力としてイベントログをアップロードします。

設定

連携と出力先：対象のAWSアカウント、リージョン、イベントバスに対して、Genesys Cloud CXのAWS EventBridge連携を設定します。利用可能なイベントサンプルで定義されていない組織固有の連携設定には、[Configure based on your system]を使用してください。
イベントの範囲：Genesys Cloudの組織で利用できる、会話、参加者、セッション、ルーティング、状態変更、転送、切断、後処理、ACWの詳細イベントを購読します。本番環境に導入する前に、管理されたテストインタラクションを使って正確なイベントタイプとペイロードのフィールドパスを確認してください。
ケース識別子：業務上のマッピングで選定した安定した会話識別子またはサービスリクエスト識別子をServiceRequestに使用します。別のシステムにサービスリクエスト識別子がある場合は、その識別子とGenesys Cloudの会話識別子の対応関係を文書化してください。
期間：プロセス分析では、まず過去3～6か月分のデータを対象にします。ほぼリアルタイムで処理する場合は、短いポーリング間隔またはストリーミングのチェックポイント間隔を設定し、再処理できる生データの受け入れ層を保持します。障害から復旧する際は期間に重なりを設け、その後、プロバイダーのイベント識別子で重複を除外します。
フィルター：部門、キュー、メディアタイプ、事業部門、インタラクションの日付など、組織内で確認済みのフィルターだけを適用します。連携元のシステムに該当する概念があり、環境内で明示的にマッピングされていない限り、Company CodeやDocument Typeのフィルターは適用しないでください。
タイムスタンプの処理：EventTimeにはソースイベントのタイムスタンプを保持します。取り込み時刻または更新時刻はLastDataUpdateに別途保存します。監査上必要な場合は元のオフセットも保持しながら、すべてのタイムスタンプを文書化されたタイムゾーン、できればUTCに統一します。
任意の属性：AgentIdとAgentNameには参加者またはユーザーの詳細、QueueNameにはルーティングデータ、WrapUpCodeには後処理データ、MediaTypeには会話またはセッションのチャネルを設定できます。EventEndTimeには明示的な終了遷移またはソースの所要時間を設定できます。
パフォーマンス：生データと正規化データをイベントの日付ごとに、可能であれば組織または部門ごとにも分割します。EventBridgeのチェックポイントを使って増分処理を行います。データウェアハウスやデータレイクへの書き込みはバッチ処理し、遅延を監視して、全履歴の繰り返しスキャンを避けてください。
信頼性：再試行、デッドレター処理、再処理手順、重複検出を設定し、配信失敗、スキーマ変更、イベントの遅延を監視します。
前提条件：Genesys Cloudの管理者権限または連携権限、AWS EventBridgeの権限、選択したイベントバスと後続のストレージへのアクセス権限、会話詳細を読み取る権限、および組織のイベント連携に必要なGenesys Cloudの機能やライセンスが必要です。会話データを保存する前に、保持期間とプライバシー要件を確認してください。

クエリの例 python

import json
import os
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional

import boto3


SOURCE_SYSTEM = class="hl-string">"Genesys Cloud CX via AWS EventBridge"
ACTIVITY_NAMES = {
    class="hl-string">"interaction_started": class="hl-string">"Interaction Started",
    class="hl-string">"conversation_routed_to_queue": class="hl-string">"Conversation Routed to Queue",
    class="hl-string">"agent_offered_interaction": class="hl-string">"Agent Offered Interaction",
    class="hl-string">"agent_accepted_interaction": class="hl-string">"Agent Accepted Interaction",
    class="hl-string">"hold_placed_on_interaction": class="hl-string">"Hold Placed on Interaction",
    class="hl-string">"hold_removed_from_interaction": class="hl-string">"Hold Removed from Interaction",
    class="hl-string">"interaction_transferred": class="hl-string">"Interaction Transferred",
    class="hl-string">"after_call_work_started": class="hl-string">"After-Call Work Started",
    class="hl-string">"wrap_up_code_assigned": class="hl-string">"Wrap-Up Code Assigned",
    class="hl-string">"after_call_work_ended": class="hl-string">"After-Call Work Ended",
    class="hl-string">"interaction_disconnected": class="hl-string">"Interaction Disconnected",
    class="hl-string">"service_request_re_opened": class="hl-string">"Service Request Re-Opened",
    class="hl-string">"customer_satisfaction_survey_sent": class="hl-string">"Customer Satisfaction Survey Sent",
}


def first_value(data: Dict[str, Any], paths: Iterable[str]) -> Optional[Any]:
    for path in paths:
        value: Any = data
        found = True
        for part in path.split(class="hl-string">"."):
            if isinstance(value, dict) and part in value:
                value = value[part]
            else:
                found = False
                break
        if found and value not in (None, class="hl-string">""):
            return value
    return None


def parse_time(value: Any) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value / 1000 if value > 100000000000 else value, tz=timezone.utc).isoformat()
    text = str(value).strip()
    if not text:
        return None
    return text


def event_detail(event: Dict[str, Any]) -> Dict[str, Any]:
    detail = event.get(class="hl-string">"detail")
    if isinstance(detail, dict):
        return detail
    if isinstance(detail, str):
        try:
            parsed = json.loads(detail)
            return parsed if isinstance(parsed, dict) else {}
        except json.JSONDecodeError:
            return {}
    return {}


def event_type(event: Dict[str, Any], detail: Dict[str, Any]) -> str:
    value = first_value(detail, [
        class="hl-string">"eventType", class="hl-string">"type", class="hl-string">"event.type", class="hl-string">"topicName", class="hl-string">"name", class="hl-string">"state", class="hl-string">"participant.state", class="hl-string">"session.state"
    ])
    return str(value or event.get(class="hl-string">"detail-type") or event.get(class="hl-string">"source") or class="hl-string">"").lower().replace(class="hl-string">"_", class="hl-string">" ")


def base_row(detail: Dict[str, Any], activity: str, event: Dict[str, Any], refresh_time: str) -> Dict[str, Any]:
    service_request = first_value(detail, [
        class="hl-string">"conversationId", class="hl-string">"conversation.id", class="hl-string">"conversation.conversationId", class="hl-string">"serviceRequest", class="hl-string">"serviceRequestId", class="hl-string">"caseId"
    ])
    event_time = parse_time(first_value(detail, [
        class="hl-string">"eventTime", class="hl-string">"timestamp", class="hl-string">"event.timestamp", class="hl-string">"conversation.startTime", class="hl-string">"participant.startTime", class="hl-string">"session.startTime", class="hl-string">"startTime"
    ]))
    event_end_time = parse_time(first_value(detail, [
        class="hl-string">"eventEndTime", class="hl-string">"endTime", class="hl-string">"event.endTime", class="hl-string">"conversation.endTime", class="hl-string">"participant.endTime", class="hl-string">"session.endTime"
    ]))
    agent_id = first_value(detail, [class="hl-string">"agentId", class="hl-string">"userId", class="hl-string">"agent.id", class="hl-string">"user.id", class="hl-string">"participant.userId", class="hl-string">"participant.user.id"])
    agent_name = first_value(detail, [class="hl-string">"agentName", class="hl-string">"userName", class="hl-string">"agent.name", class="hl-string">"user.name", class="hl-string">"participant.userName", class="hl-string">"participant.user.name"])
    queue_name = first_value(detail, [class="hl-string">"queueName", class="hl-string">"queue.name", class="hl-string">"routing.queueName", class="hl-string">"participant.queueName"])
    wrap_up_code = first_value(detail, [class="hl-string">"wrapUpCode", class="hl-string">"wrapupCode", class="hl-string">"wrapUp.code", class="hl-string">"wrapup.code", class="hl-string">"dispositionCode"])
    media_type = first_value(detail, [class="hl-string">"mediaType", class="hl-string">"media", class="hl-string">"channel", class="hl-string">"conversation.mediaType", class="hl-string">"session.mediaType"])
    return {
        class="hl-string">"ServiceRequest": service_request,
        class="hl-string">"ActivityName": activity,
        class="hl-string">"EventTime": event_time,
        class="hl-string">"EventEndTime": event_end_time,
        class="hl-string">"SourceSystem": SOURCE_SYSTEM,
        class="hl-string">"LastDataUpdate": refresh_time,
        class="hl-string">"AgentId": agent_id,
        class="hl-string">"AgentName": agent_name,
        class="hl-string">"QueueName": queue_name,
        class="hl-string">"WrapUpCode": wrap_up_code,
        class="hl-string">"MediaType": media_type,
        class="hl-string">"SourceEventId": event.get(class="hl-string">"id") or first_value(detail, [class="hl-string">"eventId", class="hl-string">"id"]),
    }


def classify(detail: Dict[str, Any], event: Dict[str, Any]) -> List[str]:
    text = event_type(event, detail)
    values = [str(first_value(detail, paths) or class="hl-string">"").lower() for paths in [
        [class="hl-string">"eventType", class="hl-string">"type"], [class="hl-string">"state"], [class="hl-string">"participant.state"], [class="hl-string">"session.state"], [class="hl-string">"event.name"], [class="hl-string">"event.type"]
    ]]
    text = class="hl-string">" ".join([text] + values)
    activities: List[str] = []
    if any(token in text for token in [class="hl-string">"conversation created", class="hl-string">"interaction started", class="hl-string">"conversation started", class="hl-string">"conversation.created"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_started"])
    if any(token in text for token in [class="hl-string">"routed", class="hl-string">"route to queue", class="hl-string">"queue joined", class="hl-string">"acd queue"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"conversation_routed_to_queue"])
    if any(token in text for token in [class="hl-string">"alerting", class="hl-string">"offered", class="hl-string">"agent offered"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"agent_offered_interaction"])
    if any(token in text for token in [class="hl-string">"connected", class="hl-string">"accepted", class="hl-string">"agent accepted", class="hl-string">"interacting"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"agent_accepted_interaction"])
    if any(token in text for token in [class="hl-string">"hold", class="hl-string">"held"]) and not any(token in text for token in [class="hl-string">"resume", class="hl-string">"unhold", class="hl-string">"off hold"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"hold_placed_on_interaction"])
    if any(token in text for token in [class="hl-string">"resume", class="hl-string">"unhold", class="hl-string">"off hold", class="hl-string">"hold removed"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"hold_removed_from_interaction"])
    if any(token in text for token in [class="hl-string">"transfer", class="hl-string">"transferred"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_transferred"])
    if any(token in text for token in [class="hl-string">"acw start", class="hl-string">"after call work started", class="hl-string">"wrapup start", class="hl-string">"wrap-up start"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"after_call_work_started"])
    if any(token in text for token in [class="hl-string">"wrap-up code", class="hl-string">"wrap up code", class="hl-string">"wrapup code", class="hl-string">"disposition"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"wrap_up_code_assigned"])
    if any(token in text for token in [class="hl-string">"acw end", class="hl-string">"after call work ended", class="hl-string">"wrapup end", class="hl-string">"wrap-up end"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"after_call_work_ended"])
    if any(token in text for token in [class="hl-string">"disconnected", class="hl-string">"disconnect", class="hl-string">"conversation ended", class="hl-string">"conversation completed"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_disconnected"])
    if any(token in text for token in [class="hl-string">"re-opened", class="hl-string">"reopened", class="hl-string">"service request reopened"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"service_request_re_opened"])
    if any(token in text for token in [class="hl-string">"csat sent", class="hl-string">"customer satisfaction survey sent", class="hl-string">"survey sent"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"customer_satisfaction_survey_sent"])
    return list(dict.fromkeys(activities))


def normalize_event(event: Dict[str, Any], refresh_time: str) -> List[Dict[str, Any]]:
    detail = event_detail(event)
    rows: List[Dict[str, Any]] = []
    for activity in classify(detail, event):
        rows.append(base_row(detail, activity, event, refresh_time))
    return rows


def extract_events(event_bus_name: str, start_time: datetime, end_time: datetime) -> List[Dict[str, Any]]:
    client = boto3.client(class="hl-string">"events", region_name=os.environ.get(class="hl-string">"AWS_REGION", class="hl-string">"[Your AWS Region]"))
    rows: List[Dict[str, Any]] = []
    next_token: Optional[str] = None
    refresh_time = datetime.now(timezone.utc).isoformat()
    while True:
        request: Dict[str, Any] = {
            class="hl-string">"EventBusName": event_bus_name,
            class="hl-string">"StartTime": start_time,
            class="hl-string">"EndTime": end_time,
            class="hl-string">"Limit": 100,
        }
        if next_token:
            request[class="hl-string">"NextToken"] = next_token
        response = client.list_events(**request)
        for event in response.get(class="hl-string">"Events", []):
            rows.extend(normalize_event(event, refresh_time))
        next_token = response.get(class="hl-string">"NextToken")
        if not next_token:
            break
    return rows


def append_reopened_and_survey_rows(rows: List[Dict[str, Any]], explicit_business_events: Iterable[Dict[str, Any]]) -> None:
    refresh_time = datetime.now(timezone.utc).isoformat()
    for event in explicit_business_events:
        detail = event_detail(event)
        for activity in [
            ACTIVITY_NAMES[class="hl-string">"service_request_re_opened"],
            ACTIVITY_NAMES[class="hl-string">"customer_satisfaction_survey_sent"],
        ]:
            if activity in classify(detail, event):
                rows.append(base_row(detail, activity, event, refresh_time))


def deduplicate(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    result: List[Dict[str, Any]] = []
    seen = set()
    for row in rows:
        key = (
            row.get(class="hl-string">"SourceEventId"), row.get(class="hl-string">"ServiceRequest"), row.get(class="hl-string">"ActivityName"),
            row.get(class="hl-string">"EventTime"), row.get(class="hl-string">"AgentId"), row.get(class="hl-string">"QueueName")
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(row)
    return result


if __name__ == class="hl-string">"__main__":
    event_bus_name = os.environ.get(class="hl-string">"EVENT_BUS_NAME", class="hl-string">"[Your EventBridge event bus]")
    start_time = datetime.fromisoformat(os.environ.get(class="hl-string">"START_TIME", class="hl-string">"[Start date and time in ISO 8601 format]").replace(class="hl-string">"Z", class="hl-string">"+00:00"))
    end_time = datetime.fromisoformat(os.environ.get(class="hl-string">"END_TIME", class="hl-string">"[End date and time in ISO 8601 format]").replace(class="hl-string">"Z", class="hl-string">"+00:00"))
    normalized_rows = extract_events(event_bus_name, start_time, end_time)
    normalized_rows = deduplicate(normalized_rows)
    output_columns = [
        class="hl-string">"ServiceRequest", class="hl-string">"ActivityName", class="hl-string">"EventTime", class="hl-string">"EventEndTime", class="hl-string">"SourceSystem",
        class="hl-string">"LastDataUpdate", class="hl-string">"AgentId", class="hl-string">"AgentName", class="hl-string">"QueueName", class="hl-string">"WrapUpCode", class="hl-string">"MediaType", class="hl-string">"SourceEventId"
    ]
    output_path = os.environ.get(class="hl-string">"OUTPUT_PATH", class="hl-string">"[Your output file path]")
    import csv
    with open(output_path, class="hl-string">"w", newline=class="hl-string">"", encoding=class="hl-string">"utf-8") as output_file:
        writer = csv.DictWriter(output_file, fieldnames=output_columns)
        writer.writeheader()
        writer.writerows(normalized_rows)

利用を開始する準備はできましたか？

このテンプレートに沿って進めることで、カスタマーサービス業務の実態を明らかにし、効率化につなげられます。

改善ガイドに戻る

テンプレートをダウンロード

問い合わせの繰り返しを減らし、今すぐカスタマーサービスを改善

顧客満足度（CSAT）を高め、初回コンタクト解決率80％を無理なく目指せます。

無料トライアルを開始

クレジットカードは不要です。数分でトライアルを有効化できます。

カスタマーサービスのデータテンプレート

カスタマーサービスのデータテンプレート

カスタマーサービスの属性

カスタマーサービスのアクティビティ

データ抽出ガイド

Genesys Cloud Analyticsの会話詳細エクスポート

ステップ

設定

クエリの例 python

Genesys Cloud Platform APIを使用するPythonスクリプト

ステップ

設定

クエリの例 python

AWS EventBridge連携によるイベントストリーミング

ステップ

設定

クエリの例 python

利用を開始する準備はできましたか？

問い合わせの繰り返しを減らし、今すぐカスタマーサービスを改善