AI Agent客服系统实战：从0到1搭建智能客服

一、开场：客服困境与AI破局

大家好，我是老金。

去年我们公司的客服团队是这样的：

• 10个客服，每天处理2000+咨询
• 重复问题占70%（查订单、问物流、问退换货）
• 客服流动率高，培训成本大
• 高峰期用户等待超过30分钟

老板说：”能不能用AI解决？”

我们花了2个月，从0到1搭建了智能客服系统。现在：

• AI处理80%的重复咨询
• 客服专注于复杂问题
• 用户等待时间降到2分钟
• 客服满意度从75%提升到92%

今天分享完整的搭建经验。

二、系统架构设计

整体架构

┌─────────────────────────────────────────────────────────┐
│                  智能客服系统架构                        │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  用户入口层                                             │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                │
│  │ 网页客服 │  │ 微信公众号│  │ APP内嵌 │                │
│  └────┬────┘  └────┬────┘  └────┬────┘                │
│       │            │            │                       │
│       └────────────┼────────────┘                       │
│                    ↓                                    │
│  ┌─────────────────────────────────────────────────┐   │
│  │              对话管理层                          │   │
│  │  • 意图识别                                      │   │
│  │  • 槽位填充                                      │   │
│  │  • 对话状态管理                                  │   │
│  │  • 多轮对话                                      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              核心AI层                            │   │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐      │   │
│  │  │ LLM大脑  │  │ 知识库   │  │ 工具集   │      │   │
│  │  │(GPT-4)   │  │ (RAG)    │  │(API调用) │      │   │
│  │  └──────────┘  └──────────┘  └──────────┘      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              业务系统层                          │   │
│  │  订单系统 │ 物流系统 │ 用户系统 │ 通知系统      │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
└─────────────────────────────────────────────────────────┘

核心模块

# 系统核心模块
class CustomerServiceAgent:
    """智能客服Agent"""

    def __init__(self, config: dict):
        # LLM核心
        self.llm = get_llm(config["model"])

        # 知识库
        self.knowledge_base = RAGSystem(
            embedder=config["embedder"],
            vector_store=config["vector_store"]
        )

        # 工具集
        self.tools = self._init_tools(config["tools"])

        # 对话管理
        self.dialog_manager = DialogManager()

        # 意图识别
        self.intent_classifier = IntentClassifier()

    def _init_tools(self, tool_configs):
        """初始化工具集"""
        tools = {}
        for config in tool_configs:
            if config["type"] == "order":
                tools["query_order"] = QueryOrderTool(config)
                tools["cancel_order"] = CancelOrderTool(config)
            elif config["type"] == "product":
                tools["search_product"] = SearchProductTool(config)
            elif config["type"] == "logistics":
                tools["query_logistics"] = QueryLogisticsTool(config)
        return tools

三、意图识别模块

意图分类设计

class IntentClassifier:
    """意图识别器"""

    INTENTS = {
        "order_query": {
            "name": "订单查询",
            "keywords": ["订单", "查订单", "订单号", "我的订单"],
            "examples": [
                "帮我查一下订单ORD12345",
                "我的订单到哪了",
                "查订单状态"
            ],
            "slots": ["order_id"],
            "priority": 1
        },
        "order_cancel": {
            "name": "订单取消",
            "keywords": ["取消订单", "退订单", "不要了"],
            "examples": [
                "我要取消订单",
                "帮我退掉这个订单"
            ],
            "slots": ["order_id", "reason"],
            "priority": 2
        },
        "product_query": {
            "name": "商品咨询",
            "keywords": ["商品", "产品", "多少钱", "有货吗"],
            "examples": [
                "这款手机多少钱",
                "有红色吗",
                "这个产品怎么样"
            ],
            "slots": ["product_name", "attribute"],
            "priority": 1
        },
        "logistics_query": {
            "name": "物流查询",
            "keywords": ["快递", "物流", "发货", "到哪了"],
            "examples": [
                "我的快递到哪了",
                "什么时候发货",
                "物流信息"
            ],
            "slots": ["order_id"],
            "priority": 1
        },
        "return_request": {
            "name": "退换货",
            "keywords": ["退货", "换货", "退款", "质量"],
            "examples": [
                "我要退货",
                "这个质量有问题想换货"
            ],
            "slots": ["order_id", "reason", "type"],
            "priority": 2
        },
        "general_qa": {
            "name": "通用问答",
            "keywords": [],
            "examples": [],
            "slots": [],
            "priority": 10
        }
    }

    def __init__(self):
        self.keyword_matcher = KeywordMatcher()
        self.semantic_classifier = SemanticClassifier()

    async def classify(self, user_input: str, context: dict = None) -> dict:
        """分类意图"""
        # 1. 关键词快速匹配
        keyword_scores = self.keyword_matcher.match(
            user_input, 
            self.INTENTS
        )

        # 2. 语义分类
        semantic_scores = await self.semantic_classifier.classify(
            user_input,
            list(self.INTENTS.keys())
        )

        # 3. 融合得分
        final_scores = self._merge_scores(
            keyword_scores,
            semantic_scores
        )

        # 4. 选择最佳意图
        best_intent = max(final_scores.items(), key=lambda x: x[1])

        return {
            "intent": best_intent[0],
            "confidence": best_intent[1],
            "all_scores": final_scores
        }

    def _merge_scores(self, keyword_scores, semantic_scores, 
                      alpha=0.3):
        """融合关键词和语义得分"""
        merged = {}
        for intent in keyword_scores:
            merged[intent] = (
                alpha * keyword_scores[intent] + 
                (1 - alpha) * semantic_scores.get(intent, 0)
            )
        return merged

槽位填充

class SlotFiller:
    """槽位填充器"""

    SLOT_PATTERNS = {
        "order_id": {
            "pattern": r"ORDd{10}",
            "description": "订单号，格式ORD+10位数字"
        },
        "phone": {
            "pattern": r"1[3-9]d{9}",
            "description": "手机号"
        },
        "product_name": {
            "extract": "llm",  # 使用LLM提取
            "description": "商品名称"
        }
    }

    async def fill_slots(self, user_input: str, 
                        intent: str, 
                        required_slots: list,
                        context: dict = None) -> dict:
        """填充槽位"""
        filled = {}
        missing = []

        for slot in required_slots:
            if slot in self.SLOT_PATTERNS:
                slot_config = self.SLOT_PATTERNS[slot]

                if slot_config.get("extract") == "llm":
                    # 使用LLM提取
                    value = await self._llm_extract(
                        user_input, slot, slot_config["description"]
                    )
                else:
                    # 正则匹配
                    match = re.search(
                        slot_config["pattern"], 
                        user_input
                    )
                    value = match.group(0) if match else None

                if value:
                    filled[slot] = value
                else:
                    missing.append(slot)

        return {
            "filled": filled,
            "missing": missing,
            "need_ask": len(missing) > 0
        }

    async def _llm_extract(self, text: str, slot_name: str, 
                          description: str) -> str:
        """使用LLM提取槽位值"""
        prompt = f"""
        从以下文本中提取"{slot_name}"（{description}）：

        文本：{text}

        如果找不到，返回null。
        只返回提取的值，不要其他内容。
        """

        result = await llm.generate(prompt)
        return result.strip() if result.lower() != "null" else None

四、对话状态管理

对话上下文

class DialogContext:
    """对话上下文"""

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.history = []
        self.current_intent = None
        self.slots = {}
        self.state = "idle"  # idle, collecting, executing, confirming
        self.user_info = {}
        self.created_at = datetime.now()

    def add_message(self, role: str, content: str):
        """添加消息"""
        self.history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

    def get_recent_history(self, n: int = 5):
        """获取最近n轮对话"""
        return self.history[-n*2:]  # 每轮包含user和assistant


class DialogManager:
    """对话管理器"""

    def __init__(self):
        self.sessions = {}  # session_id -> DialogContext

    def get_or_create_session(self, session_id: str) -> DialogContext:
        """获取或创建会话"""
        if session_id not in self.sessions:
            self.sessions[session_id] = DialogContext(session_id)
        return self.sessions[session_id]

    async def process(self, session_id: str, user_input: str) -> dict:
        """处理用户输入"""
        ctx = self.get_or_create_session(session_id)

        # 添加用户消息
        ctx.add_message("user", user_input)

        # 状态机处理
        if ctx.state == "idle":
            return await self._handle_idle(ctx, user_input)
        elif ctx.state == "collecting":
            return await self._handle_collecting(ctx, user_input)
        elif ctx.state == "confirming":
            return await self._handle_confirming(ctx, user_input)
        else:
            return await self._handle_idle(ctx, user_input)

    async def _handle_idle(self, ctx: DialogContext, user_input: str):
        """处理空闲状态"""
        # 识别意图
        intent_result = await intent_classifier.classify(
            user_input, 
            {"history": ctx.history}
        )

        ctx.current_intent = intent_result["intent"]
        intent_config = IntentClassifier.INTENTS[ctx.current_intent]

        # 填充槽位
        slot_result = await slot_filler.fill_slots(
            user_input,
            ctx.current_intent,
            intent_config["slots"],
            {"history": ctx.history}
        )

        ctx.slots.update(slot_result["filled"])

        if slot_result["need_ask"]:
            # 需要收集更多信息
            ctx.state = "collecting"
            question = self._generate_question(
                ctx.current_intent,
                slot_result["missing"][0]
            )
            return {"type": "ask", "content": question}
        else:
            # 信息完整，执行操作
            return await self._execute_intent(ctx)

    async def _handle_collecting(self, ctx: DialogContext, user_input: str):
        """处理信息收集状态"""
        intent_config = IntentClassifier.INTENTS[ctx.current_intent]
        missing_slots = [
            s for s in intent_config["slots"] 
            if s not in ctx.slots
        ]

        # 尝试填充缺失槽位
        slot_result = await slot_filler.fill_slots(
            user_input,
            ctx.current_intent,
            missing_slots,
            {"history": ctx.history}
        )

        ctx.slots.update(slot_result["filled"])

        still_missing = [
            s for s in intent_config["slots"] 
            if s not in ctx.slots
        ]

        if still_missing:
            question = self._generate_question(
                ctx.current_intent,
                still_missing[0]
            )
            return {"type": "ask", "content": question}
        else:
            return await self._execute_intent(ctx)

    def _generate_question(self, intent: str, missing_slot: str) -> str:
        """生成追问"""
        questions = {
            "order_id": "请问您的订单号是多少？",
            "phone": "请问您的手机号是多少？",
            "reason": "请问取消/退货的原因是什么？",
            "product_name": "请问您想咨询哪个商品？"
        }
        return questions.get(missing_slot, f"请问{missing_slot}是多少？")

五、工具集成

订单查询工具

class QueryOrderTool:
    """订单查询工具"""

    def __init__(self, config: dict):
        self.api_base = config["api_base"]
        self.api_key = config["api_key"]

    async def execute(self, order_id: str, phone: str = None) -> dict:
        """执行查询"""
        # 调用订单API
        async with aiohttp.ClientSession() as session:
            params = {"order_id": order_id}
            if phone:
                params["phone"] = phone

            async with session.get(
                f"{self.api_base}/orders/{order_id}",
                headers={"Authorization": f"Bearer {self.api_key}"},
                params=params
            ) as response:
                if response.status == 200:
                    order = await response.json()
                    return self._format_result(order)
                elif response.status == 404:
                    return {"success": False, "message": "订单不存在"}
                else:
                    return {"success": False, "message": "查询失败，请稍后再试"}

    def _format_result(self, order: dict) -> dict:
        """格式化查询结果"""
        status_map = {
            "pending": "待付款",
            "paid": "已付款",
            "shipped": "已发货",
            "delivered": "已签收",
            "cancelled": "已取消"
        }

        return {
            "success": True,
            "order_id": order["order_id"],
            "status": status_map.get(order["status"], order["status"]),
            "items": [
                {"name": item["name"], "quantity": item["quantity"]}
                for item in order["items"]
            ],
            "total": order["total"],
            "tracking_number": order.get("tracking_number"),
            "tracking_url": order.get("tracking_url"),
            "message": f"订单{order['order_id']}状态：{status_map.get(order['status'])}"
        }

工具调用管理

class ToolExecutor:
    """工具执行器"""

    def __init__(self, tools: dict):
        self.tools = tools
        self.execution_log = []

    async def execute(self, tool_name: str, params: dict) -> dict:
        """执行工具"""
        if tool_name not in self.tools:
            return {
                "success": False,
                "error": f"Unknown tool: {tool_name}"
            }

        tool = self.tools[tool_name]

        start_time = time.time()
        try:
            result = await tool.execute(**params)
            success = result.get("success", True)
        except Exception as e:
            result = {"success": False, "error": str(e)}
            success = False

        latency = time.time() - start_time

        # 记录执行日志
        self.execution_log.append({
            "tool": tool_name,
            "params": params,
            "result": result,
            "success": success,
            "latency": latency,
            "timestamp": datetime.now().isoformat()
        })

        return result

六、知识库建设

知识库内容

# 客服知识库内容结构
KNOWLEDGE_BASE_STRUCTURE = {
    "products": {
        "description": "产品信息",
        "items": [
            {
                "id": "prod_001",
                "name": "智能手表Pro",
                "category": "智能穿戴",
                "price": 1999,
                "features": ["心率监测", "睡眠追踪", "NFC支付"],
                "faq": [
                    "续航多久？续航约7天",
                    "防水吗？支持5ATM防水"
                ]
            }
        ]
    },
    "policies": {
        "description": "政策规则",
        "items": [
            {
                "id": "policy_001",
                "title": "退换货政策",
                "content": "签收后7天内支持无理由退货，30天内质量问题可换货...",
                "keywords": ["退货", "换货", "退款"]
            }
        ]
    },
    "procedures": {
        "description": "操作流程",
        "items": [
            {
                "id": "proc_001",
                "title": "如何申请退货",
                "steps": [
                    "登录账户，进入我的订单",
                    "找到需要退货的订单，点击申请退货",
                    "选择退货原因，提交申请",
                    "等待审核（1-2个工作日）",
                    "审核通过后，寄回商品"
                ]
            }
        ]
    }
}

知识库检索

class KnowledgeRetriever:
    """知识库检索器"""

    def __init__(self, vector_store, embedder):
        self.vector_store = vector_store
        self.embedder = embedder

    async def retrieve(self, query: str, top_k: int = 3) -> list:
        """检索相关知识"""
        # 向量化查询
        query_vector = await self.embedder.embed(query)

        # 检索
        results = await self.vector_store.search(
            query_vector,
            top_k=top_k,
            filter={"category": "customer_service"}
        )

        return results

    async def augment_prompt(self, query: str, prompt: str) -> str:
        """使用知识增强提示词"""
        # 检索相关知识
        relevant_docs = await self.retrieve(query)

        if not relevant_docs:
            return prompt

        # 构建上下文
        context = "nn".join([
            f"【{doc['title']}】n{doc['content']}"
            for doc in relevant_docs
        ])

        return f"""
参考以下知识库内容回答用户问题：

{context}

---

{prompt}
        """

七、人机协作机制

转人工判断

class HumanTransferDecider:
    """转人工判断器"""

    TRANSFER_CONDITIONS = {
        "sentiment_negative": {
            "trigger": "用户情绪负面",
            "check": lambda ctx: ctx.sentiment_score  5 and not ctx.resolved
        },
        "user_request": {
            "trigger": "用户主动要求",
            "check": lambda ctx: "人工" in ctx.last_user_input
        },
        "policy_violation": {
            "trigger": "涉及政策敏感问题",
            "check": lambda ctx: any(kw in ctx.last_user_input 
                                    for kw in ["投诉", "举报", "法律"])
        }
    }

    def should_transfer(self, context: DialogContext) -> tuple:
        """判断是否需要转人工"""
        for condition_name, condition in self.TRANSFER_CONDITIONS.items():
            if condition["check"](context):
                return True, condition["trigger"]

        return False, None


class HumanAgentQueue:
    """人工客服队列"""

    def __init__(self):
        self.queue = asyncio.Queue()
        self.agents = {}  # agent_id -> status

    async def add_to_queue(self, session_id: str, context: DialogContext,
                          reason: str):
        """加入人工队列"""
        await self.queue.put({
            "session_id": session_id,
            "context": context,
            "reason": reason,
            "timestamp": datetime.now()
        })

        # 通知用户
        return {
            "type": "transfer",
            "message": f"正在为您转接人工客服，预计等待{self._estimate_wait()}分钟..."
        }

    def _estimate_wait(self) -> int:
        """预估等待时间"""
        queue_length = self.queue.qsize()
        available_agents = sum(
            1 for status in self.agents.values() 
            if status == "available"
        )

        if available_agents == 0:
            return queue_length * 5  # 假设每个5分钟

        return max(1, queue_length // available_agents * 3)

八、效果评估

核心指标

class CustomerServiceMetrics:
    """客服系统指标"""

    def __init__(self):
        self.metrics = {
            "total_conversations": 0,
            "ai_resolved": 0,
            "human_resolved": 0,
            "avg_response_time": [],
            "user_satisfaction": [],
            "intent_accuracy": []
        }

    def record_conversation(self, result: dict):
        """记录对话结果"""
        self.metrics["total_conversations"] += 1

        if result["resolved_by"] == "ai":
            self.metrics["ai_resolved"] += 1
        else:
            self.metrics["human_resolved"] += 1

        self.metrics["avg_response_time"].append(result["response_time"])

        if "satisfaction" in result:
            self.metrics["user_satisfaction"].append(result["satisfaction"])

    def get_report(self) -> dict:
        """生成报告"""
        total = self.metrics["total_conversations"]

        return {
            "total_conversations": total,
            "ai_resolution_rate": self.metrics["ai_resolved"] / max(1, total),
            "human_transfer_rate": self.metrics["human_resolved"] / max(1, total),
            "avg_response_time": sum(self.metrics["avg_response_time"]) / max(1, len(self.metrics["avg_response_time"])),
            "avg_satisfaction": sum(self.metrics["user_satisfaction"]) / max(1, len(self.metrics["user_satisfaction"]))
        }

九、部署与优化

部署架构

# kubernetes部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: customer-service-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: customer-service-agent
  template:
    spec:
      containers:
      - name: agent
        image: customer-service-agent:v1.0
        ports:
        - containerPort: 8080
        env:
        - name: LLM_API_KEY
          valueFrom:
            secretKeyRef:
              name: llm-secrets
              key: api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 30

持续优化

十、总结

实施清单

关键成功因素

1. 知识库质量：FAQ要全、答案要准
2. 意图识别：覆盖主要场景，边界清晰
3. 人机协作：知道什么时候该转人工
4. 持续优化：根据数据反馈不断改进

下期预告

明天聊聊AI Agent代码助手——让AI帮你写代码、改Bug！

往期回顾

• AI Agent合规指南：企业级部署的法律红线
• AI Agent灾难恢复：当AI”挂了”怎么办？
• AI Agent版本管理：从Prompt版本到模型版本