---

前言

“我的订单怎么还没到？”——当用户在深夜发出这条消息时，人工客服早已下班。传统IVR语音菜单的"按1查订单、按2找人工"让用户烦躁，而简单的FAQ机器人又答非所问。

智能客服Agent的出现正在改变这一切。它不再是预设问答的"复读机"，而是一个能理解意图、调用工具、记忆上下文的智能体。当用户说"买了三天的毛衣好像小了"，Agent能自动识别退换货意图，查询订单，创建售后工单，最后告知用户"已为您申请换货，快递小哥明天上门"。

本文将基于电商客服场景，从架构设计到代码实现，完整讲解如何构建一个生产级的智能客服Agent。

---

一、系统架构总览

1.1 核心需求

一个电商客服Agent需要处理三类核心场景：

场景类型	用户提问示例	处理方式
咨询类	“你们发货用哪个快递？”	RAG检索知识库
订单查询	“帮我看看订单12345到哪了”	调用订单API
退换货	“这件衣服想换大一号”	创建售后工单

1.2 三层架构设计

1.3 核心流程

二、意图识别与实体抽取

2.1 基于Few-shot的意图分类

意图识别是Agent的大脑。我们采用LLM + Few-shot示例的方式，既能保证灵活性，又能通过示例约束输出格式。

import openai
import json

class IntentClassifier:
    def __init__(self, api_key):
        self.client = openai.OpenAI(api_key=api_key)
        self.system_prompt = """
你是一个电商客服意图识别助手。请分析用户输入，识别意图并提取关键实体。

意图类型：
1. query_order - 查询订单状态
2. return_exchange - 退换货申请  
3. knowledge_query - 咨询常见问题
4. complaint - 投诉
5. talk_to_human - 要求转人工
6. chitchat - 闲聊

输出格式必须是JSON：
{
    "intent": "意图类型",
    "entities": {
        "order_id": "提取的订单号，没有则null",
        "product": "提取的商品名，没有则null",
        "reason": "提取的原因，没有则null"
    },
    "confidence": 0.95,  // 置信度0-1
    "response": "如果置信度<0.7，这里填写需要澄清的问题"
}

示例1：
用户：帮我查一下订单12345到哪了
输出：{"intent": "query_order", "entities": {"order_id": "12345"}, "confidence": 0.98}

示例2：
用户：这件衣服想换个颜色
输出：{"intent": "return_exchange", "entities": {"product": "衣服"}, "confidence": 0.85}
"""
    
    def classify(self, user_input, history=None):
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        # 如果有对话历史，可以加入增强上下文
        if history:
            messages.insert(1, {"role": "assistant", "content": f"历史对话：{history}"})
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.1,  # 低温度保证稳定性
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        return result

2.2 置信度阈值与人工兜底

当置信度过低时，不能强行回答，应该主动澄清或转人工。

def handle_intent(user_input, user_id):
    classifier = IntentClassifier(api_key="your-key")
    result = classifier.classify(user_input)
    
    # 低置信度处理策略
    if result["confidence"] < 0.6:
        # 转人工兜底
        transfer_to_human(user_id, user_input, result)
        return "您的问题比较复杂，我为您转接人工客服..."
    
    elif result["confidence"] < 0.8:
        # 主动澄清
        return result.get("response", "您是想查询订单还是办理售后呢？")
    
    else:
        # 高置信度，进入具体处理流程
        return route_by_intent(result, user_id)

三、工具定义与调用

3.1 Function Calling定义

工具是Agent连接业务系统的桥梁。通过Function Calling，LLM可以决定何时调用哪个API。

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "查询订单状态，需要订单号",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号，例如 OD2025123456"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_knowledge",
            "description": "搜索知识库，查询常见问题",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "用户的问题关键词"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_after_sale",
            "description": "创建售后工单，需要订单号和原因",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号"
                    },
                    "reason": {
                        "type": "string",
                        "enum": ["退货", "换货", "维修", "退款"],
                        "description": "售后原因"
                    },
                    "product": {
                        "type": "string",
                        "description": "商品名称或SKU"
                    }
                },
                "required": ["order_id", "reason"]
            }
        }
    }
]

3.2 工具调用执行器

class ToolExecutor:
    def __init__(self):
        # 模拟订单API
        self.order_api = OrderAPIClient()
        # 模拟工单系统
        self.workorder_api = WorkOrderAPIClient()
        # 知识库检索
        self.knowledge_base = KnowledgeBase()
    
    def execute(self, tool_call):
        """执行工具调用"""
        tool_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        if tool_name == "query_order":
            return self.query_order(arguments["order_id"])
        elif tool_name == "search_knowledge":
            return self.search_knowledge(arguments["query"])
        elif tool_name == "create_after_sale":
            return self.create_after_sale(arguments)
        else:
            return {"error": f"未知工具：{tool_name}"}
    
    def query_order(self, order_id):
        # 调用订单系统API
        order_info = self.order_api.get_order(order_id)
        # 权限校验：只能查自己的订单
        return order_info
    
    def search_knowledge(self, query):
        # 向量检索知识库
        return self.knowledge_base.search(query)
    
    def create_after_sale(self, params):
        # 创建售后工单
        ticket_id = self.workorder_api.create(params)
        return {"ticket_id": ticket_id, "status": "created"}

3.3 RAG知识库检索

对于咨询类问题，我们采用RAG（检索增强生成）架构，从知识库中检索相关内容后生成回答。

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class KnowledgeBase:
    def __init__(self):
        # 加载embedding模型
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        # 加载知识库索引（生产环境用Milvus）
        self.index = faiss.read_index("knowledge.index")
        self.documents = self.load_documents()
    
    def search(self, query, top_k=3):
        # 向量化查询
        query_emb = self.model.encode([query]).astype('float32')
        # FAISS检索
        distances, indices = self.index.search(query_emb, top_k)
        
        # 返回相关文档
        results = []
        for idx in indices[0]:
            if idx >= 0:
                results.append({
                    "content": self.documents[idx],
                    "score": float(distances[0][list(indices[0]).index(idx)])
                })
        return results
    
    def generate_response(self, query, retrieved_docs):
        # 将检索结果作为上下文，调用LLM生成回答
        context = "\n".join([doc["content"] for doc in retrieved_docs])
        prompt = f"""
基于以下知识库内容，回答用户问题。
知识库：
{context}

用户问题：{query}
回答要准确、简洁，如果知识库中没有相关信息，请说明不知道。
"""
        # 调用LLM生成
        return self.llm.generate(prompt)

四、多轮对话管理

4.1 基于Redis的会话状态管理

多轮对话需要记忆上下文。我们使用Redis存储短期会话状态，支持分布式部署。

import redis
import json
import time

class DialogStateManager:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(
            host=redis_host, 
            port=redis_port,
            decode_responses=True
        )
        # 会话过期时间：1小时
        self.expire_time = 3600
    
    def get_state(self, session_id):
        """获取会话状态"""
        key = f"session:{session_id}"
        state = self.redis.get(key)
        if state:
            return json.loads(state)
        return {
            "session_id": session_id,
            "history": [],  # 最近5轮对话
            "context": {},  # 上下文槽位
            "last_update": time.time()
        }
    
    def update_state(self, session_id, state):
        """更新会话状态"""
        key = f"session:{session_id}"
        state["last_update"] = time.time()
        self.redis.setex(key, self.expire_time, json.dumps(state))
    
    def add_to_history(self, session_id, user_msg, bot_msg):
        """添加一轮对话到历史"""
        state = self.get_state(session_id)
        state["history"].append({
            "user": user_msg,
            "bot": bot_msg,
            "time": time.time()
        })
        # 只保留最近5轮
        if len(state["history"]) > 5:
            state["history"] = state["history"][-5:]
        self.update_state(session_id, state)
    
    def update_slot(self, session_id, slot_name, slot_value):
        """更新槽位信息"""
        state = self.get_state(session_id)
        if "slots" not in state:
            state["slots"] = {}
        state["slots"][slot_name] = slot_value
        self.update_state(session_id, state)

4.2 槽位填充机制

对于复杂流程（如退换货），需要逐步收集必要信息。

class SlotFilling:
    def __init__(self, state_manager):
        self.state_manager = state_manager
        
    def process(self, session_id, intent, entities):
        """处理槽位填充逻辑"""
        state = self.state_manager.get_state(session_id)
        
        # 定义各意图需要的槽位
        required_slots = {
            "return_exchange": ["order_id", "reason", "product"],
            "query_order": ["order_id"],
            "knowledge_query": []
        }
        
        slots_needed = required_slots.get(intent, [])
        current_slots = state.get("slots", {})
        
        # 更新本次抽取的实体
        for key, value in entities.items():
            if value and key in slots_needed:
                current_slots[key] = value
        
        # 检查缺失槽位
        missing = [slot for slot in slots_needed if slot not in current_slots]
        
        if missing:
            # 还有缺失，继续追问
            return {
                "status": "incomplete",
                "missing": missing,
                "question": self.generate_question(missing[0])
            }
        else:
            # 槽位已齐，执行动作
            return {
                "status": "complete",
                "slots": current_slots
            }
    
    def generate_question(self, missing_slot):
        """生成追问话术"""
        questions = {
            "order_id": "请提供您的订单号",
            "reason": "您需要办理退货、换货还是退款？",
            "product": "请问是哪件商品？"
        }
        return questions.get(missing_slot, "请补充信息")

五、人工兜底策略

5.1 智能转人工决策

不是所有问题都要"硬扛"。优雅的转人工是用户体验的关键 。

class HumanHandoffDecision:
    def __init__(self):
        # 强转意图词库
        self.force_transfer_keywords = ["投诉", "举报", "找领导", "12315", "赔偿"]
        # 情绪词库
        self.negative_emotion_words = ["垃圾", "差评", "退货", "退款", "差劲"]
    
    def should_transfer(self, user_input, intent_result, dialog_history):
        """判断是否需要转人工"""
        score = 0
        reasons = []
        
        # 1. 关键意图词触发
        for word in self.force_transfer_keywords:
            if word in user_input:
                score += 0.8
                reasons.append(f"触发强转词：{word}")
        
        # 2. 置信度过低
        if intent_result.get("confidence", 1.0) < 0.5:
            score += 0.7
            reasons.append("置信度过低")
        
        # 3. 重复提问（同一问题问3次以上）
        if self.is_repeating(user_input, dialog_history):
            score += 0.6
            reasons.append("重复提问")
        
        # 4. 情绪检测
        negative_count = sum(1 for w in self.negative_emotion_words if w in user_input)
        if negative_count >= 2:
            score += 0.7
            reasons.append("检测到负面情绪")
        
        # 5. 对话轮次过长（超过8轮）
        if len(dialog_history) > 8:
            score += 0.5
            reasons.append("对话过长")
        
        return {
            "should_transfer": score >= 0.8,
            "score": score,
            "reasons": reasons
        }
    
    def is_repeating(self, user_input, history):
        """检查是否重复提问"""
        if len(history) < 2:
            return False
        # 简单判断：最近3轮中有相似问题
        similar_count = 0
        for h in history[-3:]:
            if self.text_similarity(user_input, h["user"]) > 0.8:
                similar_count += 1
        return similar_count >= 2

5.2 上下文无缝移交

转人工时，必须将完整上下文同步给人工客服，避免用户重复描述 。

class HandoffContext:
    def __init__(self, state_manager):
        self.state_manager = state_manager
    
    def prepare_context(self, session_id):
        """准备转人工上下文"""
        state = self.state_manager.get_state(session_id)
        
        context = {
            "session_id": session_id,
            "user_id": state.get("user_id"),
            "intent_history": [],  # 历史意图
            "extracted_entities": state.get("slots", {}),  # 已提取实体
            "last_user_input": state["history"][-1]["user"] if state["history"] else "",
            "agent_attempts": state["history"],  # 完整人机对话
            "emotion_tag": self.detect_emotion(state["history"]),  # 情绪标签
            "timestamp": time.time()
        }
        
        # 压缩历史（保留关键信息）
        context["summary"] = self.summarize_conversation(state["history"])
        
        return context
    
    def transfer_to_human(self, session_id):
        """执行转人工"""
        context = self.prepare_context(session_id)
        
        # 调用客服分配服务
        assign_result = self.call_assign_service(context)
        
        # 将上下文存入Redis供人工客服读取
        self.redis.setex(
            f"handoff:{assign_result['agent_id']}:{session_id}",
            300,  # 5分钟有效期
            json.dumps(context)
        )
        
        return {
            "agent_id": assign_result["agent_id"],
            "estimated_wait": assign_result.get("wait_time", 0),
            "context_id": session_id
        }

六、性能优化与监控

6.1 流式响应

为了提升用户体验，可以采用SSE（Server-Sent Events）实现流式响应，让用户看到"正在思考"的过程。

from flask import Response, stream_with_context
import json

def stream_response(generator):
    """SSE流式响应生成器"""
    for chunk in generator:
        yield f"data: {json.dumps(chunk)}\n\n"
    
@app.route('/chat', methods=['POST'])
def chat():
    data = request.json
    user_input = data.get('message')
    session_id = data.get('session_id')
    
    def generate():
        # 1. 发送接收确认
        yield {"type": "status", "content": "接收到消息"}
        
        # 2. 意图识别
        yield {"type": "thinking", "content": "正在理解您的问题..."}
        intent = intent_classifier.classify(user_input)
        
        # 3. 工具调用（如果需要）
        if intent["intent"] == "query_order":
            yield {"type": "action", "content": "正在查询订单..."}
            result = tool_executor.query_order(intent["entities"]["order_id"])
        
        # 4. 生成回复
        yield {"type": "response", "content": result}
    
    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

6.2 监控指标体系

指标	说明	告警阈值
意图识别准确率	人工复核正确率	< 85%
转人工率	转人工会话占比	> 40% 或 < 5%（异常）
平均处理轮次	完成任务的平均对话数	> 8轮
用户满意度	事后评价/点赞率	< 90%
响应延迟	首包时间/完整时间	> 2s
工具调用成功率	调用订单/工单API成功率	< 99%

class MetricsCollector:
    def __init__(self):
        self.metrics = {
            "total_sessions": 0,
            "handoff_count": 0,
            "intent_accuracy": [],
            "response_times": []
        }
    
    def record_session_end(self, session_id, success=True, handoff=False):
        """会话结束记录"""
        self.metrics["total_sessions"] += 1
        if handoff:
            self.metrics["handoff_count"] += 1
        
        # 计算转人工率
        handoff_rate = self.metrics["handoff_count"] / self.metrics["total_sessions"]
        
        # 如果转人工率异常，告警
        if handoff_rate > 0.4 or handoff_rate < 0.05:
            self.send_alert(f"异常转人工率：{handoff_rate:.2%}")

七、总结与进阶思考

7.1 核心设计要点

维度	关键设计	生产建议
意图识别	Few-shot + 置信度阈值	定期更新few-shot样本，持续优化
工具调用	Function Calling + 权限校验	严格限制API权限，只允许查自己订单
多轮对话	Redis状态管理 + 槽位填充	设置过期时间，防止内存泄漏
人工兜底	智能决策 + 上下文移交	建立转人工复盘机制，闭环优化
性能	流式响应 + 缓存策略	高频问题缓存，降低LLM调用

7.2 从MVP到生产级的演进路径

7.3 踩坑经验

• 幻觉控制：LLM会"编造"订单信息。解决方案是工具调用结果必须真实，不允许LLM自己生成 。
• 权限校验：用户只能查自己的订单。每个API调用前都要校验身份。
• 成本控制：LLM调用成本高。对高频常见问题建立缓存，用规则引擎兜底。
• 数据闭环：所有失败的案例都要进入训练集，持续优化模型 。

---

写在最后：

智能客服Agent的开发是一个持续迭代的过程。从简单的意图识别开始，逐步增加工具调用、多轮对话、个性化记忆，最终目标是让用户感觉"像在和真人对话"。