1. はじめに
- 最近、Bedrockを使ったデモアプリ的なものの練習をしているが、アプリの中でプロンプトを変えて複数回LLMを呼んだり、条件に応じて処理を分岐させたりしたいと思うことがあった。
- それにはいろいろなやり方があるようだが、今回は一番メジャーっぽいLangGraphに入門してみる。
2. 成果物
2.1 仕様
- スマホに関することだけに回答するQ&Aボットを作成する。
- ユーザからの問い合わせ内容について、スマホに関することかどうかをLLMを呼んで判断する。
- スマホに関する問い合わせの場合は、再度LLMを呼んで解決策を回答する。
- スマホに関する問い合わせではない場合は、「スマホ以外のことは分かりません」と回答する。
- 今回はLangGraphの学習目的のため、Webフロントエンドなどはなし(Pythonコマンドラインでの入出力)。
2.2 動作イメージ
Pythonコマンドラインから問い合わせを入力する。
(kiro) PS C:\Kiro> python main.py
🚀 LangGraph + Bedrock スマホ問い合わせ判定デモ
==================================================
------------------------------
問い合わせ内容を入力してください（'quit'で終了）: iphoneの画面が割れた
📱 スマホ問い合わせ判定中...
🤖 判定結果: はい。
この問い合わせはiPhoneの故障に関する質問なので、スマートフォンに関する問い合わせに該当します。
💡 対策案を生成中...
💡 対策案を生成しました
📊 判定結果:
入力: iphoneの画面が割れた
スマホ関連: ✅ はい
詳細: はい。
この問い合わせはiPhoneの故障に関する質問なので、スマートフォンに関する問い合わせに該当します。
💡 対策案:
はい、iPhoneの画面が割れた問題についてご説明させていただきます。
1. 問題の原因として考えられること
- 落下や強い衝撃によって、ガラス画面が割れてしまった可能性が高いです。
- 製品の品質上の問題や、以前からの細かいひびが徐々に大きくなってしまった可能性もあります。
2. 試してもらいたい解決策
(1) 画面の修理
- 最も確実な解決策は、iPhone正規の修理サービスや、認定修理店で画面を交換してもらうことです。
- 料金は3,000円前後かかりますが、ガラス部分を新品と交換してくれるため、しっかりと修理できます。
(2) 画面保護フィルムの貼り付け
- 修理が難しい場合は、割れた部分に透明の保護フィルムを貼ると、さらに割れが広がるのを防げます。
- 100円ショップなどでも購入できる簡単なアイテムです。
(3) 一時的な対応
- 割れた部分にセロハンテープや粘着テープを貼って、切り傷などを防ぐこともできます。
- ただし、このような応急処置は長期的な解決策ではありませんので、できるだけ早めに修理に出すことをおすすめします。
大切なスマートフォンの修理はストレスがありますが、専門家に任せることで確実に治せます。ご不明な点がありましたら、遠慮なくご相談ください。
------------------------------
問い合わせ内容を入力してください（'quit'で終了）: パソコンが起動しない
📱 スマホ問い合わせ判定中...
🤖 判定結果: いいえ。
この問い合わせ内容はスマートフォンに関するものではありません。
パソコンの起動に関する問題は、スマートフォンのカテゴリに含まれません。
📊 判定結果:
入力: パソコンが起動しない
スマホ関連: ❌ いいえ
詳細: いいえ。
この問い合わせ内容はスマートフォンに関するものではありません。
パソコンの起動に関する問題は、スマートフォンのカテゴリに含まれません。
------------------------------
2.3 構成図とシーケンス
- PC内のPython アプリからLLM(Bedrock/Claude)を呼び出す（PC内にAWS認証情報設定済)。
- LangGraphを用いて、まず1回目のLLM呼び出しでスマホ関連の質問かどうかを判断し、YESの場合のみ２回目のLLM呼び出しを行い解決策を調査する。
2.4 コード
PythonコードはKiroで生成したもの。今回はなるべくコードを短く、自分が理解できる範囲のものだけになるように、仕様をなるべくシンプルにして余計な機能をつけないようにした。
Python コード (左のボタンで展開)
main.py
#!/usr/bin/env python3
"""
LangGraphとBedrockを使ったスマホ問い合わせ判定デモ
"""
import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
class GraphState(TypedDict):
"""グラフの状態を定義"""
user_input: str
is_smartphone_query: bool
classification_result: str
solution_suggestions: str
def classify_smartphone_query(state: GraphState) -> GraphState:
"""
ユーザー入力がスマホに関する問い合わせかどうかを判定するノード
"""
print("📱 スマホ問い合わせ判定中...")
# Bedrock Claude 3 Haikuを使用（コスト効率が良い）
llm = ChatBedrock(
model_id="anthropic.claude-3-haiku-20240307-v1:0",
region_name="us-east-1" # 必要に応じて変更
)
system_prompt = """あなたはカスタマーサポートの分類システムです。
ユーザーの問い合わせがスマートフォン（スマホ、携帯電話、iPhone、Android等）に関するものかどうかを判定してください。
判定基準：
- スマートフォンの機種、機能、操作方法に関する質問
- アプリの使い方、インストール、トラブル
- 通信、通話、メール設定
- バッテリー、充電、故障に関する問題
- アクセサリー（ケース、充電器等）
「はい」または「いいえ」で回答し、その後に簡潔な理由を述べてください。"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"問い合わせ内容: {state['user_input']}")
]
try:
response = llm.invoke(messages)
result = response.content
# 判定結果を解析
is_smartphone = "はい" in result.lower() or "yes" in result.lower()
print(f"🤖 判定結果: {result}")
return {
**state,
"is_smartphone_query": is_smartphone,
"classification_result": result,
"solution_suggestions": ""
}
except Exception as e:
print(f"❌ エラーが発生しました: {e}")
return {
**state,
"is_smartphone_query": False,
"classification_result": f"エラー: {str(e)}",
"solution_suggestions": ""
}
def suggest_solutions(state: GraphState) -> GraphState:
"""
スマホ問題に対する対策案を提示するノード
"""
print("💡 対策案を生成中...")
llm = ChatBedrock(
model_id="anthropic.claude-3-haiku-20240307-v1:0",
region_name="us-east-1"
)
system_prompt = """あなたはスマートフォンのテクニカルサポート専門家です。
ユーザーのスマートフォンに関する問題に対して、実用的で分かりやすい対策案を提示してください。
対策案の提示方法：
1. 問題の原因として考えられることを簡潔に説明
2. 試してもらいたい解決策を優先度順に3-5個提示
3. 各解決策は具体的な手順を含める
4. 専門用語は避け、一般ユーザーにも分かりやすい表現を使用
回答は日本語で、親しみやすく丁寧な口調で答えてください。"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"スマートフォンの問題: {state['user_input']}\n\n上記の問題に対する対策案を教えてください。")
]
try:
response = llm.invoke(messages)
suggestions = response.content
print(f"💡 対策案を生成しました")
return {
**state,
"solution_suggestions": suggestions
}
except Exception as e:
print(f"❌ 対策案生成でエラーが発生しました: {e}")
return {
**state,
"solution_suggestions": f"対策案生成エラー: {str(e)}"
}
def create_workflow():
"""ワークフローを作成"""
workflow = StateGraph(GraphState)
# ノードを追加
workflow.add_node("classify", classify_smartphone_query)
workflow.add_node("suggest_solutions", suggest_solutions)
# エントリーポイントを設定
workflow.set_entry_point("classify")
# 条件分岐を設定
def should_suggest_solutions(state: GraphState) -> str:
"""スマホ問題の場合は対策案提示、そうでなければ終了"""
if state["is_smartphone_query"]:
return "suggest_solutions"
else:
return END
# 条件付きエッジを追加
workflow.add_conditional_edges(
"classify",
should_suggest_solutions,
{
"suggest_solutions": "suggest_solutions",
END: END
}
)
# 対策案提示後は終了
workflow.add_edge("suggest_solutions", END)
return workflow.compile()
def main():
"""メイン関数"""
# .envファイルから環境変数を読み込み
load_dotenv()
print("🚀 LangGraph + Bedrock スマホ問い合わせ判定デモ")
print("=" * 50)
# AWS認証情報の確認
if not os.getenv("AWS_ACCESS_KEY_ID") and not os.getenv("AWS_PROFILE"):
print("⚠️ AWS認証情報が設定されていません。")
print(".envファイルにAWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEYまたはAWS_PROFILEを設定してください。")
return
# ワークフローを作成
app = create_workflow()
while True:
print("\n" + "-" * 30)
user_input = input("問い合わせ内容を入力してください（'quit'で終了）: ")
if user_input.lower() in ['quit', 'exit', '終了']:
print("👋 デモを終了します。")
break
if not user_input.strip():
print("❌ 入力が空です。もう一度入力してください。")
continue
# 初期状態を設定
initial_state = {
"user_input": user_input,
"is_smartphone_query": False,
"classification_result": "",
"solution_suggestions": ""
}
try:
# ワークフローを実行
result = app.invoke(initial_state)
print("\n📊 判定結果:")
print(f"入力: {result['user_input']}")
print(f"スマホ関連: {'✅ はい' if result['is_smartphone_query'] else '❌ いいえ'}")
print(f"詳細: {result['classification_result']}")
# スマホ問題の場合は対策案も表示
if result['is_smartphone_query'] and result['solution_suggestions']:
print(f"\n💡 対策案:")
print(result['solution_suggestions'])
except Exception as e:
print(f"❌ 実行エラー: {e}")
if __name__ == "__main__":
main()
3. 学習プロセス
3.1 LangGraphについて入門
- 以下のサイトなどを見て、LangGraphの概念、文法などについて入門した。
- 基本用語についてなんとなく理解した。
- ノード： 処理の単位（LLMを呼んで結果を得たりとか）
- エッジ： ノードをつなぐ線（あるノードの処理が完了後、次に実行するノードの指定）
- 条件付きエッジ： 条件に基づき次に実行するノードを分岐
- ステート： ユーザからの質問文や、処理状況フラグなどを保存しておくところ
3.2 Kiroでハンズオン
- Kiroにいろいろ作ってもらって、ノードや条件付きエッジの使用方法を学習した（以下雰囲気）。
- 仕様変更依頼するとコード(main.py)が自動で更新されるため、毎回差分を確認してコードの内容を把握するようにした（それをさぼって変更を重ねるとどんどん中身が分からなくなるため）
4. 所感
LangGraphを使ってLLMを複数回呼び出したり、条件分岐したりする雰囲気は分かったので、もう少し複雑なグラフ構成などにチャレンジしたい。