基于OpenBuddy搭建私域客服助手，告别数据上传焦虑的实战方案

# 创建虚拟环境
python -m venv openbuddy_env
source openbuddy_env/bin/activate  # Linux/Mac
# .\openbuddy_env\Scripts\activate  # Windows

# 安装依赖包
pip install torch==2.0.1 --index-url https://download.pytorch.org/whl/cu118
pip install transformers accelerate sentencepiece protobuf
pip install flask flask-cors  # 用于构建 API 服务

# download_model.py
from transformers import AutoTokenizer, AutoModelForCausalLM

model_name = "OpenBuddy/openbuddy-zephyr-7b-v14.1"

# 下载到本地目录（避免重复下载）
print("Downloading tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(
    model_name, 
    trust_remote_code=True
)

print("Downloading model (this may take 10-20 minutes)...")
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",          # 自动分配到 GPU/CPU
    load_in_8bit=True,          # 8bit 量化，节省显存
    trust_remote_code=True
)

# 保存到本地
model.save_pretrained("./openbuddy_local")
tokenizer.save_pretrained("./openbuddy_local")
print("Model saved locally.")

# 虚构的售后知识库
PRODUCT_POLICIES = {
    "XPhone 14": {
        "warranty": "1年整机保修，电池6个月",
        "return_policy": "签收后7天无理由退货（未激活）",
        "common_issues": {
            "无法开机": "长按电源键10秒强制重启；若无效请连接充电器30分钟后重试",
            "屏幕闪烁": "检查是否开启了深色模式切换动画，可在设置-辅助功能中关闭"
        }
    },
    "XPad Air": {
        "warranty": "2年主要部件保修（屏幕、主板）",
        "return_policy": "签收后15天质量问题换货",
        "common_issues": {
            "充电慢": "尝试更换原装PD充电器，检查接口是否有异物",
            "触控笔断触": "重新配对蓝牙，并拧紧笔尖"
        }
    }
}

def search_policy(product_name: str, query: str) -> str:
    """模拟 RAG 检索"""
    product = None
    for p in PRODUCT_POLICIES:
        if p.lower() in product_name.lower():
            product = p
            break
    if not product:
        return f"未找到产品 {product_name} 的信息。"
    
    policy = PRODUCT_POLICIES[product]
    
    if "保修" in query or "warranty" in query.lower():
        return f"{product} 保修政策：{policy['warranty']}"
    elif "退货" in query or "return" in query.lower():
        return f"{product} 退货政策：{policy['return_policy']}"
    elif any(k in query for k in policy['common_issues'].keys()):
        for issue, solution in policy['common_issues'].items():
            if issue in query:
                return f"故障「{issue}」的解决方法：{solution}"
    else:
        # 返回通用指引
        issues_list = "、".join(policy['common_issues'].keys())
        return f"常见问题：{issues_list}。如需详细帮助，请描述具体故障现象。"

# prompt_builder.py
def build_customer_service_prompt(user_query: str, retrieved_knowledge: str) -> str:
    system_template = """你是一个专业的电子产品售后客服，名叫「小布」。请严格遵循以下规则：
1. 只回答与电子产品售后、保修、故障排查相关的问题。
2. 如果用户询问超出范围的内容（如政治、竞争对手产品、价格比较），请礼貌拒绝并引导回售后主题。
3. 优先使用下方「知识库内容」中的信息回答，不要自行编造保修条款。
4. 回答要简洁、耐心，单次回复不超过 150 字。
5. 如果知识库中没有答案，请建议用户联系人工客服 400-882-XXXX。

【知识库内容】
{knowledge}

用户问题：{query}
客服回答："""

    return system_template.format(
        knowledge=retrieved_knowledge,
        query=user_query
    )

# inference.py
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
from knowledge_base import search_policy
from prompt_builder import build_customer_service_prompt

class OpenBuddyCustomerService:
    def __init__(self, model_path="./openbuddy_local"):
        print("Loading model...")
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path,
            device_map="auto",
            load_in_8bit=True,
            trust_remote_code=True
        )
        self.model.eval()
    
    def generate_response(self, user_message: str, product_hint: str = None) -> str:
        # Step1: 检索知识
        if product_hint:
            knowledge = search_policy(product_hint, user_message)
        else:
            # 简单地尝试从消息中提取产品名
            product_candidates = ["XPhone", "XPad", "XPhone 14", "XPad Air"]
            found = None
            for p in product_candidates:
                if p.lower() in user_message.lower():
                    found = p
                    break
            knowledge = search_policy(found, user_message) if found else "请提供具体产品型号（如 XPhone 14）。"
        
        # Step2: 构建 prompt
        prompt = build_customer_service_prompt(user_message, knowledge)
        
        # Step3: 模型推理
        inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=2048)
        inputs = {k: v.to(self.model.device) for k, v in inputs.items()}
        
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.7,
                do_sample=True,
                top_p=0.9,
                repetition_penalty=1.1
            )
        
        full_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        # 提取 "客服回答：" 之后的内容
        answer = full_response.split("客服回答：")[-1].strip()
        return answer if answer else "抱歉，我暂时无法回答这个问题，请转人工客服。"

# app.py
from flask import Flask, request, jsonify
from flask_cors import CORS
from inference import OpenBuddyCustomerService

app = Flask(__name__)
CORS(app)

# 全局加载模型（只需一次）
print("Initializing AI客服...")
cs_bot = OpenBuddyCustomerService()
print("Ready.")

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    user_msg = data.get("message", "")
    product = data.get("product", None)  # 可选，前端可传产品型号
    
    if not user_msg:
        return jsonify({"error": "消息不能为空"}), 400
    
    try:
        reply = cs_bot.generate_response(user_msg, product)
        return jsonify({"reply": reply, "status": "success"})
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "alive", "model": "OpenBuddy-Zephyr-7B"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=False)

# test_client.py
import requests

API_URL = "http://localhost:8080/chat"

def ask(question, product=None):
    resp = requests.post(API_URL, json={"message": question, "product": product})
    return resp.json()["reply"]

# 测试用例
print("[测试1] 询问保修：", ask("XPhone 14 保修多久？", product="XPhone 14"))
print("[测试2] 故障排查：", ask("屏幕一直闪怎么办？", product="XPhone 14"))
print("[测试3] 越狱问题：", ask("请问可以刷机越狱吗？", product="XPhone 14"))
print("[测试4] 无产品信息：", ask("充电器坏了"))

# 终端1: 启动服务（首次加载模型约10秒）
python app.py

# 终端2: 运行测试
python test_client.py