用 Hermes 搭建智能客服机器人的完整方案

客服场景为什么适合 AI

如果你问我 AI 在企业落地最成熟的场景是什么，我会毫不犹豫地回答：客服。

原因很简单——客服工作的特征天然适合 AI 处理：

重复率高。大部分客户问的问题高度集中在少数几个类别里。退换货怎么操作、物流到哪了、密码怎么找回、XXX 功能怎么用。你让人类客服每天回答一百遍同样的问题，效率低不说，心态也会崩。

有标准答案。绝大部分客服问题都有明确的标准回答，不需要创造性思维。这正好是语言模型擅长的——从知识库里找到对应的内容，组织成自然的语言回复。

7x24 需求。用户不会只在工作时间遇到问题。凌晨两点发现账户被锁，等到第二天上午九点才能联系客服，这体验太差了。

成本敏感。招一个客服专员一个月至少几千块，智能客服的边际成本几乎为零。

但市面上的智能客服方案要么太贵（SaaS 按量收费，用多了很肉疼），要么太傻（基于关键词匹配的老方案，用户稍微换个说法就答不上来）。用 Hermes 自建一套，能在质量和成本之间找到很好的平衡点。

整体架构设计

一个完整的智能客服系统，远不止一个聊天机器人那么简单。你需要考虑知识库管理、对话流程控制、人机协作、数据分析等多个模块。

           用户入口
          (网页/APP/微信)
              │
              ▼
      意图识别 & 路由层
     /        |        \
    ▼         ▼         ▼
FAQ 问答   业务查询   复杂投诉
(知识库)   (API调用)  (转人工)
    \         |         /
     ▼        ▼        ▼
      对话管理 & 上下文
              │
              ▼
        Hermes 推理层
              │
              ▼
      回复生成 & 质量检查
              │
              ▼
          返回用户

每个模块的职责：

意图识别：判断用户想干什么。是问常见问题、查询订单状态，还是要投诉？不同的意图走不同的处理流程。

知识库检索：维护一套产品知识库，用 RAG 技术从中检索相关内容。

业务 API 调用：查订单、查物流、查余额这类需要访问后端系统的操作。

人工转接：识别到模型处理不了的复杂问题时，平滑转接给人工客服。

对话管理：维护对话上下文，处理多轮对话中的指代消解、话题切换等。

第一步：构建知识库

知识库是智能客服的根基。你的机器人能回答什么问题、回答的质量如何，90% 取决于知识库的质量。

知识库内容来源

通常有这几个来源：

1. 产品文档：使用手册、功能说明、版本更新说明
2. FAQ 集合：历史高频问题和标准回答
3. 客服工单：过去人工客服处理过的工单记录
4. 政策文件：退换货政策、隐私协议、服务条款

知识库处理流程

import chromadb
from sentence_transformers import SentenceTransformer
import json
import hashlib

class KnowledgeBase:
    def __init__(self, persist_dir="./kb_data"):
        self.client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.client.get_or_create_collection(
            name="customer_service",
            metadata={"hnsw:space": "cosine"}
        )
        self.embedder = SentenceTransformer(
            "shibing624/text2vec-base-chinese"
        )

    def add_faq(self, question, answer, category="general"):
        """添加 FAQ 条目"""
        doc_id = hashlib.md5(question.encode()).hexdigest()
        combined = f"问题：{question}\n回答：{answer}"
        embedding = self.embedder.encode([combined]).tolist()

        self.collection.upsert(
            ids=[doc_id],
            documents=[combined],
            embeddings=embedding,
            metadatas=[{
                "question": question,
                "answer": answer,
                "category": category,
                "type": "faq"
            }]
        )

    def add_document(self, title, content, category="docs"):
        """添加文档（自动分段）"""
        chunks = self._split_document(content)
        for i, chunk in enumerate(chunks):
            doc_id = hashlib.md5(
                f"{title}_{i}".encode()
            ).hexdigest()
            embedding = self.embedder.encode([chunk]).tolist()

            self.collection.upsert(
                ids=[doc_id],
                documents=[chunk],
                embeddings=embedding,
                metadatas=[{
                    "title": title,
                    "chunk_index": i,
                    "category": category,
                    "type": "document"
                }]
            )

    def search(self, query, top_k=5):
        """检索相关知识"""
        query_embedding = self.embedder.encode([query]).tolist()
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=top_k,
            include=["documents", "metadatas", "distances"]
        )
        return [
            {
                "content": doc,
                "metadata": meta,
                "score": 1 - dist  # 转换为相似度分数
            }
            for doc, meta, dist in zip(
                results["documents"][0],
                results["metadatas"][0],
                results["distances"][0]
            )
        ]

    def _split_document(self, text, chunk_size=400,
                        overlap=50):
        """文档分段"""
        sentences = text.replace("。", "。\n").split("\n")
        chunks = []
        current_chunk = ""

        for sent in sentences:
            sent = sent.strip()
            if not sent:
                continue
            if len(current_chunk) + len(sent) > chunk_size:
                if current_chunk:
                    chunks.append(current_chunk)
                current_chunk = sent
            else:
                current_chunk += sent

        if current_chunk:
            chunks.append(current_chunk)

        return chunks

知识库数据准备示例

kb = KnowledgeBase()

# 添加 FAQ
faqs = [
    {
        "q": "如何修改收货地址",
        "a": "登录账号后，进入「个人中心」→「地址管理」，点击需要修改的地址右侧的「编辑」按钮即可修改。如果订单已发货，需要联系快递公司修改。",
        "cat": "订单"
    },
    {
        "q": "退款多久到账",
        "a": "审核通过后，退款将在 1-3 个工作日内原路退回。信用卡可能需要 3-7 个工作日。如果超过 7 个工作日未收到退款，请联系客服。",
        "cat": "退换货"
    },
    {
        "q": "密码忘记了怎么办",
        "a": "在登录页面点击「忘记密码」，输入注册时使用的手机号或邮箱，按照提示完成身份验证后即可重置密码。",
        "cat": "账户"
    }
]

for faq in faqs:
    kb.add_faq(faq["q"], faq["a"], faq["cat"])

第二步：对话流程设计

好的客服不是简单地一问一答，而是要能引导对话、收集信息、最终解决问题。

意图识别

用 Hermes 来做意图分类，比传统的关键词匹配准确得多：

async def classify_intent(self, user_input):
    """识别用户意图"""
    prompt = f"""请判断以下用户输入的意图类别。

用户输入：「{user_input}」

可选类别：
- faq: 常见问题咨询
- order_query: 查询订单/物流状态
- complaint: 投诉或不满
- refund: 退款/退换货
- account: 账户相关问题
- human: 要求转人工
- other: 无法归类

只返回类别名称，不要任何其他内容。"""

    result = await self._call_model([
        {"role": "user", "content": prompt}
    ], temperature=0.1)

    return result.strip().lower()

多轮对话管理

客服场景经常需要多轮对话来收集信息。比如用户说「我要退货」，你需要知道订单号、退货原因、是否已拆封等信息才能处理。

class ConversationManager:
    def __init__(self):
        self.sessions = {}

    def get_session(self, session_id):
        if session_id not in self.sessions:
            self.sessions[session_id] = {
                "history": [],
                "intent": None,
                "collected_info": {},
                "state": "greeting"
            }
        return self.sessions[session_id]

    async def process(self, session_id, user_input, agent):
        session = self.get_session(session_id)

        # 检测是否要求转人工
        if any(kw in user_input for kw in
               ["转人工", "人工客服", "真人"]):
            return self._transfer_to_human(session_id)

        # 第一次交互，识别意图
        if session["state"] == "greeting":
            intent = await agent.classify_intent(user_input)
            session["intent"] = intent
            session["state"] = "collecting"

        # 根据意图选择处理流程
        handler = self._get_handler(session["intent"])
        return await handler(session, user_input, agent)

    def _get_handler(self, intent):
        handlers = {
            "faq": self._handle_faq,
            "order_query": self._handle_order_query,
            "refund": self._handle_refund,
            "complaint": self._handle_complaint,
        }
        return handlers.get(intent, self._handle_faq)

    async def _handle_refund(self, session, user_input, agent):
        """退款流程：需要收集订单号和原因"""
        info = session["collected_info"]

        if "order_id" not in info:
            # 尝试从用户输入中提取订单号
            order_id = self._extract_order_id(user_input)
            if order_id:
                info["order_id"] = order_id
            else:
                return "请提供您的订单号，通常在订单详情页面可以找到。"

        if "reason" not in info:
            if session["state"] == "collecting_reason":
                info["reason"] = user_input
            else:
                session["state"] = "collecting_reason"
                return "请问您申请退款的原因是什么？"

        # 信息收集完毕，生成处理方案
        return await self._generate_refund_response(
            info, agent
        )

    def _extract_order_id(self, text):
        """从文本中提取订单号"""
        import re
        patterns = [
            r'[A-Z]{2}\d{10,}',
            r'\d{15,20}',
            r'订单号[：:]\s*(\S+)',
        ]
        for pattern in patterns:
            match = re.search(pattern, text)
            if match:
                return match.group(0)
        return None

    def _transfer_to_human(self, session_id):
        """转接人工客服"""
        return {
            "type": "transfer",
            "message": "正在为您转接人工客服，请稍候...",
            "session_data": self.sessions.get(session_id, {})
        }

第三步：回复生成与质量控制

客服场景对回复质量有特殊要求：不能出错、不能答非所问、不能承诺做不到的事情。

回复生成

async def generate_response(self, user_input, context,
                             session_history):
    """生成客服回复"""
    system_prompt = """你是一个专业的客服助手。

回答规则：
1. 基于提供的知识库内容回答，不要编造信息
2. 如果知识库中没有相关信息，诚实告知并建议转人工
3. 语气友好但专业，不要过度热情
4. 涉及退款金额、处理时间等，以知识库中的说明为准
5. 不要承诺你无法确认的事情
6. 回答要简洁，一般不超过 150 字
7. 如果用户情绪激动，先表示理解再给出方案"""

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

    # 加入对话历史
    for msg in session_history[-10:]:
        messages.append(msg)

    # 构建带知识库上下文的消息
    augmented = f"""知识库参考内容：
{context}

用户最新消息：{user_input}

请基于知识库内容回答用户问题。如果知识库中没有相关信息，请告知用户并建议其他解决方式。"""

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

    return await self._call_model(
        messages, temperature=0.3
    )

回复质量检查

在回复发出去之前，做一轮自动检查：

async def quality_check(self, reply, user_input, context):
    """检查回复质量"""
    check_prompt = f"""请检查以下客服回复的质量。

用户问题：{user_input}
知识库参考：{context}
客服回复：{reply}

检查项目：
1. 回复是否准确回答了用户的问题？
2. 回复中是否有知识库中未提及的信息？（可能是编造的）
3. 是否有不当承诺？（比如承诺具体退款时间）
4. 语气是否合适？

如果全部通过，返回 PASS。
如果有问题，返回 FAIL 并说明原因。"""

    result = await self._call_model(
        [{"role": "user", "content": check_prompt}],
        temperature=0.1
    )

    return result.strip().startswith("PASS")

第四步：接入渠道对接

智能客服需要对接多个入口：网页嵌入、微信公众号、APP 内嵌等。

网页嵌入（WebSocket 方案）

from fastapi import WebSocket
import json

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    session_id = str(id(websocket))

    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            response = await conversation_manager.process(
                session_id, message["content"], agent
            )

            if isinstance(response, dict) and \
               response.get("type") == "transfer":
                await websocket.send_json({
                    "type": "transfer",
                    "message": response["message"]
                })
            else:
                await websocket.send_json({
                    "type": "message",
                    "content": response
                })
    except Exception:
        pass
    finally:
        conversation_manager.cleanup(session_id)

前端嵌入组件

一个简单的聊天窗口组件：

<div id="chat-widget"
     style="position:fixed;bottom:20px;right:20px;">
  <div id="chat-window" style="display:none;">
    <div id="chat-messages"></div>
    <input id="chat-input" placeholder="请输入您的问题..."
           onkeypress="if(event.key==='Enter')sendMessage()">
  </div>
  <button onclick="toggleChat()">在线客服</button>
</div>

<script>
let ws = null;

function toggleChat() {
  const win = document.getElementById('chat-window');
  win.style.display = win.style.display === 'none'
    ? 'block' : 'none';
  if (!ws) connectWebSocket();
}

function connectWebSocket() {
  ws = new WebSocket('wss://your-domain.com/ws/chat');
  ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    appendMessage('assistant', data.content);
  };
}

function sendMessage() {
  const input = document.getElementById('chat-input');
  const text = input.value.trim();
  if (!text) return;
  appendMessage('user', text);
  ws.send(JSON.stringify({content: text}));
  input.value = '';
}

function appendMessage(role, content) {
  const div = document.createElement('div');
  div.className = 'msg-' + role;
  div.textContent = content;
  document.getElementById('chat-messages').appendChild(div);
}
</script>

第五步：效果监控与持续优化

上线之后不是结束，而是开始。你需要持续监控效果并迭代优化。

关键指标

需要追踪的核心指标：

• 解决率：用户问题被成功解决的比例（不转人工的比例）
• 满意度：对话结束后的用户评分
• 平均对话轮次：越少说明效率越高
• 转人工率：越低说明 AI 处理能力越强
• 响应时间：从用户发消息到收到回复的时间

未解决问题自动收集

async def log_unresolved(session_id, conversation):
    """记录未解决的问题，用于后续知识库补充"""
    summary_prompt = f"""以下是一段客服对话，
用户的问题没有得到满意解决。

对话记录：
{conversation}

请提取：
1. 用户的核心问题是什么
2. 为什么 AI 无法解决（知识库缺失/逻辑复杂/需要操作权限）
3. 建议添加什么样的知识库条目来覆盖这类问题

格式为 JSON。"""

    result = await call_hermes(summary_prompt)

    with open("unresolved_log.jsonl", "a") as f:
        f.write(json.dumps({
            "session_id": session_id,
            "analysis": result,
            "timestamp": datetime.now().isoformat()
        }, ensure_ascii=False) + "\n")

定期回顾这些未解决问题，补充知识库，是提升解决率最直接的手段。

实际效果参考

用 Hermes 搭建的客服系统实际跑下来，一些参考数据：

• FAQ 类问题解决率：85%+
• 订单查询类（接了业务 API 的）：90%+
• 投诉类（情感较复杂的）：40%，大部分需要转人工
• 平均响应时间：1-2 秒（使用 8B 量化模型本地推理）
• 日均处理对话：500-2000 轮

投诉类场景的低解决率是正常的。用户情绪激动的时候，往往不是知识库能解决的问题，需要人的同理心和灵活处理。AI 在这个环节的定位应该是「安抚情绪 + 收集信息 + 快速转接」，而不是试图独自解决。

关于 Hermes 模型的指令遵循能力和角色扮演表现，可以参考 Hermes 3 技术报告解读 里的相关分析。如果你对 Agent 架构的更多细节感兴趣，推荐看看 Hermes Agent 完全指南。

常见问题和踩坑经验

知识库更新后要重建索引。修改了 FAQ 内容之后，记得重新生成 embedding 并更新向量数据库。光改文本不重建索引，检索到的还是旧内容。

System Prompt 越具体越好。「你是客服」远不如「你是 XXX 公司的客服，负责处理 XXX 产品的售后问题，回答不超过 150 字」。约束条件要写清楚，模型才能严格遵守。

不要让模型查数据库。涉及到查订单、查余额这类操作，用 Function Calling 调后端 API，别让模型直接生成 SQL。安全问题不是闹着玩的。

做好兜底策略。模型无法回答的时候，与其给一个模棱两可的回复，不如直接告诉用户「这个问题我需要帮您转接人工客服来处理」。用户宁可多等一会也不想听废话。

有任何实际落地中的问题，欢迎到 cocoloop 社区 来交流，那里有不少做过企业客服系统的开发者可以给你提供经验。