最近在做一个闲鱼电商的智能客服项目发现这个场景真的挺有意思也踩了不少坑。今天就来聊聊怎么用xianyu autoagent这套思路从零开始搭建一个能扛住真实流量的智能客服系统。我们的目标是让机器人能处理99%的常见问题把人力解放出来。闲鱼上的客服需求和标准电商平台很不一样。首先商品五花八门从二手手机到老家具什么都有问题也千奇百怪很难用一套标准话术框死。其次用户可能用各种方言或者不规范的表达来咨询比如“这个机子成色咋样”、“包邮不啦”。更关键的是交易安全相关的问题必须精准识别和回复比如“怎么担保交易”、“遇到骗子怎么办”一旦答错可能引发纠纷。这些特殊性决定了我们的机器人不能太“死板”。技术选型规则、机器学习还是大模型在动手之前我们得先想清楚用什么技术。市面上主要有三种路子基于规则Rule-Based就是写一大堆if-else判断句。比如用户消息里包含“价格”、“多少钱”就触发价格查询流程。优点是简单、直接、可控性极高零成本冷启动快。缺点也明显维护成本随着规则数量指数级上升而且泛化能力差用户换个说法比如“最低多少能出”可能就识别不了。适合处理非常明确、固定的问题比如“运费模板怎么设置”。传统机器学习ML比如用BERT这类预训练模型做意图分类。我们需要先定义好一批意图标签如“咨询价格”、“询问运费”、“举报用户”然后收集数据、标注、训练模型。优点是泛化能力强能理解不同表达背后的相同意图维护起来比规则系统轻松。缺点是初期需要一定量的标注数据并且模型需要持续优化。大语言模型LLM直接用GPT、文心一言等模型做对话。优点是能力强大几乎能应对任何开放性问题甚至能进行创作。缺点是成本高API调用或自建GPU集群、响应速度可能较慢、可控性差可能胡说八道即“幻觉”而且对交易安全这种需要精确回答的场景风险较高。我们的选择对于闲鱼客服这种对准确性和成本控制要求高的场景我们采用了“ML为主规则兜底”的混合策略。高频、明确的意图如价格、运费、发货时间用微调后的BERT模型识别一些非常敏感或固定的流程如安全提醒、引导用户点击官方链接用规则确保万无一失对于非常开放的长尾问题可以考虑在后期引入LLM作为补充但前期先用“转人工”策略处理。核心实现一步步搭建系统骨架确定了方向我们开始动手。整个系统的后端我们用 Python 的 FastAPI 来搭建因为它异步性能好写起来也快。1. 服务框架搭建Python FastAPI首先建立一个清晰的项目结构和服务入口。# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import uvicorn from core.nlp_intent import IntentRecognizer from core.dialog_manager import DialogManager from core.async_queue import TaskQueue app FastAPI(titleXianyu AutoAgent API) # 初始化核心组件 intent_recognizer IntentRecognizer() dialog_manager DialogManager() task_queue TaskQueue() class UserMessage(BaseModel): user_id: str session_id: str message: str timestamp: Optional[int] None app.post(/chat) async def handle_chat(request: UserMessage): 处理用户消息的核心接口 try: # 1. 意图识别 intent, confidence intent_recognizer.predict(request.message) # 2. 对话状态管理获取/更新当前会话状态 session_state dialog_manager.get_state(request.session_id) # 3. 根据意图和状态生成回复策略 reply_strategy dialog_manager.get_reply(intent, session_state) # 4. 如果是复杂任务如查询订单放入异步队列 if reply_strategy.needs_async: task_id task_queue.push(request.user_id, intent) return {reply: 您的问题正在处理中请稍候..., task_id: task_id} # 5. 直接回复 return {reply: reply_strategy.response, intent: intent} except Exception as e: # 记录日志并返回友好提示 app.logger.error(fError processing message: {e}) raise HTTPException(status_code500, detail服务暂时不可用请稍后再试) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)2. 意图识别模型微调基于BERT这是智能的核心。我们选择在中文BERT基础上进行微调。数据准备与清洗来源从历史客服日志中导出初期可以手动筛选和标注几百条。清洗去除无意义的符号、统一繁体字、纠正明显错别字。对于闲鱼场景要特别注意过滤掉用户名、商品链接等隐私信息。标注定义一批核心意图标签例如ask_price,ask_shipping,report_user,negotiate,greeting,farewell。# core/nlp_intent.py import torch from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments from torch.utils.data import Dataset from typing import List, Tuple import pandas as pd class IntentDataset(Dataset): 自定义意图分类数据集 def __init__(self, texts: List[str], labels: List[int], tokenizer, max_len: int 128): self.texts texts self.labels labels self.tokenizer tokenizer self.max_len max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): text str(self.texts[idx]) label self.labels[idx] encoding self.tokenizer.encode_plus( text, add_special_tokensTrue, max_lengthself.max_len, paddingmax_length, truncationTrue, return_attention_maskTrue, return_tensorspt, ) return { input_ids: encoding[input_ids].flatten(), attention_mask: encoding[attention_mask].flatten(), labels: torch.tensor(label, dtypetorch.long) } class IntentRecognizer: 意图识别器 def __init__(self, model_path: str ./model_save): self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.tokenizer BertTokenizer.from_pretrained(model_path) self.model BertForSequenceClassification.from_pretrained(model_path).to(self.device) self.model.eval() # 意图标签映射 self.id2label {0: ask_price, 1: ask_shipping, 2: report_user} # 示例需完善 def predict(self, text: str) - Tuple[str, float]: 预测用户消息的意图 try: inputs self.tokenizer(text, return_tensorspt, paddingTrue, truncationTrue, max_length128).to(self.device) with torch.no_grad(): outputs self.model(**inputs) probabilities torch.nn.functional.softmax(outputs.logits, dim-1) predicted_class_id torch.argmax(probabilities, dim-1).item() confidence probabilities[0][predicted_class_id].item() return self.id2label.get(predicted_class_id, unknown), confidence except RuntimeError as e: # 处理GPU内存不足等运行时错误 print(fPrediction error: {e}) return error, 0.0 except Exception as e: # 其他异常如输入格式错误 print(fUnexpected error: {e}) return unknown, 0.0 # 假设我们有一个清洗好的CSV文件 def prepare_data(csv_path: str): df pd.read_csv(csv_path) texts df[text].tolist() labels df[label].tolist() # 标签已经是数字ID return texts, labels # 微调训练示例简化版 def train_model(): texts, labels prepare_data(cleaned_intent_data.csv) tokenizer BertTokenizer.from_pretrained(bert-base-chinese) model BertForSequenceClassification.from_pretrained(bert-base-chinese, num_labelslen(set(labels))) dataset IntentDataset(texts, labels, tokenizer) # 划分训练集和验证集... training_args TrainingArguments( output_dir./results, num_train_epochs3, per_device_train_batch_size16, evaluation_strategyepoch, save_strategyepoch, ) trainer Trainer( modelmodel, argstraining_args, train_datasettrain_dataset, eval_dataseteval_dataset, ) trainer.train() model.save_pretrained(./model_save) tokenizer.save_pretrained(./model_save)3. 对话状态管理Redis存储设计客服对话往往不是一问一答而是多轮的。我们需要记住上下文。比如用户先问“这个手机多少钱”接着问“包邮吗”机器人需要知道“这个手机”指代的是什么。我们用 Redis 来存储对话状态因为它快而且支持设置过期时间适合会话场景。# core/dialog_manager.py import redis import json from typing import Dict, Any, Optional from datetime import timedelta import pickle # 注意实际生产环境可能需要更高效的序列化 class DialogState: def __init__(self, session_id: str, last_intent: str None, entities: Dict[str, Any] None, turn_count: int 0): self.session_id session_id self.last_intent last_intent self.entities entities or {} # 例如 {item_id: 12345, price: 1000} self.turn_count turn_count class DialogManager: def __init__(self, redis_host: str localhost, redis_port: int 6379, ttl: int 1800): # TTL设为30分钟会话超时自动清除 self.redis_client redis.Redis(hostredis_host, portredis_port, decode_responsesFalse) self.ttl ttl def _get_key(self, session_id: str) - str: return fdialog_state:{session_id} def get_state(self, session_id: str) - Optional[DialogState]: 从Redis获取对话状态 try: data self.redis_client.get(self._get_key(session_id)) if data: # 使用pickle反序列化生产环境可考虑msgpack等 return pickle.loads(data) return None except (redis.RedisError, pickle.UnpicklingError) as e: print(fFailed to get dialog state: {e}) return None def update_state(self, session_id: str, new_state: DialogState) - bool: 更新对话状态到Redis try: data pickle.dumps(new_state) # 每次更新都刷新TTL result self.redis_client.setex(self._get_key(session_id), self.ttl, data) return result except (redis.RedisError, pickle.PicklingError) as e: print(fFailed to update dialog state: {e}) return False def clear_state(self, session_id: str) - None: 清除对话状态 try: self.redis_client.delete(self._get_key(session_id)) except redis.RedisError as e: print(fFailed to clear dialog state: {e}) def get_reply(self, current_intent: str, state: Optional[DialogState]) - Dict[str, Any]: 根据当前意图和状态生成回复策略简化版状态机逻辑 # 这是一个非常简单的示例实际逻辑会更复杂 reply_map { ask_price: {response: 这款商品的价格是XXX元。, needs_async: False}, ask_shipping: {response: 默认发XX快递全国大部分地区包邮。, needs_async: False}, } # 如果有上一轮意图可以处理多轮对话 if state and state.last_intent ask_price: if current_intent ask_shipping: return {response: 刚才您问的那件商品发货地是上海一般3天内发出。, needs_async: False} # 更新状态 if state: state.last_intent current_intent state.turn_count 1 self.update_state(state.session_id, state) return reply_map.get(current_intent, {response: 您好请问有什么可以帮您, needs_async: False})性能优化扛住真实流量系统搭起来了但能不能用还得看性能。我们主要做了两件事压力测试和敏感词过滤。1. 压力测试方案使用Locust用 Locust 这个Python工具来模拟高并发用户看看我们的服务能撑到多少QPS每秒查询率。# locustfile.py from locust import HttpUser, task, between class ChatbotUser(HttpUser): wait_time between(0.5, 2) # 用户等待时间 task def send_message(self): # 模拟发送一条典型消息 payload { user_id: test_user_001, session_id: session_abc123, message: 这个多少钱 } with self.client.post(/chat, jsonpayload, catch_responseTrue) as response: if response.status_code 200: response.success() else: response.failure(fStatus code: {response.status_code})运行命令locust -f locustfile.py然后在浏览器打开http://localhost:8089设置目标用户数和每秒生成用户数比如模拟500个并发用户观察响应时间和失败率。根据测试结果我们可能需要优化数据库查询、增加缓存、或者对模型进行轻量化。2. 敏感词过滤AC自动机实现电商平台必须过滤敏感词和广告。我们使用AC自动机算法它能在O(n)时间复杂度内检测文本中是否包含任意多个预设关键词效率非常高。# core/sensitive_filter.py from ahocorasick import Automaton # 需要 pip install pyahocorasick from typing import List, Set class SensitiveFilter: def __init__(self, sensitive_words: List[str]): self.automaton Automaton() for idx, word in enumerate(sensitive_words): # 将每个敏感词添加到自动机中 self.automaton.add_word(word, (idx, word)) self.automaton.make_automaton() def contains_sensitive(self, text: str) - Set[str]: 检查文本是否包含敏感词返回找到的敏感词集合 found_words: Set[str] set() # 遍历文本自动机将匹配所有可能的位置 for end_index, (_, original_word) in self.automaton.iter(text): found_words.add(original_word) return found_words def replace_sensitive(self, text: str, replace_char: str *) - str: 将敏感词替换为指定字符 if not text: return text result_chars list(text) # 记录需要替换的位置 for end_index, (_, original_word) in self.automaton.iter(text): start_index end_index - len(original_word) 1 for i in range(start_index, end_index 1): result_chars[i] replace_char return .join(result_chars) # 使用示例 if __name__ __main__: words [加微信, 私下交易, 违禁品] filter SensitiveFilter(words) test_text 你好我们可以加微信聊吗 print(f包含敏感词: {filter.contains_sensitive(test_text)}) # 输出: 包含敏感词: {加微信} print(f替换后: {filter.replace_sensitive(test_text)}) # 输出: 替换后: 你好我们可以***聊吗避坑指南来自实战的经验冷启动语料收集一开始没数据怎么办除了爬历史日志还可以模拟对话自己扮演买家和卖家生成一批高质量的对话。利用公开数据集找一些电商QA数据集进行迁移学习和适配。用户反馈闭环在机器人回复后加一个“是否解决”的按钮将“未解决”的对话自动收集起来作为优化样本。多租户隔离如果你的系统要服务多个不同的闲鱼卖家店铺数据必须严格隔离。可以在session_id或user_id的设计中加入租户标识如tenant_id:session_id在Redis键、数据库查询等所有数据存取层都带上这个标识确保A店铺的数据绝不会泄露给B店铺。异常会话熔断有些用户可能会故意发送乱码或大量无意义信息消耗系统资源。我们需要一个熔断机制在DialogState中增加error_count字段。当连续多次如5次识别到的意图都是unknown或置信度极低时触发熔断。触发后当前会话在接下来几分钟内系统直接返回固定提示如“您的输入暂时无法识别如需帮助请联系人工客服”并停止调用模型直到冷却期结束。延伸思考让机器人更“聪明”基础的文本对话做好了但闲鱼是图片很重要的场景。买家经常直接发一张商品图问“这里划痕影响使用吗” 未来的优化方向之一就是集成图片识别能力。我们可以引入一个视觉模型如ResNet、ViT或专用的缺陷检测模型专门处理用户上传的图片功能自动识别图片中的商品品类、成色等级、是否存在明显瑕疵如划痕、磕碰。流程用户发送图片 - 系统调用视觉API进行分析 - 将分析结果如“检测到屏幕有细微划痕”作为上下文结合文本意图生成更精准的回复如“根据图片屏幕有细微划痕对日常使用影响不大价格可适当优惠。”。挑战需要收集和标注大量的二手商品瑕疵图片并且模型需要不断迭代优化。写在最后从零搭建一个高可用的智能客服系统涉及到的技术点确实不少从NLP模型、对话管理、缓存、异步任务到性能优化和安全过滤。xianyu autoagent这个项目给我的最大体会是没有银弹必须根据实际业务场景比如闲鱼的非标品、方言、安全诉求做针对性的技术选型和设计。先用一个“ML规则”的混合模型跑起来解决80%的问题再通过数据迭代和功能扩展如图片识别去攻克剩下的20%是一个务实且有效的路径。目前这个系统在测试环境下对常见问题的自动回复率已经能达到95%以上高峰期也能稳定运行。当然还有很多细节可以打磨比如更复杂的多轮对话设计、更精细的情感分析来应对客户投诉、以及如何与人工客服无缝交接。希望这篇笔记能给你带来一些启发也欢迎一起交流探讨。