环境准备与依赖安装
在开始构建系统之前，我们需要确保本地环境已经配置好Python运行环境。本系统基于Python 3.8+开发，使用轻量级Web框架Flask来接收消息，利用SQLite进行数据存储，无需额外安装数据库服务，确保零门槛快速启动。
在终端中执行以下命令安装核心依赖库Flask：
pip install flask
安装完成后，建议新建一个项目目录，例如chat_system，并在其中创建一个名为app.py的主程序文件。接下来的所有代码都将在这个文件中编写。
数据库设计与留存逻辑实现
留存的核心在于将用户数据结构化存储。我们将使用SQLite数据库创建两张表：users表用于存储用户基本信息及标签，message_logs表用于记录交互历史，便于后续分析话术效果。
请在app.py中写入以下数据库初始化代码。这段代码会在程序启动时自动检测数据库文件是否存在，如果不存在则自动创建表结构。
```python
import sqlite3
import datetime
from flask import Flask, request, jsonify
app = Flask(__name__)
DB_NAME = "retention.db"
def init_db():
with sqlite3.connect(DB_NAME) as conn:
cursor = conn.cursor()
创建用户表，用于留存用户信息和打标签
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id user_id TEXT PRIMARY KEY,
nickname TEXT,
tag TEXT,
source TEXT,
last_active_time TEXT,
create_time TEXT
)
''')
创建消息日志表，用于记录引流交互细节
cursor.execute('''
CREATE TABLE IF NOT EXISTS message_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT,
message_content TEXT,
reply_content TEXT,
timestamp TEXT
)
''')
conn.commit()
```
话术配置与引流规则定义
为了实现精准引流，我们需要建立一套关键词匹配机制。这里使用Python字典来配置话术规则。当用户的消息包含特定关键词时，系统自动触发对应的回复。回复内容中包含引流链接或联系方式，实现“话术+引流”的闭环。
在app.py中继续添加以下配置代码。你可以根据实际业务需求修改KEYWORD_RULES字典中的内容。
```python
定义话术规则：关键词 -> 回复内容 + 用户标签
KEYWORD_RULES = {
"价格": {
"reply": "我们的全套课程仅需99元，点击链接即可查看详情：http://your-domain.com/course",
"tag": "potential_price_sensitive"
},
"教程": {
"reply": "这是为您准备的新手入门教程，加微信TechMaster领取完整PDF。",
"tag": "potential_tutorial_seeker"
},
"加群": {
"reply": "群聊二维码已发送，请长按识别加入技术交流群。",
"tag": "community_user"
}
}
默认回复，当未匹配到关键词时使用
DEFAULT_REPLY = "您好，请问是想了解课程价格还是领取技术教程？"
```
核心业务逻辑编写
这一部分是系统的引擎。我们需要编写一个处理函数，该函数接收用户ID和消息内容，执行三个步骤：匹配话术、更新用户标签（留存）、记录日志。
请将以下核心逻辑函数添加到app.py中：
```python
def handle_message(user_id, nickname, message):
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
reply = DEFAULT_REPLY
matched_tag = "default_user"
1. 话术匹配逻辑
for keyword, action in KEYWORD_RULES.items():
if keyword in message:
reply = action["reply"]
matched_tag = action["tag"]
break
2. 留存逻辑：更新或插入用户信息
with sqlite3.connect(DB_NAME) as conn:
cursor = conn.cursor()
try:
cursor.execute('''
INSERT INTO users (user_id, nickname, tag, source, last_active_time, create_time)
VALUES (?, ?, ?, ?, ?, ?)
''', (user_id, nickname, matched_tag, "api_input", current_time, current_time))
except sqlite3.IntegrityError:
如果用户已存在，更新标签和最后活跃时间
cursor.execute('''
UPDATE users
SET tag = ?, last_active_time = ?
WHERE user_id = ?
''', (matched_tag, current_time, user_id))
3. 记录交互日志
cursor.execute('''
INSERT INTO message_logs (user_id, message_content, reply_content, timestamp)
VALUES (?, ?, ?, ?)
''', (user_id, message, reply, current_time))
conn.commit()
return reply
```
API接口搭建与运行
为了让外部系统（如微信公众号、钉钉机器人或测试脚本）能够调用我们的逻辑，我们需要暴露一个HTTP接口。这里使用Flask创建一个POST接口。
在app.py的最后添加路由配置和启动命令：
```python
@app.route('/api/chat', methods=['POST'])
def chat():
获取JSON数据
data = request.get_json()
user_id = data.get('user_id', 'unknown')
nickname = data.get('nickname', 'guest')
message = data.get('message', '')
if not message:
return jsonify({"error": "Message content is required"}), 400
调用核心处理逻辑
reply_content = handle_message(user_id, nickname, message)
return jsonify({
"status": "success",
"reply": reply_content,
"user_tag": "updated_in_db"
})
if __name__ == '__main__':
init_db() 启动时初始化数据库
app.run(port=5000, debug=True)
``>
系统测试与验证
现在，所有代码已经编写完毕。我们需要验证系统是否能正确进行话术匹配和数据留存。
第一步：启动服务
在终端中运行：
python app.py
看到输出Running on http://127.0.0.1:5000即表示服务启动成功。
第二步：模拟用户发送消息（测试引流）
我们需要发送包含关键词“价格”的消息。打开新的终端窗口，使用curl命令进行测试：
```bash
curl -X POST http://127.0.0.1:5000/api/chat \
-H "Content-Type: application/json" \
-d '{"user_id": "10086", "nickname": "张三", "message": "我想咨询一下课程的价格"}'
```
预期返回结果中应包含回复内容：“我们的全套课程仅需99元...”。
第三步：模拟用户发送消息（测试留存）
再次发送一条不包含关键词的消息，测试默认回复：
```bash
curl -X POST http://127.0.0.1:5000/api/chat \
-H "Content-Type: application/json" \
-d '{"user_id": "10086", "nickname": "张三", "message": "你好"}'
```
预期返回结果为默认回复。
第四步：验证数据库数据（测试留存效果）
使用Python命令行或者SQLite查看工具打开retention.db，检查users表。你应该能看到user_id为"10086"的记录，其tag字段应被更新为"potential_price_sensitive"，last_active_time也应为最新时间。这证明系统成功实现了用户打标与留存。
扩展与部署建议
以上代码已经完成了“话术-引流-留存”的最小功能闭环（MVP）。如果你需要将其接入真实的微信生态，只需修改chat函数中的数据获取方式，解析微信XML或JSON数据包即可。数据库方面，SQLite在数据量达到百万级前性能完全足够，后续如需更高并发，可将sqlite3替换为pymysql连接MySQL，逻辑层代码无需大改。
