五金门店AI对话收银系统 - 自学习系统技术方案（纯API版）

import requests
import json

def batch_discover_synonyms(texts: list[str], store_id: int, user_id: int) -> list[dict]:
    """
    批量发现同义词（基于语义聚类）
    texts: 最多50条文本
    store_id: 门店ID（用于权限校验）
    user_id: 用户ID（用于权限校验）
    返回: [{canonical: str, synonyms: list[str], store_id: int, user_id: int, auth_allow: bool}]
    """
    prompt = f"""
你是一个同义词发现专家。以下是用户说的多条意思相同的话：
{json.dumps(texts, ensure_ascii=False)}

请识别它们表达的核心意思（canonical form），并列出所有同义词。

输出格式（JSON）：
{{
  "canonical": "一盒螺丝钉",
  "synonyms": ["一箱螺钉", "螺钉一盒", "螺丝一盒"],
  "intent_hint": "buy",
  "store_id": {store_id},
  "user_id": {user_id},
  "auth_allow": true
}}

注意：
1. canonical是最规范、最常见的说法
2. synonyms是意思完全相同的表达
3. intent_hint是根据语境推测的意图（buy/search/query等）
4. store_id和user_id必须原样传递
5. auth_allow表示当前用户是否有权限执行该操作
6. 只返回JSON，不要其他文字
"""

    response = requests.post(
        "https://api.minimax.chat/v1/text/chatcompletion_pro",
        headers={
            "Authorization": f"Bearer {MINIMAX_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "MiniMax-M2.2-highspeed",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,  # 低温度保证稳定性
            "max_tokens": 500
        }
    )

    result = json.loads(response.json()["choices"][0]["message"]["content"])
    return result

def detect_new_intent(texts: list[str], store_id: int, user_id: int) -> list[dict]:
    """
    检测是否是新意图
    texts: 用户说的多句话
    store_id: 门店ID（用于权限校验）
    user_id: 用户ID（用于权限校验）
    返回: [{text: str, is_new: bool, suggested_intent: str, reason: str, store_id: int, user_id: int, auth_allow: bool}]
    """
    prompt = f"""
你是一个意图识别专家。以下是用户说的话：
{json.dumps(texts, ensure_ascii=False)}

请判断：
1. 每句话属于哪个已知意图（buy/search/query/refund/hello/goodbye等）
2. 如果某句话不属于任何已知意图，标记为NEW意图

已知意图定义：
- buy: 购买商品
- search: 搜索商品
- query: 查询订单/价格/库存
- refund: 退货退款
- hello: 问候
- goodbye: 告别

输出格式（JSON数组）：
[
  {{
    "text": "我要买螺丝钉",
    "is_new": false,
    "intent": "buy",
    "confidence": 0.95,
    "store_id": {store_id},
    "user_id": {user_id},
    "auth_allow": true
  }},
  {{
    "text": "能不能便宜点",
    "is_new": true,
    "suggested_intent": "negotiate",
    "reason": "这是砍价/议价意图，不在已知意图中",
    "store_id": {store_id},
    "user_id": {user_id},
    "auth_allow": true
  }}
]

注意：
1. store_id和user_id必须原样传递
2. auth_allow表示当前用户是否有权限执行该操作
"""

    response = requests.post(
        "https://api.minimax.chat/v1/text/chatcompletion_pro",
        headers={
            "Authorization": f"Bearer {MINIMAX_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "MiniMax-M2.2-highspeed",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.1,
            "max_tokens": 1000
        }
    )

    return json.loads(response.json()["choices"][0]["message"]["content"])

def get_embeddings(texts: list[str], store_id: int, user_id: int) -> list[dict]:
    """
    获取文本的向量表示（用于相似度计算）
    texts: 最多100条文本
    store_id: 门店ID（用于权限校验）
    user_id: 用户ID（用于权限校验）
    返回: [{text: str, embedding: list[float], store_id: int, user_id: int, auth_allow: bool}]
    """
    response = requests.post(
        "https://api.minimax.chat/v1/embeddings",
        headers={
            "Authorization": f"Bearer {MINIMAX_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "embo-01",
            "texts": texts,
            "store_id": store_id,
            "user_id": user_id
        }
    )

    result = response.json()["data"]
    # 补充权限字段
    for i, item in enumerate(result):
        item["store_id"] = store_id
        item["user_id"] = user_id
        item["auth_allow"] = True
    return result

def validate_permission(store_id: int, user_id: int, action: str) -> bool:
    """
    校验用户是否有权限执行该操作
    store_id: 门店ID（多租户隔离）
    user_id: 用户ID
    action: 操作类型
    返回: 是否允许
    """
    # 1. 校验store_id是否属于该用户可访问范围
    # 2. 校验user_id是否有该action权限
    # 3. 返回auth_allow
    pass

# 调用示例
def call_api_with_permission(texts: list[str], store_id: int, user_id: int, action: str):
    auth_allow = validate_permission(store_id, user_id, action)
    if not auth_allow:
        return {"error": "Permission denied", "store_id": store_id, "user_id": user_id, "auth_allow": False}

    # 继续调用API...
    return {"store_id": store_id, "user_id": user_id, "auth_allow": True}

#!/usr/bin/env python3
"""
self-learning: 自学习调度脚本
用法: self-learning [command]
"""

import sys
import os
import json
import time
from datetime import datetime
from typing import Optional, List, Dict

class SelfLearningScheduler:
    """自学习调度器"""

    def __init__(self, store_id: int = None, user_id: int = None):
        self.minimax_api_key = os.environ.get("MINIMAX_API_KEY", "")
        self.store_id = store_id  # 门店ID（权限隔离）
        self.user_id = user_id    # 用户ID（操作审计）
        self.db_config = {
            "host": os.environ.get("DB_HOST", "localhost"),
            "user": os.environ.get("DB_USER", "root"),
            "password": os.environ.get("DB_PASSWORD", ""),
            "database": os.environ.get("DB_NAME", "self_learning")
        }
        self.redis_config = {
            "host": os.environ.get("REDIS_HOST", "localhost"),
            "port": int(os.environ.get("REDIS_PORT", 6379)),
            "password": os.environ.get("REDIS_PASSWORD", "")
        }

    def timestamp(self) -> str:
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def _check_permission(self, action: str) -> bool:
        """检查权限：store_id和user_id必须有值才能执行写操作"""
        if self.store_id is None or self.user_id is None:
            return False
        # 可在此扩展更细粒度的权限校验
        return True

    def sync_pending(self, store_id: int = None, user_id: int = None):
        """同步未识别的反馈"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("sync_pending"):
            raise PermissionError(f"无权限执行 sync_pending: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始同步未识别反馈... store_id={store_id}, user_id={user_id}")
        # 1. 从 feedback_log 读取 intent=NULL 的记录（带权限校验）
        # 2. 批量写入处理队列（Redis list）
        # 3. 记录同步时间戳
        print(f"[{self.timestamp()}] 完成")

    def batch_clustering(self, store_id: int = None, user_id: int = None):
        """批量同义词聚类"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("batch_clustering"):
            raise PermissionError(f"无权限执行 batch_clustering: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始同义词聚类... store_id={store_id}, user_id={user_id}")
        # 1. 读取待处理文本（出现5次以上的，带store_id/user_id过滤）
        # 2. 按意图分组，每组最多50条
        # 3. 批量调用 MiniMax API 聚类
        # 4. 写入 synonym_table（status=pending，带权限字段）
        print(f"[{self.timestamp()}] 完成")

    def detect_new_intents(self, store_id: int = None, user_id: int = None):
        """检测新意图"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("detect_new_intents"):
            raise PermissionError(f"无权限执行 detect_new_intents: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始检测新意图... store_id={store_id}, user_id={user_id}")
        # 1. 读取待分析文本（带权限过滤）
        # 2. 调用 MiniMax 意图检测 API
        # 3. 标记新意图，等待人工确认
        print(f"[{self.timestamp()}] 完成")

    def update_knowledge(self, store_id: int = None, user_id: int = None):
        """更新知识库"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("update_knowledge"):
            raise PermissionError(f"无权限执行 update_knowledge: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始更新知识库... store_id={store_id}, user_id={user_id}")
        # 1. 读取 pending 状态的同义词（带store_id/user_id过滤）
        # 2. 写入/更新 intent_mapping（带权限字段）
        # 3. 记录 model_version 快照（带权限字段）
        print(f"[{self.timestamp()}] 完成")

    def refresh_cache(self, store_id: int = None, user_id: int = None):
        """刷新Redis缓存"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("refresh_cache"):
            raise PermissionError(f"无权限执行 refresh_cache: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始刷新缓存... store_id={store_id}, user_id={user_id}")
        # 1. 连接 Redis
        # 2. 删除旧缓存 keys: intent:synonym:*
        # 3. 从 MySQL 加载新词表（带store_id/user_id过滤）
        # 4. 写入 Redis Hash
        print(f"[{self.timestamp()}] 完成")

    def ab_test_report(self, store_id: int = None, user_id: int = None):
        """A/B测试统计"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        print(f"[{self.timestamp()}] 开始A/B测试统计... store_id={store_id}, user_id={user_id}")
        # 1. 读取过去24小时的 ab_test_result（带权限过滤）
        # 2. 计算各版本准确率
        # 3. 输出报告到日志
        print(f"[{self.timestamp()}] 完成")

    def promote_winner(self, store_id: int = None, user_id: int = None):
        """升级获胜版本"""
        store_id = store_id or self.store_id
        user_id = user_id or self.user_id
        if not self._check_permission("promote_winner"):
            raise PermissionError(f"无权限执行 promote_winner: store_id={store_id}, user_id={user_id}")
        print(f"[{self.timestamp()}] 开始版本切换检查... store_id={store_id}, user_id={user_id}")
        # 1. 如果某版本连续3天领先>5%，提升其流量占比
        # 2. 更新 model_version.traffic_split（带权限字段）
        print(f"[{self.timestamp()}] 完成")

def main():
    command = sys.argv[1] if len(sys.argv) > 1 else "help"
    store_id = int(sys.argv[2]) if len(sys.argv) > 2 else None
    user_id = int(sys.argv[3]) if len(sys.argv) > 3 else None
    scheduler = SelfLearningScheduler(store_id=store_id, user_id=user_id)

    if hasattr(scheduler, command):
        getattr(scheduler, command)()
    else:
        print(f"未知命令: {command}")
        print("可用命令: sync-pending, batch-clustering, detect-new-intents, "
              "update-knowledge, refresh-cache, ab-test-report, promote-winner")
        print("示例: self-learning sync-pending 314 23")

if __name__ == "__main__":
    main()

# 1. 批量处理减少API调用次数
# 坏：每条文本单独调用
# 好：每次批量50-100条

# 2. 缓存复用
# 用Redis缓存Embedding结果
# 相同文本只计算一次

# 3. 关键词快速过滤
# 先用关键词匹配，命中则不调用API
# 只对模糊输入调用API

# 4. 闲时处理
# 聚类任务安排在凌晨2点
# 避开API限流时段