こちらの続きです。
導入
順不同ですが、次はこちらをMLflowを使う形に魔改造してウォークスルーしてみます。
Customer Support Agent概要
上記のノートブック上段部を邦訳して抜粋。
LangGraphを使用したインテリジェントなカスタマーサポートエージェントの構築
概要
このチュートリアルでは、LangGraphを使用してインテリジェントなカスタマーサポートエージェントを作成する方法を示します。LangGraphは、複雑な言語モデルワークフローを構築するための強力なツールです。このエージェントは、顧客の問い合わせを分類し、感情を分析し、適切な応答を提供するか、必要に応じて問題をエスカレーションするように設計されています。
動機
今日の急速に変化するビジネス環境では、効率的で正確なカスタマーサポートが重要です。顧客との最初のやり取りを自動化することで、応答時間を大幅に短縮し、全体的な顧客満足度を向上させることができます。このプロジェクトは、先進的な言語モデルとグラフベースのワークフローを組み合わせて、さまざまな顧客の問い合わせに対応できる高度なサポートシステムを作成する方法を紹介することを目的としています。
主要コンポーネント
- 状態管理: 各顧客のやり取りの状態を定義および管理するためにTypedDictを使用。
- 問い合わせの分類: 顧客の問い合わせを技術的、請求、一般のカテゴリに分類。
- 感情分析: 顧客の問い合わせの感情的なトーンを判断。
- 応答生成: 問い合わせのカテゴリと感情に基づいて適切な応答を作成。
- エスカレーションメカニズム: 否定的な感情を持つ問い合わせを自動的に人間のエージェントにエスカレーション。
- ワークフローグラフ: LangGraphを使用して柔軟で拡張可能なワークフローを作成。
方法の詳細
- 初期設定: 環境を設定し、必要なライブラリをインポート。
- 状態定義: 問い合わせ情報、カテゴリ、感情、応答を保持する構造を作成。
- ノード関数: 分類、感情分析、応答生成のための個別の関数を実装。
- グラフ構築: StateGraphを使用してワークフローを定義し、サポートプロセスを表すノードとエッジを追加。
- 条件付きルーティング: カテゴリと感情に基づいて問い合わせをルーティングするロジックを実装。
- ワークフローのコンパイル: グラフを実行可能なアプリケーションにコンパイル。
- 実行: 顧客の問い合わせをワークフローで処理し、結果を取得。
結論
このチュートリアルは、LangGraphのパワーと柔軟性を示し、複雑なAI駆動のワークフローを作成する方法を紹介します。自然言語処理の能力と構造化されたグラフベースのアプローチを組み合わせることで、幅広い問い合わせに効率的に対応できるカスタマーサポートエージェントを作成しました。このシステムは、特定のビジネスニーズに合わせてさらに拡張およびカスタマイズすることができ、既存のカスタマーサポートツールやデータベースと統合して、さらに高度なやり取りを実現する可能性があります。
ここで紹介したアプローチは、カスタマーサポート以外にも広く応用でき、言語モデルを効果的にオーケストレーションして、さまざまな分野で複雑な多段階の問題を解決する方法を示しています。
このチュートリアルはタイトル通りLangGraphを用いて構築されています。
チュートリアル内でも提示されていますが、以下のようなグラフ構造を構築して問い合わせ内容のカテゴリや感情種類から処理の内容を決定して返すエージェントとなっています。
それでは、mlflowにロギングしてDatabricks上で利用可能な形に実装してみます。
実装と実行
Databricks上でノートブックを作成し、LangChain/LangGraph関連とMlflow最新版をインストールします。なお、クラスタはサーバレスを利用しています。
%pip install -q -U langchain-core==0.3.13 langchain-databricks==0.1.1 langchain_community==0.3.3 langgraph==0.2.39
%pip install -q -U typing-extensions
%pip install -q -U "mlflow-skinny[databricks]==2.17.1"
dbutils.library.restartPython()
MLflowのカスタムチャットモデルとして、エージェントを実装します。
実のところLangChain/LangGraphはmlflowの標準フレーバーでサポートされているため、カスタムチャットモデルを作らなくてもロギングは可能です。
チャットインターフェースを備える方がModel Servingを利用する際など便利なのでカスタムチャットモデルとして実装しています。
%%writefile "./customer_support_agent.py"
import uuid
from typing import List, Optional, Dict, TypedDict
import mlflow
from mlflow.pyfunc import ChatModel
from mlflow.models import set_model
from mlflow.entities import SpanType
from mlflow.types.llm import (
ChatResponse,
ChatMessage,
ChatParams,
ChatChoice,
)
from langgraph.graph import StateGraph, END
from langchain_databricks import ChatDatabricks
from langchain_core.prompts import ChatPromptTemplate
class State(TypedDict):
endpoint: str
params: dict
query: str
category: str
sentiment: str
response: str
class CustomerSupportAgent(ChatModel):
def __init__(self):
"""
CustomerSupportAgentクラスの初期化メソッド。
"""
self.models = {}
self.models_config = {}
self.app = None
def load_context(self, context):
"""
エージェントのコンテキストをロードし、モデル構成を設定します。
引数:
context: モデル構成を含むコンテキスト。
"""
self.models = context.model_config.get("models", {})
self.models_config = context.model_config
self.app = self.build_graph()
def predict(
self, context, messages: List[ChatMessage], params: Optional[ChatParams] = None
) -> ChatResponse:
"""
指定されたチャットメッセージとパラメータに対する応答を予測します。
引数:
context: 予測のためのコンテキスト。
messages (List[ChatMessage]): チャットメッセージのリスト。
params (Optional[ChatParams]): チャットのためのオプションのパラメータ。
戻り値:
ChatResponse: チャットモデルからの応答。
"""
with mlflow.start_span(name="Agent", span_type=SpanType.AGENT) as root_span:
root_span.set_inputs(messages)
attributes = {**params.to_dict(), **self.models_config, **self.models}
root_span.set_attributes(attributes)
endpoint = self._get_model_endpoint("agent")
query = messages[-1].content
response = self.run_customer_support(query, endpoint, params.to_dict())
output = ChatResponse(
choices=[
ChatChoice(
index=0,
message=ChatMessage(
role="assistant",
content=response.get("response"),
),
)
],
usage={},
metadata={
"category": response.get("category"),
"sentiment": response.get("sentiment"),
},
model=endpoint,
)
root_span.set_outputs(output)
return output
def build_graph(self):
"""
グラフを構築します。
戻り値:
StateGraph: 構築されたグラフ。
"""
workflow = StateGraph(State)
workflow.add_node("categorize", self.categorize)
workflow.add_node("analyze_sentiment", self.analyze_sentiment)
workflow.add_node("handle_technical", self.handle_technical)
workflow.add_node("handle_billing", self.handle_billing)
workflow.add_node("handle_general", self.handle_general)
workflow.add_node("escalate", self.escalate)
workflow.add_edge("categorize", "analyze_sentiment")
workflow.add_conditional_edges(
"analyze_sentiment",
self.route_query,
{
"handle_technical": "handle_technical",
"handle_billing": "handle_billing",
"handle_general": "handle_general",
"escalate": "escalate",
},
)
workflow.add_edge("handle_technical", END)
workflow.add_edge("handle_billing", END)
workflow.add_edge("handle_general", END)
workflow.add_edge("escalate", END)
workflow.set_entry_point("categorize")
return workflow.compile()
@mlflow.trace(span_type=SpanType.CHAIN)
def categorize(self, state: State) -> State:
"""
顧客の問い合わせをカテゴリに分類します。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
prompt = ChatPromptTemplate.from_template(
"Categorize the following customer query into one of these categories: "
"Technical, Billing, General."
"You must return only one category without any other sentences.\n\n"
"Query: {query}"
)
chain = prompt | ChatDatabricks(
endpoint=state.get("endpoint"), **state.get("params")
)
category = chain.invoke({"query": state["query"]}).content
return {"category": category}
@mlflow.trace(span_type=SpanType.CHAIN)
def analyze_sentiment(self, state: State) -> State:
"""
顧客の問い合わせの感情を分析します。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
prompt = ChatPromptTemplate.from_template(
"Analyze the sentiment of the following customer query. "
"Respond with either 'Positive', 'Neutral', or 'Negative'."
"You must return only one word without any other sentences.\n\n"
"Query: {query}"
)
chain = prompt | ChatDatabricks(
endpoint=state.get("endpoint"), **state.get("params")
)
sentiment = chain.invoke({"query": state["query"]}).content
return {"sentiment": sentiment}
@mlflow.trace(span_type=SpanType.CHAIN)
def handle_technical(self, state: State) -> State:
"""
技術サポートの応答を提供します。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
prompt = ChatPromptTemplate.from_template(
"Provide a technical support response to the following query: {query}"
)
chain = prompt | ChatDatabricks(
endpoint=state.get("endpoint"), **state.get("params")
)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
@mlflow.trace(span_type=SpanType.CHAIN)
def handle_billing(self, state: State) -> State:
"""
請求サポートの応答を提供します。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
prompt = ChatPromptTemplate.from_template(
"Provide a billing support response to the following query: {query}"
)
chain = prompt | ChatDatabricks(
endpoint=state.get("endpoint"), **state.get("params")
)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
@mlflow.trace(span_type=SpanType.CHAIN)
def handle_general(self, state: State) -> State:
"""
一般的なサポートの応答を提供します。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
prompt = ChatPromptTemplate.from_template(
"Provide a general support response to the following query: {query}"
)
chain = prompt | ChatDatabricks(
endpoint=state.get("endpoint"), **state.get("params")
)
response = chain.invoke({"query": state["query"]}).content
return {"response": response}
@mlflow.trace(span_type=SpanType.UNKNOWN)
def escalate(self, state: State) -> State:
"""
問い合わせを人間のエージェントにエスカレートします。
引数:
state (State): 現在の状態。
戻り値:
State: 更新された状態。
"""
return {
"response": "This query has been escalated to a human agent due to its negative sentiment."
}
@mlflow.trace(span_type=SpanType.TOOL)
def route_query(self, state: State) -> str:
"""
問い合わせを感情とカテゴリに基づいてルーティングします。
引数:
state (State): 現在の状態。
戻り値:
str: ルーティング先のノード名。
"""
if state["sentiment"] == "Negative":
return "escalate"
elif state["category"] == "Technical":
return "handle_technical"
elif state["category"] == "Billing":
return "handle_billing"
else:
return "handle_general"
@mlflow.trace(name="Customer Support", span_type=SpanType.CHAIN)
def run_customer_support(self, query, endpoint, params):
"""
顧客サポートのワークフローを実行します。
引数:
query (str): 顧客の問い合わせ。
endpoint (str): モデルのエンドポイント。
params (dict): モデルのパラメータ。
戻り値:
dict: ワークフローの実行結果。
"""
return self.app.invoke({"query": query, "endpoint": endpoint, "params": params})
def _get_model_endpoint(self, role: str) -> str:
"""
指定された役割のモデルエンドポイントを取得します。
引数:
role (str): モデルエンドポイントを取得する役割。
戻り値:
str: モデルエンドポイント。
"""
role_config = self.models.get(role, {})
return role_config.get("endpoint")
set_model(CustomerSupportAgent())
大部分の処理はLangGraphのノード用メソッド定義です。
build_graphメソッド内でこれらのメソッドを利用してグラフを定義しています。
構築したグラフの流れは上記図のように、
- 問い合わせをTechnical, Billing, Generalの3種類に分類
- 問い合わせの感情をPositive, Neutral, Negativeの３種に分類
- カテゴリと感情分析からどのノードを実行するか判定
- 適切なノードで問い合わせに対する回答を生成
となります。
次にカスタムチャットモデルをMLflowに保管・登録します。
モデル設定にLLMのエンドポイント名を指定していますが、前回同様Llama 3.2 3Bモデルを用いたMosaic AI Model Servingエンドポイントを指定しました。
import mlflow
# Databricks Unity Catalogを利用してモデル管理
mlflow.set_registry_uri("databricks-uc")
model_config = {
"models": {
"agent": {
"endpoint": "llama_v3_2_3b_instruct_endpoint",
},
},
}
input_example = {
"messages": [
{
"role": "user",
"content": "My internet connection keeps dropping. Can you help?",
}
]
}
registered_model_name = "training.llm.customer_support_agent"
with mlflow.start_run():
model_info = mlflow.pyfunc.log_model(
"model",
python_model="customer_support_agent.py",
model_config=model_config,
input_example=input_example,
registered_model_name=registered_model_name,
)
モデルが無事保管できたら、ロードして実際に使ってみます。
import mlflow
from mlflow import MlflowClient
from langchain_core.runnables.graph import MermaidDrawMethod
from IPython.display import display, Image
client = MlflowClient()
versions = [
mv.version for mv in client.search_model_versions(f"name='{registered_model_name}'")
]
agent = mlflow.pyfunc.load_model(f"models:/{registered_model_name}/{versions[0]}")
def run_customer_support(query):
result = agent.predict(
{
"messages": [{"role": "user", "content": query}],
"temerature": 0.1,
"max_tokens": 1000,
}
)
print(f"Query: {query}")
print(f"Category: {result['metadata']['category']}")
print(f"Sentiment: {result['metadata']['sentiment']}")
print(f"Response: {result['choices'][0]['message']['content']}")
print("\n")
# escalate
print("--- escalate case ---")
query = "My internet connection keeps dropping. Can you help?"
run_customer_support(query)
# handle_technical
print("--- handle_technical case ---")
query = "I need help talking to ChatGPT as technical"
run_customer_support(query)
# handle_billing
print("--- handle_billing case ---")
query = "where can i find my receipt as billing case?"
run_customer_support(query)
# handle_general
print("--- handle_general case ---")
query = "What are your business hours?"
run_customer_support(query)
実行結果
--- escalate case ---
Query: My internet connection keeps dropping. Can you help?
Category: Technical
Sentiment: Negative
Response: This query has been escalated to a human agent due to its negative sentiment.
--- handle_technical case ---
Query: I need help talking to ChatGPT as technical
Category: Technical
Sentiment: Positive
Response: Here's a technical support response:
**Subject: Assistance with ChatGPT Conversational Interface**
**Dear User,**
Thank you for reaching out to our technical support team. We're here to help you troubleshoot and resolve any issues with our conversational AI, ChatGPT.
**Issue Description:**
You're experiencing difficulty communicating with ChatGPT. Specifically, you're seeking assistance with initiating or maintaining conversations with the chat interface.
**Technical Details:**
1. **OS Compatibility:** ChatGPT is compatible with Windows 10, macOS High Sierra (or later), and Android 5.0 (or later).
2. **Browser Requirements:** A modern web browser (e.g., Google Chrome, Mozilla Firefox, Safari) with JavaScript enabled is necessary for interacting with ChatGPT.
3. **Internet Connection:** A stable internet connection (broadband or Wi-Fi) is required to access ChatGPT.
**Troubleshooting Steps:**
1. Ensure your device meets the minimum system requirements and that your browser is up-to-date.
2. Clear your browser's cache and cookies to start a fresh conversation.
3. Check your internet connection and try interacting with ChatGPT again.
4. If you're using a mobile device, try opening ChatGPT through our mobile app (available for download from the App Store or Google Play Store).
**Additional Tips:**
* Make sure to speak clearly and slowly when interacting with ChatGPT.
* Try using a neutral tone and phrasing to ensure effective communication.
* If you're experiencing difficulties with a specific conversation or topic, feel free to provide more details so our support team can offer personalized guidance.
**Next Steps:**
If you've tried the troubleshooting steps above and are still experiencing issues, please reply to this email with your specific scenario or a detailed description of the problem you're encountering. Our technical support team will be happy to assist you further.
Thank you for your patience, and we look forward to helping you connect with ChatGPT successfully.
Best regards,
[Your Name]
Technical Support Team
--- handle_billing case ---
Query: where can i find my receipt as billing case?
Category: Billing
Sentiment: Neutral
Response: Here is a possible billing support response to the query:
**Subject: Receipt for Billing Case**
**Dear [Customer Name],**
Thank you for reaching out to us regarding your inquiry about the receipt for your billing case. We appreciate your patience and understanding as we assist you with your request.
To locate your receipt, please follow these steps:
1. Log in to your [billing system/platform] account using your [username/email] and [password].
2. Click on the "My Account" or "Billing History" section.
3. Select the billing case or invoice for which you are looking for the receipt.
4. The receipt should be displayed on the screen, or you can download it as a PDF or CSV file.
If you are unable to find the receipt in the above steps, please contact our Billing Support team, and we will be happy to assist you further. We may also need some additional information from you to help locate the receipt.
Alternatively, you can also check the "Order History" section in our website/app, where you may be able to find a downloadable receipt.
If you still have issues, please reply to this email or give us a call at [phone number] so that we can assist you directly.
Thank you for your cooperation and understanding.
**Best regards,**
[Your Name]
Billing Support Team
[Company Name]
[Contact Information]
--- handle_general case ---
Query: What are your business hours?
Category: General
Sentiment: Neutral
Response: Here is a general support response:
"Thank you for reaching out to us. Our business hours are:
* Monday - Friday: [insert hours, e.g. 9:00 AM - 5:00 PM EST]
* Saturday: [insert hours, e.g. 10:00 AM - 4:00 PM EST]
* Sunday: [insert hours, e.g. closed]
We are available to respond to your inquiries and support requests during these hours. If you have a urgent matter outside of our regular hours, please contact us through our [insert contact method, e.g. phone number or email] and we will do our best to assist you as soon as possible.
Please note that our hours may be subject to change, especially during holidays or special events. If you're unsure about our hours or need assistance outside of our regular hours, feel free to ask and we'll do our best to help."
日本語に翻訳すると以下の通り。
--- エスカレートケース ---
クエリ: インターネット接続が頻繁に切れます。助けていただけますか？
カテゴリ: 技術的
感情: ネガティブ
応答: このクエリはネガティブな感情のため、人間のエージェントにエスカレートされました。
--- 技術的なケースの処理 ---
クエリ: ChatGPTと技術的に話すのを手伝ってほしい
カテゴリ: 技術的
感情: ポジティブ
応答: こちらは技術サポートの応答です：
**件名: ChatGPT会話インターフェースのサポート**
**ユーザー様へ**
技術サポートチームにご連絡いただきありがとうございます。ChatGPTの会話AIに関する問題のトラブルシューティングと解決をお手伝いします。
**問題の説明:**
ChatGPTとのコミュニケーションに問題があるようです。特に、チャットインターフェースとの会話の開始や維持に関するサポートを求めています。
**技術的な詳細:**
1. **OSの互換性:** ChatGPTはWindows 10、macOS High Sierra（またはそれ以降）、およびAndroid 5.0（またはそれ以降）に対応しています。
2. **ブラウザの要件:** ChatGPTと対話するには、JavaScriptが有効な最新のウェブブラウザ（例：Google Chrome、Mozilla Firefox、Safari）が必要です。
3. **インターネット接続:** ChatGPTにアクセスするには、安定したインターネット接続（ブロードバンドまたはWi-Fi）が必要です。
**トラブルシューティングの手順:**
1. デバイスが最低システム要件を満たしていることを確認し、ブラウザが最新であることを確認してください。
2. ブラウザのキャッシュとクッキーをクリアして、新しい会話を開始してください。
3. インターネット接続を確認し、再度ChatGPTと対話してみてください。
4. モバイルデバイスを使用している場合は、App StoreまたはGoogle Play Storeからダウンロードできるモバイルアプリを使用してChatGPTを開いてみてください。
**追加のヒント:**
* ChatGPTと対話する際は、明確でゆっくりと話すようにしてください。
* 効果的なコミュニケーションを確保するために、中立的なトーンと表現を使用してみてください。
* 特定の会話やトピックに問題がある場合は、詳細を提供していただければ、サポートチームが個別のガイダンスを提供できます。
**次のステップ:**
上記のトラブルシューティング手順を試しても問題が解決しない場合は、具体的なシナリオや遭遇している問題の詳細な説明を添えてこのメールに返信してください。技術サポートチームがさらにサポートいたします。
ご理解とご協力に感謝し、ChatGPTとの接続が成功することをお祈りしています。
よろしくお願いいたします。
[あなたの名前]
技術サポートチーム
--- 請求ケースの処理 ---
クエリ: 請求ケースの領収書はどこで見つけられますか？
カテゴリ: 請求
感情: 中立
応答: クエリに対する請求サポートの応答は以下の通りです：
**件名: 請求ケースの領収書**
**[お客様の名前]様**
請求ケースの領収書に関するお問い合わせありがとうございます。ご協力とご理解に感謝いたします。
領収書を見つけるには、以下の手順に従ってください：
1. [請求システム/プラットフォーム]アカウントに[ユーザー名/メール]と[パスワード]を使用してログインします。
2. 「マイアカウント」または「請求履歴」セクションをクリックします。
3. 領収書を探している請求ケースまたは請求書を選択します。
4. 領収書が画面に表示されるか、PDFまたはCSVファイルとしてダウンロードできます。
上記の手順で領収書が見つからない場合は、請求サポートチームにご連絡ください。領収書を見つけるために追加の情報が必要な場合もあります。
また、ウェブサイト/アプリの「注文履歴」セクションでもダウンロード可能な領収書を見つけることができます。
それでも問題が解決しない場合は、このメールに返信するか、[電話番号]にお電話いただければ、直接サポートいたします。
ご協力とご理解に感謝いたします。
**よろしくお願いいたします。**
[あなたの名前]
請求サポートチーム
[会社名]
[連絡先情報]
--- 一般的なケースの処理 ---
クエリ: 営業時間は何時ですか？
カテゴリ: 一般
感情: 中立
応答: こちらは一般的なサポートの応答です：
"お問い合わせいただきありがとうございます。私たちの営業時間は以下の通りです：
* 月曜日 - 金曜日: [営業時間を挿入、例：午前9時 - 午後5時 EST]
* 土曜日: [営業時間を挿入、例：午前10時 - 午後4時 EST]
* 日曜日: [営業時間を挿入、例：休業]
これらの時間内にお問い合わせやサポートリクエストに対応いたします。通常の営業時間外に緊急の問題がある場合は、[連絡方法を挿入、例：電話番号またはメール]でご連絡いただければ、できるだけ早く対応いたします。
営業時間は、特に休日や特別なイベントの際に変更される場合があります。営業時間に不明な点がある場合や、通常の営業時間外にサポートが必要な場合は、お気軽にお問い合わせください。できる限り対応いたします。"
問い合わせの内容に基づいて処理を分岐し、適切な処理で回答させるようになっていますね。
ちなみにmlflow tracingにも対応しているのでどのルートを辿って終了したのか追うこともできます。
まとめ
GenAI Agentsのチュートリアル:5. Customer Support AgentをMLflowのカスタムモデルを使うように魔改造して実行してみました。
これによってMLflowによるモデル（エージェント）管理やMosaic AI Model ServingによるDatabricks上でのデプロイを行うことができます。
実務的には、分岐後の処理は専門の別エージェントが担うようにするべきでしょう。
例えば技術的な質問に対しては専用のDBを裏側に構えたRAGエージェントを用意するべきでしょうし、そのエージェントは別で構築して今回のエージェントから呼び出す形にするのが自然だと思います。
比較的シンプルなチュートリアルではありますが、拡張していくといろいろ面白い用途に使えそうですね。