AI托管店铺系统设计：从客流识别到库存预测的工程实战

先问一句：开实体店的朋友，有没有经历过这种状态——

每天开门第一件事是看昨天的营业额，但根本不知道客流为什么涨、为什么跌；顾客在微信上问“有没有货”，回复慢了就被隔壁抢走；进货全凭感觉，旺季备货不足，淡季货堆成山；晚上关门后，店里安全全靠运气。

这类问题单独拆开看都不算大，可它们凑在一起，就成了压在小店老板身上的日常琐碎。传统的收银系统、进销存软件、摄像头监控，每个都能解决一个点，但彼此是割裂的。数据要人工汇总，报表要自己看，决策要自己想。

《木头Ai店管家》这类“一站式AI托管店铺”方案想解决的，正是这个割裂问题：把客流、车流、客服、库存、安防监控、经营分析这些原本分散的系统统一交给一套智能体（AI Agent）体系来接管，让数据自动流转，让 AI 辅助决策，让商家只关注“怎么把生意做好”，而不是把时间耗在“整理数据”和“盯系统”上。

本文会从一个相对完整的工程视角拆解这套系统：整体架构怎么搭、核心模块怎么设计、关键技术点是什么、数据结构怎么组织、落地时有哪些坑，以及一套可以直接参考的代码示例。无论你是正在做门店数字化产品的开发者，还是想用 AI 改造自家生意模型的店主，这篇文章都能给你一个可执行的思路框架。

1. 背景与核心概念

1.1 什么是“AI托管店铺”

先把它拆成三层理解：

• 感知层：通过摄像头、IoT传感器、收银系统、电商订单接口，采集店内客流、门口车流、商品库存、顾客咨询、交易流水等实时数据。
• 决策层：AI 对这些数据做统计分析、预测建模、异常识别，并生成经营建议。例如“明天是周末，建议把高毛利奶茶备货量提升 30%”“今天下午 3 点客流锐减，可能和周边学校放假有关系”。
• 执行层：AI 不只是给建议，它还能自动执行一部分动作。比如自动回复顾客的常见咨询、自动生成补货单、定时切换夜间安防模式、自动把经营日报推送到老板微信。

所以“AI托管”不是远程人工替你看店，而是构建一套感知—决策—执行闭环的软件系统。它本质上是一个面向实体零售场景的 AI Agent 系统，主线是数据流，副线是业务自动化。

1.2 它解决什么核心问题

传统门店经营有三个痛点很难靠人工解决：

第一，数据孤岛。客流数据在摄像头里，库存数据在进销存软件里，客服记录在手机微信里，销售数据在收银机里。老板想做一次完整的经营复盘，需要手动导出好几套报表，再用 Excel 加工半天。AI 店管家要做的，就是把这些数据源统一接入，建成一套实时更新的数据中台。

第二，响应不及时。顾客晚上 11 点留言咨询，人工客服已经下班，等第二天回复时顾客可能已经买了别家。AI 客服的价值不是替代人工，而是兜住非工作时间的高频重复咨询。

第三，决策靠经验拍脑袋。老手店主靠直觉确实能撑起一家店，但直觉很难复制到第二家、第三家店。AI 基于历史数据做预测和归因，至少可以让决策从“我觉得”变成“数据告诉我”。

1.3 适合哪些场景落地

这套系统不是大型商场的专属，反而越是个体小店，边际价值越明显。常见应用场景包括：

• 连锁奶茶店、咖啡店：客流预测、单品备货优化、员工排班建议。
• 社区便利店、超市：库存周转分析、临期商品提醒、自动补货。
• 餐饮门店：等位时长分析、车流与客流关联分析、差评归因。
• 服装店、数码店：顾客逛店路线热力分析、推荐搭配话术辅助。
• 汽修店、4S 店：门口车流统计、进店转化率计算、预约保养提醒。

从技术角度说，这套系统的复杂度不高，但“杂”：涉及计算机视觉、自然语言处理、时间序列预测、数据可视化、消息推送、定时任务等多个方向。这也是为什么它在工程上更适合用“模块化 + 智能体编排”的方式来做，而不是一个单体巨无霸应用。

2. 系统整体架构设计

2.1 分层架构图（文字版）

整个系统可以分为五个层级，每层职责单一，层与层之间通过消息队列或 API 解耦。

┌─────────────────────────────────────────┐ │ 展示层（Web/小程序/大屏） │ │ 经营看板、实时监控、AI对话建议、报表导出 │ └─────────────────────────────────────────┘ ┌─────────────────────────────────────────┐ │ 智能体层（AI Agent 编排） │ │ 客流分析Agent 库存预测Agent 客服Agent │ │ 安防巡检Agent 经营建议Agent │ └─────────────────────────────────────────┘ ┌─────────────────────────────────────────┐ │ 服务层（业务能力封装） │ │ 数据接入 消息推送 定时调度 权限管理 │ └─────────────────────────────────────────┘ ┌─────────────────────────────────────────┐ │ 数据层（存储与计算） │ │ MySQL PostgreSQL Redis Minio │ └─────────────────────────────────────────┘ ┌─────────────────────────────────────────┐ │ 采集层（边缘设备与插件） │ │ 摄像头 IoT传感器 收银系统 电商平台API │ └─────────────────────────────────────────┘

这个分层的好处是：摄像头、收银机这些底层设备经常换品牌，但只要接入层做好适配，上面的 AI 逻辑不用改动；AI 模型升级也只影响智能体层，不会牵一发动全身。

2.2 核心数据流

一次完整的“AI 托管”动作，数据是这样流转的：

1. 摄像头采集的视频帧通过边缘盒子上的模型推理，得到“进店人数”“经过人数”“停留时长”等结构化数据。
2. 这些数据上报到服务层，写入 PostgreSQL 的客流明细表。
3. 库存系统同步当天库存快照到 MySQL。
4. 定时任务在每天 22:00 触发库存预测 Agent，读取近 90 天的销售流水和库存快照，用 Prophet 或 XGBoost 预测未来 7 天销量，生成补货建议。
5. 补货建议通过企业微信/钉钉机器人推送给店长。
6. 店长在看板上看到建议，可以一键确认或修改，确认后自动生成采购单。

整个链路里，AI 不是站在流程外面给建议，而是插在流水线中间：数据进来，经过模型加工，产出可直接执行的业务动作。这就是“托管”和普通 BI 报表软件的本质区别。

2.3 模块划分

从工程实现角度，可以拆成以下独立模块：

模块	职责	关键技术点
视频采集与识别	客流、车流、停留时长、排队长度	YOLOv8、ByteTrack、DeepStream
数据接入层	对接收银、电商、ERP 系统	REST API、消息队列、数据清洗
业务数据仓库	统一建模存储	PostgreSQL、MySQL、Redis
AI 客服	自动应答、话术推荐、知识库检索	LangChain、RAG、向量数据库
库存预测	销量预测、补货建议、临期提醒	Prophet、XGBoost、LSTM
经营报告	日报、周报、异常告警	ECharts、定时任务、消息推送
远程看店	实时画面、夜间巡检、异常告警	WebRTC、RTSP、目标检测
决策辅助	归因分析、对标数据、营销建议	统计建模、OLAP

这些模块可以分阶段上线。MVP 阶段最推荐先做“视频客流统计 + 经营看板 + AI 日报”，因为这三个功能投入最小、见效最快，也最能积累数据。等数据量跑起来之后，再上库存预测和 AI 客服，模型效果才会好。

3. 开发环境与工具选型

3.1 技术栈参考

这里给出的不是唯一答案，而是一套经过验证的组合。版本号请根据你实际安装时的官方文档为准，不同时间下载的版本会有差异，但整体选型思路不变。

层次	推荐工具	说明
后台框架	Spring Boot / FastAPI	Java 适合重业务系统，Python 适合 AI 服务
视觉推理	YOLOv8 + ByteTrack	目标检测 + 多目标跟踪，客流计数的经典组合
AI 编排	LangChain / 自研 Agent	管理多轮对话、工具调用、记忆
向量库	Milvus / Chroma / pgvector	存放商品知识、FAQ 的向量索引
关系库	MySQL + PostgreSQL	MySQL 存业务单据，PostgreSQL 存时序客流数据
缓存	Redis	实时客流缓存、分布式锁、热点数据
存储	MinIO / 阿里云 OSS	存储录像片段、截图、报表文件
任务调度	XXL-Job / APScheduler	定时生成报表、执行预测任务
消息通知	企业微信机器人 / 钉钉机器人 / 邮件	把 AI 结果推给老板
前端	Vue 3 + ECharts	管理后台 + 数据大屏

3.2 生产环境的硬件参考

摄像头方面，普通的 400 万像素网络摄像头就够用，关键不是像素，而是安装位置和角度。客流统计摄像头建议装在正对店门的位置，离地 2.5 到 3 米，俯视角度 30 到 45 度，避免逆光和遮挡。

推理设备建议用边缘计算盒子，而不是把所有视频流都传到云服务器。一路 1080P 视频流传到云端每天要产生几十 GB 的流量，成本很高。更经济的做法是：在店里放一个 Jetson Orin Nano 或者类似算力的边缘盒子，本地跑 YOLO 模型，只把识别结果（JSON 格式）上传服务器，视频流只在需要远程看店时按需拉取。

3.3 本地开发环境搭建

以 Python 服务端为例，开发环境建议这样准备：

# 创建虚拟环境 python3 -m venv venv source venv/bin/activate # 安装基础依赖（版本按实际环境调整） pip install fastapi uvicorn pip install sqlalchemy pymysql pip install redis pip install opencv-python pip install ultralytics pip install pandas numpy pip install prophet pip install langchain langchain-openai

需要提醒的是：Prophet 在 Windows 上的安装有时会遇到编译问题，如果安装失败，可以考虑直接用 XGBoost 或 statsmodels 替代，或者使用 Docker 环境。视觉部分如果用 GPU 推理，还需要单独安装 CUDA 版本的 PyTorch。

4. 核心功能模块设计与实现

这一节是全文的重点，我们把核心模块逐一拆开：先讲这个模块在业务上解决什么问题，再讲实现思路，最后给出可运行的代码骨架。

4.1 客流分析与车流统计模块

4.1.1 业务价值

客流是店铺经营的“第一手数据”。进店人数与路过人数的比值，就是进店转化率；进店人数除以成交单数，就是成交转化率。这两个指标能直接反映门头吸引力、店员接待能力和商品陈列是否合理。

车流统计主要服务于汽修店、洗车店、临街餐饮店。门口经过多少辆车、有多少辆减速观望、有多少辆拐进店，是衡量“线下曝光”的关键指标。

4.1.2 实现思路

视频流接入后，先抽帧，再对每一帧做人形检测，然后通过跟踪算法（ByteTrack）给每个目标一个 ID。判断“进店”的方法很简单：定义一个虚拟围栏（一条直线或者一个矩形区域），当检测目标的中心点从区域外移动到区域内时，计为一次“进店”事件；反向移动则计为“出店”。车流统计使用相同的逻辑，只是检测类别换成车辆。

4.1.3 代码示例

下面是一个最简化的“进店检测”逻辑，使用 YOLOv8 做检测、ByteTrack 做跟踪。实际生产环境还需要处理摄像头角度校正、光线变化、目标遮拦等问题。

# 文件路径：vision/counter.py import cv2 import numpy as np from ultralytics import YOLO from collections import defaultdict class StoreCounter: def __init__(self, weights_path="yolov8n.pt", enter_line_y=0.6): self.model = YOLO(weights_path) # 每条进店线用“目标中心点是否跨过水平线 y = H * enter_line_y”判断 self.enter_line_y = enter_line_y self.track_history = defaultdict(list) # track_id -> 中心点历史 self.enter_count = 0 self.exit_count = 0 def process_frame(self, frame): H, W = frame.shape[:2] line_y = int(H * self.enter_line_y) results = self.model.track(frame, persist=True, tracker="bytetrack.yaml", classes=[0]) # classes=[0] 表示只检测 person 类，YOLO 的默认类别 0 是 person for box in results[0].boxes: if box.id is None: continue track_id = int(box.id.item()) x1, y1, x2, y2 = box.xyxy[0].tolist() center_x = (x1 + x2) / 2 center_y = (y1 + y2) / 2 self.track_history[track_id].append((center_x, center_y)) # 只保留最近 20 帧，避免列表无限增长 if len(self.track_history[track_id]) > 20: self.track_history[track_id].pop(0) # 如果前一个点在线的上方，当前点在线的下方，说明目标向下穿过线 if len(self.track_history[track_id]) >= 2: prev_y = self.track_history[track_id][-2][1] curr_y = center_y if prev_y < line_y <= curr_y: self.enter_count += 1 elif prev_y > line_y >= curr_y: self.exit_count += 1 # 画一条虚拟计数线，方便调试 cv2.line(frame, (0, line_y), (W, line_y), (0, 255, 0), 2) cv2.putText(frame, f"Enter: {self.enter_count} Exit: {self.exit_count}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) return frame

这段代码的核心逻辑不复杂，但在真实场景中有几个关键工程点需要处理：

• 重复计数问题：同一目标在线的附近来回徘徊时，会反复触发事件。生产环境通常会对同一个 track_id 设置冷却时间，比如 30 秒内只计一次。
• 漏检问题：人流量大时，后面的人可能被挡住。这种情况要考虑把摄像头斜向安装，或者用双目摄像头做深度信息。
• 性能优化：边缘盒子算力有限，可以把视频分辨率降到 640x640 推理，同时把推理帧率控制在每秒 5 帧左右，完全足够统计客流。

4.1.4 数据上报

识别结果不能只停在内存里，需要写入数据库。上报接口可以设计成 POST JSON 格式：

{ "store_id": "store_001", "device_id": "cam_front_01", "ts": "2025-06-01 14:30:00", "event_type": "enter", "track_id": 23, "image_url": "https://minio.example.com/clips/2025/06/01/xxx.jpg" }

4.2 AI 客服模块

4.2.1 业务价值

店铺的咨询消息通常 80% 是重复问题：什么时候营业、有没有停车位、某款商品还有没有货、支不支持外卖、能不能开发票。这些问题的答案相对固定，非常适合用知识库 + 大模型的方式做自动回复。

4.2.2 实现思路

AI 客服不再建议用传统的意图识别 + 槽位填充方式，因为维护成本太高。更好的方案是 RAG（检索增强生成）：

1. 把店铺的常见问答、商品手册、售后政策整理成文档，切分后向量化，存入向量数据库。
2. 用户提问后，先从向量库检索最相关的片段。
3. 把检索结果拼进 Prompt，让大模型基于检索内容生成回答。

核心代码用 LangChain 实现如下。

# 文件路径：service/ai_chat.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.chains import RetrievalQA # 1. 准备知识文档，实际场景从数据库加载店铺信息、商品表、FAQ def load_store_docs(store: dict, products: list, faqs: list): text = f"店铺名称：{store['name']}\n营业时间：{store['open_hours']}\n地址：{store['address']}\n" for p in products: text += f"商品：{p['name']}，价格：{p['price']}元，库存：{p['stock']}，简介：{p['desc']}\n" for faq in faqs: text += f"Q：{faq['q']}\nA：{faq['a']}\n" return text def build_qa_chain(store_text: str): splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40) chunks = splitter.split_text(store_text) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectordb = Chroma.from_texts(chunks, embeddings, persist_directory="./chroma_store") llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2) qa = RetrievalQA.from_chain_type( llm=llm, retriever=vectordb.as_retriever(search_kwargs={"k": 4}) ) return qa # 示例调用 if __name__ == "__main__": store_info = {"name": "木头奶茶店", "open_hours": "09:00-22:00", "address": "科技路 88 号"} products = [ {"name": "招牌珍珠奶茶", "price": 15, "stock": 50, "desc": "销量第一，茶味浓郁"}, {"name": "杨枝甘露", "price": 18, "stock": 30, "desc": "芒果+西柚，季节限定"}, ] faqs = [ {"q": "可以停车吗", "a": "门口有免费停车位，营业时间内都可以停。"}, {"q": "可以开发票吗", "a": "可以，收银台扫码填写抬头即可。"}, ] doc = load_store_docs(store_info, products, faqs) qa = build_qa_chain(doc) response = qa.invoke("你好，你们有珍珠奶茶吗？多少钱？") print(response["result"])

这里有两个问题要特别注意。

第一，库存信息要动态注入。直接把整个商品表一次性放进向量库是偷懒的做法，因为库存数量每分钟都在变化。更好的做法是：商品固定信息（名字、描述、价格）作为静态知识进向量库，但库存数量在收到咨询时实时查一次数据库，在 Prompt 里临时注入。

第二，答非所问的风险控制。大模型面对未知问题时可能“编造”政策。建议在 Prompt 中明确约束：“如果知识库中没有相关信息，请回复‘我需要帮您转接人工客服’，不要自行猜测。”同时，当用户连续两次收到自动回复后询问“转人工”，应直接把会话状态改成人工接待模式。

4.3 库存分析与自动补货模块

4.3.1 业务价值

库存管理最怕两件事：备货不足导致缺货损失销售额，备货过多导致资金占用和商品过期。AI 补货的核心是预测“未来 7 天每天大概能卖多少”，再结合现有库存、采购提前期和安全库存计算采购数量。

4.3.2 实现思路

一个完整补货建议的计算流程是这样的：

1. 读取该商品最近 90 天的日销量。
2. 读取未来 7 天的预测特征（星期几、是否节假日、天气预报、附近是否有活动）。
3. 用 Prophet 或 XGBoost 预测未来 7 天销量。
4. 补货量 = max(0, 预测销量 + 安全库存 - 现有库存 - 在途库存)。

下面用 Prophet 给一个完整示例。Prophet 对业务时序数据非常友好，不需要手动做太多特征工程。

# 文件路径：service/inventory_predict.py import pandas as pd from prophet import Prophet def predict_sales(sales_df: pd.DataFrame, forecast_days: int = 7): """ sales_df 必须包含两列： - ds: 日期，格式 2025-05-01 - y: 当日销量 """ model = Prophet( daily_seasonality=False, weekly_seasonality=True, yearly_seasonality=False, changepoint_prior_scale=0.05 ) model.fit(sales_df) future = model.make_future_dataframe(periods=forecast_days) forecast = model.predict(future) return forecast.tail(forecast_days) # 示例数据 demo_data = pd.DataFrame({ "ds": pd.date_range("2025-03-01", periods=60, freq="D").strftime("%Y-%m-%d"), "y": [12, 15, 14, 18, 22, 25, 20, 16, 13, 15, 17, 19, 23, 27, 22, 18, 15, 14, 16, 20, 24, 26, 21, 17, 14, 15, 18, 21, 25, 23, 19, 16, 13, 14, 17, 20, 24, 28, 23, 18, 15, 12, 14, 18, 21, 25, 27, 22, 17, 14, 13, 16, 19, 23, 26, 24, 20, 17, 14, 15] }) if __name__ == "__main__": result = predict_sales(demo_data, 7) print(result[["ds", "yhat", "yhat_lower", "yhat_upper"]])

要注意：Prophet 这类模型对噪声很敏感，如果店铺经常搞促销、销量波动极大，预测区间会变得很宽。生产环境建议对低价促销日做标记处理，或者直接用 XGBoost 加入“促销标记”“天气”“节假日”等特征。

补货量计算代码：

def recommend_purchase(forecast_df, current_stock, in_transit_stock, safety_stock=10): total_need = forecast_df["yhat"].sum() purchase_qty = max(0, total_need + safety_stock - current_stock - in_transit_stock) return int(purchase_qty) # 假设预测未来7天销量总和是 120，当前库存 50，在途 20，安全库存 10 # 推荐采购 = 120 + 10 - 50 - 20 = 60

4.4 24 小时看店与异常告警模块

4.4.1 业务价值

“24 小时帮您看店”并不是让 AI 一直盯着监控画面，而是让 AI 做三件事：

• 营业时间内：识别排队过长、店员离岗、地面杂物（如积水）等异常。
• 非营业时间：识别闯入、明火、烟雾、非法逗留。
• 每天定时：生成安全巡检报告，包括门窗状态、画面异常、设备离线提醒。

4.4.2 实现思路

夜间安防模式的核心是在边缘盒子上运行一个多分类检测模型，检测类别包括 person（在非营业时间出现）、fire、smoke，当检测到异常目标且置信度超过阈值时，立即截图并发送告警到老板手机。

代码设计上需要增加一个“事件冷却器”：同一目标触发告警后，5 分钟内不重复告警，避免老板被通知轰炸。

# 文件路径：service/security_guard.py import time class SecurityGuard: def __init__(self, cooldown_seconds=300): self.cooldown_seconds = cooldown_seconds self.last_alert_time = {} def on_detect(self, camera_id, event_type, confidence): now = time.time() key = f"{camera_id}_{event_type}" if key in self.last_alert_time: if now - self.last_alert_time[key] < self.cooldown_seconds: return False # 冷却期内，不重复告警 self.last_alert_time[key] = now # 实际工程中在这里调用通知服务 # notify.send(store_owner, f"告警：{camera_id} 检测到 {event_type}") return True

4.4.3 告警分级

生产环境的告警一定要分级，否则真正重要的事件会被淹没在噪音里：

级别	示例	通知方式
警告	设备离线、网络断开	企业微信消息
重要	营业时间检测到烟雾、夜间检测到闯入	电话/短信 + 微信
提示	客流异常波动、库存低于阈值	每日报表汇总

4.5 数据看板与经营日报模块

4.5.1 业务价值

看板的本质是把复杂数据变成一眼能看懂的图表。对于店主来说，最关心的不是技术指标，而是四个字：赚了没有。所以日报必须围绕人、货、场三条线组织：

• 人：进店人数、成交订单数、成交率、会员新增。
• 货：动销率、库存周转天数、缺货 SKU 数、临期商品数。
• 场：平均停留时长、客流高峰时段、当日营收对比上周同期。

4.5.2 前端可视化

推荐使用 Vue 3 + ECharts，核心就是几个常见图表类型：折线图看客流趋势、柱状图看时段分布、仪表盘看今日完成率、热力图看一周时段热度。

一个简化版的数据接口设计如下：

{ "date": "2025-06-01", "revenue": 5680, "target": 6000, "customers": 312, "orders": 186, "conversion_rate": 0.596, "hot_hours": [{"hour": 12, "count": 45}, {"hour": 13, "count": 52}] }

前端用 ECharts 柱状图展示“时段客流分布”的代码很简单：

// 文件路径：src/components/HourlyChart.vue const option = { xAxis: { type: 'category', data: hours }, yAxis: { type: 'value', name: '客流' }, series: [{ type: 'bar', data: counts, itemStyle: { color: function(params) { const max = Math.max(...counts); return params.value === max ? '#ff6b35' : '#5470c6'; } } }] };

把高峰时段用不同颜色高亮，店主一眼就能看出“中午 12 点最忙”。

4.5.3 经营日报推送

日报由定时任务触发，每天 22:00 统计当天所有数据，用 AI 生成一段 200 字左右的总结和建议，然后推送消息。生成日报的 Prompt 是关键，尽量要求模型给出“可执行建议”，而不是空话。

一个经验是：日报不只给结论，还要给“指标解释”。很多店主不是看不懂数字，而是不知道这个数字意味着什么。比如“今天的成交率是 59.6%”，需要解释为“每 100 个进店顾客里有约 60 人购买了商品，这个比例高于上周平均值 53%，说明今天的促销或店员推荐效果不错”。

5. 数据库与数据流设计

5.1 核心表结构

整个系统的表比较多，这里只设计最核心的几张，用于说明数据组织思路。

店铺表（store）

CREATE TABLE store ( id BIGINT PRIMARY KEY AUTO_INCREMENT, store_name VARCHAR(64) NOT NULL, owner_name VARCHAR(32), open_time TIME, close_time TIME, address VARCHAR(255), longitude DECIMAL(10, 7), latitude DECIMAL(10, 7), created_at DATETIME DEFAULT CURRENT_TIMESTAMP );

客流事件表（traffic_event）

客流是高频写入数据，建议按天分表或者使用 PostgreSQL 做按时间分区。这里给出通用建表结构：

CREATE TABLE traffic_event ( id BIGINT PRIMARY KEY AUTO_INCREMENT, store_id VARCHAR(32) NOT NULL, camera_id VARCHAR(64) NOT NULL, event_type ENUM('enter', 'exit', 'passby') NOT NULL, track_id INT, confidence DECIMAL(5, 4), snapshot_url VARCHAR(255), event_time DATETIME NOT NULL, INDEX idx_store_time (store_id, event_time), INDEX idx_camera_time (camera_id, event_time) );

商品表（product）

CREATE TABLE product ( id BIGINT PRIMARY KEY AUTO_INCREMENT, store_id VARCHAR(32) NOT NULL, sku_code VARCHAR(32) UNIQUE NOT NULL, product_name VARCHAR(128) NOT NULL, category VARCHAR(64), price DECIMAL(10, 2) NOT NULL, cost_price DECIMAL(10, 2), current_stock INT DEFAULT 0, safety_stock INT DEFAULT 10, status TINYINT DEFAULT 1, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );

库存流水表（inventory_flow）

库存流水的价值在于追溯每一次变动来源（销售、补货、盘点、报损）。

CREATE TABLE inventory_flow ( id BIGINT PRIMARY KEY AUTO_INCREMENT, product_id BIGINT NOT NULL, change_type ENUM('sale', 'purchase', 'return', 'damage', 'check') NOT NULL, change_qty INT NOT NULL, before_stock INT, after_stock INT, biz_order_no VARCHAR(64), created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_product_time (product_id, created_at) );

AI 建议表（ai_suggestion）

AI 生成的所有建议都应当落库，目的是方便复盘：模型建议是否被采纳、效果如何。

CREATE TABLE ai_suggestion ( id BIGINT PRIMARY KEY AUTO_INCREMENT, store_id VARCHAR(32) NOT NULL, suggestion_type ENUM('inventory', 'marketing', 'security', 'operation') NOT NULL, suggestion_text TEXT, relate_entity VARCHAR(128), suggested_at DATETIME, status ENUM('pending', 'adopted', 'ignored') DEFAULT 'pending', feedback_result TEXT );

5.2 数据流需要注意的两件事

第一，设备时间同步。摄像头、边缘盒子、收银机都有各自的系统时钟，如果不做 NTP 时钟同步，数据到达服务器后的时间戳可能错乱，直接影响客流高峰分析和库存预测的准确性。设备上报数据时，统一以服务器时间为准，或者强制要求设备开启 NTP。

第二，数据补偿机制。门店网络不可能 100% 稳定。设备离线再上线后，本地缓存的数据要能重新补传。最简单的方式是：边缘盒子本地落一份 SQLite，每条记录增加一个sync_status字段，网络恢复后按时间顺序补传。

6. 完整落地方案：从零搭建最小可运行系统

6.1 项目结构

下面演示的是一个名为wood-ai-manager的后端服务骨架，基于 FastAPI 编写，涵盖“客流上报 + 数据查询 + 日报生成”这条最小链路。

wood-ai-manager/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── models.py # SQLAlchemy 模型 │ ├── schemas.py # Pydantic 请求/响应模型 │ ├── api/ │ │ ├── traffic.py # 客流上报接口 │ │ ├── dashboard.py # 看板数据接口 │ │ └── report.py # 日报生成接口 │ ├── services/ │ │ ├── store_ai.py # AI 汇总逻辑 │ │ └── push.py # 消息推送 │ └── config.py # 配置管理 ├── requirements.txt └── README.md

6.2 核心依赖

fastapi==0.115.0 uvicorn==0.30.6 sqlalchemy==2.0.35 pymysql==1.1.1 pydantic==2.9.2 redis==5.0.8 requests==2.32.3

6.3 FastAPI 入口与配置

# 文件路径：app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api import traffic, dashboard, report from app.config import settings app = FastAPI(title="木头Ai店管家", version="0.1.0") app.add_middleware( CORSMiddleware, allow_origins=settings.allowed_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(traffic.router, prefix="/api/v1/traffic", tags=["客流"]) app.include_router(dashboard.router, prefix="/api/v1/dashboard", tags=["看板"]) app.include_router(report.router, prefix="/api/v1/report", tags=["日报"]) @app.get("/health") def health(): return {"status": "ok"}

# 文件路径：app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str = "mysql+pymysql://root:password@localhost:3306/wood_ai?charset=utf8mb4" redis_url: str = "redis://localhost:6379/0" webhook_url: str = "" allowed_origins: list = ["http://localhost:3000"] class Config: env_file = ".env" settings = Settings()

6.4 模型与请求校验

# 文件路径：app/models.py from sqlalchemy import Column, BigInteger, String, DateTime, Integer, Text from sqlalchemy.orm import declarative_base from sqlalchemy.sql import func Base = declarative_base() class TrafficEvent(Base): __tablename__ = "traffic_event" id = Column(BigInteger, primary_key=True, autoincrement=True) store_id = Column(String(32), index=True) camera_id = Column(String(64)) event_type = Column(String(16)) track_id = Column(Integer, nullable=True) confidence = Column(Integer, nullable=True) event_time = Column(DateTime, server_default=func.now())

# 文件路径：app/schemas.py from pydantic import BaseModel from datetime import datetime class TrafficEventCreate(BaseModel): store_id: str camera_id: str event_type: str track_id: int | None = None confidence: float | None = None event_time: datetime | None = None

6.5 客流上报接口

设备端的逻辑是：检测到一次进出店事件，就向服务器 POST 一条 JSON 数据。这里使用 Redis 做简单的批量缓冲，避免设备短时间上报过多请求打满数据库。

# 文件路径：app/api/traffic.py import json import redis from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from app.schemas import TrafficEventCreate from app.models import TrafficEvent from app.config import settings router = APIRouter() redis_client = redis.Redis.from_url(settings.redis_url) @router.post("/report") def report_traffic(event: TrafficEventCreate): # 先写入 Redis 列表，由后台任务批量落库 redis_client.rpush("traffic:events", json.dumps(event.model_dump(), default=str)) return {"code": 0, "msg": "ok"}

实际的批量落库任务可以用服务启动时挂起的后台线程完成：

# 文件路径：app/services/batch_worker.py import json import time import redis from sqlalchemy.orm import sessionmaker from app.models import TrafficEvent from app.config import settings from app.db import engine SessionLocal = sessionmaker(bind=engine) redis_client = redis.Redis.from_url(settings.redis_url) def batch_insert_worker(): while True: # 每次最多取 500 条数据 pipe = redis_client.pipeline() for _ in range(500): pipe.lpop("traffic:events") items = pipe.execute() events = [] for item in items: if item: data = json.loads(item) events.append(TrafficEvent(**data)) if events: session = SessionLocal() try: session.bulk_save_objects(events) session.commit() finally: session.close() time.sleep(1)

这个小设计的价值在于：客流事件是高频写入类型，数据库单条 INSERT 在高并发下效率低，批量写入能够大幅降低数据库压力。

6.6 日报生成接口

日报生成是“AI 托管”体验的关键节点。以下代码演示如何从数据库汇总当天数据，再调用大模型生成自然语言建议。

# 文件路径：app/api/report.py from fastapi import APIRouter from datetime import date from app.services.store_ai import generate_daily_report router = APIRouter() @router.get("/daily") def daily_report(store_id: str, report_date: date): report = generate_daily_report(store_id, report_date) return {"code": 0, "data": report}

# 文件路径：app/services/store_ai.py from datetime import date from sqlalchemy import func from app.db import SessionLocal from app.models import TrafficEvent def generate_daily_report(store_id: str, report_date: date): session = SessionLocal() enter_count = session.query(func.count(TrafficEvent.id)).filter( TrafficEvent.store_id == store_id, TrafficEvent.event_type == "enter", func.date(TrafficEvent.event_time) == report_date ).scalar() # 其他指标如营收、订单量从对应表查询，这里省略 metrics = { "store_id": store_id, "date": str(report_date), "enter_count": enter_count, "order_count": 0, "revenue": 0, "conversion_rate": 0.0, } session.close() # 这里调用大模型生成日报总结 prompt = f""" 你是这家店铺的经营顾问。今天我店的数据指标如下： - 进店人数：{metrics['enter_count']} - 订单数：{metrics['order_count']} - 营业额：{metrics['revenue']}元 - 转化率：{metrics['conversion_rate']} 请生成一份 200 字以内的中文经营日报，包含： 1. 今日经营状况总结 2. 一个值得关注的现象 3. 一条可落地的明日改善建议 """ # 实际工程中在此调用大模型 SDK # 后续真接入时，会发现 prompt 里指标越具体，输出越有价值 summary = "（示例）今日进店客流平稳，建议明日重点关注午市高峰时段的服务效率。" return {"metrics": metrics, "summary": summary}

6.7 运行与验证

# 启动服务 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

测试客流上报：

curl -X POST http://localhost:8000/api/v1/traffic/report \ -H "Content-Type: application/json" \ -d '{ "store_id": "store_001", "camera_id": "cam_front_01", "event_type": "enter", "track_id": 23, "confidence": 0.92 }'

访问日报接口：

curl "http://localhost:8000/api/v1/report/daily?store_id=store_001&report_date=2025-06-01"

到这里，一个最小可运行的后端闭环就完成了：设备上传统一接口、Redis 缓冲削峰、后台任务批量落库、接口聚合数据、AI 生成建议。后面的工作全部是在这个骨架上填充业务细节。

7. 常见问题与排查思路

在实际开发这套系统的过程中，最常遇到的问题集中在视频识别、数据同步和 AI 输出质量三块。

问题现象	常见原因	解决思路
客流计数严重偏高	单目标被重复计数	给 track_id 增加冷却时间；检查摄像头安装角度是否过小
客流计数严重偏低	人流量大时遮挡严重	调大输入分辨率；换装角度；使用两个摄像头交叉覆盖
夜间误报频繁	树叶晃动、动物、光影变化被识别成“人”	提高置信度阈值；叠加画面差异检测；只在 ROI 区域检测
AI 客服回复了错误信息	知识库内容过时；模型幻觉	限定检索知识；Prompt 中明确约束“不知道就转人工”
库存预测结果波动大	促销日未做标记	给模型增加“促销”特征，或使用节假日参数作为额外回归变量
服务器收到设备数据有延迟	门店网络不稳定	边缘盒子本地缓存 + 断点续传
日报建议过于空泛	大模型 Prompt 缺少结构化指标	把指标拆细，要求先给结论再给证据
消息推送频繁打扰老板	没有告警分级	不同级别走不同渠道；增加冷却时间

这里单独说一下排查流程。遇到类似“客流数据对不上”的问题时，建议从下往上逐层验证：

1. 先用 RTSP 播放器直接看摄像头视频，确定视频画面正常。
2. 把边缘盒子的检测结果输出成带标注的视频，人工比对检测框是否准确。
3. 检查事件判断逻辑，先看是“检测不到”还是“检测到了但没计数”。
4. 查看 Redis 队列积压情况；如果积压严重，检查批量任务是否正常消费。
5. 最后确认数据库记录和接口返回数据是否一致。

8. 最佳实践与工程建议

8.1 模型与算力平衡

视频识别模型不是越大越准，边缘场景要在精度和帧率之间取平衡。YOLOv8n 在 Jetson Orin Nano 上可以跑到接近实时帧率，精度对客流统计也够用；如果换成 YOLOv8s 或更大模型，精度提升有限，但帧率下降明显。建议先跑小模型验证业务闭环，再考虑升级。

8.2 数据是 AI 的燃料

这套系统最大的价值不是代码本身，而是每天积累的数据。没有历史数据，库存预测只能拍脑袋；没有顾客咨询记录，AI 客服的学习无从谈起。所以落地时第一位的工作不是上模型，而是把各种设备的数据接入做扎实，哪怕刚开始用最朴素的方式存下来也可以。

什么意思呢？比如一开始客流统计的置信度阈值设置是 0.6，后来发现某个摄像头环境特殊，阈值要调到 0.7。如果你没有记录每天的“原始检测框数量”“有效进店数”“人工复核数”，就很难判断调整阈值是否真的改善了准确率。生产环境里一定要记录“模型原始输出”和“业务事件”两个层次，方便后续做回溯和调优。

8.3 提示词工程（Prompt Engineering）

AI 经营建议的质量高度依赖 Prompt 结构。推荐一个稳定的三段式模版：背景数据 + 分析要求 + 输出格式限制。

下面是一个推荐的 Prompt 骨架：

你是拥有20年零售管理经验的门店经营顾问。请根据以下今日经营数据进行分析并给出建议。 【今日数据】 - 进店人数：315 人 - 订单数量：186 单 - 成交转化率：59% - 客单价：30.5 元 - 上周同期转化率：53% - 客流高峰时段：12:00-13:00、18:00-19:00 【分析要求】 1. 找出今日与上周同期相比的 2 个关键变化。 2. 结合高峰时段给出排班和活动建议。 3. 所有建议必须是可执行的具体动作，禁止空话。 【输出格式】 中文，200 字以内，分三点输出，每点加粗小标题。

使用这种结构化 Prompt，生成结果的质量会比“帮我写个今日总结”稳定得多。

8.4 权限与安全边界

AI 托管系统会接触大量门店经营隐私数据，包括摄像头画面、顾客对话、销售流水。系统上线前必须做好权限隔离：

• 不同角色只能看对应数据。店长可以看全店经营数据，店员只能看与自己相关的排班和待办。
• 摄像头原始视频是高度敏感数据。建议只留存事件截图和结构化统计结果，不长期保留完整录像；必须保留时，设置明确的过期清理策略。
• 所有具备写操作能力的接口都要走服务端校验，不能在 Web 端直接拼接 SQL。
• 设备上报接口要使用 Token 鉴权，防止伪造数据污染模型训练。

8.5 灰度发布与人工兜底

AI 建议永远只是建议，不能让系统完全自动执行高风险的业务动作。比如自动补货，建议设计成“AI 推荐 + 店长确认”模式，等模型准确率稳定运行几个月后，再逐步放开对低风险商品的自动执行。库存盘点、价格调整这类动作，始终保留人工审批环节。在 AI 系统的落地过程中，“信任”是需要逐步建立的——数据积累越久、建议被采纳后效果被验证的次数越多，系统才能获得更大的自动化权限。

8.6 日志与监控

作为长期运行的线上系统，必须关注三个层面的监控：

• 设备层：摄像头离线率、边缘盒子 CPU/内存使用率、GPU 温度。
• 数据层：数据上报延迟、队列积压数量、缺失率。
• 业务层：日报生成成功率、接口响应时间、AI 建议采纳率。

一旦发现设备离线率超过 5%，就要主动检查硬件和网络，否则 AI 分析的数据基础就会出现偏差。

9. 总结与学习路线

现在回头看《木头Ai店管家》这个项目，它的本质不是某一个 AI 算法，而是一套将视频识别、自然语言处理、时序预测、消息通知整合到一起的门店数字化系统。能把它从 PPT 变成真正可运行的服务的开发者，需要具备的不是某一个方向的深技术，而是全局视野：知道摄像头的数据怎样变成客流报表，知道客流报表怎样驱动库存预测，知道库存预测怎样转化成补货单，最终帮老板把生意做得更轻松。

如果从零开始学习这套技术，推荐的路线是：

第一步，先跑通 FastAPI 或 Spring Boot 的简单 CRUD，掌握数据接入和接口设计。 第二步，学习 YOLOv8 的目标检测和 ByteTrack 跟踪，重点训练自己场景下的行人、车辆、商品检测模型。 第三步，掌握 Prophet 或 XGBoost 做时间序列预测，拿店铺真实流水数据反复测试。 第四步，学习 LangChain 和 RAG，会用向量库管理知识，能做出一版能回答问题的 AI 客服。 第五步，把这几块通过消息队列和定时任务串起来，做成一个完整的自动化闭环。

如果不需要自己开发，而是想直接理解这套系统的价值，那这篇文章的意义在于：下一次当你听到“AI 店铺托管”这个概念时，你会知道它背后是具体的技术栈和工程链路，而不是一个模糊的“魔法盒子”。

技术迭代很快，模型会变，框架会变，但底层的数据闭环思路不会变：先感知，再分析，再决策，最后执行。把这个链路刻在脑子里，无论未来用什么新模型，都能搭出自己的 AI 店管家。