Pokepay与ChatGPT集成实战：构建智能支付对话系统的技术方案

4次阅读

没有评论

共计 2850 个字符，预计需要花费 8 分钟才能阅读完成。

每次大促期间，支付系统的客服工单总是爆炸式增长。用户反复询问 ” 付款失败怎么办 ”、” 退款何时到账 ” 这类高频问题，多语言支持更是让客服团队疲于奔命。人工响应延迟经常超过 15 分钟，严重影响用户体验。

在评估了主流对话系统方案后，我们做了如下对比：

Rasa：
优势：完全自建，数据可控
劣势：需要大量标注数据训练意图识别模型
Dialogflow：
优势：谷歌提供的多语言支持
劣势：定制化能力弱，不适合复杂支付逻辑
ChatGPT：
天然支持 50+ 种语言
上下文理解能力强，适合处理支付场景的复杂咨询
通过 system prompt 可快速定义业务规则

在 Pokepay 商户后台需要设置两个关键 webhook：

payment_created – 支付创建事件
payment_refunded – 退款完成事件

配置示例（Python Flask 路由）：

@app.route('/webhook/pokepay', methods=['POST'])
def handle_webhook():
    # 验证 JWT 签名
    signature = request.headers.get('X-Pokepay-Signature')
    verify_jwt(signature, request.data)

    event = request.json
    if event['type'] == 'payment.created':
        process_payment(event['data'])
    elif event['type'] == 'refund.completed':
        process_refund(event['data'])

支付咨询的典型状态流转：

stateDiagram
    [*] --> Greeting
    Greeting --> PaymentQuery: 用户询问支付问题
    PaymentQuery --> RefundGuide: 询问退款流程
    PaymentQuery --> FailureHelp: 支付失败咨询
    RefundGuide --> [*]: 超时结束
    FailureHelp --> PaymentRetry: 建议重新支付

关键状态处理逻辑：

class ChatStateMachine:
    def __init__(self):
        self.current_state = 'GREETING'

    def transition(self, user_input):
        if '退款' in user_input:
            self.current_state = 'REFUND'
            return generate_refund_guide()

        if '支付失败' in user_input:
            self.current_state = 'FAILURE'
            return check_failure_reason()

使用正则表达式屏蔽银行卡号等敏感信息：

import re

def sanitize_input(text):
    # 过滤银行卡号
    text = re.sub(r'\b[0-9]{4}[-]?[0-9]{4}[-]?[0-9]{4}[-]?[0-9]{4}\b', 
                 '[CARD]', text)
    # 过滤 CVV
    text = re.sub(r'\b[0-9]{3}\b', '[CVV]', text)
    return text

Webhook 处理与对话管理核心代码：

import jwt
from datetime import datetime, timedelta

# JWT 验签
def verify_jwt(signature, payload):
    try:
        public_key = open('pokepay_public.pem').read()
        decoded = jwt.decode(signature, public_key, algorithms=['RS256'])
        return decoded['merchant_id'] == current_merchant
    except Exception as e:
        log_error(f"JWT 验证失败: {str(e)}")
        raise

# 对话 Session 管理
class DialogSession:
    def __init__(self, session_id):
        self.session_id = session_id
        self.context = {'start_time': datetime.now(),
            'last_active': datetime.now(),
            'history': []}

    def add_message(self, role, content):
        self.context['history'].append({
            'role': role, 
            'content': sanitize_input(content)
        })
        self.context['last_active'] = datetime.now()

通过异步处理 + 缓存实现 P99<800ms：

# 使用 Redis 缓存对话上下文
import redis
r = redis.Redis(host='redis-cluster', port=6379)

def get_context(session_id):
    cached = r.get(f"session:{session_id}")
    return json.loads(cached) if cached else None

def save_context(session_id, context, ttl=3600):
    r.setex(f"session:{session_id}", ttl, json.dumps(context))

压测结果：

并发量	平均延迟	P99 延迟
100	210ms	650ms
500	380ms	790ms

所有支付相关接口强制 HTTPS
敏感数据不落盘（对话日志中的卡号用 [REDACTED] 替换）
定期进行漏洞扫描

在调用 ChatGPT 前清洗用户输入：

def safe_prompt(user_input):
    blacklist = ['system', 'sudo', 'root']
    for word in blacklist:
        if word in user_input.lower():
            raise SecurityError("非法指令")

    return f""" 你是一个支付助手，请用中文回答关于 Pokepay 的问题。当前用户咨询：{user_input}
    回答要求：不超过 100 字，不要解释支付原理 """