【2026年版】AI カスタマーサポート自動化の導入ガイド：チャットボットから音声AIまで

# RAGベースカスタマーサポートボットの実装例
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams
import hashlib

class CustomerSupportBot:
    def __init__(self):
        self.llm = OpenAI()
        self.vector_db = QdrantClient(url="http://localhost:6333")
        self.collection_name = "support_knowledge"
        self.model = "gpt-4o"
        self.embedding_model = "text-embedding-3-large"

    def create_knowledge_base(self, documents: list[dict]):
        """ナレッジベースの構築"""
        self.vector_db.create_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(
                size=3072,
                distance=Distance.COSINE,
            ),
        )

        for doc in documents:
            embedding = self._get_embedding(doc['content'])
            doc_id = hashlib.md5(
                doc['content'].encode()
            ).hexdigest()

            self.vector_db.upsert(
                collection_name=self.collection_name,
                points=[{
                    'id': doc_id,
                    'vector': embedding,
                    'payload': {
                        'content': doc['content'],
                        'category': doc['category'],
                        'source': doc['source'],
                    },
                }],
            )

    def _get_embedding(self, text: str) -> list[float]:
        """テキストの埋め込みベクトルを取得"""
        response = self.llm.embeddings.create(
            model=self.embedding_model,
            input=text,
        )
        return response.data[0].embedding

    def _search_knowledge(self, query: str, limit: int = 5):
        """関連するナレッジを検索"""
        query_embedding = self._get_embedding(query)
        results = self.vector_db.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=limit,
            score_threshold=0.7,
        )
        return results

    def respond(self, user_message: str,
                conversation_history: list = None):
        """ユーザーメッセージに応答"""
        # 関連ナレッジの検索
        knowledge = self._search_knowledge(user_message)
        context = "\n\n".join([
            f"【{r.payload['category']}】{r.payload['content']}"
            for r in knowledge
        ])

        # システムプロンプト
        system_prompt = f"""あなたは丁寧で専門的なカスタマーサポートAIです。
以下のルールに従って回答してください：

1. 必ずナレッジベースの情報に基づいて回答する
2. 不明な場合は「確認いたします」と回答し、エスカレーションを提案する
3. 敬語を使い、簡潔で分かりやすい表現を心がける
4. 個人情報の取り扱いに注意する
5. 回答の最後に「他にご質問はございますか？」と確認する

【参照するナレッジベース】
{context}
"""

        messages = [{"role": "system", "content": system_prompt}]

        # 会話履歴の追加
        if conversation_history:
            messages.extend(conversation_history)

        messages.append({"role": "user", "content": user_message})

        # LLMによる応答生成
        response = self.llm.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.3,
            max_tokens=1000,
        )

        return {
            'response': response.choices[0].message.content,
            'sources': [r.payload['source'] for r in knowledge],
            'confidence': knowledge[0].score if knowledge else 0,
        }

# エスカレーション判定の実装
class EscalationManager:
    def __init__(self):
        self.escalation_rules = {
            'low_confidence': 0.5,    # 信頼度がこの値以下
            'negative_sentiment': -0.3, # 感情スコアがこの値以下
            'max_turns': 5,            # 最大会話ターン数
            'keywords': [
                '上司', 'マネージャー', '責任者',
                'クレーム', '解約', '返金', '訴訟',
            ],
        }

    def should_escalate(self, context: dict) -> dict:
        """エスカレーションの必要性を判定"""
        reasons = []

        # 信頼度チェック
        if context.get('confidence', 1.0) < self.escalation_rules['low_confidence']:
            reasons.append('回答の信頼度が低い')

        # 感情分析チェック
        if context.get('sentiment', 0) < self.escalation_rules['negative_sentiment']:
            reasons.append('お客様の感情がネガティブ')

        # 会話ターン数チェック
        if context.get('turn_count', 0) >= self.escalation_rules['max_turns']:
            reasons.append('会話が長期化している')

        # キーワードチェック
        message = context.get('last_message', '').lower()
        for keyword in self.escalation_rules['keywords']:
            if keyword in message:
                reasons.append(f'エスカレーションキーワード検出: {keyword}')
                break

        return {
            'should_escalate': len(reasons) > 0,
            'reasons': reasons,
            'priority': 'high' if len(reasons) >= 2 else 'medium',
        }

# 感情分析によるサポート品質向上
from transformers import pipeline
import numpy as np

class SentimentAnalyzer:
    def __init__(self):
        self.analyzer = pipeline(
            "sentiment-analysis",
            model="nlptown/bert-base-multilingual-uncased-sentiment",
            device=0,
        )
        self.emotion_analyzer = pipeline(
            "text-classification",
            model="bhadresh-savani/distilbert-base-uncased-emotion",
            device=0,
        )

    def analyze(self, text: str) -> dict:
        """テキストの感情を分析"""
        sentiment = self.analyzer(text)[0]
        emotion = self.emotion_analyzer(text)[0]

        # スコアを-1〜1に正規化
        star_rating = int(sentiment['label'].split()[0])
        normalized_score = (star_rating - 3) / 2

        return {
            'sentiment_score': normalized_score,
            'sentiment_label': self._get_label(normalized_score),
            'emotion': emotion['label'],
            'emotion_confidence': emotion['score'],
        }

    def _get_label(self, score: float) -> str:
        if score > 0.3:
            return 'positive'
        elif score < -0.3:
            return 'negative'
        return 'neutral'

    def track_conversation_sentiment(
        self, messages: list[str]
    ) -> dict:
        """会話全体の感情推移を追跡"""
        scores = []
        for msg in messages:
            analysis = self.analyze(msg)
            scores.append(analysis['sentiment_score'])

        return {
            'average_sentiment': np.mean(scores),
            'sentiment_trend': 'improving' if scores[-1] > scores[0]
                              else 'declining',
            'min_sentiment': min(scores),
            'max_sentiment': max(scores),
        }

# 最小構成のFAQボット
from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    session_id: str

class ChatResponse(BaseModel):
    response: str
    sources: list[str]
    confidence: float
    escalate: bool

bot = CustomerSupportBot()
escalation = EscalationManager()

@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """チャットエンドポイント"""
    # AI応答の生成
    result = bot.respond(request.message)

    # エスカレーション判定
    esc_result = escalation.should_escalate({
        'confidence': result['confidence'],
        'last_message': request.message,
    })

    return ChatResponse(
        response=result['response'],
        sources=result['sources'],
        confidence=result['confidence'],
        escalate=esc_result['should_escalate'],
    )