共计 3240 个字符，预计需要花费 9 分钟才能阅读完成。
背景分析
移动支付场景中，用户越来越依赖自然语言交互。特别是在跨境支付场景下，多语言支持成为刚需。传统支付系统需要用户通过复杂的表单和按钮操作，而智能对话系统可以更直观地理解用户意图，比如：
- 用户说 ” 给美国的 John 转账 100 美元 ”，系统需要准确识别收款人、金额和币种
- 在争议处理场景中，用户描述 ” 上周三的咖啡店扣款有问题 ”，系统需关联具体交易
技术选型
在实现自然语言交互时，开发者通常面临两个选择：
- 自建 NLP 模型
- 优点：数据完全自主可控
-
缺点：需要大量标注数据，多语言支持成本高
-
使用 ChatGPT API
- 优点：开箱即用的多语言能力，意图识别准确率高
- 缺点：API 调用成本需要考虑
对于大多数中小型支付系统，ChatGPT API 在开发效率和效果上更具优势。
核心实现
API 异步调用示例
import aiohttp
from typing import Optional, Dict
async def call_chatgpt(
prompt: str,
session: aiohttp.ClientSession,
max_retries: int = 3
) -> Optional[Dict]:
"""调用 ChatGPT API with 重试机制"""
headers = {"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2 # 支付场景需要低随机性
}
for attempt in range(max_retries):
try:
async with session.post(
"https://api.openai.com/v1/chat/completions",
json=payload,
headers=headers,
timeout=5
) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
await asyncio.sleep(2 ** attempt) # 指数退避
else:
break
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == max_retries - 1:
raise
return None支付指令解析
支付指令需要双重验证：
-
正则提取关键信息
import re def parse_payment_command(text: str) -> Dict: """解析支付指令""" # 匹配模式：转账 [金额][币种] 给[收款人] pattern = r"转账 \s*(\d+\.?\d*)\s*([A-Z]{3})?\s* 给 \s*(.+)" match = re.search(pattern, text) if not match: raise ValueError("无法识别的支付指令格式") amount = float(match.group(1)) currency = match.group(2) or "CNY" # 默认人民币 recipient = match.group(3).strip() return {"amount": amount, "currency": currency, "recipient": recipient} -
金额校验
def validate_payment(amount: float, user_balance: float) -> bool: """验证支付金额""" if amount <= 0: raise ValueError("金额必须大于零") if amount > user_balance: raise ValueError("余额不足") if amount > 50000: # 大额支付需要额外验证 return False return True
对话状态管理
使用 Redis 存储对话上下文：
import json
import redis
redis_client = redis.Redis(host='localhost', port=6379, db=0)
class DialogManager:
def __init__(self, session_id: str):
self.session_id = session_id
def save_context(self, messages: list):
"""保存对话上下文"""
redis_client.set(f"pokepay:session:{self.session_id}",
json.dumps(messages),
ex=3600 # 1 小时过期
)
def load_context(self) -> list:
"""加载对话上下文"""
data = redis_client.get(f"pokepay:session:{self.session_id}")
return json.loads(data) if data else []安全合规
敏感数据过滤
根据 PCI DSS 标准，必须过滤支付卡号等信息：
def sanitize_input(text: str) -> str:
"""过滤敏感信息"""
# 过滤信用卡号
text = re.sub(r"\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14})\b", "[CARD]", text)
# 过滤 CVV
text = re.sub(r"\b[0-9]{3,4}\b(?=[^\s]*(安全码 |cvv))", "[CVV]", text)
return text日志脱敏
存储日志前进行脱敏处理：
def log_safe_message(session_id: str, message: str):
"""记录脱敏日志"""
safe_msg = sanitize_input(message)
logger.info(f"{session_id}: {safe_msg}")性能优化
Prompt 压缩技巧
- 缩写固定回复模板
- 使用代码代替自然语言描述格式要求
- 将长示例转换为简写形式
冷启动优化
实现 API 预热：
async def warmup_chatgpt():
"""启动时预热 ChatGPT"""
async with aiohttp.ClientSession() as session:
await call_chatgpt("ping", session)避坑指南
Prompt 注入防御
- 严格区分系统 prompt 和用户输入
- 对用户输入进行转义
def safe_prompt(user_input: str) -> str: """构建安全 prompt""" escaped = user_input.replace("```", "'") return f""" 你是一个支付助手，只处理支付相关请求。用户说：{escaped} 请提取支付金额、币种和收款人："""
金额一致性检查
在多轮对话中校验金额是否一致：
class PaymentValidator:
def __init__(self):
self.amount = None
def check_consistency(self, new_amount: float) -> bool:
"""检查金额是否一致"""
if self.amount is None:
self.amount = new_amount
return True
return abs(self.amount - new_amount) < 0.01扩展思考
未来可以考虑将对话系统与风控系统结合：
- 实时分析对话中的风险关键词（如 ” 退款 ”、” 争议 ”）
- 当检测到高风险对话时，自动触发人工审核
- 结合用户历史行为评估当前请求的可信度
通过这种深度集成，可以在保持良好用户体验的同时，有效降低支付风险。
正文完
