电商AI客服系统架构实战，不接平台API的本地混合方案

联鹿AI智能客服

479人浏览 · 2026-08-17 11:40:54

联鹿AI智能客服 · 2026-08-17 11:40:54 发布

做电商AI客服，绕不开一个问题，就是怎么对接平台。

主流思路是走平台官方API。淘宝有千牛开放平台，京东有JOS，拼多多有商家API。听起来很美，但实际操作中问题不少。平台API开放程度有限，很多关键能力比如查看买家头像、读取聊天记录上下文、发送图片消息，要么没开放，要么有严格权限门槛。各平台API协议不统一，对接一套就要重写一套适配层。更麻烦的是，平台政策随时可能收紧，API接口说下线就下线。

有没有另一种思路？有的。这篇文章分享一种不依赖平台官方API的电商AI客服架构方案，通过RPA直接操作接待页面，结合本地部署和云端API调用，实现40多个电商平台的统一接入。这套方案已在联鹿AI智能客服的实际电商客服场景中落地运行。

一、三层分离的整体架构

整套系统分为三层，RPA接入层、本地调度层、云端推理层。三层之间通过消息队列解耦，各自独立扩展。

云端推理层在最上面，对接GPT、DeepSeek、千问、GLM等多个大模型API。中间是本地调度层，跑在商家自己的电脑上，负责消息处理、上下文管理和prompt组装。最底层是RPA接入层，通过浏览器自动化同时操作多个平台的客服接待窗口。

为什么这么设计？

核心考量有两个。

第一是数据安全，商家的聊天记录、客户信息、订单数据全部在本地处理，不经过任何第三方服务器，只有脱敏后的prompt才会发到云端API。

第二是平台无关性，RPA操作的是浏览器页面，跟人工客服的操作完全一样。平台有没有开放API、API权限多大，都不影响接入。

二、RPA接入层，不碰API也能操作接待页面

RPA（Robotic Process Automation）的核心思路很直白，就是用程序模拟人工在浏览器上的操作。但电商客服场景对RPA的要求比普通自动化高很多，需要解决几个关键技术问题。

2.1 消息采集，从DOM中提取结构化信息

每个电商平台的客服接待页面结构都不一样。淘宝千牛、京东咚咚、拼多多商家版、抖音飞鸽，它们的DOM结构、CSS类名、消息加载机制各不相同。

我们的做法是为每个平台维护一套页面解析规则，包含消息容器选择器、消息气泡选择器、买家昵称选择器、商品卡片选择器等。解析器读取DOM后，输出统一的消息结构。

难点在于电商平台的页面是动态加载的，DOM结构经常变。我们设计了多层兜底机制来应对这个问题。

class IncomingMessage:
    """统一消息结构"""
    session_id: str        # 会话标识
    buyer_id: str          # 买家标识
    platform: str          # 来源平台
    msg_type: str          # text / image / video / system
    content: str           # 文本内容
    image_urls: list       # 图片URL列表
    order_context: dict    # 当前订单上下文（如有）
    timestamp: int         # 消息时间戳

第一层，优先用CSS选择器定位。第二层，主选择器失效时，切换到XPath定位。第三层，选择器全部失效时，用文本特征加位置特征做模糊匹配。第四层，所有方式都失败时，触发报警通知运维人员手动更新选择器。

实际运行中，平台页面改版大约每季度发生一次，大部分情况下主选择器仍然有效，需要手动更新的情况并不多。

2.2 消息发送，模拟真人输入

发送消息不是简单地在输入框里塞文字。电商平台对消息发送有各种检测机制，输入太快会被判定为机器人，直接设置input.value不会触发平台的事件监听，有些平台需要模拟键盘事件才能激活发送按钮。

我们的处理方式是模拟真实的人工输入流程。

async def send_message(self, text, images=None):
    """模拟真人发送消息"""
    # 定位输入框并点击聚焦
    input_box = await self.find_input_box()
    await input_box.click()
    await asyncio.sleep(random.uniform(0.1, 0.3))
    
    # 逐字输入文本，模拟真人打字速度
    for char in text:
        await input_box.send_keys(char)
        await asyncio.sleep(random.uniform(0.02, 0.08))
    
    # 模拟按下Enter发送
    await input_box.send_keys(Keys.ENTER)
    
    # 如果有图片，先发送图片
    if images:
        for img_path in images:
            await self._upload_and_send_image(img_path)

关键点在于输入间隔是随机的，不是固定延迟。这样从平台的角度看，跟人工打字的行为特征是一致的。

2.3 多窗口管理，同时接待多个平台

一个商家可能同时开着淘宝、京东、拼多多、抖音四五个平台的接待窗口。RPA需要同时监控多个浏览器窗口或标签页，及时捕获新消息。

我们使用基于事件驱动的调度模型。每个窗口注册一个消息监听器，当检测到新消息时，将消息放入统一的消息队列。调度管理器按优先级和到达顺序从队列中取消息，分配给处理引擎。

class WindowScheduler:
    def __init__(self):
        self.message_queue = PriorityQueue()
        self.windows = {}  # platform_id -> WindowController
    
    async def on_new_message(self, platform_id, message):
        """新消息到达时入队"""
        # 优先级排序，售后问题 > 售前咨询 > 普通闲聊
        priority = self._calculate_priority(message)
        self.message_queue.put((priority, message))
    
    async def dispatch_loop(self):
        """调度主循环"""
        while True:
            priority, msg = await self.message_queue.get()
            response = await self.engine.process(msg)
            await self.windows[msg.platform].send_message(response)

三、本地调度层，消息处理与智能编排

消息从RPA层采集上来后，进入本地调度层。这一层跑在商家自己的电脑上，负责所有"思考"的工作。

3.1 意图识别与路由

不是所有消息都需要调用大模型。简单的关键词匹配就能处理的问题，比如自动回复"在的""稍等"，没必要浪费API调用。

本地调度层维护了一个多级处理管道。买家消息先过规则引擎做关键词和正则匹配，命中则直接回复预设话术。未命中的进入意图分类器，这是一个轻量级本地模型。简单意图走知识库检索回复，复杂意图才组装prompt调用云端API。

这样做的效果是，大约30%到40%的消息在本地就能处理，不需要调用云端API。对于token费用按量计费的场景，这个比例直接影响商家的运营成本。

3.2 上下文管理

电商客服的对话有明确的上下文关联。买家问"这个有货吗"，必须知道"这个"指的是哪个商品。

我们的上下文管理器会维护每个会话的状态，包括当前咨询的商品信息、关联订单信息、最近N轮对话记录、当前意图状态和已收集的槽位信息。

class SessionContext:
    session_id: str
    buyer_id: str
    platform: str
    current_product: dict    # 当前咨询的商品（从页面提取）
    order_info: dict         # 关联订单信息
    message_history: list    # 最近N轮对话记录
    intent_state: str        # 售前/售后/投诉...
    slot_values: dict        # 已收集的槽位信息

商品信息和订单信息是从接待页面的DOM中实时提取的。买家打开哪个商品页面，系统就自动读取哪个商品的标题、价格、SKU、库存等数据，注入到上下文中。

3.3 Prompt工程与模型编排

调用云端API时，prompt的质量直接决定回复质量。我们为不同场景设计了不同的prompt模板。

# 售前咨询prompt模板（简化版）
PRE_SALE_PROMPT = """
你是一个{category}品类的电商客服。
当前商品: {product_title}
商品价格: {price}
库存状态: {stock_status}

买家的问题: {buyer_question}

已知商品信息:
{product_knowledge}

回复要求:
1. 简洁，不超过100字
2. 口语化，像真人客服
3. 不确定的信息不要编造
"""

# 售后处理prompt模板（简化版）
AFTER_SALE_PROMPT = """
你是一个电商售后客服。
买家订单号: {order_id}
订单状态: {order_status}
买家问题类型: {issue_type}

买家描述: {buyer_description}

处理规则:
- 退货退款: 确认订单状态，引导买家发起退货申请
- 物流问题: 查询最新物流状态，给出预计到达时间
- 质量问题: 先安抚，收集图片证据，根据金额决定补偿方案
"""

模型选择上，系统支持按场景路由到不同的模型。售前咨询用DeepSeek，性价比高。复杂售后用通义千问，推理能力强。简单问答用本地小模型，零成本。商家可以根据实际效果和成本自行调整。

3.4 图片处理能力

电商客服场景中，买家经常会发图片。"这个颜色有没有实拍图""收到的货跟图片不一样你看"，这些都是高频场景。

本地调度层集成了图像识别能力。买家发来的图片，调用视觉模型识别内容，判断瑕疵、色差、商品型号等，辅助售后诉求分析。商家配置的商品图，则根据买家需求自动匹配发送，比如买家问有没有白色的图，系统自动发送白色款的商品图。

需要说明的是，目前只支持图片识别，不支持视频识别。遇到买家发视频反馈问题，系统会自动转交人工处理。

四、云端推理层的成本控制

大模型API按token计费，每次调用都有成本。如何在不影响回复质量的前提下降低调用量？我们做了几层优化。

第一层是本地缓存。相同或相似的问题，直接从缓存中取答案。电商客服场景中大量问题是重复的，什么时候发货、能便宜吗、有运费险吗，缓存命中率通常在40%以上。

第二层是批量合并。短时间内多条同类消息需要处理时，合并成一次API调用。比如同时有3个买家问同一个商品的库存，只需要查一次。

第三层是降级策略。API调用失败或超时时，自动降级到预设的兜底话术，或者转为人工接待，避免买家长时间等待。

class APIController:
    def __init__(self, config):
        self.cache = LocalCache(max_size=10000)
        self.rate_limiter = RateLimiter(config.max_rpm)
        self.fallback = FallbackStrategy()
    
    async def call_llm(self, prompt, context):
        # 先查缓存
        cache_key = self._compute_simhash(prompt)
        cached = self.cache.get(cache_key, threshold=0.9)
        if cached:
            return cached
        
        # 限流检查
        if not self.rate_limiter.acquire():
            return self.fallback.get_response(context)
        
        # 调用API
        try:
            response = await self.llm_client.chat(
                model=context.model,
                messages=prompt,
                timeout=10
            )
            self.cache.set(cache_key, response)
            return response
        except Exception:
            return self.fallback.get_response(context)