中期專案：智慧客服聊天機器人

2025 iThome 鐵人賽

DAY 15

Build on AWS

從零開始的AWS AI之路：用Bedrock與SageMaker打造智慧應用的30天實戰系列第 15 篇

17th鐵人賽 aws bedrock aws sagemaker

MichaelHo

2025-09-29 14:11:02

24 瀏覽

分享至

前言

前陣子已經針對 aws bedrock 和 asw sageMaker 做各種介紹，我們要開始整合各方知識實作一個專案
過往我在公司也有過過智能客服聊天機器人，但那時候是以 openai 的 api 實作，今天我們嘗試用 aws
的服務建立起來，不同的是，我這次用 python 實作，(當時是以 go 去完成該項目)。今天建立這樣的主題
也同時是給自己一個挑戰

首先我們要達成的專案目標

理解客戶問題並提供準確回答
記住對話歷史，保持對話連貫性
整合企業知識庫（FAQ、產品資訊）
處理多輪對話
提供友善的使用者介面

架構思維

使用者 -> Streamlit UI -> 

Lambda/API Gateway -> Bedrock Claude

Lambda/API Gateway ->  DynamoDB (對話歷史) -> S3 (知識庫文件)

核心組件

對話管理：維護對話狀態和歷史
意圖識別：理解使用者問題類型
知識檢索：從知識庫中找到相關資訊
回應生成：使用 Claude 生成自然的回答

開始專案

step 0: 啟動專案

uv init

安裝依賴

uv add boto3 streamlit python-dotenv

step 1 : 準備知識庫

(以下知識庫靈感和資料本身是參考 ai 幫我生成的)

knowledge_base.py

FAQ_DATA = {
    "產品資訊": {
        "產品有哪些特色？": "我們的產品具備 AI 智能分析、雲端同步、多平台支援等特色。",
        "支援哪些作業系統？": "支援 Windows、macOS、Linux 以及 iOS、Android 行動裝置。",
        "價格方案？": "我們提供免費版、專業版（NT$299/月）和企業版（客製化報價）。"
    },
    "技術支援": {
        "如何重設密碼？": "請點擊登入頁面的「忘記密碼」連結，輸入註冊信箱後，系統會寄送重設連結。",
        "資料如何備份？": "系統每天自動備份，您也可以在設定中手動匯出資料。",
        "遇到錯誤怎麼辦？": "請先確認網路連線，然後嘗試重新整理。若問題持續，請聯繫技術支援。"
    },
    "帳戶管理": {
        "如何升級方案？": "登入後進入「帳戶設定」→「訂閱方案」，選擇想要的方案並完成付款。",
        "可以取消訂閱嗎？": "可以隨時取消，取消後該計費週期結束前仍可使用付費功能。",
        "如何更改帳戶資訊？": "請到「個人資料」頁面進行修改，變更信箱需要驗證。"
    }
}

def search_knowledge_base(query: str) -> str:
    """簡單的知識庫搜尋"""
    query_lower = query.lower()
    
    # 關鍵字匹配
    keywords = {
        "價格|費用|方案": "產品資訊",
        "密碼|登入|帳號": "帳戶管理",
        "錯誤|問題|bug": "技術支援",
        "備份|資料": "技術支援",
        "特色|功能": "產品資訊"
    }
    
    relevant_info = []
    for pattern, category in keywords.items():
        if any(kw in query_lower for kw in pattern.split("|")):
            if category in FAQ_DATA:
                relevant_info.extend(FAQ_DATA[category].values())
    
    return "\n".join(relevant_info) if relevant_info else ""

step 2 : 對話管理器

conversation_manager.py

import json
from datetime import datetime
from typing import List, Dict

class ConversationManager:
    def __init__(self):
        self.conversations = {}
    
    def create_session(self, session_id: str):
        """建立新的對話會話"""
        self.conversations[session_id] = {
            "messages": [],
            "created_at": datetime.now().isoformat(),
            "updated_at": datetime.now().isoformat()
        }
    
    def add_message(self, session_id: str, role: str, content: str):
        """新增訊息到對話歷史"""
        if session_id not in self.conversations:
            self.create_session(session_id)
        
        self.conversations[session_id]["messages"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        self.conversations[session_id]["updated_at"] = datetime.now().isoformat()
    
    def get_history(self, session_id: str, limit: int = 10) -> List[Dict]:
        """取得對話歷史（最近 N 則）"""
        if session_id not in self.conversations:
            return []
        
        messages = self.conversations[session_id]["messages"]
        return messages[-limit:] if len(messages) > limit else messages
    
    def clear_session(self, session_id: str):
        """清除對話會話"""
        if session_id in self.conversations:
            del self.conversations[session_id]
    
    def format_history_for_claude(self, session_id: str) -> List[Dict]:
        """將歷史格式化為 Claude API 格式"""
        history = self.get_history(session_id)
        return [
            {"role": msg["role"], "content": msg["content"]}
            for msg in history
        ]

step 3 : 智慧客服核心

chatbot.py

import boto3
import json
from knowledge_base import search_knowledge_base
from conversation_manager import ConversationManager

class SmartCustomerServiceBot:
    def __init__(self):
        self.bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
        self.model_id = "anthropic.claude-3-sonnet-20240229-v1:0"
        self.conversation_manager = ConversationManager()
        
        # 系統提示詞
        self.system_prompt = """你是一位專業且友善的客服助理。你的任務是：

            1. 理解客戶的問題和需求
            2. 根據提供的知識庫資訊回答問題
            3. 如果知識庫中沒有相關資訊，誠實告知並建議聯繫人工客服
            4. 保持對話自然流暢，記住之前的對話內容
            5. 使用繁體中文回答
            6. 態度親切、有耐心，避免使用過於技術性的術語

            回答格式要求：
            - 簡潔明瞭，避免冗長
            - 如有步驟說明，請使用編號列出
            - 重要資訊可以適當強調
        """
    
    def chat(self, session_id: str, user_message: str) -> str:
        """處理使用者訊息並回傳回應"""
        
        # 1. 搜尋知識庫
        knowledge = search_knowledge_base(user_message)
        
        # 2. 建構提示詞
        if knowledge:
            context = f"\n\n相關知識庫資訊：\n{knowledge}"
        else:
            context = "\n\n（知識庫中未找到直接相關資訊）"
        
        # 3. 取得對話歷史
        history = self.conversation_manager.format_history_for_claude(session_id)
        
        # 4. 建構完整訊息
        messages = history + [
            {
                "role": "user",
                "content": user_message + context
            }
        ]
        
        # 5. 呼叫 Bedrock API
        try:
            response = self.bedrock.invoke_model(
                modelId=self.model_id,
                body=json.dumps({
                    "anthropic_version": "bedrock-2023-05-31",
                    "max_tokens": 1000,
                    "temperature": 0.7,
                    "system": self.system_prompt,
                    "messages": messages
                })
            )
            
            # 6. 解析回應
            response_body = json.loads(response['body'].read())
            assistant_message = response_body['content'][0]['text']
            
            # 7. 儲存對話歷史
            self.conversation_manager.add_message(session_id, "user", user_message)
            self.conversation_manager.add_message(session_id, "assistant", assistant_message)
            
            return assistant_message
            
        except Exception as e:
            return f"抱歉，系統發生錯誤：{str(e)}"
    
    def reset_conversation(self, session_id: str):
        """重置對話"""
        self.conversation_manager.clear_session(session_id)

使用 streamlit user interface

app.py

import streamlit as st
import uuid
from chatbot import SmartCustomerServiceBot

# 頁面設定
st.set_page_config(
    page_title="智慧客服助理",
    page_icon="🤖",
    layout="wide"
)

# 初始化
if 'session_id' not in st.session_state:
    st.session_state.session_id = str(uuid.uuid4())

if 'bot' not in st.session_state:
    st.session_state.bot = SmartCustomerServiceBot()

if 'messages' not in st.session_state:
    st.session_state.messages = []

# 頁面標題
st.title("🤖 智慧客服助理")
st.markdown("---")

# 側邊欄
with st.sidebar:
    st.header("📋 功能選單")
    
    if st.button("🔄 開始新對話"):
        st.session_state.bot.reset_conversation(st.session_state.session_id)
        st.session_state.messages = []
        st.session_state.session_id = str(uuid.uuid4())
        st.rerun()
    
    st.markdown("---")
    st.subheader("💡 常見問題")
    quick_questions = [
        "產品有哪些特色？",
        "如何重設密碼？",
        "價格方案有哪些？",
        "如何升級方案？"
    ]
    
    for question in quick_questions:
        if st.button(question, key=f"quick_{question}"):
            st.session_state.messages.append({"role": "user", "content": question})
            with st.spinner("思考中..."):
                response = st.session_state.bot.chat(st.session_state.session_id, question)
                st.session_state.messages.append({"role": "assistant", "content": response})
            st.rerun()
    
    st.markdown("---")
    st.info("💬 提示：您可以問我關於產品、技術支援或帳戶管理的問題！")

# 顯示對話歷史
chat_container = st.container()
with chat_container:
    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

# 使用者輸入
if prompt := st.chat_input("請輸入您的問題..."):
    # 顯示使用者訊息
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)
    
    # 取得並顯示助理回應
    with st.chat_message("assistant"):
        with st.spinner("思考中..."):
            response = st.session_state.bot.chat(st.session_state.session_id, prompt)
            st.markdown(response)
    
    st.session_state.messages.append({"role": "assistant", "content": response})
    st.rerun()

# 頁尾
st.markdown("---")
st.caption("Powered by Amazon Bedrock & Claude 3")

跑起來

uv run streamlit run app.py

優化方向與目標

整合 RAG

# 使用 Amazon Bedrock Knowledge Base
from boto3 import client

bedrock_agent = client('bedrock-agent-runtime')

def query_knowledge_base(query: str, kb_id: str):
    response = bedrock_agent.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={'text': query},
        retrievalConfiguration={
            'vectorSearchConfiguration': {
                'numberOfResults': 5
            }
        }
    )
    return response['retrievalResults']

加入意圖分類

def classify_intent(user_message: str) -> str:
    """使用 Claude 分類使用者意圖"""
    intents_prompt = """
    分析以下使用者訊息，判斷意圖類別：
    - product_info: 產品資訊查詢
    - technical_support: 技術支援
    - account_management: 帳戶管理
    - complaint: 投訴抱怨
    - other: 其他
    
    只回傳意圖類別名稱。
    
    使用者訊息：{message}
    """
    # 呼叫 Claude 進行分類
    # ...

加入情緒分析

def analyze_sentiment(text: str) -> str:
    """分析使用者情緒"""
    # 使用 Claude 分析情緒
    # 根據情緒調整回應策略
    # 這裏就自由發揮 --> 如果需要的話
    pass

品質監控

class QualityMonitor:
    def __init__(self):
        self.metrics = {
            'total_conversations': 0,
            'avg_response_time': 0,
            'user_satisfaction': 0
        }
    
    def log_conversation(self, session_id, metrics):
        """記錄對話品質指標"""
        # 儲存到 CloudWatch 或 DynamoDB
        pass
    
    def generate_report(self):
        """生成品質報告"""
        pass