用 Qwen2.5 搭建店铺 AI 客服

2025-11-26 16:56:34
文章摘要
商家总被饿了么、微信群的催单、改地址等消息追着跑？这套国内能落地的 AI 客服系统，自动接住 80% 重复问题，30 秒响应，不用来回切换平台，省人力还让顾客满意，帮店长从忙乱救火变轻松监督。

一、工具选择:用什么搭起来?

二、技术方案

  1. 数据入口
  2. 智能中台
  3. 自动操作层

三、实操示例

  1. 部署 Qwen2.5 大模型.
  2. 对接饿了么 API
  3. 搭建 FastAPI 服务
  4. 上线运行

总结:AI客服的本质是什么?



早上 8 点，饿了么后台弹出 3 条催单，

中午 12 点，微信群里 5 个人问能不能改地址。

下午 3 点，又有人在群里说菜品少了一份。

晚上 8 点，还有顾客私信，我要退单，骑手态度不好。

这些消息分散在饿了么商家后台、微信群、私聊窗口，店长一天要在 3 个平台之间切来切去，手忙脚乱。

更要命的是，70% 的问题其实都是重复的，催单、改地址、退款、补菜。但每次都得人工回复，回复慢了顾客不满意，回复快了容易出错。

用 AI 客服的本质是把重复性高、规则明确的客服工作自动化，让店长从救火队员变成监督员。

AI 负责接住 80% 的常规问题，剩下 20% 的疑难杂症再由人工介入。这样既提升响应速度，又降低人力成本。


一、工具选择：用什么搭起来?

这套 AI 客服系统的核心是，本地部署的大模型 + 饿了么 API + 企业微信机器人。全部工具都可以在国内环境落地，成本可控。


1. 大模型 Qwen2.5 7B/14B

阿里通义千问开源的 Qwen2.5 是目前国产大模型中最适合做客服的选择。

为什么选 Qwen2.5?

支持 128K 长文本：可以一次性读取整个订单历史和聊天记录，做出准确判断

中文理解优秀：能理解各种口语化表达

本地部署友好：7B 版本只需 16GB 显存，一台普通 GPU 服务器搞定，14B 版本效果更好但需要 32GB 显存

模型地址： https：//huggingface.co/Qwen/Qwen2.5-7B-Instruct


2. 饿了么 API + 企微机器人（数据入口）

饿了么开放平台 API提供了完整的商家 API，可以获取：订单状态（已接单、配送中、已完成），顾客评价（满意、不满意），售后请求（退款、投诉），订单详情（菜品、地址、备注）

同时可以执行操：作修改订单备注，申请退款，标记异常订单

企业微信机器人可以：接收群内顾客的留言、投诉，自动回复消息，@店长处理疑难问题


3. 接口服务：FastAPI

用 FastAPI 搭建一个轻量级的中间层服务，负责：接收饿了么和企业微信的消息，调用 Qwen2.5 分析意图，执行对应操作，返回结果


4. 数据存储：SQLite

用 SQLite 存储订单缓存和聊天记录，方便 AI 查询历史信息做判断。


二、技术方案

整个系统分三层：数据入口层、智能中台层、自动操作层。就像一个餐厅的前厅（接单）、后厨（做菜）、传菜员（上菜），各司其职。


第一层 数据入口（接住碎片化消息）

任务：把来自不同渠道的消息统一接入

顾客的消息来自两个地方：

1. 饿了么商家后台

饿了么会推送事件到商家的回调接口（需要在开放平台配置），包括：新订单通知，顾客催单，申请退款，修改备注请求

这些事件都是 JSON 格式，比如催单事件：

{
  "order_id"： "1234567890"，
  "event_type"： "urge_order"，
  "message"： "师傅快点，我赶时间"，
  "timestamp"： "2025-11-25 12：30：00"
}



2. 企业微信群机器人

企业微信群里，顾客会直接在群里留言：我要改地址，能不能退款。

企业微信机器人通过 WebHook 接收群消息：

{
  "msgtype"： "text"，
  "text"： {
    "content"： "我的订单 1234567890 能改个地址吗?改成 XX 小区 5 号楼"
  }，
  "from"： {
    "userid"： "user123"，
    "name"： "张三"
  }
}



统一入口

用 FastAPI 搭建一个统一的消息接收服务：

from fastapi import FastAPI， Request

app = FastAPI（）

# 接收饿了么回调
@app.post（"/eleme/callback"）
async def eleme_callback（request： Request）：
    data = await request.json（）
    event_type = data.get（'event_type'）
    order_id = data.get（'order_id'）
    message = data.get（'message'）
    
    # 转交给 AI 中台处理
    await process_message（source="eleme"， order_id=order_id， message=message）
    return {"status"： "ok"}

# 接收企业微信消息
@app.post（"/wechat/callback"）
async def wechat_callback（request： Request）：
    data = await request.json（）
    content = data['text']['content']
    user_name = data['from']['name']
    
    # 转交给 AI 中台处理
    await process_message（source="wechat"， user_name=user_name， message=content）
    return {"status"： "ok"}

现在，不管消息从哪来，都统一进入 process_message 函数，由 AI 中台统一处理。


第二层 智能中台

任务：理解顾客意图，生成操作指令

这是整个系统的"大脑"。Qwen2.5 负责四件事：

1. 识别顾客意图：催单、退单、改地址、投诉、咨询菜品

2. 校验是否可执行：比如订单已配送完成就不能改地址

3. 生成操作指令：输出结构化的 JSON 指令，告诉系统下一步干什么

4. 编写客服回复：生成礼貌、专业的回复文本


实现方式：

from transformers import AutoModelForCausalLM， AutoTokenizer
import json

# 加载 Qwen2.5 7B
model_path = "./qwen2.5-7b"
model = AutoModelForCausalLM.from_pretrained（
    model_path，
    torch_dtype="auto"，
    device_map="auto"
）
tokenizer = AutoTokenizer.from_pretrained（model_path）

async def process_message（source， order_id=None， user_name=None， message=None）：
    # 1. 查询订单信息（如果有订单号）
    order_info = None
    if order_id：
        order_info = await get_order_info（order_id） # 调用饿了么 API
    
    # 2. 构建提示词
    system_prompt = """你是一个餐饮店的 AI 客服助手。你的任务是：
1. 识别顾客的意图（催单、退款、改地址、投诉、咨询等）
2. 判断该操作是否可执行
3. 输出 JSON 格式的操作指令
4. 生成礼貌的客服回复

输出格式：
{
  "intent"： "催单 | 退款 | 改地址 | 投诉 | 咨询"，
  "executable"： true/false，
  "reason"： "如果不可执行，说明原因"，
  "action"： {
    "type"： "call_api | notify_manager | reply_only"，
    "api"： "eleme_order_update | eleme_refund | ..."，
    "params"： {...}
  }，
  "reply"： "给顾客的回复文本"
}
"""
    
    user_prompt = f"""
顾客消息： {message}
订单信息： {json.dumps（order_info， ensure_ascii=False） if order_info else "无订单信息"}

请分析顾客意图并生成操作指令。
"""
    
    messages = [
        {"role"： "system"， "content"： system_prompt}，
        {"role"： "user"， "content"： user_prompt}
    ]
    
    # 3. 调用 Qwen2.5
    text = tokenizer.apply_chat_template（messages， tokenize=False， add_generation_prompt=True）
    model_inputs = tokenizer（[text]， return_tensors="pt"）.to（model.device）
    generated_ids = model.generate（**model_inputs， max_new_tokens=1024）
    response = tokenizer.batch_decode（generated_ids， skip_special_tokens=True）[0]
    
    # 4. 解析 AI 输出
    result = json.loads（response）
    
    # 5. 执行操作
    await execute_action（result['action']， order_id）
    
    # 6. 发送回复
    await send_reply（source， result['reply']， user_name）


第三层 自动操作层（执行指令）

任务：把 AI 的决策变成实际操作

根据 Qwen2.5 输出的指令，调用对应的 API：

1. 饿了么 API 操作

import requests
import hashlib
import time

# 饿了么 API 配置
ELEME_API_BASE = "https：//openapi.ele.me/v2"
CONSUMER_KEY = "your_consumer_key"
CONSUMER_SECRET = "your_consumer_secret"

def calculate_signature（url， params）：
    """计算饿了么 API 签名"""
    # 1. 参数排序
    sorted_params = sorted（params.items（））
    param_str = "&".join（[f"{k}={v}" for k， v in sorted_params]）
    
    # 2. 拼接字符串
    string_to_sign = f"{url}?{param_str}{CONSUMER_SECRET}"
    
    # 3. UTF-8 编码后转 HEX
    hex_str = string_to_sign.encode（'utf-8'）.hex（）
    
    # 4. 计算 SHA1
    sig = hashlib.sha1（hex_str.encode（'utf-8'））.hexdigest（）
    return sig

async def eleme_api_call（endpoint， method="GET"， params=None）：
    """调用饿了么 API"""
    url = f"{ELEME_API_BASE}{endpoint}"
    
    # 添加系统级参数
    params = params or {}
    params['consumer_key'] = CONSUMER_KEY
    params['timestamp'] = int（time.time（））
    
    # 计算签名
    params['sig'] = calculate_signature（url， params）
    
    # 发送请求
    if method == "GET"：
        response = requests.get（url， params=params）
    elif method == "POST"：
        response = requests.post（url， params=params）
    
    return response.json（）

async def get_order_info（order_id）：
    """获取订单信息"""
    return await eleme_api_call（f"/order/{order_id}/"， method="GET"）

async def update_order_remark（order_id， remark）：
    """修改订单备注"""
    # 注：实际 API 端点需参考饿了么文档
    return await eleme_api_call（
        f"/order/{order_id}/remark"，
        method="POST"，
        params={"remark"： remark}
    ）


2. 企业微信机器人操作

async def send_wechat_message（webhook_url， content， mentioned_list=None）：
    """发送企业微信消息"""
    data = {
        "msgtype"： "text"，
        "text"： {
            "content"： content，
            "mentioned_list"： mentioned_list or []  # @某人
        }
    }
    requests.post（webhook_url， json=data）

async def send_reply（source， reply， user_name=None）：
    """统一回复接口"""
    if source == "wechat"：
        # 回复到企业微信群
        await send_wechat_message（WECHAT_WEBHOOK_URL， reply）
    elif source == "eleme"：
        # 饿了么没有直接回复接口，可以通过客服消息推送
        # 这里简化处理，实际需要调用饿了么的客服消息接口
        print（f"回复给顾客： {reply}"）


3. 完整的执行函数

async def execute_action（action， order_id=None）：
    """执行 AI 生成的操作指令"""
    action_type = action['type']
    
    if action_type == "call_api"：
        api_name = action['api']
        params = action['params']
        
        if api_name == "eleme_order_update"：
            # 修改订单（这里简化，实际需要根据具体字段调用不同 API）
            await update_order_remark（order_id， params.get（'new_address'））
        
        elif api_name == "eleme_refund"：
            # 申请退款
            # 注：需要调用饿了么退款 API
            pass
    
    elif action_type == "notify_manager"：
        # 通知店长
        await send_wechat_message（
            WECHAT_WEBHOOK_URL，
            f"有一个问题需要人工处理，订单号： {order_id}"，
            mentioned_list=["@店长"]
        ）
    
    elif action_type == "reply_only"：
        # 仅回复，不执行操作
        pass


三、实操示例

现在我们来真刀真枪搭一个。目标：半天时间，搭一个能用的店铺 AI 客服系统。

我们把这个过程拆成四个步骤。


第一步 部署 Qwen2.5 大模型（30 分钟）

任务：把 Qwen2.5 7B 跑起来

# 1. 安装依赖
pip install transformers torch accelerate

# 2. 下载模型（约 15GB）
# 如果访问不了 HuggingFace，用镜像站 hf-mirror.com
huggingface-cli download Qwen/Qwen2.5-7B-Instruct --local-dir ./qwen2.5-7b


测试模型：

from transformers import AutoModelForCausalLM， AutoTokenizer

model_path = "./qwen2.5-7b"
model = AutoModelForCausalLM.from_pretrained（
    model_path，
    torch_dtype="auto"，
    device_map="auto"
）
tokenizer = AutoTokenizer.from_pretrained（model_path）

prompt = "顾客说：'师傅快点，我赶时间'，这是什么意图?"
messages = [
    {"role"： "system"， "content"： "你是一个餐饮店 AI 客服。"}，
    {"role"： "user"， "content"： prompt}
]
text = tokenizer.apply_chat_template（messages， tokenize=False， add_generation_prompt=True）
model_inputs = tokenizer（[text]， return_tensors="pt"）.to（model.device）

generated_ids = model.generate（**model_inputs， max_new_tokens=128）
response = tokenizer.batch_decode（generated_ids， skip_special_tokens=True）[0]
print（response）

提示： 如果显存不够，加上 load_in_8bit=True 参数，显存占用减半。


第二步 对接饿了么 API（40 分钟）

任务：能获取订单信息、修改订单

步骤 1：注册饿了么开放平台

去 https://open.ele.me 注册商家账号，申请 API 权限。会得到：

● consumer_key：你的应用 ID

● consumer_secret：签名密钥


步骤 2：写签名函数

饿了么的签名算法比较特殊，需要严格按照文档来：

import hashlib
import time
import requests
from urllib.parse import quote

ELEME_API_BASE = "https：//openapi.ele.me/v2"
CONSUMER_KEY = "your_consumer_key"
CONSUMER_SECRET = "your_consumer_secret"

def calculate_signature（url， params）：
    """
    饿了么 API 签名计算：
    1. 参数排序
    2. 拼接： url + ? + 参数 + secret
    3. UTF-8 编码转 HEX
    4. 计算 SHA1
    """
    # 1. 参数排序并 URL 编码
    sorted_params = sorted（params.items（））
    param_str = "&".join（[f"{k}={quote（str（v）， safe=''）}" for k， v in sorted_params]）
    
    # 2. 拼接字符串（注意 URL 末尾要加 /）
    if not url.endswith（'/'）：
        url += '/'
    string_to_sign = f"{url}?{param_str}{CONSUMER_SECRET}"
    
    # 3. UTF-8 编码后转 HEX
    hex_str = string_to_sign.encode（'utf-8'）.hex（）
    
    # 4. 计算 SHA1
    sig = hashlib.sha1（hex_str.encode（'utf-8'））.hexdigest（）
    return sig

def eleme_api_call（endpoint， method="GET"， params=None， data=None）：
    """调用饿了么 API"""
    url = f"{ELEME_API_BASE}{endpoint}"
    
    # 添加系统级参数
    params = params or {}
    params['consumer_key'] = CONSUMER_KEY
    params['timestamp'] = int（time.time（））
    
    # 计算签名
    params['sig'] = calculate_signature（url， params）
    
    # 发送请求
    if method == "GET"：
        response = requests.get（url， params=params）
    elif method == "POST"：
        response = requests.post（url， params=params， data=data）
    
    return response.json（）


步骤 3：测试 API

# 获取餐厅评价
restaurant_id = "your_restaurant_id"
result = eleme_api_call（
    f"/rating/restaurant/{restaurant_id}/"，
    method="GET"，
    params={"limit"： 10}
）
print（result）

如果能正常返回数据，说明 API 对接成功了。

提示： 饿了么签名算法容易出错，一定要严格按照文档的顺序和编码规则来。如果遇到签名错误，检查：

1. URL 末尾是否加了 /

2. 参数是否做了 URL 编码

3. 编码后是否转成了 HEX


第三步 搭建 FastAPI 服务（50 分钟）

任务：搭建消息接收和处理服务

from fastapi import FastAPI， Request
import json
import sqlite3

app = FastAPI（）

# 初始化数据库（存储订单缓存）
def init_db（）：
    conn = sqlite3.connect（'orders.db'）
    cursor = conn.cursor（）
    cursor.execute（'''
        CREATE TABLE IF NOT EXISTS orders （
            order_id TEXT PRIMARY KEY，
            status TEXT，
            address TEXT，
            remark TEXT，
            updated_at TEXT
        ）
    '''）
    conn.commit（）
    conn.close（）

init_db（）

# 接收饿了么回调
@app.post（"/eleme/callback"）
async def eleme_callback（request： Request）：
    data = await request.json（）
    
    # 解析事件
    event_type = data.get（'event_type'）
    order_id = data.get（'order_id'）
    message = data.get（'message'， ''）
    
    # 根据事件类型处理
    if event_type == "urge_order"：
        # 催单
        await handle_urge_order（order_id， message）
    elif event_type == "refund_request"：
        # 退款请求
        await handle_refund（order_id）
    
    return {"status"： "ok"}

# 接收企业微信消息
@app.post（"/wechat/callback"）
async def wechat_callback（request： Request）：
    data = await request.json（）
    
    content = data['text']['content']
    user_name = data['from']['name']
    
    # 提取订单号（简单正则匹配）
    import re
    order_match = re.search（r'\d{10，}'， content）
    order_id = order_match.group（） if order_match else None
    
    # 处理消息
    await handle_customer_message（order_id， content， user_name， source="wechat"）
    
    return {"status"： "ok"}

async def handle_customer_message（order_id， message， user_name， source）：
    """统一消息处理"""
    # 1. 查询订单信息
    order_info = None
    if order_id：
        result = eleme_api_call（f"/order/{order_id}/"， method="GET"）
        if result['code'] == 200：
            order_info = result['data']
    
    # 2. 调用 Qwen2.5 分析意图
    system_prompt = """你是一个餐饮店的 AI 客服。请分析顾客意图并输出 JSON 格式的操作指令。

输出格式：
{
  "intent"： "催单 | 退款 | 改地址 | 投诉 | 咨询"，
  "executable"： true/false，
  "reason"： "如果不可执行，说明原因"，
  "action"： {
    "type"： "call_api | notify_manager | reply_only"，
    "details"： "具体操作说明"
  }，
  "reply"： "给顾客的回复文本"
}
"""
    
    user_prompt = f"""
顾客消息： {message}
订单信息： {json.dumps（order_info， ensure_ascii=False） if order_info else "无订单信息"}

请分析并生成操作指令。
"""
    
    messages_for_ai = [
        {"role"： "system"， "content"： system_prompt}，
        {"role"： "user"， "content"： user_prompt}
    ]
    
    text = tokenizer.apply_chat_template（messages_for_ai， tokenize=False， add_generation_prompt=True）
    model_inputs = tokenizer（[text]， return_tensors="pt"）.to（model.device）
    generated_ids = model.generate（**model_inputs， max_new_tokens=1024）
    response = tokenizer.batch_decode（generated_ids， skip_special_tokens=True）[0]
    
    # 3. 解析 AI 输出
    try：
        result = json.loads（response）
    except：
        # 如果解析失败，返回默认回复
        result = {
            "reply"： "不好意思，我没理解您的意思，请联系店长处理~ @店长"
        }
    
    # 4. 发送回复
    if source == "wechat"：
        await send_wechat_message（WECHAT_WEBHOOK_URL， result['reply']）
    
    return result

# 企业微信发送消息
async def send_wechat_message（webhook_url， content）：
    import requests
    data = {
        "msgtype"： "text"，
        "text"： {"content"： content}
    }
    requests.post（webhook_url， json=data）

# 启动服务
if __name__ == "__main__"：
    import uvicorn
    uvicorn.run（app， host="0.0.0.0"， port=8000）


启动服务：

python main.py

现在服务已经在 http：//localhost：8000 运行了。

提示： 企业微信机器人 WebHook 需要在企业微信后台创建群机器人获取。饿了么回调地址需要在开放平台后台配置（配置成 http：//你的服务器IP：8000/eleme/callback）。


上线运行

部署到服务器：

1. 租一台云服务器（腾讯云、阿里云都可以），配置:4 核 CPU + 16GB 内存 + 16GB 显存的 GPU

2. 把代码和模型上传到服务器

3. 配置 Nginx 反向代理（可选，用于 HTTPS）

4. 在饿了么开放平台和企业微信后台配置回调地址

5. 启动服务

监控和日志：

import logging

logging.basicConfig（
    level=logging.INFO，
    format='%（asctime）s - %（levelname）s - %（message）s'，
    handlers=[
        logging.FileHandler（'ai_customer_service.log'），
        logging.StreamHandler（）
    ]
）

# 在关键位置加日志
logging.info（f"收到消息: {message}"）
logging.info（f"AI 分析结果: {result}"）
logging.info（f"执行操作: {action}"）

现在，你的 AI 客服已经可以 7×24 小时自动响应顾客消息了。


总结：AI 客服的本质是什么?

搭完这套系统，你会发现AI 客服的本质不是替代人，而是筛选 + 分流 + 加速。

AI 客服就像一个杠杆——它把店长从重复劳动中解放出来，让店长有时间分析数据，优化菜品，培训员工，维护重要客户关系。

这套系统搭完，店长的工作量减少，顾客满意度提升，投诉响应速度从平均 15 分钟缩短到 30 秒。这不是替代人，这是让人变得更强大。

声明：该内容由作者自行发布，观点内容仅供参考，不代表平台立场；如有侵权，请联系平台删除。