經過本週的深度學習,我們掌握了多輪對話、記憶管理、API 整合、文件處理、圖片識別和錯誤處理等核心技術。今天讓我們將這些技能整合,打造一個功能完整的智能客服系統!
我們的客服系統將具備以下核心能力:
smart_customer_service/
├── main.py # 主程式入口
├── core/
│ ├── __init__.py
│ ├── service_manager.py # 客服管理核心
│ ├── conversation_engine.py # 對話引擎
│ └── knowledge_manager.py # 知識管理器
├── modules/
│ ├── __init__.py
│ ├── ticket_system.py # 工單系統
│ ├── faq_handler.py # FAQ 處理器
│ ├── order_tracker.py # 訂單追蹤
│ └── feedback_collector.py # 回饋收集器
├── integrations/
│ ├── __init__.py
│ ├── crm_integration.py # CRM 系統整合
│ ├── notification_service.py # 通知服務
│ └── analytics_service.py # 分析服務
├── workflows/
│ ├── __init__.py
│ └── customer_service_workflow.py # LangGraph 客服工作流程
├── data/
│ ├── faq_database.json # FAQ 資料庫
│ ├── product_info.json # 產品資訊
│ └── service_policies.json # 服務政策
└── utils/
├── __init__.py
├── intent_classifier.py # 意圖分類器
└── sentiment_analyzer.py # 情感分析器
from typing import Dict, List, Optional, Any
from datetime import datetime
import json
from core.conversation_engine import ConversationEngine
from core.knowledge_manager import KnowledgeManager
from modules.ticket_system import TicketSystem
from utils.intent_classifier import IntentClassifier
from utils.sentiment_analyzer import SentimentAnalyzer
class CustomerServiceManager:
"""智能客服管理核心"""
def __init__(self):
self.conversation_engine = ConversationEngine()
self.knowledge_manager = KnowledgeManager()
self.ticket_system = TicketSystem()
self.intent_classifier = IntentClassifier()
self.sentiment_analyzer = SentimentAnalyzer()
# 載入基礎資料
self._load_service_data()
# 客服統計
self.session_stats = {
'total_conversations': 0,
'resolved_issues': 0,
'escalated_tickets': 0,
'average_satisfaction': 0.0
}
def _load_service_data(self):
"""載入客服相關資料"""
try:
# 載入 FAQ
with open('data/faq_database.json', 'r', encoding='utf-8') as f:
faq_data = json.load(f)
self.knowledge_manager.load_faq(faq_data)
# 載入產品資訊
with open('data/product_info.json', 'r', encoding='utf-8') as f:
product_data = json.load(f)
self.knowledge_manager.load_products(product_data)
except FileNotFoundError as e:
print(f"⚠️ 資料檔案不存在: {e}")
def start_conversation(self, customer_id: str, initial_message: str) -> Dict[str, Any]:
"""開始客服對話"""
self.session_stats['total_conversations'] += 1
# 分析初始意圖和情感
intent = self.intent_classifier.classify(initial_message)
sentiment = self.sentiment_analyzer.analyze(initial_message)
# 建立對話上下文
context = {
'customer_id': customer_id,
'session_start': datetime.now(),
'intent': intent,
'sentiment': sentiment,
'conversation_history': []
}
# 生成歡迎回應
welcome_response = self._generate_welcome_message(intent, sentiment)
return {
'session_id': f"cs_{customer_id}_{int(datetime.now().timestamp())}",
'response': welcome_response,
'context': context,
'suggested_actions': self._get_suggested_actions(intent),
'priority_level': self._calculate_priority(intent, sentiment)
}
def _generate_welcome_message(self, intent: str, sentiment: float) -> str:
"""生成個性化歡迎訊息"""
base_welcome = "您好!我是智能客服助手,很高興為您服務。"
# 根據意圖調整歡迎訊息
intent_messages = {
'complaint': '我了解您遇到了問題,我會盡全力幫助您解決。',
'inquiry': '我會為您提供詳細的資訊和解答。',
'technical_support': '我會協助您解決技術問題。',
'order_status': '我可以幫您查詢訂單狀態和物流資訊。',
'general': '請告訴我您需要什麼協助?'
}
intent_msg = intent_messages.get(intent, intent_messages['general'])
# 根據情感調整語氣
if sentiment < -0.3: # 負面情感
tone_adjustment = "我非常理解您的困擾,讓我們一起來解決這個問題。"
elif sentiment > 0.3: # 正面情感
tone_adjustment = "感謝您選擇我們的服務!"
else:
tone_adjustment = ""
return f"{base_welcome} {intent_msg} {tone_adjustment}".strip()
def _get_suggested_actions(self, intent: str) -> List[str]:
"""根據意圖獲取建議操作"""
suggestions = {
'complaint': [
'📋 建立投訴工單',
'📞 轉接人工客服',
'📄 查看解決方案',
'💰 申請退款/換貨'
],
'inquiry': [
'❓ 查看常見問題',
'📚 瀏覽產品資訊',
'💬 詳細諮詢',
'📝 預約服務'
],
'technical_support': [
'🔧 故障排除指南',
'📱 遠程協助',
'📋 技術工單',
'📞 專家支援'
],
'order_status': [
'📦 查詢物流',
'📋 訂單詳情',
'🔄 修改訂單',
'❌ 取消訂單'
]
}
return suggestions.get(intent, ['💬 繼續對話', '📋 建立工單', '📞 人工服務'])
def _calculate_priority(self, intent: str, sentiment: float) -> str:
"""計算優先級"""
priority_scores = {
'complaint': 3,
'technical_support': 2,
'order_status': 2,
'inquiry': 1
}
base_score = priority_scores.get(intent, 1)
# 負面情感提升優先級
if sentiment < -0.5:
base_score += 2
elif sentiment < -0.2:
base_score += 1
if base_score >= 4:
return 'urgent'
elif base_score >= 3:
return 'high'
elif base_score >= 2:
return 'medium'
else:
return 'low'
import google.generativeai as genai
import json
import os
from typing import Dict
genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
model = genai.GenerativeModel('gemini-pro')
class IntentClassifier:
"""客服意圖分類器"""
def __init__(self):
self.intent_examples = {
'complaint': [
'產品有問題', '服務不滿意', '要投訴', '很生氣', '不滿意',
'退款', '換貨', '品質很差', '服務態度不好'
],
'inquiry': [
'想了解', '請問', '如何', '什麼是', '介紹一下',
'價格多少', '有什麼功能', '規格', '尺寸'
],
'technical_support': [
'不能用', '故障', '壞了', '錯誤', '無法連接',
'系統問題', '軟體', '硬體', '設定', '安裝'
],
'order_status': [
'訂單', '物流', '出貨', '配送', '到貨時間',
'追蹤', '運費', '地址', '收件'
]
}
def classify(self, text: str) -> str:
"""分類客戶意圖"""
try:
prompt = f"""
分析以下客戶訊息的意圖類別:
訊息:{text}
可能的意圖類別:
- complaint: 投訴、抱怨、退換貨
- inquiry: 一般詢問、產品諮詢
- technical_support: 技術問題、故障排除
- order_status: 訂單查詢、物流追蹤
- general: 一般對話、其他
只回答意圖類別名稱,不要額外說明。
"""
response = model.generate_content(prompt)
intent = response.text.strip().lower()
# 驗證回應是否為有效意圖
valid_intents = ['complaint', 'inquiry', 'technical_support', 'order_status', 'general']
return intent if intent in valid_intents else 'general'
except Exception as e:
print(f"意圖分類失敗: {e}")
return self._fallback_classify(text)
def _fallback_classify(self, text: str) -> str:
"""備用分類方法"""
text_lower = text.lower()
# 關鍵詞匹配
for intent, keywords in self.intent_examples.items():
if any(keyword in text_lower for keyword in keywords):
return intent
return 'general'
from langgraph.graph import StateGraph, END
from typing import TypedDict, Dict, Any, Literal
from core.service_manager import CustomerServiceManager
from modules.ticket_system import TicketSystem
from modules.faq_handler import FAQHandler
from modules.order_tracker import OrderTracker
class CustomerServiceState(TypedDict):
customer_id: str
user_message: str
intent: str
sentiment: float
context: Dict[str, Any]
faq_results: Dict[str, Any]
order_info: Dict[str, Any]
ticket_info: Dict[str, Any]
final_response: str
next_action: str
satisfaction_score: float
# 初始化各個模組
service_manager = CustomerServiceManager()
ticket_system = TicketSystem()
faq_handler = FAQHandler()
order_tracker = OrderTracker()
def analyze_customer_request(state: CustomerServiceState) -> CustomerServiceState:
"""分析客戶請求"""
user_message = state["user_message"]
# 使用服務管理器分析請求
analysis = service_manager.start_conversation(
state["customer_id"],
user_message
)
return {
**state,
"intent": analysis['context']['intent'],
"sentiment": analysis['context']['sentiment'],
"context": analysis['context']
}
def handle_faq_query(state: CustomerServiceState) -> CustomerServiceState:
"""處理 FAQ 查詢"""
user_message = state["user_message"]
# 搜尋相關 FAQ
faq_results = faq_handler.search_faq(user_message)
if faq_results['found']:
response = f"""
📚 **根據常見問題解答:**
**問題:** {faq_results['question']}
**解答:** {faq_results['answer']}
**相關資訊:**
{chr(10).join(['• ' + info for info in faq_results.get('related_info', [])])}
還有其他問題嗎?我很樂意為您解答!
"""
next_action = "complete"
else:
response = "很抱歉,我沒有找到直接相關的解答。讓我為您建立一個諮詢工單,會有專人為您處理。"
next_action = "create_ticket"
return {
**state,
"faq_results": faq_results,
"final_response": response,
"next_action": next_action
}
def handle_order_inquiry(state: CustomerServiceState) -> CustomerServiceState:
"""處理訂單查詢"""
customer_id = state["customer_id"]
user_message = state["user_message"]
# 查詢客戶訂單
order_info = order_tracker.get_customer_orders(customer_id)
if order_info['orders']:
# 格式化訂單資訊
orders_text = []
for order in order_info['orders'][:3]: # 顯示最近3筆
orders_text.append(f"""
📦 **訂單 {order['order_id']}**
• 狀態:{order['status']}
• 金額:NT$ {order['amount']:,}
• 預計到貨:{order['estimated_delivery']}
""")
response = f"""
📋 **您的訂單資訊:**
{chr(10).join(orders_text)}
💡 如需查看特定訂單詳情或物流追蹤,請提供訂單編號。
"""
next_action = "complete"
else:
response = "很抱歉,我沒有找到您的訂單記錄。請確認您的會員帳號,或提供訂單編號讓我為您查詢。"
next_action = "create_ticket"
return {
**state,
"order_info": order_info,
"final_response": response,
"next_action": next_action
}
def handle_complaint(state: CustomerServiceState) -> CustomerServiceState:
"""處理投訴"""
customer_id = state["customer_id"]
user_message = state["user_message"]
sentiment = state["sentiment"]
# 建立投訴工單
ticket_info = ticket_system.create_ticket({
'customer_id': customer_id,
'type': 'complaint',
'priority': 'high' if sentiment < -0.3 else 'medium',
'description': user_message,
'category': '客戶投訴'
})
response = f"""
🎫 **投訴工單已建立**
工單編號:{ticket_info['ticket_id']}
優先級:{ticket_info['priority']}
預計處理時間:{ticket_info['estimated_resolution']}
我們非常重視您的意見,專責人員會在 {ticket_info['estimated_response_time']} 內與您聯繫。
同時,您可以:
• 📞 撥打客服專線:0800-123-456
• 💬 透過即時通訊與我們對話
• 📧 發送郵件至:service@company.com
再次為造成您的不便深表歉意。
"""
return {
**state,
"ticket_info": ticket_info,
"final_response": response,
"next_action": "complete"
}
def handle_technical_support(state: CustomerServiceState) -> CustomerServiceState:
"""處理技術支援"""
user_message = state["user_message"]
# 嘗試提供基礎技術解決方案
technical_solutions = {
'連接問題': [
'1. 檢查網路連線是否正常',
'2. 重新啟動路由器',
'3. 確認防火牆設定',
'4. 嘗試更換網路環境'
],
'軟體問題': [
'1. 重新啟動應用程式',
'2. 清除瀏覽器快取',
'3. 更新到最新版本',
'4. 檢查系統相容性'
],
'登入問題': [
'1. 確認帳號密碼正確',
'2. 使用忘記密碼功能',
'3. 清除瀏覽器 Cookie',
'4. 嘗試不同的瀏覽器'
]
}
# 簡單的關鍵詞匹配
solution_key = None
for key in technical_solutions:
if any(keyword in user_message for keyword in key.split()):
solution_key = key
break
if solution_key:
solutions = technical_solutions[solution_key]
response = f"""
🔧 **技術支援建議**
針對您的 {solution_key},請嘗試以下解決方案:
{chr(10).join(solutions)}
如果問題仍未解決,我可以:
• 📋 為您建立技術支援工單
• 📞 安排技術人員回電
• 💻 提供遠程協助服務
請告訴我您需要哪種協助?
"""
next_action = "await_user_choice"
else:
response = "我已了解您的技術問題。讓我為您建立技術支援工單,我們的技術團隊會盡快為您解決。"
next_action = "create_ticket"
return {
**state,
"final_response": response,
"next_action": next_action
}
def route_intent_processing(state: CustomerServiceState) -> Literal["faq", "order", "complaint", "technical", "general"]:
"""根據意圖路由處理"""
intent = state["intent"]
routing_map = {
'inquiry': 'faq',
'order_status': 'order',
'complaint': 'complaint',
'technical_support': 'technical'
}
return routing_map.get(intent, 'faq')
def create_customer_service_workflow():
"""建立客服工作流程"""
workflow = StateGraph(CustomerServiceState)
# 添加節點
workflow.add_node("analyze", analyze_customer_request)
workflow.add_node("faq", handle_faq_query)
workflow.add_node("order", handle_order_inquiry)
workflow.add_node("complaint", handle_complaint)
workflow.add_node("technical", handle_technical_support)
# 設定流程
workflow.set_entry_point("analyze")
# 條件路由
workflow.add_conditional_edges(
"analyze",
route_intent_processing,
{
"faq": "faq",
"order": "order",
"complaint": "complaint",
"technical": "technical"
}
)
# 結束節點
workflow.add_edge("faq", END)
workflow.add_edge("order", END)
workflow.add_edge("complaint", END)
workflow.add_edge("technical", END)
return workflow.compile()
from workflows.customer_service_workflow import create_customer_service_workflow
import uuid
def main():
"""智能客服系統主程式"""
print("🤖 智能客服系統")
print("🎯 整合多輪對話、知識庫、工單系統、多模態處理")
print("=" * 60)
app = create_customer_service_workflow()
# 客戶身份識別
customer_id = input("請輸入您的會員編號(或按Enter使用訪客身份):").strip()
if not customer_id:
customer_id = f"guest_{uuid.uuid4().hex[:8]}"
print(f"👤 客戶編號:{customer_id}")
print("💬 客服助手已就緒,請描述您需要的協助...")
print("💡 範例:'我的訂單什麼時候到貨?'、'產品故障怎麼辦?'、'想要退換貨'")
print("-" * 60)
session_count = 0
while True:
try:
user_message = input(f"\n💬 您:").strip()
if user_message.lower() in ['quit', 'exit', '結束', '再見']:
print("👋 感謝您的使用!如有其他問題,歡迎隨時回來。")
break
if not user_message:
continue
session_count += 1
print("🔍 分析處理中...")
# 執行客服工作流程
initial_state = {
"customer_id": customer_id,
"user_message": user_message,
"intent": "",
"sentiment": 0.0,
"context": {},
"faq_results": {},
"order_info": {},
"ticket_info": {},
"final_response": "",
"next_action": "",
"satisfaction_score": 0.0
}
result = app.invoke(initial_state)
print(f"\n🤖 客服助手:{result['final_response']}")
# 顯示服務資訊
if result.get('intent'):
intent_names = {
'inquiry': '一般諮詢',
'complaint': '投訴處理',
'order_status': '訂單查詢',
'technical_support': '技術支援'
}
print(f"🎯 服務類型:{intent_names.get(result['intent'], '一般服務')}")
# 滿意度調查(每3次互動一次)
if session_count % 3 == 0:
satisfaction = input("\n⭐ 請為本次服務評分 (1-5星,按Enter跳過):").strip()
if satisfaction.isdigit() and 1 <= int(satisfaction) <= 5:
print("📊 感謝您的評分!我們會持續改進服務品質。")
print("-" * 60)
except KeyboardInterrupt:
print("\n👋 感謝您的使用!")
break
except Exception as e:
print(f"\n❌ 系統發生錯誤:{e}")
print("🔄 請重新描述您的問題,我們會繼續為您服務。")
if __name__ == "__main__":
main()
第三週我們將進入進階應用開發:
恭喜完成第二週的學習!你已經能夠建立功能完整的 AI 助理系統。下週我們將探索更進階的技術,讓你的 AI 助理變得更加智能和實用!🎉