网吧AI助手核心模块 - 基于本地LLM的意图识别与任务执行
import json
import sqlite3
from datetime import datetime, timedelta

class NetbarAIAssistant:
    """网吧AI助手核心类 - 本地化部署版本"""
    
    def __init__(self, db_path="netbar_data.db"):
         本地数据库：数据不上云，完全自主可控
        self.conn = sqlite3.connect(db_path)
        self._init_db()
        
    def _init_db(self):
        """初始化本地会员数据库"""
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS members (
                id INTEGER PRIMARY KEY,
                name TEXT,
                phone TEXT,
                expire_date TEXT,
                project TEXT
            )
        ''')
        self.conn.commit()
    
    def nlu_parse(self, user_input: str):
        """
        模拟NLU意图识别（实际生产环境可接入DeepSeek等LLM API）
        将用户自然语言转化为结构化指令
        """
         简单规则匹配示例，实际可用大模型增强
        if "查" in user_input and "到期" in user_input:
            return {"intent": "query_expiry", "entities": self._extract_name(user_input)}
        elif "续费" in user_input:
            return {"intent": "renew", "entities": self._extract_renew_info(user_input)}
        elif "汇总" in user_input:
            return {"intent": "summary", "entities": {}}
        else:
            return {"intent": "unknown", "entities": {}}
    
    def execute(self, intent_result: dict):
        """执行意图解析后的业务指令"""
        intent = intent_result["intent"]
        if intent == "query_expiry":
            name = intent_result["entities"].get("name")
            return self._query_expiry(name)
        elif intent == "renew":
            return self._do_renew(intent_result["entities"])
        elif intent == "summary":
            return self._daily_summary()
        else:
            return "抱歉，我没理解您的意思，请再说一遍～"
    
    def chat(self, user_input: str):
        """对话式入口 - 像聊天一样操作业务"""
        intent = self.nlu_parse(user_input)
        result = self.execute(intent)
         记录操作日志：AI操作全程可追溯
        self._log_operation(user_input, result)
        return result
    
    def _query_expiry(self, name):
        cursor = self.conn.execute(
            "SELECT name, expire_date FROM members WHERE name LIKE ?", (f"%{name}%",)
        )
        rows = cursor.fetchall()
        if not rows:
            return f"未找到客户 {name}"
        return "\n".join([f"{r[0]} 到期时间：{r[1]}" for r in rows])
    
    def _do_renew(self, entities):
         续费逻辑
        return f"已为 {entities.get('name')} 续费 {entities.get('months', 1)} 个月"
    
    def _daily_summary(self):
        today = datetime.now().strftime("%Y-%m-%d")
        return f"今日到期客户：{self._get_today_expired()} 人"
    
    def _log_operation(self, input_msg, output_msg):
        """操作日志：手动操作和AI操作都能追溯"""
        print(f"[LOG] AI执行 | 输入：{input_msg} | 输出：{output_msg[:50]}")


 使用示例：一句话完成续费查询
if __name__ == "__main__":
    assistant = NetbarAIAssistant()
    
     模拟对话式交互
    print(assistant.chat("查一下先锋网吧加速器什么时候到期"))
    print(assistant.chat("给先锋网吧续费一个月"))
    print(assistant.chat("汇总一下今天有哪些客户到期"))