摘要

本文通过一个完整的智能客服机器人案例，演示如何使用 OpenClaw 构建企业级客服系统。文章涵盖意图识别、槽位填充、多轮对话、知识库问答等核心功能，帮助开发者掌握 OpenClaw 在客服场景的应用。通过详细的系统设计和代码实现，让读者了解智能客服机器人的完整构建过程。🤖

---

1. 引言 - 智能客服概述

1.1 客服系统痛点

传统客服系统面临诸多挑战，人工客服难以满足现代服务需求：

痛点	传统客服	OpenClaw智能客服
响应慢	排队等待	即时响应
成本高	人力成本	自动化降本
质量不稳定	依赖个人能力	标准化服务
无法7×24	轮班制	全天候服务
数据难沉淀	分散记录	知识积累

1.2 系统架构设计

1.3 核心功能规划

功能模块	核心能力	技术实现
意图识别	理解用户意图	NLU分类
槽位填充	提取关键信息	实体识别
多轮对话	复杂场景处理	状态机 + DST
知识问答	FAQ自动回答	语义匹配 + RAG
人机协作	无缝转人工	智能路由

---

2. 意图识别模块

2.1 意图定义与管理

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import re

class IntentCategory(Enum):
    """意图类别"""
    CONSULT = "consult"       # 咨询类
    COMPLAINT = "complaint"   # 投诉类
    SERVICE = "service"       # 服务类
    CHITCHAT = "chitchat"     # 闲聊类

@dataclass
class Intent:
    """意图定义"""
    id: str
    name: str
    category: IntentCategory
    description: str
    examples: List[str] = field(default_factory=list)
    slots: List[str] = field(default_factory=list)  # 需要的槽位
    responses: List[str] = field(default_factory=list)
    priority: int = 0

class IntentManager:
    """意图管理器"""
    
    def __init__(self):
        self.intents: Dict[str, Intent] = {}
        self.patterns: Dict[str, List[str]] = {}
    
    def register_intent(self, intent: Intent):
        """注册意图"""
        self.intents[intent.id] = intent
        
        # 生成模式
        self.patterns[intent.id] = self._generate_patterns(intent.examples)
    
    def _generate_patterns(self, examples: List[str]) -> List[str]:
        """生成匹配模式"""
        patterns = []
        
        for example in examples:
            # 简化：直接使用示例作为模式
            # 实际应使用模板生成
            pattern = example.lower()
            patterns.append(pattern)
        
        return patterns
    
    def recognize(self, text: str) -> List[Dict]:
        """识别意图"""
        text_lower = text.lower()
        
        results = []
        
        for intent_id, intent in self.intents.items():
            score = self._calculate_score(text_lower, intent)
            
            if score > 0:
                results.append({
                    "intent_id": intent_id,
                    "intent_name": intent.name,
                    "category": intent.category.value,
                    "confidence": score,
                    "required_slots": intent.slots
                })
        
        # 按置信度排序
        results.sort(key=lambda x: x["confidence"], reverse=True)
        
        return results
    
    def _calculate_score(self, text: str, intent: Intent) -> float:
        """计算意图匹配分数"""
        patterns = self.patterns.get(intent.id, [])
        
        max_score = 0
        
        for pattern in patterns:
            # 简单匹配
            if pattern in text:
                score = len(pattern) / len(text)
                max_score = max(max_score, score)
            else:
                # 关键词匹配
                keywords = pattern.split()
                matched = sum(1 for kw in keywords if kw in text)
                score = matched / len(keywords) * 0.5
                max_score = max(max_score, score)
        
        return max_score
    
    def get_intent(self, intent_id: str) -> Optional[Intent]:
        """获取意图"""
        return self.intents.get(intent_id)

# 使用示例
im = IntentManager()

# 注册意图
im.register_intent(Intent(
    id="query_order",
    name="查询订单",
    category=IntentCategory.CONSULT,
    description="用户查询订单状态",
    examples=[
        "我的订单到哪了",
        "查询订单状态",
        "订单发货了吗",
        "我想看一下我的订单"
    ],
    slots=["order_id"],
    responses=[
        "请提供您的订单号，我帮您查询",
        "好的，请告诉我订单号"
    ]
))

im.register_intent(Intent(
    id="return_goods",
    name="退货退款",
    category=IntentCategory.SERVICE,
    description="用户申请退货退款",
    examples=[
        "我要退货",
        "申请退款",
        "商品不满意想退货",
        "怎么退款"
    ],
    slots=["order_id", "reason"],
    responses=[
        "请提供订单号和退货原因",
        "好的，请告诉我订单号"
    ]
))

# 识别意图
result = im.recognize("我想查一下我的订单到哪了")
print(f"识别结果: {result}")

2.2 意图分类器

from typing import Dict, List, Tuple
import numpy as np

class IntentClassifier:
    """意图分类器"""
    
    def __init__(self):
        self.intent_manager: IntentManager = None
        self.vocabulary: Dict[str, int] = {}
        self.model = None
    
    def train(self, training_data: List[Tuple[str, str]]):
        """训练分类器"""
        # 构建词汇表
        vocab = set()
        for text, intent_id in training_data:
            words = text.lower().split()
            vocab.update(words)
        
        self.vocabulary = {word: i for i, word in enumerate(vocab)}
        
        # 构建特征矩阵
        X = []
        y = []
        
        intent_to_idx = {}
        idx = 0
        
        for text, intent_id in training_data:
            features = self._text_to_features(text)
            X.append(features)
            
            if intent_id not in intent_to_idx:
                intent_to_idx[intent_id] = idx
                idx += 1
            
            y.append(intent_to_idx[intent_id])
        
        self.idx_to_intent = {v: k for k, v in intent_to_idx.items()}
        
        # 训练模型（简化：使用简单的最近邻）
        self.X_train = np.array(X)
        self.y_train = np.array(y)
    
    def _text_to_features(self, text: str) -> np.ndarray:
        """文本转特征向量"""
        features = np.zeros(len(self.vocabulary))
        
        words = text.lower().split()
        for word in words:
            if word in self.vocabulary:
                features[self.vocabulary[word]] = 1
        
        return features
    
    def predict(self, text: str) -> List[Dict]:
        """预测意图"""
        if self.X_train is None:
            return []
        
        features = self._text_to_features(text)
        
        # 计算与训练样本的相似度
        similarities = []
        
        for i, train_features in enumerate(self.X_train):
            sim = np.dot(features, train_features) / (
                np.linalg.norm(features) * np.linalg.norm(train_features) + 1e-8
            )
            similarities.append((i, sim))
        
        # 找最相似的
        similarities.sort(key=lambda x: x[1], reverse=True)
        
        # 聚合结果
        intent_scores = {}
        for idx, sim in similarities[:5]:
            intent_id = self.idx_to_intent[self.y_train[idx]]
            intent_scores[intent_id] = intent_scores.get(intent_id, 0) + sim
        
        # 归一化
        total = sum(intent_scores.values())
        results = []
        
        for intent_id, score in intent_scores.items():
            results.append({
                "intent_id": intent_id,
                "confidence": score / total if total > 0 else 0
            })
        
        results.sort(key=lambda x: x["confidence"], reverse=True)
        
        return results

# 使用示例
classifier = IntentClassifier()

# 训练数据
training_data = [
    ("我的订单到哪了", "query_order"),
    ("查询订单状态", "query_order"),
    ("订单发货了吗", "query_order"),
    ("我要退货", "return_goods"),
    ("申请退款", "return_goods"),
    ("怎么退款", "return_goods"),
    ("你好", "greeting"),
    ("在吗", "greeting"),
]

classifier.train(training_data)

# 预测
result = classifier.predict("我想查订单")
print(f"预测结果: {result}")

---

3. 槽位填充模块

3.1 槽位定义

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from enum import Enum
import re

class SlotType(Enum):
    """槽位类型"""
    TEXT = "text"
    NUMBER = "number"
    DATE = "date"
    PHONE = "phone"
    ORDER_ID = "order_id"
    ENUM = "enum"

@dataclass
class SlotDefinition:
    """槽位定义"""
    name: str
    type: SlotType
    description: str
    required: bool = True
    prompt: str = ""
    validation_pattern: str = ""
    enum_values: List[str] = field(default_factory=list)

@dataclass
class SlotValue:
    """槽位值"""
    name: str
    value: Any
    confidence: float = 1.0
    source: str = "extracted"  # extracted, user_provided, default

class SlotFiller:
    """槽位填充器"""
    
    def __init__(self):
        self.slot_definitions: Dict[str, SlotDefinition] = {}
        self.extractors: Dict[SlotType, callable] = {
            SlotType.TEXT: self._extract_text,
            SlotType.NUMBER: self._extract_number,
            SlotType.DATE: self._extract_date,
            SlotType.PHONE: self._extract_phone,
            SlotType.ORDER_ID: self._extract_order_id,
            SlotType.ENUM: self._extract_enum
        }
    
    def register_slot(self, slot: SlotDefinition):
        """注册槽位"""
        self.slot_definitions[slot.name] = slot
    
    def fill_slots(self, text: str, required_slots: List[str]) -> Dict[str, SlotValue]:
        """填充槽位"""
        result = {}
        
        for slot_name in required_slots:
            slot_def = self.slot_definitions.get(slot_name)
            
            if not slot_def:
                continue
            
            extractor = self.extractors.get(slot_def.type)
            
            if extractor:
                value = extractor(text, slot_def)
                
                if value is not None:
                    result[slot_name] = SlotValue(
                        name=slot_name,
                        value=value,
                        confidence=0.8
                    )
        
        return result
    
    def _extract_text(self, text: str, slot_def: SlotDefinition) -> Optional[str]:
        """提取文本"""
        # 简化：返回整个文本
        return text if text else None
    
    def _extract_number(self, text: str, slot_def: SlotDefinition) -> Optional[float]:
        """提取数字"""
        match = re.search(r'\d+\.?\d*', text)
        if match:
            return float(match.group())
        return None
    
    def _extract_date(self, text: str, slot_def: SlotDefinition) -> Optional[str]:
        """提取日期"""
        patterns = [
            r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?',
            r'\d{1,2}月\d{1,2}日',
            r'今天|明天|后天|昨天'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, text)
            if match:
                return match.group()
        
        return None
    
    def _extract_phone(self, text: str, slot_def: SlotDefinition) -> Optional[str]:
        """提取电话"""
        match = re.search(r'1[3-9]\d{9}', text)
        if match:
            return match.group()
        return None
    
    def _extract_order_id(self, text: str, slot_def: SlotDefinition) -> Optional[str]:
        """提取订单号"""
        # 假设订单号格式：字母+数字，长度10-20
        match = re.search(r'[A-Z]{2}\d{10,}', text.upper())
        if match:
            return match.group()
        
        # 尝试提取纯数字订单号
        match = re.search(r'\d{10,}', text)
        if match:
            return match.group()
        
        return None
    
    def _extract_enum(self, text: str, slot_def: SlotDefinition) -> Optional[str]:
        """提取枚举值"""
        for value in slot_def.enum_values:
            if value in text:
                return value
        return None
    
    def validate_slot(self, slot_name: str, value: Any) -> bool:
        """验证槽位值"""
        slot_def = self.slot_definitions.get(slot_name)
        
        if not slot_def:
            return False
        
        if slot_def.validation_pattern:
            return bool(re.match(slot_def.validation_pattern, str(value)))
        
        return True
    
    def get_missing_slots(self, filled_slots: Dict[str, SlotValue], 
                         required_slots: List[str]) -> List[str]:
        """获取缺失的槽位"""
        missing = []
        
        for slot_name in required_slots:
            slot_def = self.slot_definitions.get(slot_name)
            
            if slot_def and slot_def.required:
                if slot_name not in filled_slots:
                    missing.append(slot_name)
        
        return missing
    
    def get_prompt(self, slot_name: str) -> str:
        """获取槽位提示"""
        slot_def = self.slot_definitions.get(slot_name)
        
        if slot_def:
            return slot_def.prompt
        
        return f"请提供{slot_name}"

# 使用示例
sf = SlotFiller()

# 注册槽位
sf.register_slot(SlotDefinition(
    name="order_id",
    type=SlotType.ORDER_ID,
    description="订单号",
    required=True,
    prompt="请提供您的订单号",
    validation_pattern=r'[A-Z]{2}\d{10,}'
))

sf.register_slot(SlotDefinition(
    name="phone",
    type=SlotType.PHONE,
    description="手机号",
    required=True,
    prompt="请提供您的手机号"
))

sf.register_slot(SlotDefinition(
    name="reason",
    type=SlotType.ENUM,
    description="退货原因",
    required=True,
    prompt="请选择退货原因：质量问题、不喜欢、发错货",
    enum_values=["质量问题", "不喜欢", "发错货", "其他"]
))

# 填充槽位
text = "我的订单号是AB1234567890，手机是13800138000"
slots = sf.fill_slots(text, ["order_id", "phone", "reason"])

print(f"填充结果: {[(k, v.value) for k, v in slots.items()]}")

# 检查缺失槽位
missing = sf.get_missing_slots(slots, ["order_id", "phone", "reason"])
print(f"缺失槽位: {missing}")

if missing:
    print(f"提示: {sf.get_prompt(missing[0])}")

---

4. 多轮对话模块

4.1 对话状态追踪

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import time

class DialogState(Enum):
    """对话状态"""
    IDLE = "idle"
    INTENT_RECOGNIZED = "intent_recognized"
    SLOT_FILLING = "slot_filling"
    CONFIRMING = "confirming"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class DialogContext:
    """对话上下文"""
    session_id: str
    user_id: str
    state: DialogState = DialogState.IDLE
    current_intent: Optional[str] = None
    slots: Dict[str, SlotValue] = field(default_factory=dict)
    history: List[Dict] = field(default_factory=list)
    turn_count: int = 0
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

class DialogStateTracker:
    """对话状态追踪器"""
    
    def __init__(self, intent_manager: IntentManager, slot_filler: SlotFiller):
        self.intent_manager = intent_manager
        self.slot_filler = slot_filler
        self.contexts: Dict[str, DialogContext] = {}
    
    def create_context(self, session_id: str, user_id: str) -> DialogContext:
        """创建对话上下文"""
        context = DialogContext(
            session_id=session_id,
            user_id=user_id
        )
        
        self.contexts[session_id] = context
        return context
    
    def get_context(self, session_id: str) -> Optional[DialogContext]:
        """获取对话上下文"""
        return self.contexts.get(session_id)
    
    def process_input(self, session_id: str, text: str) -> Dict:
        """处理用户输入"""
        context = self.get_context(session_id)
        
        if not context:
            return {"error": "会话不存在"}
        
        # 更新上下文
        context.turn_count += 1
        context.updated_at = time.time()
        
        # 记录历史
        context.history.append({
            "role": "user",
            "content": text,
            "timestamp": time.time()
        })
        
        # 根据当前状态处理
        if context.state == DialogState.IDLE:
            return self._handle_idle(context, text)
        
        elif context.state == DialogState.SLOT_FILLING:
            return self._handle_slot_filling(context, text)
        
        elif context.state == DialogState.CONFIRMING:
            return self._handle_confirming(context, text)
        
        else:
            return self._handle_idle(context, text)
    
    def _handle_idle(self, context: DialogContext, text: str) -> Dict:
        """处理空闲状态"""
        # 识别意图
        intents = self.intent_manager.recognize(text)
        
        if not intents:
            return {
                "response": "抱歉，我没有理解您的意思，请重新描述",
                "state": "idle"
            }
        
        # 选择最可能的意图
        top_intent = intents[0]
        
        if top_intent["confidence"] < 0.3:
            return {
                "response": "您是想" + "还是".join([i["intent_name"] for i in intents[:3]]) + "？",
                "state": "clarifying",
                "candidates": intents[:3]
            }
        
        # 设置当前意图
        context.current_intent = top_intent["intent_id"]
        context.state = DialogState.INTENT_RECOGNIZED
        
        # 获取意图定义
        intent = self.intent_manager.get_intent(top_intent["intent_id"])
        
        if not intent:
            return {
                "response": "抱歉，系统出现错误",
                "state": "error"
            }
        
        # 尝试填充槽位
        filled_slots = self.slot_filler.fill_slots(text, intent.slots)
        context.slots.update(filled_slots)
        
        # 检查缺失槽位
        missing = self.slot_filler.get_missing_slots(context.slots, intent.slots)
        
        if missing:
            context.state = DialogState.SLOT_FILLING
            prompt = self.slot_filler.get_prompt(missing[0])
            
            return {
                "response": prompt,
                "state": "slot_filling",
                "missing_slots": missing
            }
        else:
            context.state = DialogState.CONFIRMING
            
            return {
                "response": self._generate_confirmation(context),
                "state": "confirming"
            }
    
    def _handle_slot_filling(self, context: DialogContext, text: str) -> Dict:
        """处理槽位填充状态"""
        intent = self.intent_manager.get_intent(context.current_intent)
        
        if not intent:
            return {"error": "意图不存在"}
        
        # 填充槽位
        missing = self.slot_filler.get_missing_slots(context.slots, intent.slots)
        
        if missing:
            filled = self.slot_filler.fill_slots(text, missing)
            context.slots.update(filled)
        
        # 再次检查缺失
        missing = self.slot_filler.get_missing_slots(context.slots, intent.slots)
        
        if missing:
            prompt = self.slot_filler.get_prompt(missing[0])
            
            return {
                "response": prompt,
                "state": "slot_filling",
                "missing_slots": missing
            }
        else:
            context.state = DialogState.CONFIRMING
            
            return {
                "response": self._generate_confirmation(context),
                "state": "confirming"
            }
    
    def _handle_confirming(self, context: DialogContext, text: str) -> Dict:
        """处理确认状态"""
        # 检查用户确认
        confirm_keywords = ["是的", "确认", "对", "没问题", "好的"]
        cancel_keywords = ["不是", "取消", "不对", "重新"]
        
        if any(kw in text for kw in confirm_keywords):
            context.state = DialogState.EXECUTING
            
            return {
                "response": "好的，正在为您处理...",
                "state": "executing",
                "action": "execute"
            }
        
        elif any(kw in text for kw in cancel_keywords):
            context.state = DialogState.IDLE
            context.current_intent = None
            context.slots.clear()
            
            return {
                "response": "好的，已取消。请问还有什么可以帮您？",
                "state": "idle"
            }
        
        else:
            return {
                "response": "请确认是否正确？回复'是'确认，回复'否'重新填写",
                "state": "confirming"
            }
    
    def _generate_confirmation(self, context: DialogContext) -> str:
        """生成确认信息"""
        intent = self.intent_manager.get_intent(context.current_intent)
        
        parts = [f"您要{intent.name}，"]
        
        for slot_name, slot_value in context.slots.items():
            slot_def = self.slot_filler.slot_definitions.get(slot_name)
            if slot_def:
                parts.append(f"{slot_def.description}：{slot_value.value}，")
        
        parts.append("确认吗？")
        
        return "".join(parts)

# 使用示例
dst = DialogStateTracker(im, sf)

# 创建会话
context = dst.create_context("session_001", "user_001")

# 处理对话
response1 = dst.process_input("session_001", "我要退货")
print(f"机器人: {response1['response']}")

response2 = dst.process_input("session_001", "AB1234567890")
print(f"机器人: {response2['response']}")

response3 = dst.process_input("session_001", "质量问题")
print(f"机器人: {response3['response']}")

response4 = dst.process_input("session_001", "是的")
print(f"机器人: {response4['response']}")

---

5. 知识问答模块

5.1 FAQ知识库

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import re

@dataclass
class FAQItem:
    """FAQ条目"""
    id: str
    question: str
    answer: str
    keywords: List[str] = field(default_factory=list)
    category: str = ""
    tags: List[str] = field(default_factory=list)

class FAQKnowledgeBase:
    """FAQ知识库"""
    
    def __init__(self):
        self.faqs: Dict[str, FAQItem] = {}
        self.keyword_index: Dict[str, List[str]] = {}
    
    def add_faq(self, faq: FAQItem):
        """添加FAQ"""
        self.faqs[faq.id] = faq
        
        # 更新关键词索引
        for keyword in faq.keywords:
            if keyword not in self.keyword_index:
                self.keyword_index[keyword] = []
            self.keyword_index[keyword].append(faq.id)
    
    def search(self, query: str, top_k: int = 5) -> List[Tuple[FAQItem, float]]:
        """搜索FAQ"""
        query_lower = query.lower()
        
        scores = {}
        
        # 关键词匹配
        for keyword, faq_ids in self.keyword_index.items():
            if keyword in query_lower:
                for faq_id in faq_ids:
                    scores[faq_id] = scores.get(faq_id, 0) + 1
        
        # 问题相似度
        for faq_id, faq in self.faqs.items():
            # 简单的词重叠度
            question_words = set(faq.question.lower().split())
            query_words = set(query_lower.split())
            
            overlap = len(question_words & query_words)
            
            if overlap > 0:
                scores[faq_id] = scores.get(faq_id, 0) + overlap * 0.5
        
        # 排序
        sorted_faqs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        
        results = []
        for faq_id, score in sorted_faqs[:top_k]:
            faq = self.faqs.get(faq_id)
            if faq:
                results.append((faq, score))
        
        return results
    
    def get_answer(self, query: str) -> Optional[Dict]:
        """获取答案"""
        results = self.search(query, top_k=1)
        
        if results:
            faq, score = results[0]
            
            # 归一化分数
            confidence = min(score / 5, 1.0)
            
            return {
                "question": faq.question,
                "answer": faq.answer,
                "confidence": confidence,
                "source": "faq"
            }
        
        return None

# 使用示例
faq_kb = FAQKnowledgeBase()

# 添加FAQ
faq_kb.add_faq(FAQItem(
    id="faq_001",
    question="如何修改收货地址？",
    answer="您可以在订单详情页面点击"修改地址"按钮进行修改。如果订单已发货，请联系客服处理。",
    keywords=["收货地址", "修改地址", "地址"],
    category="订单"
))

faq_kb.add_faq(FAQItem(
    id="faq_002",
    question="退货运费谁承担？",
    answer="质量问题退货，运费由商家承担；非质量问题退货，运费由买家承担。",
    keywords=["退货", "运费", "承担"],
    category="售后"
))

faq_kb.add_faq(FAQItem(
    id="faq_003",
    question="如何联系客服？",
    answer="您可以通过以下方式联系客服：1. 在线客服（点击右下角图标）；2. 客服热线：400-xxx-xxxx；3. 邮箱：service@example.com",
    keywords=["联系客服", "客服", "热线"],
    category="服务"
))

# 搜索
result = faq_kb.get_answer("怎么修改收货地址")
print(f"问题: {result['question']}")
print(f"答案: {result['answer']}")
print(f"置信度: {result['confidence']:.2f}")

5.2 智能问答引擎

from typing import Dict, List, Optional

class IntelligentQA:
    """智能问答引擎"""
    
    def __init__(self, faq_kb: FAQKnowledgeBase):
        self.faq_kb = faq_kb
        self.confidence_threshold = 0.5
    
    def answer(self, question: str) -> Dict:
        """回答问题"""
        # 尝试FAQ匹配
        faq_result = self.faq_kb.get_answer(question)
        
        if faq_result and faq_result["confidence"] >= self.confidence_threshold:
            return {
                "success": True,
                "answer": faq_result["answer"],
                "source": "faq",
                "confidence": faq_result["confidence"],
                "need_human": False
            }
        
        # 尝试知识库问答（RAG）
        # rag_result = self.rag_query(question)
        
        # 如果置信度不够，转人工
        return {
            "success": False,
            "answer": "抱歉，这个问题我暂时无法回答，正在为您转接人工客服...",
            "source": "fallback",
            "confidence": 0,
            "need_human": True
        }
    
    def batch_answer(self, questions: List[str]) -> List[Dict]:
        """批量回答"""
        return [self.answer(q) for q in questions]
    
    def get_suggestions(self, question: str) -> List[str]:
        """获取相关问题建议"""
        results = self.faq_kb.search(question, top_k=5)
        
        return [faq.question for faq, _ in results]

# 使用示例
qa = IntelligentQA(faq_kb)

# 提问
result = qa.answer("退货谁出运费")
print(f"答案: {result['answer']}")
print(f"来源: {result['source']}")

# 获取建议
suggestions = qa.get_suggestions("退货")
print(f"相关问题: {suggestions}")

---

6. 人机协作模块

6.1 智能路由

from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class TransferReason(Enum):
    """转人工原因"""
    LOW_CONFIDENCE = "low_confidence"
    COMPLEX_INTENT = "complex_intent"
    USER_REQUEST = "user_request"
    EMOTION_NEGATIVE = "emotion_negative"
    SYSTEM_ERROR = "system_error"

@dataclass
class HumanAgent:
    """人工客服"""
    id: str
    name: str
    skills: List[str]
    status: str  # online, busy, offline
    current_load: int = 0
    max_load: int = 5

class IntelligentRouter:
    """智能路由"""
    
    def __init__(self):
        self.agents: Dict[str, HumanAgent] = {}
        self.transfer_rules: List[Dict] = []
    
    def register_agent(self, agent: HumanAgent):
        """注册客服"""
        self.agents[agent.id] = agent
    
    def add_transfer_rule(self, rule: Dict):
        """添加转人工规则"""
        self.transfer_rules.append(rule)
    
    def should_transfer(self, context: DialogContext, qa_result: Dict) -> Tuple[bool, TransferReason]:
        """判断是否需要转人工"""
        # 规则1：置信度过低
        if qa_result.get("confidence", 1) < 0.3:
            return True, TransferReason.LOW_CONFIDENCE
        
        # 规则2：用户明确要求
        if context.history:
            last_msg = context.history[-1].get("content", "")
            transfer_keywords = ["转人工", "人工客服", "真人"]
            if any(kw in last_msg for kw in transfer_keywords):
                return True, TransferReason.USER_REQUEST
        
        # 规则3：复杂意图
        intent = context.current_intent
        complex_intents = ["complaint", "refund_high_value"]
        if intent in complex_intents:
            return True, TransferReason.COMPLEX_INTENT
        
        # 规则4：情绪负面
        # emotion = self.detect_emotion(context)
        # if emotion == "negative":
        #     return True, TransferReason.EMOTION_NEGATIVE
        
        return False, None
    
    def route_to_agent(self, context: DialogContext, reason: TransferReason) -> Optional[HumanAgent]:
        """路由到人工客服"""
        # 找到最合适的客服
        available_agents = [
            agent for agent in self.agents.values()
            if agent.status == "online" and agent.current_load < agent.max_load
        ]
        
        if not available_agents:
            return None
        
        # 按负载排序
        available_agents.sort(key=lambda x: x.current_load)
        
        # 选择负载最低的
        selected = available_agents[0]
        
        # 更新负载
        selected.current_load += 1
        
        return selected
    
    def transfer(self, session_id: str, reason: TransferReason) -> Dict:
        """执行转人工"""
        context = self.contexts.get(session_id)
        
        if not context:
            return {"success": False, "message": "会话不存在"}
        
        agent = self.route_to_agent(context, reason)
        
        if agent:
            return {
                "success": True,
                "agent_id": agent.id,
                "agent_name": agent.name,
                "message": f"正在为您转接{agent.name}，请稍候..."
            }
        else:
            return {
                "success": False,
                "message": "当前客服繁忙，请稍后再试或留言"
            }

# 使用示例
router = IntelligentRouter()

# 注册客服
router.register_agent(HumanAgent(
    id="agent_001",
    name="客服小王",
    skills=["订单", "售后"],
    status="online",
    current_load=2
))

router.register_agent(HumanAgent(
    id="agent_002",
    name="客服小李",
    skills=["投诉", "技术"],
    status="online",
    current_load=1
))

# 判断是否转人工
context = DialogContext(session_id="session_001", user_id="user_001")
qa_result = {"confidence": 0.2}

should, reason = router.should_transfer(context, qa_result)
if should:
    print(f"需要转人工，原因: {reason.value}")

---

7. 最佳实践

7.1 系统设计原则

原则	说明	实践
用户优先	体验至上	快速响应 + 准确理解
人机协作	无缝衔接	智能路由
持续学习	不断优化	日志分析 + 模型更新
安全合规	数据保护	敏感信息处理

7.2 常见问题

问题	原因	解决方案
意图识别不准	训练数据不足	增加样本 + 规则补充
槽位提取失败	表述多样	多模式匹配
对话卡死	状态异常	超时重置

---

8. 总结

8.1 核心要点

本文通过完整的智能客服机器人案例，展示了 OpenClaw 在客服场景的应用：

模块	核心功能	技术要点
意图识别	理解用户	NLU分类
槽位填充	信息提取	实体识别
多轮对话	场景处理	DST
知识问答	FAQ回答	语义匹配
人机协作	无缝转接	智能路由

8.2 下一步学习

• 第81篇：OpenClaw 行业应用：金融风控系统

---

参考资料

• OpenClaw 官方文档
• 对话系统设计

---