生成AIとインクリメンタル処理を用いたカスタマーサポートの効率化

:::note info
**注意**
こちらは、タイトルにあるシナリオにおいてDatabricksの機能をどのように活用できるのかに関する思考実験とプロトタイピングの内容となっています。
:::


発端は最近のこちらの新機能のリリースです。

https://qiita.com/taka_yayoi/items/16ec9516309b775784ba

特にストリーミングテーブルが気になりまして。以下の点がモヤモヤしていました。

- そもそも、ストリーミングテーブルとは何か
- どう使ったらいいのか

色々触っていてどうにか理解できたとなったので、何かで実践したくなりました。

以下で説明はされています。ただし、結構難解。

https://docs.databricks.com/ja/delta-live-tables/index.html#streaming-table

> ストリーミングテーブルは、ストリーミングまたは増分データ処理の追加サポートを備えたDeltaテーブルです。ストリーミングテーブルを使用すると、各行を1回だけ処理して、増大するデータセットを処理できます。ほとんどのデータセットは時間の経過とともに継続的に増加するため、ストリーミングテーブルはほとんどの取り込みワークロードに適しています。ストリーミングテーブルは、データの鮮度と低遅延を必要とするパイプラインに最適です。ストリーミングテーブルは、新しいデータが到着するたびに結果を増分計算して、更新のたびにすべてのソースデータを完全に再計算する必要がなく、結果を最新の状態に保つことができるため、大規模な変換にも役立ちます。ストリーミングテーブルは、追加専用のデータソース用に設計されています。

自分なりに読み解くと、

- ストリーミングテーブルのデータソースは追記型(ストレージにファイルが追加される、メッセージングサービスからメッセージが届く)
- このようなケースで、ソースにあるファイルやメッセージをすべて処理するのは非効率的
- ストリーミングテーブルを用いることで、追記された**データのみ**を取り込めるようになる。これが上の**各行を1回だけ処理して**の意味。

と理解しました。ストリーミングテーブルはソースデータを**どこまで処理したのか**を記憶してくれます。Spark構造化ストリーミングにおけるチェックポイントの機能が組み込まれているということですね。

# ストリーミングテーブルと生成AIによるカスタマーサポートの効率化

前振りが長くなりましたが、そういったデータソースがあるシナリオで簡単な仕組みを作ってみたいと思った訳です。それで思いついたのが、タイトルにあるカスタマーサポートです。オンライン窓口からやってくる問い合わせはまさに追記型です。そして、今のDatabricksの提供機能なら、生成AIの活用を含めエンドツーエンドで効率化のための仕組みが作れることにも気づきました。

今回使うDatabricksの機能は以下の通りです:

- [Databricks Apps](https://docs.databricks.com/ja/dev-tools/databricks-apps/index.html): 問い合わせ画面を作ります。
- [Delta Live Tables](https://docs.databricks.com/ja/delta-live-tables/index.html): 新規に登録された問い合わせを処理するデータパイプラインを実装します。
- [AI関数](https://docs.databricks.com/ja/large-language-models/ai-functions.html): 生成AIを用いて問い合わせのテキストを処理します。
- [Databricksジョブ](https://docs.databricks.com/ja/jobs/index.html): 問い合わせが登録されたらデータパイプラインを起動するようにオーケストレートします。
- [AI/BIダッシュボード](https://docs.databricks.com/ja/dashboards/index.html): 問い合わせの処理結果を可視化します。

![Screenshot 2024-11-08 at 17.17.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5ec7ccbd-80f3-75fe-7747-48f21bea5c7c.png)

## Databricks Appsによるフロントエンド画面の構築

構築したアプリの実装は以下の通りです。

```py
import streamlit as st
import pandas as pd
from databricks.sdk import WorkspaceClient
from io import BytesIO, StringIO
from datetime import datetime

w = WorkspaceClient()

now = datetime.now()
filename = now.strftime('%Y%m%d_%H%M%S') + '.csv'

volume_file_path = f'/Volumes/taka_yayoi_catalog/customer_service/inquiries/{filename}.csv'

st.set_page_config(layout="wide")

st.header("カスタマーサポート問い合わせ窓口")

def form_callback(inquiry_input):    

    strIO = StringIO(f"inquiry\n{inquiry_input}\n")
    binary_data = BytesIO(bytes(strIO.getvalue(), encoding='utf-8'))

    # Upload a file to a volume.
    w.files.upload(volume_file_path, binary_data, overwrite = True)


with st.form(key="my_form",clear_on_submit=True):
    
    st.write("お問い合わせ内容を入力してください。")
    
    inquiry_input = st.text_input('問い合わせ内容', key='inquiry')
    
    submitted = st.form_submit_button("送信")
    if submitted:
        st.write("お問い合わせ内容", inquiry_input)
        form_callback(inquiry_input)
```

画面はこんな感じで。

![Screenshot 2024-11-08 at 16.47.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/73b854c5-f0cb-0d42-cbdd-697bc526fe43.png)

Databricks SDKを使って、問い合わせ内容をUnity Catalogのボリュームに書き出すようにしています。アプリのサービスプリンシパルに書き込み権限を与えておきます。
![Screenshot 2024-11-08 at 16.49.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/825b85ba-a6f5-210a-06cc-0f1e35f7fcb3.png)

うまく動作すれば、問い合わせ登録の都度ファイルが生成されます。今回、個別のファイルに書き出しているのはジョブの[ファイル到着トリガー](https://docs.databricks.com/ja/jobs/file-arrival-triggers.html)を使うためです。

![Screenshot 2024-11-08 at 16.49.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7ce8b298-63d6-8fd2-d15c-aad569884d19.png)

## Delta Live TablesとAI関数による生成AIを用いたインクリメンタルなテキスト処理

ここが個人的には面白かった部分。元々、Delta Live Tables(DLT)はデータパイプラインを効率的に開発、運用できるようにするためのフレームワークでしたが、そこにSQLからLLMを呼び出せる[AI関数](https://docs.databricks.com/ja/large-language-models/ai-functions.html)が入ってきたことで、テキストデータも柔軟に処理できるようになりました。

ここでは、以下のような処理を実装します。

1. 新規に登録された問い合わせのテキストをインクリメンタル(新規のファイルのみを読み込む)に読み込むストリーミングテーブルを宣言
1. ストリーミングテーブルのデータに対して、緊急度の判別、感情分析、製品名の抽出、解答案の作成を行い、結果をマテリアライズドビューとして格納

**ストリーミングテーブル**

```sql
CREATE OR REPLACE STREAMING TABLE inquiries AS
SELECT
  inquiry
FROM
  STREAM (
    read_files("/Volumes/taka_yayoi_catalog/customer_service/inquiries", format => "csv", header => "true")
  )
```

**マテリアライズドビュー**

```sql
CREATE OR REPLACE MATERIALIZED VIEW analyzed_inquiries AS
SELECT
  inquiry,
  ai_classify(inquiry, ARRAY("緊急", "通常")) AS urgency, -- 緊急度
  ai_analyze_sentiment(inquiry) AS sentiment, -- 感情分析
  ai_extract(
    inquiry,
    array('prodcut', 'service')
  ) AS product, -- 製品、サービス名
  ai_query(
      'databricks-meta-llama-3-1-70b-instruct',
      CONCAT('問い合わせに対する解答案を作成してください。問い合わせ: ', inquiry)
    ) AS response_draft
FROM
  LIVE.inquiries
```

これだけで上の処理が実装できました。[`ai_query`関数](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/functions/ai_query)では基盤モデルAPIのエンドポイントを指定して、より柔軟な指示を与えています。

パイプラインを実行して動作確認します。

![Screenshot 2024-11-08 at 16.56.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e6fdea35-85a1-9305-4e8d-51dfcee0bd53.png)

期待した結果が生成されています。

![Screenshot 2024-11-08 at 16.57.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3c910295-7db5-47cb-ff3a-4ee7f60d5073.png)

## ファイル到着トリガーによる処理の自動化

ここまでで大体出来上がっていますが、パイプライン処理を手動で起動しなくてはならないという難点があります。そこで、Databricksジョブのファイル到着トリガーの出番です。名前の通り、特定の場所にファイルが到着したらジョブを起動する機能です。

ここでは一つのタスクのみですが、ジョブは複数のタスクから構成することができます。上で定義したパイプラインを設定します。

![Screenshot 2024-11-08 at 16.59.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/60b29558-03b2-8896-1389-a8f034403903.png)

トリガー設定画面でファイル到着トリガーを指定します。`/Volumes/taka_yayoi_catalog/customer_service/inquiries/`へのファイル到着を監視して、新規ファイルが作成されたらパイプラインを起動するように設定します。

![Screenshot 2024-11-08 at 16.27.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ffe0dc13-d450-2726-d648-14fa15ae2aba.png)

アプリの画面から問い合わせを登録します。

![Screenshot 2024-11-08 at 17.02.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cc2cb3c9-75e2-ea5a-0f49-7b09ee2071a5.png)

少しするとジョブが起動します。

![Screenshot 2024-11-08 at 17.03.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9a822a1e-94a2-246e-ac91-bd4d0130839d.png)

問い合わせと処理結果が更新されました。

![Screenshot 2024-11-08 at 17.07.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1f997122-fc99-82f8-a7fc-687d39a0dc50.png)

## ダッシュボードによる問い合わせの可視化

あとは、概要を把握できるようにダッシュボードを作成します。テーブルを表示している画面から簡単にダッシュボードを作成することができます。

![Screenshot 2024-11-08 at 16.29.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a00778ad-ca7b-3972-7193-cc37b74d180a.png)

![Screenshot 2024-11-08 at 17.09.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/64f0ebb4-d636-d994-2b7a-ec4854af7e1c.png)

ここ数年でテキストの活用の幅が本当に広がりましたね。

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)