门店层级多租户AI智能体架构设计
版本：v1.0
日期：2026-04-30
概述：一个门店为一个集合，门店内不同人员共享权限允许的记忆和技能，同时保留私有记忆，不同门店之间完全独立隔离
一、整体架构
1.1 架构拓扑图
┌─────────────────────────────────────────────────────────────────┐
│ Store A（门店A·完全隔离） │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 员工小张 │ │ 员工小李 │ │ 员工小王 │ │
│ │ ├─私有记忆 │ │ ├─私有记忆 │ │ ├─私有记忆 │ │
│ │ └─私有技能 │ │ └─私有技能 │ │ └─私有技能 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ ▲ ▲ ▲ │
│ └──────────────┴──────────────┘ │
│ 共享层（门店A） │
│ ├─ 门店共享记忆（客户资料/商品档案） │
│ └─ 门店共享技能（销售话术/盘点流程） │
└─────────────────────────────────────────────────────────────────┘
│
✕（完全隔离）
│
┌─────────────────────────────────────────────────────────────────┐
│ Store B（门店B·完全隔离） │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ 员工小赵 │ │ 员工小周 │ ... │
│ │ ├─私有记忆 │ │ ├─私有记忆 │ │
│ │ └─私有技能 │ │ └─私有技能 │ │
│ └─────────────┘ └─────────────┘ │
│ ▲ ▲ │
│ └──────────────┴────── 共享层（门店B） │
└─────────────────────────────────────────────────────────────────┘
1.2 设计原则
| 原则 | 说明 |
|---|---|
| 门店隔离 | 不同门店之间完全隔离，不可互相访问对方数据 |
| 层级记忆 | L1私有 > L2门店共享 > L3系统全局，加载优先级递减 |
| 技能继承 | 员工技能 = 私有技能 + 门店共享技能 + 全局技能 |
| 单一实例 | 一个 Hermes Gateway 实例，通过 session 隔离实现逻辑多租户 |
二、记忆层级设计
2.1 三层记忆模型
| 层级 | 名称 | 可见范围 | 可写范围 | 用途 |
|---|---|---|---|---|
| L1 | 员工私有记忆 | 仅员工本人 | 仅员工本人 | 个人偏好、工作习惯、私人备注 |
| L2 | 门店共享记忆 | 门店全员 | 门店全员 | 客户资料、商品档案、门店配置 |
| L3 | 系统全局记忆 | 全系统 | 管理员 | 通用五金知识、公司政策 |
2.2 加载优先级
用户提问
│
▼
L1 员工私有记忆 ──▶ 注入（最高优先级，覆盖后续同名key）
L2 门店共享记忆 ──▶ 注入
L3 系统全局记忆 ──▶ 注入（最低优先级）
│
▼
AIAgent 处理
注意：同一 memory_key 高优先级会覆盖低优先级。例如员工设置了私有别名"L1_商品A库存=100"，则该员工的回答中以此为准。
2.3 记忆示例
L1 员工私有记忆：
员工小张:
- 偏好: 喜欢用语音播报库存预警
- 习惯: 每周一上午盘库
- 备注客户: "李总喜欢抹零"
L2 门店共享记忆：
门店A共享:
- 客户档案: { "李总": { "信用额度": 50000, "账期": 30天 } }
- 商品档案: { "螺纹钢Φ12": { "安全库存": 200, "供应商": "张三供应商" } }
- 门店配置: { "营业时间": "8:00-20:00", "负责人": "王店长" }
L3 系统全局记忆：
系统全局:
- 五金行业知识库
- 公司合规政策
- 系统公告
三、Session 路由策略
3.1 路由逻辑
# 微信用户 → 映射到唯一的 (store_id, employee_id)
# session key = f"wechat_{openid}"
# 消息路由流程
gateway:
wechat:
# 解析消息中的 openid
extract_openid: true
session:
# 自动创建新 session
auto_create: true
# session 过期时间（小时）
ttl: 720 # 30天
memory:
# 加载顺序：员工私有 → 门店共享 → 系统全局
load_order:
- scope: "employee" # L1
source: "memory_employee"
- scope: "store" # L2
source: "memory_store"
- scope: "global" # L3
source: "memory_global"
# 隔离规则
isolation:
# 门店间完全隔离
store_boundary: "strict"
# 员工间：私有隔离，共享互通
employee_rule: "private_isolated_shared_public"
3.2 用户映射表
CREATE TABLE user_store_mapping (
openid VARCHAR(64) PRIMARY KEY, -- 微信 openid
store_id VARCHAR(32) NOT NULL, -- 门店ID
employee_id VARCHAR(32) NOT NULL, -- 员工ID（门店内唯一）
employee_name VARCHAR(64), -- 员工姓名
role ENUM('owner','manager','staff'), -- 角色
created_at DATETIME DEFAULT NOW(),
updated_at DATETIME DEFAULT NOW() ON UPDATE NOW(),
-- 联合唯一约束
UNIQUE KEY uk_store_employee (store_id, employee_id)
);
四、数据模型
4.1 记忆表设计
-- L1: 员工私有记忆（仅本人可见/可写）
CREATE TABLE memory_employee (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
employee_id VARCHAR(32) NOT NULL,
memory_key VARCHAR(128) NOT NULL,
memory_value TEXT NOT NULL,
created_at DATETIME DEFAULT NOW(),
updated_at DATETIME DEFAULT NOW() ON UPDATE NOW(),
UNIQUE KEY uk_employee_key (employee_id, memory_key),
INDEX idx_employee (employee_id)
);
-- L2: 门店共享记忆（门店全员可见/可写）
CREATE TABLE memory_store (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
store_id VARCHAR(32) NOT NULL,
memory_key VARCHAR(128) NOT NULL,
memory_value TEXT NOT NULL,
updated_by VARCHAR(32), -- 最后更新人
created_at DATETIME DEFAULT NOW(),
updated_at DATETIME DEFAULT NOW() ON UPDATE NOW(),
UNIQUE KEY uk_store_key (store_id, memory_key),
INDEX idx_store (store_id)
);
-- L3: 系统全局记忆（所有门店可见/仅管理员可写）
CREATE TABLE memory_global (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
memory_key VARCHAR(128) NOT NULL UNIQUE,
memory_value TEXT NOT NULL,
updated_by VARCHAR(32), -- 最后更新管理员
created_at DATETIME DEFAULT NOW(),
updated_at DATETIME DEFAULT NOW() ON UPDATE NOW(),
UNIQUE KEY uk_global_key (memory_key)
);
4.2 技能表设计
-- 技能定义表
CREATE TABLE skills (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
skill_code VARCHAR(64) NOT NULL UNIQUE,
skill_name VARCHAR(128),
skill_type ENUM('global','store','employee') NOT NULL,
scope_id VARCHAR(32), -- store_id 或 employee_id，global 为 NULL
content TEXT, -- 技能内容或脚本路径
enabled TINYINT DEFAULT 1,
created_at DATETIME DEFAULT NOW(),
updated_at DATETIME DEFAULT NOW() ON UPDATE NOW()
);
-- 技能示例
INSERT INTO skills (skill_code, skill_name, skill_type, scope_id) VALUES
-- L3 全局技能
('global_wujin_knowledge', '五金行业知识库', 'global', NULL),
('global_policy', '公司合规政策', 'global', NULL),
-- L2 门店共享技能
('store_sales_script', '销售话术模板', 'store', 'STORE_A'),
('store_inventory_process', '盘点流程指引', 'store', 'STORE_A'),
-- L1 员工私有技能
('employee_personal_note', '个人快捷备注', 'employee', 'EMP_A001');
五、技能共享设计
5.1 技能加载层级
skills:
# L3: 系统全局技能（所有门店所有员工）
global:
- 五金产品知识库
- 公司合规政策
- 通用AI能力
# L2: 门店共享技能（门店全员可用）
store_shared:
- 五金产品知识库（门店版）
- 销售话术模板
- 盘点流程指引
- 库存预警规则
# L1: 员工私有技能（仅本人可用）
employee_private:
- 个人快捷命令
- 私人笔记
- 个人工作习惯配置
5.2 技能加载算法
def load_employee_skills(employee_id: str, store_id: str) -> list:
"""加载某员工的完整技能列表"""
skills = []
# L3: 全局技能
global_skills = db.query(
"SELECT * FROM skills WHERE skill_type = 'global' AND enabled = 1"
)
skills.extend(global_skills)
# L2: 门店共享技能
store_skills = db.query(
"SELECT * FROM skills WHERE skill_type = 'store' AND scope_id = ? AND enabled = 1",
store_id
)
skills.extend(store_skills)
# L1: 员工私有技能
emp_skills = db.query(
"SELECT * FROM skills WHERE skill_type = 'employee' AND scope_id = ? AND enabled = 1",
employee_id
)
skills.extend(emp_skills)
return skills
六、隔离规则总结
6.1 访问控制矩阵
| 操作 | 本店员工 | 跨店员工 | 管理员 |
|---|---|---|---|
| 读自己的私有记忆 | ✅ | ✅ | ✅ |
| 写自己的私有记忆 | ✅ | ✅ | ✅ |
| 读本店共享记忆 | ✅ | ❌ | ✅ |
| 写本店共享记忆 | ✅ | ❌ | ✅ |
| 读他店共享记忆 | ❌ | ❌ | ❌ |
| 写他店共享记忆 | ❌ | ❌ | ❌ |
| 读系统全局记忆 | ✅ | ✅ | ✅ |
| 写系统全局记忆 | ❌ | ❌ | ✅ |
6.2 记忆隔离示意
┌─────────────────────────────────────────────────────────────┐
│ 系统全局 (L3) │
│ [ 全员可见 ] ─────────────────────────────────────────▶ │
│ [ 仅管理员写 ] │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 门店A共享层 (L2) │
│ │
│ 员工A ──▶ 可见 ✅ ──▶ 可写 ✅ │
│ 员工B ──▶ 可见 ✅ ──▶ 可写 ✅ │
│ 员工C ──▶ 可见 ✅ ──▶ 可写 ✅ │
│ │
│ 门店B ──▶ 可见 ❌ ──▶ 可写 ❌ │
│ 门店C ──▶ 可见 ❌ ──▶ 可写 ❌ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 员工A私有层 (L1) │
│ │
│ 员工A ──▶ 可见 ✅ ──▶ 可写 ✅ │
│ 员工B ──▶ 可见 ❌ ──▶ 可写 ❌ │
│ 门店B全员 ──▶ 可见 ❌ ──▶ 可写 ❌ │
└─────────────────────────────────────────────────────────────┘
七、技术实现架构
7.1 系统拓扑
微信用户消息
│
▼
┌──────────────────────────────────────────────────────────────┐
│ Hermes Gateway（单实例） │
│ 解析消息 → 提取 openid → 查询用户档案 │
└────────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────┐
│ Session Router │
│ │
│ session_key = "wechat_{openid}" │
│ 自动创建 / 自动 resume │
│ │
│ → AIAgent 处理（单一实例，session 隔离） │
└────────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────┐
│ 上下文注入层（记忆加载） │
│ │
│ 1. 查 user_store_mapping 获取 (store_id, employee_id) │
│ 2. 加载 L1 私有记忆（employee_id） │
│ 3. 加载 L2 门店共享记忆（store_id） │
│ 4. 加载 L3 系统全局记忆 │
│ │
│ 注入顺序：L1 → L2 → L3 │
└────────────────────────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────┐
│ AIAgent │
│ 基于注入的上下文进行处理 │
│ 回复中不会泄露其他门店/员工的私密信息 │
└──────────────────────────────────────────────────────────────┘
7.2 核心代码示例
# AIAgent 启动时加载用户上下文
def load_user_context(openid: str) -> dict:
"""根据 openid 加载完整上下文"""
# 1. 查用户属于哪个门店
user = db.query_one(
"SELECT * FROM user_store_mapping WHERE openid = ?",
openid
)
if not user:
raise ValueError(f"未知用户: {openid}")
# 2. 加载三层记忆
private_memory = db.query(
"SELECT memory_key, memory_value FROM memory_employee "
"WHERE employee_id = ?",
user.employee_id
)
store_memory = db.query(
"SELECT memory_key, memory_value, updated_by FROM memory_store "
"WHERE store_id = ?",
user.store_id
)
global_memory = db.query(
"SELECT memory_key, memory_value FROM memory_global"
)
# 3. 加载技能
skills = load_employee_skills(user.employee_id, user.store_id)
# 4. 构建上下文
return {
"employee_id": user.employee_id,
"employee_name": user.employee_name,
"store_id": user.store_id,
"role": user.role,
"private_memory": private_memory, # L1
"store_memory": store_memory, # L2
"global_memory": global_memory, # L3
"skills": skills,
}
# 系统提示词组装
def build_system_prompt(ctx: dict) -> str:
"""组装带记忆层级的系统提示词"""
prompt_parts = []
# 1. 基础身份
prompt_parts.append(f"""
你是「{ctx['store_id']}门店」的「{ctx['employee_name']}」（{ctx['role']}角色）。
当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%s')}
""")
# 2. L3 系统全局（最先注入，最低优先级）
if ctx['global_memory']:
prompt_parts.append("【系统全局知识】")
for row in ctx['global_memory']:
prompt_parts.append(f" {row['memory_key']}: {row['memory_value']}")
# 3. L2 门店共享
if ctx['store_memory']:
prompt_parts.append("【本店共享资料】（全员可见，可更新）")
for row in ctx['store_memory']:
prompt_parts.append(f" {row['memory_key']}: {row['memory_value']} [更新人:{row['updated_by']}]")
# 4. L1 员工私有（最后注入，最高优先级）
if ctx['private_memory']:
prompt_parts.append("【你的私人笔记】（仅你可见）")
for row in ctx['private_memory']:
prompt_parts.append(f" {row['memory_key']}: {row['memory_value']}")
return "\n".join(prompt_parts)
八、跨门店协作（可选扩展）
8.1 临时借调场景
员工临时借调到其他门店时，授予临时访问权限：
-- 临时跨店权限表
CREATE TABLE cross_store_access (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
employee_id VARCHAR(32) NOT NULL,
store_id VARCHAR(32) NOT NULL, -- 被借调到的门店
access_level ENUM('read','read_write') DEFAULT 'read',
valid_from DATE NOT NULL,
valid_to DATE NOT NULL,
reason VARCHAR(256),
approved_by VARCHAR(32),
created_at DATETIME DEFAULT NOW(),
INDEX idx_employee (employee_id),
INDEX idx_store (store_id),
INDEX idx_valid (valid_from, valid_to)
);
def has_cross_store_access(employee_id: str, store_id: str) -> bool:
"""检查员工是否有跨店访问权限"""
result = db.query_one("""
SELECT access_level FROM cross_store_access
WHERE employee_id = ?
AND store_id = ?
AND valid_from <= CURDATE()
AND valid_to >= CURDATE()
""", employee_id, store_id)
return result is not None
九、部署架构
9.1 单实例多租户部署
┌─────────────────────────────────────────────────────────────┐
│ Hermes Gateway（单实例） │
│ │
│ 一个 Gateway 进程，服务所有门店 │
│ 通过 session 隔离 + 记忆层级注入实现逻辑多租户 │
│ │
│ ┌─ memory_employee (L1) ─── per employee │
│ ├─ memory_store (L2) ────── per store │
│ └─ memory_global (L3) ─────── global │
└────────────────────────┬──────────────────────────────────┘
│
▼
┌──────────────────────┐
│ MySQL（主数据库） │
│ │
│ user_store_mapping │
│ memory_employee (L1) │
│ memory_store (L2) │
│ memory_global (L3) │
│ skills │
│ cross_store_access │
└──────────────────────┘
9.2 优势
| 优势 | 说明 |
|---|---|
| 资源节省 | 一个 Gateway 实例，服务所有门店 |
| 运维简单 | 无需为每个门店部署独立服务 |
| 线性扩展 | 新增门店无需重启服务，配置即可 |
| 逻辑隔离 | session 隔离 + 记忆层级，安全可靠 |
十、与现有架构结合
10.1 集成你的 MySQL + ClickHouse 架构
当前你已有 MySQL 主库和 ClickHouse 分析库，这套多租户架构可以复用：
┌──────────────────────────────────────────────────────────────┐
│ MySQL 主库 │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 业务表 │ │ 多租户表 │ │
│ │ (orders等) │ │ (本架构新增) │ │
│ └─────────────────┘ └─────────────────┘ │
└────────────────────────┬───────────────────────────────────┘
│ DTS CDC
▼
┌──────────────────────────────────────────────────────────────┐
│ ClickHouse（分析库） │
│ │
│ - 订单统计（按门店隔离查询） │
│ - 库存预测 │
│ - 财务月报 │
│ - 供应商/客户分析 │
│ │
│ WHERE store_id = 'STORE_A' -- 自然隔离 │
└──────────────────────────────────────────────────────────────┘
10.2 实施步骤
| 阶段 | 时间 | 任务 |
|---|---|---|
| Phase 1 | 第1周 | 数据库表创建（user_store_mapping, memory_*, skills等） |
| Phase 2 | 第2周 | Hermes Gateway session 路由改造 |
| Phase 3 | 第3周 | 记忆加载层开发（L1/L2/L3注入逻辑） |
| Phase 4 | 第4周 | 技能加载层开发 |
| Phase 5 | 第5周 | 跨店协作（可选）开发 |
| Phase 6 | 第6周 | 测试与调优 |
文档版本：v1.0
更新日期：2026-04-30
架构：Hermes Gateway 单实例 + MySQL 多租户记忆层