RAGで問い合わせ対応をラクにするWebアプリをつくってみた

この記事はCommune Advent Calendar 2024シリーズ1、23日目の記事です。コミューン株式会社でデータサイエンティストをしているsuk1yak1が担当します。

ある部署に対する社内での問い合わせ対応が増えていたのをきっかけに、RAG（Retrieval-Augmented Generation）を活用して問い合わせ対応をラクにする社内向けWebアプリをつくってみたので、その内容を紹介します。

課題感
どんなWebアプリをつくったか？
利用した技術
実装詳細
導入した効果
- 定量的な効果
- 定性的な効果
まとめ

課題感

社内のある業務で質問に対してエビデンスを付与して回答を作成するという業務があるのですが，その回答の品質が担当者のスキルに依存気味で、過去の対応履歴はあるもののナレッジが十分に活用できていないという課題がありました。例えば、過去に対応した担当者が退職・異動した場合や対応履歴が散在していて検索性が悪いケースが挙げられます。

どんなWebアプリをつくったか？

中心となる技術としてRAG（Retrieval-Augmented Generation）を採用しました。理由は、比較的品質の高い対応履歴データが蓄積されており、それを活用すれば一定レベル以上の回答を生成できる見込みがあったためです。社内では、オープンな場所でドキュメンテーションを蓄積する文化が根付いており、そのおかげで質の高い対応履歴データを活用できる環境が整っていました。

また、データソースや問い合わせ対応の業務の実態に合わせて下記機能を実装しました。

1. まとめて回答機能

大量の問い合わせが一度にまとめてくるケースもあったため、表形式でまとめて入力し回答ボタンを押すと非同期 x 並列処理でまとめて回答が返ってくるようにしました。

2. 好みのUIを選択できるように

1のまとめて回答機能の他にUIとして親しみのある一問一答 x ストリーミング形式で回答を出力するチャットボット形式のUIも用意し、個々人で使いやすい方を選択できるようにしました。利用状況を確認したところ、まとめて回答・一問一答の両方が使われており、2つのUIを用意した効果はありそうでした。

3. データソースをニアリアルタイムで更新・反映できる

プロダクトが日々進化していることもあり、問い合わせ内容によっては、過去と現在では出すべき回答が異なっているケースが散見されました。新機能や改善されて変更のあった機能に関するQ&Aをすぐにアプリに取り入れられるようにしたり、古い回答は参照されないようにするために、Google SheetsをInputとして採用し、アプリの立ち上げ時に最新の状況から読み込むようにしました。

データサイエンティストが介在しなくても、問い合わせ業務の担当者がデータソースを更新できるHuman-in-the-Loopのような形を目指しました。

4. 回答のもとになっているデータソースが簡単に確認できるようにする

どのデータソースを元に回答したか参照できるように、UI上でデータソースも必要に応じて確認できるようにし、ハルシネーションが起きていないかや過去の回答との整合性チェックをアプリ上でできるようにしました。

5. LLM-as-a-Judgeの導入

まとめて回答を生成した際に重点的に人間の目でチェックすべき回答はどれか？わかるようにLLMによる評価スコアとその理由も出力するようにしました。

利用した技術

データソース：Google Sheets
Webフレームワーク：Streamlit
デプロイ・ホスティング：Cloud Run
ベクトル検索サービス：Voyager
Embedding モデル：text-embedding-3-small
LLM：GPT-4o, GPT-4o mini
認証：Identity-Aware Proxy（IAP）
モニタリング：langfuse (今回は触れません)

実装詳細

1. データの読み込み

Google Sheets API を利用して特定のGoogle Sheetの特定のシートの決まった範囲を読み込み、polarsのDataFrameとして読み込むようにしました。

データセットはシンプルで過去の問い合わせ(question)とその回答(answer)と回答日(answer_date)の3カラムのみにしています。

import polars as pl
from google.auth import default
from googleapiclient.discovery import build

def fetch_google_sheet(sheet_id:str, range_name:str) -> list[list[str]]:
    creds, _ = default()
    service = build("sheets", "v4", credentials=creds)
    sheet = service.spreadsheets()
    result = sheet.values().get(spreadsheetId=sheet_id, range=range_name).execute()
    values = result.get("values", [])
    return values

sheet_id = "hogefuga"
range_name = "data!A1:C"
values = fetch_google_sheet(sheet_id, range_name)

def fill_empty(values: list[list[str]]) -> list[list[str]]:
    max_length = max(len(row) for row in values)
    uniform_values = [row + [""] * (max_length - len(row)) for row in values]
    return uniform_values

df_qa = pl.DataFrame(fill_empty(values)[1:], orient="row", schema=values[0]).select(
    ["question", "answer", "answer_date"]
)

Google Sheets APIの認証は、Cloud Runでのデプロイ時のサービスアカウントを利用しました。サービスアカウントのメールアドレスに読み込み対象のGoogle Sheetの編集権限を付与しておく必要があります。

2. RAGの実装アプローチ

Embedding modelはOpenAIの text-embedding-3-small を利用しました。理由としては、アプリ立ち上げの際にリアルタイムにEmbeddingの生成も行う必要があり、メモリなどの少ないWebアプリでもレイテンシを小さくするためコストはかかりますが外部のAPIに頼ることにしました。

ベクトル検索はSpotify社が開発したVoyagerを利用しました。下記のようなクラスをアプリ立ち上げ時にインスタンス化しIndexを構築しています。

from abc import ABC, abstractmethod
from typing import Any, Tuple

import numpy as np
from openai import OpenAI
from voyager import Index, Space

class TextEmbedder(ABC):
    @abstractmethod
    def embed(self, texts: list[str]) -> np.ndarray:
        pass

class OpenAIEmbedder(TextEmbedder):
    def __init__(self, model_name: str = "text-embedding-3-small"):
        self.client = OpenAI()
        self.model_name = model_name

    def embed(self, texts: list[str]) -> np.ndarray:
        result_data = self.client.embeddings.create(input=texts, model=self.model_name).data
        embeddings = np.array([data.embedding for data in result_data])
        return embeddings

class VoyagerIndex:
    def __init__(
        self,
        embedder: TextEmbedder,
        embedddings: np.ndarray,
    ):
        self.embedder = embedder
        self.index = Index(Space.Cosine, num_dimensions=embedddings.shape[1])
        self.index.add_items(embedddings)

    def get_similar_items(
        self, text: str, top_k: int = 3
    ) -> Tuple[np.ndarray[Any, np.dtype[np.uint64]], np.ndarray[Any, np.dtype[np.float32]]]:
        neighbors, distances = self.index.query(self.embedder.embed([text])[0], k=top_k)
        return (neighbors, distances)

Generation部分は、OpenAIの gpt-4o を利用しています。

3. LLM-as-a-Judgeによる回答の定量評価

LLM-as-a-Judgeのプロンプトは、Googleが公開してくれているMetric prompt templatesのPointwise question answering qualityのプロンプトを参考に、和訳したり、Structured Outputsを用いてスコアと一緒に理由も出力するようにしたりなど調整しました。

スコア(1~5)は参考程度で速度重視にしたかったため、OpanAIの gpt-4o mini を利用しています。

検証してみた体感としては、5のときは納得のいく回答で、1~4のときは既存のデータソースでは回答が難しいケースでした。また、回答のソースが一部不足している質問&回答結果に対しては2~4の間でスコアがブレやすかったです。(0 or 1の2値分類に変更してもよさそうです。)

4. おまけ：WebUI

Streamlitで実装した際に工夫したポイントをいくつか紹介します。

表形式での入力

Streamlitのdata_editorとformを組み合わせて使いました。他の表形式のデータ(Google SheetやExcel)からそのままコピペして入力することができるのが、data_editorの特色です。また、まとめて入力する際に1セル編集するごとにStreamlitの仕様で読み込みが発生してしまうのですが、form化することで都度発生する読み込みを回避しています。

import pandas as pd
import streamlit as st

DEFAULT_ROW_NUM = 100

with st.form("inputs", border=False, enter_to_submit=False):
    df_question = pd.DataFrame(
        {
            "質問": [None] * DEFAULT_ROW_NUM,
            "回答形式": [None] * DEFAULT_ROW_NUM,
        }
    )
    editable_table = st.data_editor(df_question, num_rows="dynamic", use_container_width=True)
    generate_answer_button = st.form_submit_button("回答生成")

ローディング処理

生成にはどうしても時間がかかってしまうため、ローディング画面を追加してちゃんと動いているよということを伝えられるようにしました。完了した際にメッセージを残すかどうかで2パターンの実装を行いました。

import streamlit as st

# 完了したことをUI上で残して伝えたい場合(まとめて生成画面で採用)
with st.status("回答を生成中です...") as status:
    hogehuga()
    status.update(label="回答を生成しました！")

# ローディング中のみ表示したい場合(チャット形式の一問一答で採用)
with st.spinner("回答を作成中です..."):
    hogehuga()

ロールの追加&参照したデータソースの表示

Streamlitのchat_messageではデフォルトはuser or assistantの2つのロールとそれに対応した2つのアイコンのみが用意されています。

今回は、LLM-as-a-Judgeの結果を別ロールとして見せたかったため、systemロールとして別アイコンで表示するようにしました。また、参照したデータソースはデフォルトで全て表示されていると見づらいためexpanderで表示するようにしました。

import streamlit as st

dict_avatar = {
    "system": "🔍",
}
chat_container = st.container()
for msg in st.session_state.messages:
    with chat_container.chat_message(msg["role"], avatar=dict_avatar.get(msg["role"])):
        option = msg.get("option")
        if option == "expander":
            with st.expander("Retrieval Q&A"):
                st.write(msg["content"])
        else:
            st.write(msg["content"])

messagesの追加は下記のようなコードで行います。

import streamlit as st

st.session_state.messages.append({"role": "user", "content": user_prompt})
st.session_state.messages.append({"role": "assistant", "content": response_text})
st.session_state.messages.append({"role": "assistant", "content": list_retrieval_qa, "option": "expander"})
st.session_state.messages.append(
    {"role": "system", "content": f"{response_eval_qa_text_score} \\n\\n{response_eval_qa_text_reason}"}
)

テーマカラー設定

.streamlit./config.toml を定義することで、テーマカラーなどを変えることができます。

今回はCommune社内で定められているカラーコードの一部を元にテーマカラーを設定しました。

[theme]
primaryColor="#15BDEB"
backgroundColor="#FFFFFF"
secondaryBackgroundColor="#A1E5F7"
textColor="#131313"

導入した効果

2024年11月は約1,000件の問い合わせに対して回答を生成していました。実際に問い合わせ業務で活用いただいた担当者から以下のような声をいただきました。特に定性的な意見で精神的な負担が軽くなったと感じていただけていたのがとても嬉しかったです。

定量的な効果

回答工数25%削減：問い合わせのうち半分がアプリで回答することができ、その回答工数が1/2になったイメージ
レビュー工数80%削減：簡単な問い合わせ対応に対するレビューが1問5分かかっていたのが1問1分になった

定性的な効果

頼れる仕組みがあるのはメンターがいるようなものなので、心強い
参照先を探すみたいな億劫な作業が削減されてるので、気持ち的な削減幅が大きい
過去の問い合わせ履歴を探したり、ナレッジを検索したり、回答を考えたりなど、脳の負荷が結構かかっていたのがサクッと終わって精神的な負担がめちゃくちゃ軽減されている

まとめ

今後はデータソースの自動収集やembeddingを差分更新にするなどの効率化、他の業務への応用を進めていく予定です。

RAG（Retrieval-Augmented Generation）は、既存のナレッジを活用し、効率化と品質向上を両立する手法だと実装して改めて感じました。この記事が、皆様の業務改善やAI活用のきっかけとなれば幸いです。

最後までお読みいただきありがとうございました。