企业客服自动化

本文介绍如何使用 OpenClaw 构建企业级智能客服系统，提升客户服务效率。

系统架构

┌─────────────────────────────────────────────────────────┐
│                    OpenClaw 客服系统                      │
│                                                         │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────────┐    │
│  │ 知识库   │ │ 意图识别 │ │ 工单系统 │ │ 数据分析    │    │
│  └────┬────┘ └────┬────┘ └────┬────┘ └──────┬──────┘    │
│       └───────────┼───────────┼─────────────┘           │
│                   ▼           ▼                          │
│           ┌───────────────────────┐                     │
│           │    统一消息路由引擎    │                      │
│           └───────────┬───────────┘                     │
└───────────────────────┼─────────────────────────────────┘
                        ▼
    ┌───────────────────────────────────────────┐
    │              接入渠道                       │
    │  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐  │
    │  │微信 │ │飞书 │ │钉钉 │ │网页 │ │电话 │  │
    │  └─────┘ └─────┘ └─────┘ └─────┘ └─────┘  │
    └───────────────────────────────────────────┘

核心功能

知识库构建

知识库结构

{
  "knowledge_base": {
    "categories": [
      {
        "id": "product",
        "name": "产品相关",
        "questions": 156
      },
      {
        "id": "order",
        "name": "订单问题",
        "questions": 89
      },
      {
        "id": "payment",
        "name": "支付问题",
        "questions": 67
      },
      {
        "id": "account",
        "name": "账户问题",
        "questions": 45
      },
      {
        "id": "technical",
        "name": "技术支持",
        "questions": 123
      }
    ],
    "total_questions": 480
  }
}

问答对管理

async def add_qa_pair(question, answer, metadata=None):
    """添加问答对"""
    # 生成问题变体
    variations = await generate_question_variations(question)

    # 生成向量嵌入
    embeddings = await generate_embeddings([question] + variations)

    # 存储到知识库
    await knowledge_db.insert({
        "question": question,
        "variations": variations,
        "answer": answer,
        "embeddings": embeddings,
        "metadata": metadata or {},
        "created_at": datetime.now()
    })

知识库检索

async def search_knowledge(query, top_k=3):
    """检索知识库"""
    # 语义检索
    query_embedding = await generate_embedding(query)
    semantic_results = await vector_search(query_embedding, top_k=top_k * 2)

    # 关键词检索
    keyword_results = await keyword_search(query, top_k=top_k * 2)

    # 混合排序
    merged = merge_results(semantic_results, keyword_results)

    return merged[:top_k]

意图识别

意图分类

{
  "intents": {
    "product_inquiry": {
      "name": "产品咨询",
      "examples": [
        "这个产品有什么功能",
        "产品价格是多少",
        "有优惠活动吗"
      ],
      "response": "product_info"
    },
    "order_status": {
      "name": "订单查询",
      "examples": [
        "我的订单到哪了",
        "订单什么时候发货",
        "物流信息"
      ],
      "response": "order_query"
    },
    "refund_request": {
      "name": "退款申请",
      "examples": [
        "我要退款",
        "申请退货",
        "商品有问题想退"
      ],
      "response": "refund_process"
    },
    "complaint": {
      "name": "投诉建议",
      "examples": [
        "我要投诉",
        "服务太差了",
        "不满意"
      ],
      "response": "complaint_handle",
      "priority": "high"
    }
  }
}

意图识别服务

async def detect_intent(message):
    """识别用户意图"""
    prompt = f"""
    分析以下用户消息的意图：

    消息：{message}

    可能的意图：
    - product_inquiry: 产品咨询
    - order_status: 订单查询
    - refund_request: 退款申请
    - complaint: 投诉建议
    - general: 一般咨询

    返回 JSON 格式：
    {{
        "intent": "意图类型",
        "confidence": 0.95,
        "entities": {{"key": "value"}}
    }}
    """

    result = await generate_response(prompt)
    return parse_json(result)

工单系统

工单创建

async def create_ticket(user_id, issue, priority="normal"):
    """创建工单"""
    ticket = {
        "id": generate_ticket_id(),
        "user_id": user_id,
        "issue": issue,
        "priority": priority,
        "status": "open",
        "created_at": datetime.now(),
        "updates": []
    }

    # 自动分配
    ticket["assignee"] = await auto_assign(ticket)

    # 保存工单
    await ticket_db.insert(ticket)

    # 通知相关人员
    await notify_assignee(ticket)

    return ticket

自动分配

async def auto_assign(ticket):
    """自动分配工单"""
    # 获取客服负载
    agents = await get_agent_workload()

    # 根据技能匹配
    matched_agents = filter_by_skill(agents, ticket.issue)

    # 选择负载最低的
    selected = min(matched_agents, key=lambda x: x.workload)

    return selected.id

工单流转

{
  "workflow": {
    "id": "ticket-flow",
    "name": "工单处理流程",
    "states": {
      "open": {
        "next": ["in_progress", "closed"],
        "actions": ["assign", "auto_reply"]
      },
      "in_progress": {
        "next": ["pending", "resolved", "escalated"],
        "actions": ["reply", "transfer"]
      },
      "pending": {
        "next": ["in_progress", "closed"],
        "actions": ["follow_up"]
      },
      "resolved": {
        "next": ["closed", "reopened"],
        "actions": ["survey"]
      },
      "escalated": {
        "next": ["in_progress", "resolved"],
        "actions": ["notify_manager"]
      },
      "closed": {
        "next": ["reopened"],
        "actions": ["archive"]
      }
    }
  }
}

多渠道接入

渠道配置

{
  "channels": {
    "wechat": {
      "enabled": true,
      "type": "official_account",
      "app_id": "${WECHAT_APP_ID}",
      "app_secret": "${WECHAT_APP_SECRET}",
      "welcome_message": "您好，欢迎咨询！请问有什么可以帮您？"
    },
    "feishu": {
      "enabled": true,
      "type": "bot",
      "app_id": "${FEISHU_APP_ID}",
      "app_secret": "${FEISHU_APP_SECRET}"
    },
    "webchat": {
      "enabled": true,
      "type": "widget",
      "theme": "light",
      "position": "bottom-right"
    }
  }
}

统一消息处理

async def handle_message(channel, user_id, message):
    """统一消息处理"""
    # 1. 获取用户上下文
    context = await get_user_context(user_id)

    # 2. 意图识别
    intent = await detect_intent(message)

    # 3. 检索知识库
    knowledge = await search_knowledge(message)

    # 4. 生成回复
    if intent.confidence > 0.8:
        response = await generate_response_from_intent(intent, knowledge)
    else:
        response = await generate_general_response(message, knowledge)

    # 5. 情感检测
    sentiment = await analyze_sentiment(message)
    if sentiment.negative:
        await alert_supervisor(user_id, message, sentiment)

    # 6. 记录对话
    await save_conversation(user_id, message, response)

    return response

情感分析

情感检测

async def analyze_sentiment(message):
    """情感分析"""
    prompt = f"""
    分析以下文本的情感：

    文本：{message}

    返回 JSON：
    {{
        "sentiment": "positive/negative/neutral",
        "score": 0.8,
        "emotions": ["frustrated", "angry"],
        "urgency": "high/medium/low"
    }}
    """

    return await generate_response(prompt)

负面情绪处理

{
  "workflow": {
    "id": "negative-sentiment-handle",
    "name": "负面情绪处理",
    "trigger": {
      "type": "event",
      "event": "sentiment.negative"
    },
    "steps": [
      {
        "id": "check-history",
        "type": "action",
        "action": "getUserHistory",
        "params": {
          "user_id": "{{event.user_id}}"
        }
      },
      {
        "id": "decide",
        "type": "condition",
        "condition": "{{check-history.complaint_count}} > 2",
        "true": {
          "action": "escalateToManager",
          "params": {
            "user_id": "{{event.user_id}}",
            "message": "{{event.message}}"
          }
        },
        "false": {
          "action": "sendApology",
          "params": {
            "user_id": "{{event.user_id}}"
          }
        }
      }
    ]
  }
}

数据分析

服务指标

async def calculate_metrics(period="day"):
    """计算服务指标"""
    metrics = {
        "total_conversations": await count_conversations(period),
        "avg_response_time": await avg_response_time(period),
        "resolution_rate": await resolution_rate(period),
        "customer_satisfaction": await satisfaction_score(period),
        "top_issues": await top_issues(period, limit=10),
        "channel_distribution": await channel_distribution(period)
    }

    return metrics

报表生成

{
  "report": {
    "title": "客服日报",
    "period": "2026-03-13",
    "metrics": {
      "total_conversations": 1234,
      "avg_response_time": "45秒",
      "resolution_rate": "92%",
      "satisfaction_score": 4.5
    },
    "highlights": [
      "今日咨询量较昨日增长 15%",
      "退款问题占比最高，达 23%",
      "微信渠道响应时间最短"
    ],
    "recommendations": [
      "建议增加退款流程说明文档",
      "考虑增加晚间客服人员"
    ]
  }
}

人机协作

转人工规则

{
  "transfer_rules": [
    {
      "condition": "intent.confidence < 0.5",
      "action": "transfer_to_human",
      "message": "这个问题比较复杂，我帮您转接人工客服..."
    },
    {
      "condition": "user.request_human == true",
      "action": "transfer_to_human",
      "message": "好的，正在为您转接人工客服..."
    },
    {
      "condition": "sentiment.negative && sentiment.score > 0.8",
      "action": "transfer_to_human",
      "message": "我理解您的心情，让我为您转接专属客服..."
    },
    {
      "condition": "issue_category == 'complaint'",
      "action": "transfer_to_human",
      "message": "您的反馈很重要，我将为您转接客服主管..."
    }
  ]
}

客服辅助

async def assist_agent(agent_id, conversation):
    """客服辅助"""
    # 实时建议
    suggestions = await generate_suggestions(conversation)

    # 知识库推荐
    relevant_knowledge = await search_knowledge(conversation.last_message)

    # 相似案例
    similar_cases = await find_similar_cases(conversation)

    return {
        "suggestions": suggestions,
        "knowledge": relevant_knowledge,
        "similar_cases": similar_cases
    }

最佳实践

知识库维护

1. 定期更新 - 每周审核问答对
2. 用户反馈 - 根据用户反馈优化答案
3. 版本管理 - 保留历史版本便于回溯

服务质量

1. 响应时间 - 设置响应时间目标
2. 解决率 - 追踪首次解决率
3. 满意度 - 定期收集用户评价

团队协作

1. 技能培训 - 定期培训客服团队
2. 知识共享 - 建立内部知识库
3. 绩效激励 - 设置服务质量奖励

常见问题

机器人回答不准确

优化知识库和意图识别：

{
  "optimization": {
    "add_variations": true,
    "update_embeddings": true,
    "review_low_confidence": true
    }
}

用户不愿意用机器人

改善用户体验：

{
  "ux_improvements": {
    "quick_replies": true,
    "personality": "friendly",
    "transfer_easy": true
  }
}

高峰期响应慢

扩容和优化：

{
  "scaling": {
    "auto_scale": true,
    "cache_responses": true,
    "async_processing": true
  }
}

下一步

• 飞书机器人开发
• WhatsApp 自动化
• 企业安全