客留（Keliu）是一个面向门店商家和小微企业的 AI 客户留存平台。它解决的不是“把客户信息存起来”这么简单的问题，而是帮助店主看清楚：哪些客户正在变冷，哪些客户值得重点维护，什么时候该做召回，应该发什么样的跟进内容。

我之前做类似客户运营工具时踩过几个典型痛点：客户数据散在 Excel、微信、收银系统里；店员知道“某些老客好久没来了”，但没有量化提醒；营销文案每次都靠人工临时写；免费试用、套餐升级、支付订单又经常和业务系统割裂。客留就是围绕这些问题，把客户、到店、AI 评分、营销、订阅支付和后台管理串成一个 SaaS 闭环。

项目比较有特色的地方有三点：

• AI 不是独立聊天框，而是嵌进客户流失评分、CLV、跟进文案和经营问答里。
• 计费不是摆设，后端已经有套餐、配额、支付宝沙箱订单和管理员管控。
• 工程形态完整，包含 Vue 3 前端、FastAPI 后端、MySQL、Redis、Celery、Elasticsearch、Dify 工作流和 Docker Compose。

目录

• 运行效果
• 技术栈选择
• 整体数据流
• 核心功能拆解
• 本地部署流程
• 常见启动报错与解决
• 优化记录
• 当前不足与后续迭代
• 开发收获与配套资源

运行效果

上图是本地运行后用 Playwright 抓取的登录页，支持邮箱登录、手机号登录和微信扫码入口。

注册页保留了邮箱和手机号两种模式，适合国内商家场景，也方便后续扩展第三方身份绑定。

客户导入示例图展示了项目对批量客户数据的处理入口。实际部署后，可以把这里替换为客户列表、仪表盘或营销活动页截图。

技术栈选择

这套项目的技术选型偏“够用、清晰、容易扩展”。没有为了炫技引入过重的框架，而是围绕 SaaS 后台常见需求来选。

层级	本项目选择	同类方案	选择理由
前端框架	Vue 3 + TypeScript	React、Svelte	Vue 组合式 API 对后台表单、列表、弹窗很顺手，TypeScript 约束接口字段
构建工具	Vite	Webpack、Rspack	冷启动快，配置少，适合中小型管理台
UI 组件	Element Plus	Ant Design Vue、Naive UI	表格、表单、弹窗、分页等后台组件齐全
状态管理	Pinia	Vuex、Redux	API 简洁，适合认证态、通知、用户信息等轻量全局状态
后端框架	FastAPI	Django、Flask、NestJS	类型提示友好，Pydantic 校验清晰，异步接口开发效率高
ORM/迁移	SQLAlchemy 2 + Alembic	Django ORM、Prisma	Python 后端生态成熟，迁移脚本可控
异步任务	Celery + Redis	RQ、APScheduler	适合批量评分、预约营销、定时任务
AI 工作流	Dify + DeepSeek API	LangChain、直接调用 OpenAI/其他模型	工作流可视化，方便把 AI 逻辑交给运营或产品继续调
搜索分析	Elasticsearch	Meilisearch、数据库 LIKE	预留客户搜索和分析扩展空间
支付	支付宝沙箱	微信支付、Stripe	国内场景更贴近商家，沙箱方便开发验证

整体数据流

通俗理解，客留的数据流是：商家在前端录入或导入客户，后端写入数据库；客户到店记录触发 AI 评分；评分结果进入客户详情、仪表盘和营销模块；营销活动通过异步任务分发；订阅支付决定商家能使用多少 AI、营销和导出额度。

项目目录也按这个思路拆分：

backend/
  app/routers/      # HTTP API 入口
  app/services/     # 业务逻辑：客户、AI、营销、计费、认证
  app/models/       # SQLAlchemy 数据模型
  app/schemas/      # Pydantic 请求/响应结构
  app/tasks/        # Celery 异步和定时任务
frontend/
  src/views/        # 页面：仪表盘、客户、营销、支付、后台
  src/api/          # Axios API 封装
  src/stores/       # Pinia 状态
dify-workflows/     # Dify 工作流 DSL

核心功能一：登录、注册与角色路由

这个模块的作用是把普通商家、未创建店铺的新用户、平台管理员分流到不同页面。实现上，前端用 Vue Router 的 meta 字段标记页面权限，路由守卫里读取 Pinia 中的登录态和角色。

关键代码如下，重点是 requiresAuth、requiresStore 和 roles：

// frontend/src/router/index.ts
{
  path: '/dashboard',
  component: () => import('@/views/Dashboard/DashboardView.vue'),
  meta: { requiresAuth: true, requiresStore: true },
}

{
  path: '/admin',
  component: () => import('@/views/Admin/AdminDashboard.vue'),
  meta: { requiresAuth: true, roles: ['super_admin'] },
}

router.beforeEach(async (to, _from, next) => {
  const auth = useAuthStore()

  if (to.meta.requiresAuth && !auth.isLoggedIn) {
    return next(`/auth/signin?redirect=${encodeURIComponent(to.fullPath)}`)
  }

  if (to.meta.requiresStore && auth.isLoggedIn && !auth.hasStore) {
    return next('/onboarding')
  }

  if (to.meta.roles && auth.isLoggedIn) {
    const required = to.meta.roles as string[]
    const hasRole = auth.roles.some((r) => required.includes(r))
    if (!hasRole) return next('/dashboard')
  }

  next()
})

后端认证同时支持邮箱、手机号、验证码、密码、刷新令牌、二维码登录和微信绑定。登录接口还加了基础限流，避免验证码和密码接口被刷。

# backend/app/routers/auth.py
@router.post("/send-code", response_model=MessageResponse)
@limiter.limit("3/minute")
async def send_code(request: Request, data: SendCodeRequest):
    code = await send_verification_code(data.identity_account, purpose=data.purpose)
    if settings.ENVIRONMENT == "development" and code:
        return MessageResponse(message=f"code sent: {code}")
    return MessageResponse(message="code sent")

@router.post("/login/password", response_model=TokenResponse)
@limiter.limit("5/minute")
async def login_password(request: Request, response: Response, data: LoginByPasswordRequest, db: AsyncSession = Depends(get_db)):
    token_response = await login_by_password(account=data.identity_account, password=data.password, db=db)
    _set_refresh_cookie(response, token_response.refresh_token)
    return public_token_response(token_response)

这里的取舍是：开发环境会直接返回验证码，降低本地调试成本；生产环境则应接真实短信/邮箱服务，并把刷新令牌放在 HttpOnly Cookie 里，降低前端脚本读取令牌的风险。

核心功能二：客户管理与批量导入

客户模块负责沉淀门店最重要的数据资产：客户资料、联系方式、授权状态、到店记录、消费金额和服务反馈。批量导入使用 CSV 作为最低成本入口，适合从 Excel 或收银系统导出的历史数据。

导入流程分三步：

1. 创建导入任务，把进度写入 Redis。
2. 读取 CSV，清洗字段，校验手机号、重复客户和必填项。
3. 入库客户和到店记录，同步 Elasticsearch，并触发 AI 评分任务。

精简后的关键代码如下：

# backend/app/services/import_service.py
async def process_csv_import(task_id: str, file_path: str, store_id: str, db_session_factory) -> dict:
    df = pd.read_csv(file_path)
    raw_rows = df.to_dict(orient="records")

    cleaned_rows = []
    errors = []
    for index, row in enumerate(raw_rows, start=1):
        normalized = {
            key.strip().lower().replace(" ", "_"): value
            for key, value in row.items()
            if isinstance(key, str)
        }
        cleaned, row_errors = clean_row(normalized, existing_phones, csv_phones_seen, index)
        if cleaned:
            cleaned_rows.append(cleaned)
        errors.extend(row_errors)

    progress["success"] = len(cleaned_rows)
    progress["failed"] = len(errors)
    await set_import_progress(task_id, progress)

体验上的取舍是：先支持 CSV，比一开始适配所有 Excel 样式更稳；进度放 Redis，可以让前端轮询任务状态；失败行不直接中断全量导入，而是记录错误，避免一行脏数据拖垮整批数据。

核心功能三：AI 流失评分与 CLV

AI 评分模块的目标不是给出一个玄学分数，而是把客户行为转成可解释的风险信号。系统优先调用 Dify 流失预测工作流；如果未配置 Dify 或调用失败，就回退到本地规则引擎。

本地规则主要看三个维度：

• 到店间隔：越久没来，风险越高，权重 40%。
• 到店频率：月均到店越少，风险越高，权重 30%。
• 消费趋势：最近消费下降越明显，风险越高，权重 30%。

# backend/app/services/ai_service.py
async def calculate_churn_score(customer_id: str, db: AsyncSession, force_refresh: bool = False) -> ChurnScoreResponse:
    cache_key = f"churn:{customer_id}"
    cached = await cache_get(cache_key) if not force_refresh else None
    if cached:
        return ChurnScoreResponse(**cached)

    # Dify 优先；失败后回退本地规则
    if settings.DIFY_CHURN_API_KEY:
        dify_result = await dify_service.predict_churn(
            customer_name=customer.name,
            store_name=store.name if store else "店铺",
            visit_history=visits_json,
            total_spent=total_spent,
            days_ago=days_since,
            visit_count=total,
        )

    churn_score = round(
        recency_score * 0.4 +
        frequency_score * 0.3 +
        trend_score * 0.3,
        1,
    )

    response = ChurnScoreResponse(...)
    await cache_set(cache_key, response.model_dump(), TTL_LONG)  # TTL_LONG = 3600 秒
    return response

这块的成本取舍比较关键。AI 评分如果每次打开客户详情都调用模型，成本和延迟都会上升；所以项目做了 1 小时缓存，并且只把最近 20 条到店记录传给 Dify。这样牺牲了一点历史完整性，换来更稳定的响应速度和更低的 token 消耗。

# backend/app/services/ai_service.py
visits_json = str([
    {"date": str(v.visited_at), "service": v.service_type, "amount": float(v.amount)}
    for v in visits[:20]
])

核心功能四：营销活动与异步分发

营销模块把 AI 建议落到实际动作上：商家可以创建短信、邮件或微信触达任务，选择立即发送或预约发送。每次活动都会生成发送日志，后续可以继续扩展送达率、点击率和复购率。

预约活动不应该依赖用户一直停留在页面上，所以项目使用 Celery Beat 每分钟扫描到期活动，再交给 Worker 分发。

# backend/app/tasks/campaign_task.py
@celery_app.task(bind=True)
def dispatch_due_campaigns(self):
    async def _run():
        result = await db.execute(
            select(Campaign).where(
                Campaign.status == "scheduled",
                Campaign.scheduled_at <= datetime.utcnow(),
            )
        )
        campaigns = result.scalars().all()
        for campaign in campaigns:
            await _dispatch_campaign(campaign.id, db)

    loop.run_until_complete(_run())

这里的取舍是：分钟级扫描足够满足大多数门店营销，不需要一开始就引入更复杂的延迟队列；真正上线后，如果营销量变大，可以改成更精确的任务调度或消息队列。

核心功能五：订阅、配额与支付宝沙箱支付

商业化模块负责回答两个问题：用户买了什么套餐，以及当前还能不能继续使用某个功能。

当前项目内置免费版、基础版、专业版三档套餐：

# backend/app/services/billing_service.py
PLANS = {
    "free": {
        "price_cents": 0,
        "customer_limit": 1000,
        "ai_daily_limit": 10,
        "campaign_daily_limit": 1,
        "has_export": False,
    },
    "basic": {
        "price_cents": 1990,
        "customer_limit": 2000,
        "ai_daily_limit": -1,
        "campaign_daily_limit": 50,
        "has_export": True,
    },
    "professional": {
        "price_cents": 4990,
        "customer_limit": 5000,
        "ai_daily_limit": -1,
        "campaign_daily_limit": 100,
        "has_export": True,
    },
}

配额检查统一从 check_quota 进入，避免每个业务模块重复写套餐判断。

async def check_quota(store_id: str, db: AsyncSession, quota_type: str) -> Subscription:
    sub = await _get_or_create_sub(store_id, db)
    plan = get_plan(sub.plan_name)
    await _reset_daily_if_needed(sub)

    if quota_type == "ai":
        limit = int(plan["ai_daily_limit"])
        if limit > 0 and sub.ai_used_today >= limit:
            raise HTTPException(status_code=402, detail="今日 AI 次数已用完")
        sub.ai_used_today += 1
        await db.commit()
        return sub

支付流程如下：

支付相关逻辑的取舍是：先接支付宝沙箱，验证订单、验签、回跳和查单闭环；生产环境再补齐异步通知重试、订单对账和更多支付渠道。

本地部署流程

下面是可复现的本地启动流程。Windows PowerShell 用户建议使用 npm.cmd，避免执行策略拦截 npm.ps1。

1. 准备环境

推荐环境：

• Node.js 20+，本机验证时为 Node v24.15.0、npm 11.12.1。
• Python 3.11+。
• Docker Desktop，用来启动 MySQL、Redis、Elasticsearch。
• Git。

如果本机暂时没有 Python，也可以优先用 Docker 跑基础服务和前端，后端等 Python 环境装好后再启动。

2. 拉取代码

git clone https://github.com/hejie0373-cloud/my-project.git
cd my-project

3. 配置环境变量

copy .env.example .env
copy backend\.env.example backend\.env
copy frontend\.env.example frontend\.env

如果使用 docker compose 里的 MySQL，主机端口默认是 3307，后端本地运行时建议把 backend\.env 改成：

DB_URL=mysql+aiomysql://root:change-me@localhost:3307/keliudb
REDIS_URL=redis://localhost:6379/0
ES_URL=http://localhost:9200
SECRET_KEY=change-this-in-local-dev
ENVIRONMENT=development
AUTO_CREATE_TABLES=false

AI 和支付可以先留空。留空时，Dify 相关能力会走本地规则或模板兜底；支付宝支付需要沙箱参数完整后才能真实跳转。

4. 启动基础服务

docker compose up -d mysql redis elasticsearch
docker compose ps

5. 启动后端

cd backend
python -m venv .venv
.\.venv\Scripts\activate
pip install -r requirements.txt
alembic upgrade head
python run.py

默认后端地址：

http://127.0.0.1:8009

验证后端：

Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8009/health

期望返回：

{"status":"ok","version":"1.0.0"}

6. 启动前端

cd ..\frontend
npm.cmd install
npm.cmd run dev

默认前端地址：

http://localhost:5173

如果 5173 已被占用，可以换端口：

npm.cmd run dev -- --host 127.0.0.1 --port 5188 --strictPort

7. 一键 Docker 启动

如果希望所有服务都走 Docker，需要先确认根目录 .env 和 backend\.env 里的数据库密码一致，然后执行：

docker compose up -d --build
docker compose ps

常见启动报错与解决

npm.ps1 被 PowerShell 拦截

报错：

npm : 无法加载文件 C:\Program Files\nodejs\npm.ps1，因为在此系统上禁止运行脚本。

解决：

npm.cmd install
npm.cmd run dev

或者调整 PowerShell 执行策略，但对新手来说直接用 npm.cmd 更稳。

前端访问显示 ERR_CONNECTION_REFUSED

报错：

127.0.0.1 拒绝了我们的连接请求。
ERR_CONNECTION_REFUSED

排查：

1. 确认 Vite 是否正在运行。
2. 确认端口是否是 5173，如果换成了 5188，浏览器也要打开 5188。
3. 如果同时开了多个项目，注意不要把别的项目页面当成客留页面。

解决：

cd frontend
npm.cmd run dev -- --host 127.0.0.1 --port 5188 --strictPort

MySQL 连接失败

常见报错：

sqlalchemy.exc.OperationalError: Can't connect to MySQL server on 'localhost'

排查重点是端口。docker-compose.yml 默认把容器 3306 映射到主机 3307，所以后端本地运行时不能盲目写 localhost:3306。

解决：

DB_URL=mysql+aiomysql://root:change-me@localhost:3307/keliudb

Redis 连接失败

常见报错：

redis.exceptions.ConnectionError: Error connecting to localhost:6379

解决：

docker compose up -d redis
docker compose ps redis

Dify 调用失败或余额不足

可能出现：

Dify 401: unauthorized
Dify 429: too many requests
Dify workflow failed: insufficient_quota
Dify call failed: timed out

排查：

1. DIFY_API_URL 是否指向正确的 Dify 服务。
2. DIFY_CHURN_API_KEY、DIFY_COPY_API_KEY、DIFY_INSIGHT_API_KEY 是否填对。
3. Dify 里绑定的模型供应商是否还有余额。
4. 工作流 Start 节点变量名是否和代码一致。

本项目有规则引擎兜底，所以 Dify 不通时流失评分仍然可以返回基础结果；但文案和洞察质量会下降。

接口限流触发

验证码和密码登录接口有限流：

429 Too Many Requests

开发时等待 1 分钟再试即可。生产环境建议把限流 key 从纯 IP 扩展为 IP + 账号维度，并接入网关层限流。

Playwright 或 npm 包下载失败

本地受限网络里可能看到：

npm ERR! FetchError: request to https://registry.npmmirror.com/@playwright%2fcli failed
reason: connect EACCES

解决思路：

• 优先确认网络和 npm registry。
• CI 或内网环境中预装依赖，不在运行时临时下载。
• 截图类工具只作为文档辅助，不要加入生产启动链路。

优化记录

当前项目还没有完整线上压测数据，所以这里不编造“上线后提升 80%”之类的数字。下面只记录已经在代码里落地、或本地验证过的优化点。

优化点	优化前	优化后	当前可量化结果
AI 评分缓存	每次查看都可能调用模型	Redis 缓存 1 小时	同一客户 1 小时内重复评分不再调用 Dify
Dify 输入裁剪	可能传全部到店历史	只传最近 20 条 visit	token 消耗随历史记录减少，需线上账单继续统计
前端依赖拆包	大 vendor 包影响加载	按 vue-vendor、element-plus、echarts、visual-vendor 拆包	构建配置已落地，具体体积以 npm run build 输出为准
异步评分	导入客户后同步评分会阻塞	Celery 后台重算	导入接口更快返回，评分延后完成
Dify 失败兜底	AI 服务异常会中断业务	回退本地规则/模板	AI 服务不可用时核心流程仍能走通
本地前端启动	Vite 默认端口可能冲突	支持显式 --port 5188 --strictPort	本地验证 Vite ready 约 0.6-0.9 秒

线上使用时需要额外关注预算和并发：

• Dify/模型调用要记录 prompt_tokens、completion_tokens、耗时和失败原因。
• 高风险客户批量评分建议分批处理，避免短时间打满模型限流。
• 免费版的 AI 每日次数限制可以保护预算，但也要给用户明确提示。
• Redis 缓存命中率需要监控，否则缓存写了但没命中，成本还是会飙。

当前不足

项目已经具备完整雏形，但还不是一个可以直接大规模商业化的版本。

• 业务页截图还需要在完整后端数据和测试账号下补齐，当前文档只放了登录、注册和导入样例。
• AI 评分解释还可以更细，比如把“为什么高风险”拆成更清楚的维度卡片。
• 营销活动还缺少送达率、点击率、转化率和活动后复购分析。
• 支付模块目前以支付宝沙箱为主，生产还要补异步通知重试、对账和异常订单处理。
• 权限模型已经有角色基础，但还可以继续细分到店铺员工、财务、运营等角色。
• 自动化测试覆盖了部分认证、二维码和导入逻辑，仍需扩展到支付、营销和管理员流程。

后续迭代方向

短期优先做这些：

1. 补齐商家端仪表盘、客户详情、营销活动、支付页的完整演示截图。
2. 增加 Demo 账号和种子数据，让读者克隆后能直接体验完整流程。
3. 给 AI 评分增加解释卡片和“建议动作已采纳/忽略”的反馈入口。
4. 给营销活动增加效果统计，形成“触达后是否复购”的闭环。

中长期可以继续做：

1. 多门店、多员工、多角色权限。
2. 微信支付、Stripe 或企业付款渠道。
3. 模型渠道自动切换：高价值客户用更强模型，普通批处理用低价模型。
4. 更完整的监控：接口耗时、Celery 队列长度、Redis 命中率、模型成本、支付失败率。

开发收获

这个项目最大的收获是把“AI 应用”从一个单点能力做成了业务闭环。真正有价值的不是调用一次模型，而是把模型输出放到客户管理、营销触达、配额计费、后台管理这些具体流程里。

从工程角度看，这个项目也很适合作为全栈 SaaS 学习样例：前端有真实管理台页面，后端有清晰业务分层，数据库有迁移，异步任务有 Celery，支付和 AI 都有可继续扩展的接口。

配套资源

• 开源仓库：https://github.com/hejie0373-cloud/my-project
• 在线演示：暂未提供公开地址，建议部署后补充 Vercel/服务器链接。
• Dify 工作流：dify-workflows/
• 本地截图：docs/keliu-signin.png、docs/keliu-signup.png

欢迎讨论两个问题：

1. 门店客户留存场景里，AI 评分应该更重视消费金额，还是更重视最近一次到店时间？
2. 免费版套餐应该限制 AI 次数，还是限制客户数量，哪种对商家更容易理解？