1. 项目概述当SQLite遇见二次智能最近在折腾一个挺有意思的玩意儿我把它叫做“基于SQLite的QIS结果路由”。简单说就是让任何一台设备——你的笔记本电脑、手机甚至是树莓派——都能在不依赖任何外部服务器的情况下跑起一套具备“二次智能”的决策系统。这听起来有点玄乎但核心逻辑其实很接地气把复杂的决策逻辑从云端黑盒里拽出来塞进一个轻量级的SQLite数据库里让它在你手边的设备上就能独立运行、自主判断。“QIS”在这里指的是“二次智能系统”。它不是要替代大语言模型那种生成式AI而是专注于另一件事基于既定规则和实时数据进行多轮、递进式的推理与决策。想象一下你不是在问AI“写一首诗”而是在设计一个系统它能根据当前的天气、你的日程、交通状况和历史偏好自动为你规划出今天最佳的出行路线和衣物搭配。这个决策过程往往不是一步到位的它需要像人一样先评估A条件得出中间结论再结合B条件进行二次判断最终形成一个综合方案——这就是“二次”智能的意味。而“结果路由”则是这个系统的神经中枢。它负责管理决策流程当一个事件触发比如“早上8点”系统需要调用哪些规则这些规则产生的初步结论应该流向哪个下一步的判断节点最终的结果又该如何执行或呈现传统做法要么写死在一大坨if-else代码里难以维护要么依赖云端微服务编排延迟和隐私都是问题。所以这个项目的核心创新点就是用SQLite来承载这一切。为什么是SQLite因为它无处不在从安卓、iOS到Windows、macOS乃至各种嵌入式系统它几乎是唯一一个“开箱即用”、零配置的数据库引擎。它不需要单独安装数据库服务一个文件就是全部。我们把智能决策的规则库、状态机、工作流定义全部用关系型数据表来建模和存储。这样一来任何能运行SQLite的设备就瞬间获得了一个本地的、私密的、可离线运行的智能决策引擎。没有网络延迟没有服务器费用没有数据出域的风险真正的“Any Device, No Server Required”。2. 核心架构设计用关系模型表达智能逻辑把智能逻辑塞进数据库表这听起来像是个行为艺术但当你拆解开来会发现关系模型其实异常贴合。关键在于我们要对“智能决策”这个过程进行解构和重塑。2.1 逻辑解构从过程式代码到声明式数据传统的决策逻辑通常是过程式的。比如用Python写可能就是一系列嵌套的判断语句if weather ‘rain‘: if schedule ‘meeting‘: decision ‘take_umbrella_and_leave_early‘ else: decision ‘stay_home‘ elif temperature 30: decision ‘wear_tshirt‘ ...这种代码的缺点是僵化。每加一个新条件比如“空气质量指数”你都得去修改代码逻辑很容易出错也难以直观地看到所有规则的全貌。我们的思路是将其声明化、数据化。我们把一条决策规则拆解成几个核心组成部分触发条件什么情况下这条规则该被评估例如事件类型“早晨唤醒” 且 时间 07:00评估因子需要检查哪些数据例如weather.today.outlook, calendar.events.first.start_time判断逻辑因子之间满足什么关系例如outlook ‘rainy‘ AND start_time - current_time 30分钟输出结果如果满足产生什么结论例如action ‘suggest_umbrella‘, priority ‘high‘结果路由这个结论应该交给谁哪个下游规则做进一步处理例如route_to ‘commute_planning_rule‘这样一来每一条规则就变成了数据库里的一行记录。整个决策知识库就是一个可以随时增删改查的数据表。2.2 数据表设计构建规则引擎的骨架基于上面的解构我设计了几个核心表。这里以SQLite的建表语句为例你可以清晰地看到整个模型的骨架。1. 规则定义表 (decision_rules)这是大脑皮层存储所有具体的规则。CREATE TABLE decision_rules ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_name TEXT NOT NULL UNIQUE, -- 规则唯一标识如 ‘morning_weather_check‘ description TEXT, -- 规则描述 trigger_event TEXT, -- 触发事件类型如 ‘scheduled‘, ‘sensor_update‘, ‘user_input‘ trigger_condition_sql TEXT, -- 用SQL片段描述的触发条件灵活且强大 is_active BOOLEAN DEFAULT 1, -- 是否启用 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );注意trigger_condition_sql字段是这个设计的精髓。它存储的是一段像“SELECT 1 FROM sensor_data WHERE temperature 30 AND timestamp datetime(‘now‘, ‘-5 minutes‘)”这样的SQL片段。系统会动态执行它如果返回有结果则触发该规则。这赋予了规则极大的动态性可以直接查询最新的设备传感器数据或其他状态表。2. 规则因子表 (rule_factors)这是神经突触定义每条规则需要关注哪些输入数据。CREATE TABLE rule_factors ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id INTEGER NOT NULL, factor_key TEXT NOT NULL, -- 因子键名如 ‘current_temperature‘ data_source TEXT NOT NULL, -- 数据来源如 ‘table:sensor_readings‘, ‘env:USER_LOCATION‘ retrieval_sql TEXT, -- 如何获取该因子的SQL可为空若data_source已明确 FOREIGN KEY (rule_id) REFERENCES decision_rules(id) ON DELETE CASCADE );实操心得data_source的设计支持多种形式。table:前缀表示从其他表查询env:表示从程序运行环境变量或上下文中获取甚至可以是http:虽然离线场景慎用或file:。这使得系统能融合设备本地数据库、系统状态、配置文件等多种数据源。3. 判断逻辑与动作表 (rule_logic_actions)这是决策神经元将因子关联起来并指定满足条件后的输出和流向。CREATE TABLE rule_logic_actions ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id INTEGER NOT NULL, logic_expression TEXT NOT NULL, -- 逻辑表达式如 ‘?1 “rainy“ AND ?2 30‘ action_type TEXT NOT NULL, -- 动作类型 ‘set_output‘, ‘call_function‘, ‘route_to‘ action_value TEXT NOT NULL, -- 动作值输出内容、函数名、或下游规则名 route_target_rule_name TEXT, -- 如果action_type是‘route_to‘这里指定目标规则名 priority INTEGER DEFAULT 0, -- 同一规则内多条逻辑的优先级 FOREIGN KEY (rule_id) REFERENCES decision_rules(id) ON DELETE CASCADE );这里的logic_expression使用占位符?1、?2它们会按顺序被rule_factors检索出的实际值替换。然后系统会用一个小型的表达式求值器比如用Python的eval在安全沙箱内或自己实现一个简单的解析器来计算布尔结果。2.3 工作流引擎SQL驱动的状态机单个规则是神经元多个规则通过路由串联起来就形成了工作流或状态机。rule_logic_actions表中的route_target_rule_name字段就是定义路由的关键。整个决策过程由一个执行引擎驱动它本质上是一个循环监听trigger_event如定时器、数据插入事件。根据事件扫描decision_rules表找到所有被触发的、活跃的规则。对每条规则获取其所有因子 (rule_factors) 的当前值。按优先级评估该规则下的所有逻辑 (rule_logic_actions)。执行第一个评估为真的逻辑所定义的动作。如果动作是route_to则将route_target_rule_name作为新的触发事件或直接将其id加入待执行队列跳转到步骤2形成链式或树状的推理。如果是set_output则将该结果存入结果表并可能结束当前推理链。这个引擎可以用任何支持SQLite的语言实现Python, JavaScript, Rust等核心逻辑就是拼装和执行SQL以及一个简单的表达式求值循环。整个系统的“智能”就体现在这些表数据的关系和流动中。3. 实现细节与核心代码拆解理论说得再多不如一行代码。下面我用Python因其易读性来展示这个路由引擎的核心实现片段。选择Python的sqlite3标准库意味着这个方案在绝大多数设备上都能零依赖运行。3.1 引擎初始化与规则加载首先我们需要一个引擎类来管理数据库连接和规则缓存。import sqlite3 import threading from typing import Any, Dict, List, Optional import json import re class QIS_SQLiteEngine: def __init__(self, db_path: str ‘:memory:‘): 初始化引擎。db_path为SQLite数据库文件路径‘:memory:‘表示内存数据库。 self.db_path db_path # 使用check_same_threadFalse需谨慎这里为简单演示。生产环境建议用连接池。 self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.row_factory sqlite3.Row # 返回字典样式的行 self._rule_cache {} # 缓存规则定义避免频繁查询 self._lock threading.RLock() # 简单线程锁 self._init_tables() # 确保表存在 def _init_tables(self): 创建核心数据表如果不存在。 # 这里省略了完整的建表SQL与上一节设计一致 cursor self.conn.cursor() cursor.executescript(‘‘‘ CREATE TABLE IF NOT EXISTS decision_rules (...); CREATE TABLE IF NOT EXISTS rule_factors (...); CREATE TABLE IF NOT EXISTS rule_logic_actions (...); CREATE TABLE IF NOT EXISTS execution_logs (...); -- 用于日志记录 ‘‘‘) self.conn.commit()注意事项SQLite的默认连接对象不是线程安全的。在简单的单线程脚本或像idle这样的场景下没问题但如果你的应用可能多线程访问数据库需要使用锁如threading.RLock()来保护连接或者每个线程创建自己的连接。这里为了代码清晰使用了check_same_threadFalse并配合一个简单的锁这适用于轻量级并发。对于更复杂的场景建议采用连接池模式。3.2 核心执行循环触发、评估、路由这是引擎最核心的方法它实现了一次完整的规则触发与执行流程。def execute_on_trigger(self, trigger_event: str, context: Optional[Dict] None) - List[Dict]: 根据触发事件执行相关规则。 Args: trigger_event: 触发事件类型如 ‘scheduled_morning‘ context: 可选执行上下文可用于向因子注入额外数据 Returns: 本次执行产生的所有输出结果列表 if context is None: context {} all_outputs [] with self._lock: # 1. 查找被触发的活跃规则 cursor self.conn.cursor() # 注意这里演示了trigger_condition_sql的动态执行。实际使用需严格防范SQL注入。 # 我们假设trigger_condition_sql来自可信的配置源。 cursor.execute(‘‘‘ SELECT id, rule_name, trigger_condition_sql FROM decision_rules WHERE trigger_event ? AND is_active 1 ‘‘‘, (trigger_event,)) triggered_rules cursor.fetchall() for rule_row in triggered_rules: rule_id, rule_name, condition_sql rule_row # 2. 检查触发条件SQL如果存在 should_execute True if condition_sql: try: # 执行条件SQL期望返回至少一行则认为触发 cond_cursor self.conn.execute(condition_sql) should_execute cond_cursor.fetchone() is not None except sqlite3.Error as e: print(f“规则 ‘{rule_name}‘ 触发条件SQL执行错误: {e}“) should_execute False if not should_execute: continue # 3. 执行这条规则 rule_output self._execute_single_rule(rule_id, rule_name, context) if rule_output: all_outputs.append(rule_output) # 4. 处理路由如果输出指定了下一个规则则递归或迭代执行 if ‘next_rule‘ in rule_output: next_trigger rule_output[‘next_rule‘] # 避免无限循环可以在这里加入已执行规则栈的检查 deeper_outputs self.execute_on_trigger(next_trigger, context) all_outputs.extend(deeper_outputs) return all_outputs3.3 单条规则执行因子获取与逻辑判断_execute_single_rule方法是智能发生的具体场所。def _execute_single_rule(self, rule_id: int, rule_name: str, context: Dict) - Optional[Dict]: 执行单条规则返回输出字典或None。 cursor self.conn.cursor() # 1. 获取本规则的所有因子 cursor.execute(‘SELECT factor_key, data_source, retrieval_sql FROM rule_factors WHERE rule_id ? ORDER BY id‘, (rule_id,)) factors cursor.fetchall() factor_values {} for f_key, f_source, f_sql in factors: value self._retrieve_factor_value(f_source, f_sql, context) factor_values[f_key] value # 2. 获取本规则的所有逻辑-动作对按优先级排序 cursor.execute(‘‘‘ SELECT logic_expression, action_type, action_value, route_target_rule_name FROM rule_logic_actions WHERE rule_id ? ORDER BY priority ASC ‘‘‘, (rule_id,)) logic_actions cursor.fetchall() # 3. 按顺序评估逻辑表达式 for logic_expr, action_type, action_value, route_target in logic_actions: if self._evaluate_logic(logic_expr, factor_values): # 逻辑为真执行动作 output {‘rule‘: rule_name, ‘action_type‘: action_type, ‘value‘: action_value} if action_type ‘set_output‘: output[‘result‘] action_value self._log_execution(rule_id, True, logic_expr, output) return output elif action_type ‘route_to‘ and route_target: output[‘next_rule‘] route_target self._log_execution(rule_id, True, logic_expr, output) return output elif action_type ‘call_function‘: # 这里可以集成一个安全的函数调用机制 # 例如预定义一些函数名到实际函数的映射 print(f“规则 ‘{rule_name}‘ 调用函数: {action_value}“) output[‘function_called‘] action_value self._log_execution(rule_id, True, logic_expr, output) return output # 找到第一个为真的逻辑就执行并返回 break else: # 所有逻辑都不为真 self._log_execution(rule_id, False, None, None) return None3.4 关键支撑方法因子获取与逻辑求值_retrieve_factor_value和_evaluate_logic是两个关键辅助函数它们决定了系统的灵活性和安全性。def _retrieve_factor_value(self, data_source: str, retrieval_sql: Optional[str], context: Dict) - Any: 根据数据源描述获取因子值。 # 处理上下文中的覆盖值用于测试或动态注入 if data_source.startswith(‘ctx:‘): key data_source[4:] return context.get(key) # 从数据库表查询 elif data_source.startswith(‘table:‘): if retrieval_sql: try: cursor self.conn.execute(retrieval_sql) row cursor.fetchone() # 假设查询返回单行单列的值或做相应处理 return row[0] if row else None except sqlite3.Error as e: print(f“因子查询SQL错误 [{data_source}]: {e}“) return None else: # 如果没有提供SQL尝试从source中解析表名和字段简化处理 # 例如 ‘table:weather.current_temp‘ parts data_source.split(‘:‘)[1].split(‘.‘) if len(parts) 2: table, column parts cursor self.conn.execute(f‘SELECT {column} FROM {table} ORDER BY ROWID DESC LIMIT 1‘) row cursor.fetchone() return row[0] if row else None return None # 环境变量或固定值示例 elif data_source.startswith(‘env:‘): import os key data_source[4:] return os.environ.get(key) elif data_source.startswith(‘literal:‘): # 处理字面量如 ‘literal:42‘, ‘literal:“hello“‘ value_str data_source[8:] try: return json.loads(value_str) # 尝试解析JSON except: return value_str # 否则返回字符串 else: # 未知数据源 return None def _evaluate_logic(self, logic_expr: str, factor_values: Dict[str, Any]) - bool: 评估逻辑表达式。这是一个简化的安全实现。 if not logic_expr: return False # 将占位符 ?1, ?2 替换为实际值 # 首先我们需要知道占位符的顺序。这里简单按出现顺序替换。 # 更健壮的做法是在存储时解析并保存参数顺序。 expr_to_eval logic_expr value_list list(factor_values.values()) # 简单的占位符替换生产环境需要更严谨的解析 for i, val in enumerate(value_list, start1): placeholder f‘?{i}‘ # 将值安全地转换为表达式中的字面量 if isinstance(val, str): # 字符串需要加引号并转义内部引号 safe_val json.dumps(val) elif val is None: safe_val ‘None‘ else: safe_val str(val) expr_to_eval expr_to_eval.replace(placeholder, safe_val) # **安全警告**直接使用eval是危险的如果logic_expr来自不可信源。 # 此处仅为演示。生产环境应 # 1. 使用限制性的解析器如ast.literal_eval配合自定义操作符。 # 2. 或使用像 simpleeval 这样的安全库。 # 3. 或自己实现一个简单的表达式解析器支持 , !, , , AND, OR等。 try: # 极度简化的安全措施只允许有限的字符 if re.match(r‘^[0-9\s\.\\‘\[\]\{\}\(\)\\-\*/!|NoneTrueFalse\s]$‘, expr_to_eval): result eval(expr_to_eval, {“__builtins__“: {}}, {}) return bool(result) else: print(f“表达式包含不安全字符: {logic_expr}“) return False except Exception as e: print(f“逻辑表达式求值错误 ‘{expr_to_eval}‘: {e}“) return False核心安全提示_evaluate_logic函数中的eval使用是本演示最大的安全隐患。逻辑表达式来自数据库如果被恶意篡改可能执行任意代码。在生产环境中你必须替换这部分。推荐方案使用ast.literal_eval并配合一个自定义的、安全的操作符映射字典只允许比较和逻辑运算。引入simpleeval库它是一个专门设计用于安全评估表达式的库。自己实现一个微型解析器如果逻辑表达式语法很简单比如只支持ANDOR自己写一个也并不复杂。安全永远是第一位的。4. 实战应用构建一个本地天气决策助手光说不练假把式。让我们用这个框架在本地比如你的笔记本电脑快速实现一个智能天气决策助手。它不联网仅基于你预先存储或手动输入的天气数据以及你的个人日历来给出建议。4.1 场景设定与数据准备假设我们有一个SQLite数据库文件qis_weather.db。我们需要先往里面灌入一些基础数据。1. 模拟天气数据表我们创建一个表来存储可以是手动输入或从某个本地文件导入的天气预报。-- 在 qis_weather.db 中执行 CREATE TABLE weather_forecast ( date TEXT PRIMARY KEY, -- ‘2023-10-27‘ outlook TEXT, -- ‘sunny‘, ‘rainy‘, ‘cloudy‘, ‘snowy‘ high_temp INTEGER, -- 最高温 low_temp INTEGER, -- 最低温 precipitation_prob INTEGER -- 降水概率百分比 ); INSERT INTO weather_forecast VALUES (‘2023-10-27‘, ‘rainy‘, 18, 12, 80); INSERT INTO weather_forecast VALUES (‘2023-10-28‘, ‘cloudy‘, 20, 15, 30);2. 模拟个人日历表CREATE TABLE personal_calendar ( id INTEGER PRIMARY KEY, event_date TEXT, event_time TEXT, description TEXT, location TEXT, is_outdoor INTEGER DEFAULT 0 ); INSERT INTO personal_calendar VALUES (1, ‘2023-10-27‘, ‘09:30‘, ‘团队晨会‘, ‘公司会议室‘, 0); INSERT INTO personal_calendar VALUES (2, ‘2023-10-27‘, ‘14:00‘, ‘客户拜访‘, ‘市中心咖啡厅‘, 0); INSERT INTO personal_calendar VALUES (3, ‘2023-10-28‘, ‘11:00‘, ‘公园跑步‘, ‘中央公园‘, 1);4.2 定义决策规则现在我们向decision_rules等表插入我们的“智能”规则。这些规则定义了助手的行为逻辑。规则1早晨天气检查-- 规则定义每天早上7点触发检查今日天气 INSERT INTO decision_rules (rule_name, description, trigger_event, trigger_condition_sql) VALUES ( ‘morning_weather_check‘, ‘早晨检查天气并根据天气情况给出初步建议‘, ‘scheduled_morning‘, -- 假设有一个外部调度器在7:00发送此事件 NULL -- 无条件触发 ); -- 因子需要今日天气展望和降水概率 INSERT INTO rule_factors (rule_id, factor_key, data_source, retrieval_sql) VALUES (last_insert_rowid(), ‘today_outlook‘, ‘table:weather_forecast‘, “SELECT outlook FROM weather_forecast WHERE date date(‘now‘)“), (last_insert_rowid(), ‘precip_prob‘, ‘table:weather_forecast‘, “SELECT precipitation_prob FROM weather_forecast WHERE date date(‘now‘)“); -- 逻辑与动作 INSERT INTO rule_logic_actions (rule_id, logic_expression, action_type, action_value, route_target_rule_name, priority) VALUES (last_insert_rowid(), ‘?1 “rainy“ AND ?2 50‘, ‘route_to‘, ‘high_rain_alert‘, ‘rainy_day_planning‘, 0), (last_insert_rowid(), ‘?1 “snowy“‘, ‘set_output‘, ‘今日有雪建议穿戴防滑靴和厚外套。‘, NULL, 1), (last_insert_rowid(), ‘?1 “sunny“‘, ‘set_output‘, ‘今日晴好适合户外活动。‘, NULL, 2), (last_insert_rowid(), ‘1 1‘, ‘set_output‘, ‘今日天气一般请按常规划。‘, NULL, 3); -- 默认规则规则2雨天行程规划-- 接上一条规则的路由目标 -- 首先插入规则定义 INSERT INTO decision_rules (rule_name, description, trigger_event) VALUES (‘rainy_day_planning‘, ‘针对雨天的详细行程规划‘, ‘route_trigger‘); -- 注意trigger_event这里可以是任意标识由路由触发 -- 因子需要今日是否有户外活动 INSERT INTO rule_factors (rule_id, factor_key, data_source, retrieval_sql) VALUES (last_insert_rowid(), ‘has_outdoor_event‘, ‘table:personal_calendar‘, “SELECT 1 FROM personal_calendar WHERE event_date date(‘now‘) AND is_outdoor 1 LIMIT 1“); -- 逻辑与动作 INSERT INTO rule_logic_actions (rule_id, logic_expression, action_type, action_value, priority) VALUES (last_insert_rowid(), ‘?1 1‘, ‘set_output‘, ‘今天有户外活动且会下雨建议取消或移至室内并务必携带雨具。‘, 0), (last_insert_rowid(), ‘1 1‘, ‘set_output‘, ‘今天有雨但无户外行程。出门请带伞注意通勤安全。‘, 1);4.3 运行与测试现在我们用Python脚本连接这个数据库并触发规则执行。def main(): # 初始化引擎连接到我们的数据库文件 engine QIS_SQLiteEngine(‘qis_weather.db‘) # 模拟早晨7点的定时触发 print(“ 早晨7点天气决策助手启动 “) results engine.execute_on_trigger(‘scheduled_morning‘) for output in results: if output[‘action_type‘] ‘set_output‘: print(f“建议{output[‘value‘]}“) elif output[‘action_type‘] ‘route_to‘: print(f“触发进一步分析{output[‘next_rule‘]}“) # 引擎内部已递归执行这里打印路由结果 # 在实际日志或返回结果中能看到深层输出 if __name__ ‘__main__‘: main()运行这个脚本假设今天是2023-10-27你会看到类似这样的输出 早晨7点天气决策助手启动 触发进一步分析rainy_day_planning 建议今天有雨但无户外行程。出门请带伞注意通勤安全。系统首先触发了morning_weather_check规则发现天气是雨天且降水概率高于是路由到rainy_day_planning规则。该规则检查日历发现今天没有户外活动我们的日历里10月27日都是室内活动于是给出了最终的出行建议。整个过程中没有任何网络请求所有数据查询和逻辑判断都在本地的SQLite文件中完成响应是毫秒级的。你可以随时修改personal_calendar表添加一个户外事件然后重新运行脚本建议就会变成“建议取消或移至室内”。5. 性能优化、扩展与安全考量在单机设备上运行一个完整的规则引擎性能和资源占用是需要仔细权衡的。同时如何让这个系统更强大、更安全也是实际应用必须面对的。5.1 性能优化策略规则与因子缓存在引擎初始化时一次性将活跃的decision_rules、rule_factors、rule_logic_actions加载到内存中。SQLite的IO对于频繁的规则查询会成为瓶颈。可以使用字典或类来缓存这些元数据仅当规则发生变化时通过监听某个版本号或更新时间戳才刷新缓存。因子值批量预取对于一条规则下的多个因子如果它们的data_source相同例如都来自table:weather_forecast可以尝试合并retrieval_sql或使用更高效的JOIN查询一次性获取而不是执行多次独立的SQL查询。连接池与读写分离如果应用并发量稍高可以考虑使用像sqlite3的连接池如sqlite3.Connection配合线程局部存储或者采用主数据库文件只读、每个线程/进程使用自己的内存数据库临时拷贝的模式来缓解写竞争。索引优化确保decision_rules(trigger_event, is_active)、rule_factors(rule_id)、rule_logic_actions(rule_id, priority)等关键查询字段上有索引。SQLite创建索引非常简单CREATE INDEX idx_rules_trigger ON decision_rules(trigger_event, is_active);。表达式预编译对于logic_expression可以将其解析成抽象语法树AST或可执行的函数对象并缓存起来。每次执行时只需替换参数值并执行编译好的函数这比每次都要解析字符串并调用eval或其安全替代品快得多。5.2 系统能力扩展支持更复杂的事件触发目前的trigger_event是简单的字符串匹配。可以扩展为支持复合事件或模式匹配。例如在decision_rules表中增加一个trigger_pattern字段存储类似“sensor.temperature 30 AND time.between(14:00, 16:00)”的表达式由引擎统一解析和监听。集成外部动作执行action_type目前有set_output、route_to、call_function。可以扩展call_function的能力使其能调用预注册的Python函数、执行系统命令需极其谨慎、发送本地通知如桌面弹窗、甚至通过HTTP请求与本地其他服务交互在确保安全的前提下。引入概率与模糊逻辑当前的逻辑是布尔判断。可以扩展rule_logic_actions表增加confidence置信度字段。逻辑表达式可以返回一个概率值系统综合多个规则的置信度给出一个概率化的建议这更贴近现实世界的“智能”。规则版本管理与A/B测试为规则增加version和effective_time/expire_time字段。可以同时部署多个版本的规则通过上下文如用户ID哈希来决定启用哪一套实现无感的规则A/B测试和灰度发布。可视化规则编辑器既然所有逻辑都数据化了那么为其开发一个简单的Web或桌面GUI来编辑这些表数据就变得非常可行。通过下拉框选择数据源、拖拽生成逻辑树最终生成对应的SQL和表达式存入数据库大大降低了业务人员定义规则的门槛。5.3 安全与可靠性加固彻底弃用不安全的eval如前所述这是头等大事。必须使用安全的表达式求值库如simpleeval。配置simpleeval允许的操作符和函数白名单彻底杜绝代码注入。import simpleeval def safe_eval(expr, names): evaluator simpleeval.SimpleEval() # 只允许最基本的操作 evaluator.operators {cls: op for cls, op in evaluator.operators.items() if cls in (simpleeval.Operator.add, ...)} # 自定义白名单 evaluator.functions {} # 不允许任何函数或只允许安全的数学函数 evaluator.names names try: return evaluator.eval(expr) except Exception: return False严格审计retrieval_sql和trigger_condition_sql这些字段直接执行SQL。必须确保它们来自完全可信的来源如经过审核的后台管理系统。可以考虑使用参数化查询设计一种模板语法让retrieval_sql变成“SELECT value FROM sensor WHERE type {sensor_type}”然后在引擎内部用参数安全替换{sensor_type}。白名单表与列维护一个允许访问的表和列的白名单在执行动态SQL前进行解析和校验。最小权限连接如果可能使用只读模式打开SQLite数据库文件uriTrue模式下使用modero参数限制动态SQL的破坏能力。完善的日志与回滚execution_logs表应详细记录每次规则触发的输入、输出、耗时、是否出错。对于涉及状态修改的动作如更新某个表应考虑在数据库事务中执行一旦某个规则链中的动作失败可以回滚整个链保持状态一致性。规则循环检测规则之间相互路由可能产生死循环A-B-C-A。在执行引擎中需要维护一个“已执行规则栈”或“已访问规则集合”当检测到循环时立即终止并记录错误。6. 适用场景与局限性分析经过上面的拆解这个“SQLite上的二次智能路由”方案的轮廓已经非常清晰了。它不是一个万能AI但在特定场景下它能提供一种极其轻量、可控、私密的自动化决策能力。6.1 理想的应用场景边缘设备与物联网IoT智能家居网关、工业边缘计算盒子。设备本地根据传感器数据温度、湿度、运动触发规则控制继电器、发送本地警报无需永远在线和云端交互响应更快隐私无忧。离线优先的移动应用旅行规划APP、户外运动助手。用户提前下载目的地天气、地图兴趣点数据到本地SQLite。APP根据用户的实时GPS位置、时间、本地存储的POI信息离线给出游览建议、风险提示。个人自动化与效率工具类似于本地版的“IFTTT”或“快捷指令”。你可以定义这样的规则“如果当前时间在晚上10点后且手机连接家庭Wi-Fi且静音模式未开启则自动将手机设为静音”。所有逻辑和状态存储在本地完全由你控制。游戏或模拟器的AI逻辑一些单机游戏中的NPC行为树或状态机可以用这套系统来驱动。将NPC的感知看见玩家、血量低、行为攻击、逃跑、巡逻定义为规则和路由修改游戏难度只需修改数据库里的几个参数无需重新编译代码。原型验证与业务规则管理在开发一个复杂的业务系统前可以用这个框架快速将产品经理写的业务规则“如果用户是新注册且来自XX渠道则发放Y类优惠券”实现成一个可运行的原型方便验证逻辑正确性。后期甚至可以直接将这套规则引擎集成到生产系统中。6.2 固有的局限性无法进行真正的“学习”这里的“智能”是静态的、基于预设规则的。它不会从数据中自我进化也无法处理训练数据中没有出现过的新模式。它本质是一个高度可配置的、声明式的自动化系统而非机器学习模型。复杂逻辑的表达成本虽然用SQL和表达式能描述很多逻辑但对于极其复杂、多变的决策流例如涉及大量循环、递归、动态规划用关系表来定义可能会变得非常冗长和难以维护。这时候传统的编程语言仍然是更合适的工具。性能天花板对于每秒需要处理成千上万次事件、规则库包含数万条规则的超高性能场景纯SQLite和Python解释器可能成为瓶颈。虽然优化能提升很多但其设计初衷仍是轻量级、中等负载的场景。分布式与状态同步这套系统天然是单机版的。如果你需要在多个设备间同步规则和状态需要自己构建一套同步机制例如通过SQLite的WAL文件同步或定义一套上游的规则发布服务。这增加了复杂性。6.3 我的实操心得与避坑指南在几个小项目中实际应用这个模式后我积累了一些“踩坑”得来的经验规则调试是痛点当规则链不按预期运行时定位问题很麻烦。务必在早期就实现一个详细的、可追溯的执行日志系统。不仅要记录最终结果还要记录每一步的因子取值、逻辑表达式求值过程、路由路径。最好能有一个“回放”功能输入一个事件能一步步可视化看到规则的执行流。版本控制你的规则库decision_rules这些表就是你的代码。一定要用版本控制系统如Git来管理它们的变更。可以定期将整个数据库导出为SQL dump文件或者设计一个将规则导出为JSON/YAML格式的脚本方便diff和回滚。从简单开始渐进复杂不要一开始就试图设计一个包罗万象的规则系统。从一个非常具体的、高价值的场景开始比如“电脑锁屏时自动静音”实现它、跑通它然后再慢慢添加更多的规则和更复杂的路由。这有助于你早期就发现框架设计上的缺陷。为“未知”留出出口总会有一些情况是你的规则集覆盖不到的。一定要设计一个“默认规则”或“降级动作”。当所有规则都不匹配时系统应该有一个明确的、安全的默认行为而不是静默失败或抛出异常。性能测试要趁早即使你觉得规则不多也应在开发中期就进行压力测试。模拟高频事件触发看看引擎的响应时间和内存占用。你可能会发现某些retrieval_sql写得不好没有用上索引或者某个logic_expression解析成了性能黑洞。早发现早优化。这个项目的魅力在于它用最简单、最普遍的技术SQLite实现了一个听起来很“智能”的概念。它把决策的权力从云端还给了设备从黑盒变成了白盒。当你看到那些冰冷的数据库记录在你的引擎驱动下像活过来一样串联、判断、给出建议时你会感受到一种不同于调用云API的、实实在在的掌控感。这或许就是“二次智能”的精髓不在于有多深奥的算法而在于将确定的逻辑以灵活、可管理的方式部署在每一个需要它的角落。
基于SQLite的本地化二次智能决策系统设计与实现
发布时间：2026/5/28 14:11:08
1. 项目概述当SQLite遇见二次智能最近在折腾一个挺有意思的玩意儿我把它叫做“基于SQLite的QIS结果路由”。简单说就是让任何一台设备——你的笔记本电脑、手机甚至是树莓派——都能在不依赖任何外部服务器的情况下跑起一套具备“二次智能”的决策系统。这听起来有点玄乎但核心逻辑其实很接地气把复杂的决策逻辑从云端黑盒里拽出来塞进一个轻量级的SQLite数据库里让它在你手边的设备上就能独立运行、自主判断。“QIS”在这里指的是“二次智能系统”。它不是要替代大语言模型那种生成式AI而是专注于另一件事基于既定规则和实时数据进行多轮、递进式的推理与决策。想象一下你不是在问AI“写一首诗”而是在设计一个系统它能根据当前的天气、你的日程、交通状况和历史偏好自动为你规划出今天最佳的出行路线和衣物搭配。这个决策过程往往不是一步到位的它需要像人一样先评估A条件得出中间结论再结合B条件进行二次判断最终形成一个综合方案——这就是“二次”智能的意味。而“结果路由”则是这个系统的神经中枢。它负责管理决策流程当一个事件触发比如“早上8点”系统需要调用哪些规则这些规则产生的初步结论应该流向哪个下一步的判断节点最终的结果又该如何执行或呈现传统做法要么写死在一大坨if-else代码里难以维护要么依赖云端微服务编排延迟和隐私都是问题。所以这个项目的核心创新点就是用SQLite来承载这一切。为什么是SQLite因为它无处不在从安卓、iOS到Windows、macOS乃至各种嵌入式系统它几乎是唯一一个“开箱即用”、零配置的数据库引擎。它不需要单独安装数据库服务一个文件就是全部。我们把智能决策的规则库、状态机、工作流定义全部用关系型数据表来建模和存储。这样一来任何能运行SQLite的设备就瞬间获得了一个本地的、私密的、可离线运行的智能决策引擎。没有网络延迟没有服务器费用没有数据出域的风险真正的“Any Device, No Server Required”。2. 核心架构设计用关系模型表达智能逻辑把智能逻辑塞进数据库表这听起来像是个行为艺术但当你拆解开来会发现关系模型其实异常贴合。关键在于我们要对“智能决策”这个过程进行解构和重塑。2.1 逻辑解构从过程式代码到声明式数据传统的决策逻辑通常是过程式的。比如用Python写可能就是一系列嵌套的判断语句if weather ‘rain‘: if schedule ‘meeting‘: decision ‘take_umbrella_and_leave_early‘ else: decision ‘stay_home‘ elif temperature 30: decision ‘wear_tshirt‘ ...这种代码的缺点是僵化。每加一个新条件比如“空气质量指数”你都得去修改代码逻辑很容易出错也难以直观地看到所有规则的全貌。我们的思路是将其声明化、数据化。我们把一条决策规则拆解成几个核心组成部分触发条件什么情况下这条规则该被评估例如事件类型“早晨唤醒” 且 时间 07:00评估因子需要检查哪些数据例如weather.today.outlook, calendar.events.first.start_time判断逻辑因子之间满足什么关系例如outlook ‘rainy‘ AND start_time - current_time 30分钟输出结果如果满足产生什么结论例如action ‘suggest_umbrella‘, priority ‘high‘结果路由这个结论应该交给谁哪个下游规则做进一步处理例如route_to ‘commute_planning_rule‘这样一来每一条规则就变成了数据库里的一行记录。整个决策知识库就是一个可以随时增删改查的数据表。2.2 数据表设计构建规则引擎的骨架基于上面的解构我设计了几个核心表。这里以SQLite的建表语句为例你可以清晰地看到整个模型的骨架。1. 规则定义表 (decision_rules)这是大脑皮层存储所有具体的规则。CREATE TABLE decision_rules ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_name TEXT NOT NULL UNIQUE, -- 规则唯一标识如 ‘morning_weather_check‘ description TEXT, -- 规则描述 trigger_event TEXT, -- 触发事件类型如 ‘scheduled‘, ‘sensor_update‘, ‘user_input‘ trigger_condition_sql TEXT, -- 用SQL片段描述的触发条件灵活且强大 is_active BOOLEAN DEFAULT 1, -- 是否启用 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );注意trigger_condition_sql字段是这个设计的精髓。它存储的是一段像“SELECT 1 FROM sensor_data WHERE temperature 30 AND timestamp datetime(‘now‘, ‘-5 minutes‘)”这样的SQL片段。系统会动态执行它如果返回有结果则触发该规则。这赋予了规则极大的动态性可以直接查询最新的设备传感器数据或其他状态表。2. 规则因子表 (rule_factors)这是神经突触定义每条规则需要关注哪些输入数据。CREATE TABLE rule_factors ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id INTEGER NOT NULL, factor_key TEXT NOT NULL, -- 因子键名如 ‘current_temperature‘ data_source TEXT NOT NULL, -- 数据来源如 ‘table:sensor_readings‘, ‘env:USER_LOCATION‘ retrieval_sql TEXT, -- 如何获取该因子的SQL可为空若data_source已明确 FOREIGN KEY (rule_id) REFERENCES decision_rules(id) ON DELETE CASCADE );实操心得data_source的设计支持多种形式。table:前缀表示从其他表查询env:表示从程序运行环境变量或上下文中获取甚至可以是http:虽然离线场景慎用或file:。这使得系统能融合设备本地数据库、系统状态、配置文件等多种数据源。3. 判断逻辑与动作表 (rule_logic_actions)这是决策神经元将因子关联起来并指定满足条件后的输出和流向。CREATE TABLE rule_logic_actions ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id INTEGER NOT NULL, logic_expression TEXT NOT NULL, -- 逻辑表达式如 ‘?1 “rainy“ AND ?2 30‘ action_type TEXT NOT NULL, -- 动作类型 ‘set_output‘, ‘call_function‘, ‘route_to‘ action_value TEXT NOT NULL, -- 动作值输出内容、函数名、或下游规则名 route_target_rule_name TEXT, -- 如果action_type是‘route_to‘这里指定目标规则名 priority INTEGER DEFAULT 0, -- 同一规则内多条逻辑的优先级 FOREIGN KEY (rule_id) REFERENCES decision_rules(id) ON DELETE CASCADE );这里的logic_expression使用占位符?1、?2它们会按顺序被rule_factors检索出的实际值替换。然后系统会用一个小型的表达式求值器比如用Python的eval在安全沙箱内或自己实现一个简单的解析器来计算布尔结果。2.3 工作流引擎SQL驱动的状态机单个规则是神经元多个规则通过路由串联起来就形成了工作流或状态机。rule_logic_actions表中的route_target_rule_name字段就是定义路由的关键。整个决策过程由一个执行引擎驱动它本质上是一个循环监听trigger_event如定时器、数据插入事件。根据事件扫描decision_rules表找到所有被触发的、活跃的规则。对每条规则获取其所有因子 (rule_factors) 的当前值。按优先级评估该规则下的所有逻辑 (rule_logic_actions)。执行第一个评估为真的逻辑所定义的动作。如果动作是route_to则将route_target_rule_name作为新的触发事件或直接将其id加入待执行队列跳转到步骤2形成链式或树状的推理。如果是set_output则将该结果存入结果表并可能结束当前推理链。这个引擎可以用任何支持SQLite的语言实现Python, JavaScript, Rust等核心逻辑就是拼装和执行SQL以及一个简单的表达式求值循环。整个系统的“智能”就体现在这些表数据的关系和流动中。3. 实现细节与核心代码拆解理论说得再多不如一行代码。下面我用Python因其易读性来展示这个路由引擎的核心实现片段。选择Python的sqlite3标准库意味着这个方案在绝大多数设备上都能零依赖运行。3.1 引擎初始化与规则加载首先我们需要一个引擎类来管理数据库连接和规则缓存。import sqlite3 import threading from typing import Any, Dict, List, Optional import json import re class QIS_SQLiteEngine: def __init__(self, db_path: str ‘:memory:‘): 初始化引擎。db_path为SQLite数据库文件路径‘:memory:‘表示内存数据库。 self.db_path db_path # 使用check_same_threadFalse需谨慎这里为简单演示。生产环境建议用连接池。 self.conn sqlite3.connect(db_path, check_same_threadFalse) self.conn.row_factory sqlite3.Row # 返回字典样式的行 self._rule_cache {} # 缓存规则定义避免频繁查询 self._lock threading.RLock() # 简单线程锁 self._init_tables() # 确保表存在 def _init_tables(self): 创建核心数据表如果不存在。 # 这里省略了完整的建表SQL与上一节设计一致 cursor self.conn.cursor() cursor.executescript(‘‘‘ CREATE TABLE IF NOT EXISTS decision_rules (...); CREATE TABLE IF NOT EXISTS rule_factors (...); CREATE TABLE IF NOT EXISTS rule_logic_actions (...); CREATE TABLE IF NOT EXISTS execution_logs (...); -- 用于日志记录 ‘‘‘) self.conn.commit()注意事项SQLite的默认连接对象不是线程安全的。在简单的单线程脚本或像idle这样的场景下没问题但如果你的应用可能多线程访问数据库需要使用锁如threading.RLock()来保护连接或者每个线程创建自己的连接。这里为了代码清晰使用了check_same_threadFalse并配合一个简单的锁这适用于轻量级并发。对于更复杂的场景建议采用连接池模式。3.2 核心执行循环触发、评估、路由这是引擎最核心的方法它实现了一次完整的规则触发与执行流程。def execute_on_trigger(self, trigger_event: str, context: Optional[Dict] None) - List[Dict]: 根据触发事件执行相关规则。 Args: trigger_event: 触发事件类型如 ‘scheduled_morning‘ context: 可选执行上下文可用于向因子注入额外数据 Returns: 本次执行产生的所有输出结果列表 if context is None: context {} all_outputs [] with self._lock: # 1. 查找被触发的活跃规则 cursor self.conn.cursor() # 注意这里演示了trigger_condition_sql的动态执行。实际使用需严格防范SQL注入。 # 我们假设trigger_condition_sql来自可信的配置源。 cursor.execute(‘‘‘ SELECT id, rule_name, trigger_condition_sql FROM decision_rules WHERE trigger_event ? AND is_active 1 ‘‘‘, (trigger_event,)) triggered_rules cursor.fetchall() for rule_row in triggered_rules: rule_id, rule_name, condition_sql rule_row # 2. 检查触发条件SQL如果存在 should_execute True if condition_sql: try: # 执行条件SQL期望返回至少一行则认为触发 cond_cursor self.conn.execute(condition_sql) should_execute cond_cursor.fetchone() is not None except sqlite3.Error as e: print(f“规则 ‘{rule_name}‘ 触发条件SQL执行错误: {e}“) should_execute False if not should_execute: continue # 3. 执行这条规则 rule_output self._execute_single_rule(rule_id, rule_name, context) if rule_output: all_outputs.append(rule_output) # 4. 处理路由如果输出指定了下一个规则则递归或迭代执行 if ‘next_rule‘ in rule_output: next_trigger rule_output[‘next_rule‘] # 避免无限循环可以在这里加入已执行规则栈的检查 deeper_outputs self.execute_on_trigger(next_trigger, context) all_outputs.extend(deeper_outputs) return all_outputs3.3 单条规则执行因子获取与逻辑判断_execute_single_rule方法是智能发生的具体场所。def _execute_single_rule(self, rule_id: int, rule_name: str, context: Dict) - Optional[Dict]: 执行单条规则返回输出字典或None。 cursor self.conn.cursor() # 1. 获取本规则的所有因子 cursor.execute(‘SELECT factor_key, data_source, retrieval_sql FROM rule_factors WHERE rule_id ? ORDER BY id‘, (rule_id,)) factors cursor.fetchall() factor_values {} for f_key, f_source, f_sql in factors: value self._retrieve_factor_value(f_source, f_sql, context) factor_values[f_key] value # 2. 获取本规则的所有逻辑-动作对按优先级排序 cursor.execute(‘‘‘ SELECT logic_expression, action_type, action_value, route_target_rule_name FROM rule_logic_actions WHERE rule_id ? ORDER BY priority ASC ‘‘‘, (rule_id,)) logic_actions cursor.fetchall() # 3. 按顺序评估逻辑表达式 for logic_expr, action_type, action_value, route_target in logic_actions: if self._evaluate_logic(logic_expr, factor_values): # 逻辑为真执行动作 output {‘rule‘: rule_name, ‘action_type‘: action_type, ‘value‘: action_value} if action_type ‘set_output‘: output[‘result‘] action_value self._log_execution(rule_id, True, logic_expr, output) return output elif action_type ‘route_to‘ and route_target: output[‘next_rule‘] route_target self._log_execution(rule_id, True, logic_expr, output) return output elif action_type ‘call_function‘: # 这里可以集成一个安全的函数调用机制 # 例如预定义一些函数名到实际函数的映射 print(f“规则 ‘{rule_name}‘ 调用函数: {action_value}“) output[‘function_called‘] action_value self._log_execution(rule_id, True, logic_expr, output) return output # 找到第一个为真的逻辑就执行并返回 break else: # 所有逻辑都不为真 self._log_execution(rule_id, False, None, None) return None3.4 关键支撑方法因子获取与逻辑求值_retrieve_factor_value和_evaluate_logic是两个关键辅助函数它们决定了系统的灵活性和安全性。def _retrieve_factor_value(self, data_source: str, retrieval_sql: Optional[str], context: Dict) - Any: 根据数据源描述获取因子值。 # 处理上下文中的覆盖值用于测试或动态注入 if data_source.startswith(‘ctx:‘): key data_source[4:] return context.get(key) # 从数据库表查询 elif data_source.startswith(‘table:‘): if retrieval_sql: try: cursor self.conn.execute(retrieval_sql) row cursor.fetchone() # 假设查询返回单行单列的值或做相应处理 return row[0] if row else None except sqlite3.Error as e: print(f“因子查询SQL错误 [{data_source}]: {e}“) return None else: # 如果没有提供SQL尝试从source中解析表名和字段简化处理 # 例如 ‘table:weather.current_temp‘ parts data_source.split(‘:‘)[1].split(‘.‘) if len(parts) 2: table, column parts cursor self.conn.execute(f‘SELECT {column} FROM {table} ORDER BY ROWID DESC LIMIT 1‘) row cursor.fetchone() return row[0] if row else None return None # 环境变量或固定值示例 elif data_source.startswith(‘env:‘): import os key data_source[4:] return os.environ.get(key) elif data_source.startswith(‘literal:‘): # 处理字面量如 ‘literal:42‘, ‘literal:“hello“‘ value_str data_source[8:] try: return json.loads(value_str) # 尝试解析JSON except: return value_str # 否则返回字符串 else: # 未知数据源 return None def _evaluate_logic(self, logic_expr: str, factor_values: Dict[str, Any]) - bool: 评估逻辑表达式。这是一个简化的安全实现。 if not logic_expr: return False # 将占位符 ?1, ?2 替换为实际值 # 首先我们需要知道占位符的顺序。这里简单按出现顺序替换。 # 更健壮的做法是在存储时解析并保存参数顺序。 expr_to_eval logic_expr value_list list(factor_values.values()) # 简单的占位符替换生产环境需要更严谨的解析 for i, val in enumerate(value_list, start1): placeholder f‘?{i}‘ # 将值安全地转换为表达式中的字面量 if isinstance(val, str): # 字符串需要加引号并转义内部引号 safe_val json.dumps(val) elif val is None: safe_val ‘None‘ else: safe_val str(val) expr_to_eval expr_to_eval.replace(placeholder, safe_val) # **安全警告**直接使用eval是危险的如果logic_expr来自不可信源。 # 此处仅为演示。生产环境应 # 1. 使用限制性的解析器如ast.literal_eval配合自定义操作符。 # 2. 或使用像 simpleeval 这样的安全库。 # 3. 或自己实现一个简单的表达式解析器支持 , !, , , AND, OR等。 try: # 极度简化的安全措施只允许有限的字符 if re.match(r‘^[0-9\s\.\\‘\[\]\{\}\(\)\\-\*/!|NoneTrueFalse\s]$‘, expr_to_eval): result eval(expr_to_eval, {“__builtins__“: {}}, {}) return bool(result) else: print(f“表达式包含不安全字符: {logic_expr}“) return False except Exception as e: print(f“逻辑表达式求值错误 ‘{expr_to_eval}‘: {e}“) return False核心安全提示_evaluate_logic函数中的eval使用是本演示最大的安全隐患。逻辑表达式来自数据库如果被恶意篡改可能执行任意代码。在生产环境中你必须替换这部分。推荐方案使用ast.literal_eval并配合一个自定义的、安全的操作符映射字典只允许比较和逻辑运算。引入simpleeval库它是一个专门设计用于安全评估表达式的库。自己实现一个微型解析器如果逻辑表达式语法很简单比如只支持ANDOR自己写一个也并不复杂。安全永远是第一位的。4. 实战应用构建一个本地天气决策助手光说不练假把式。让我们用这个框架在本地比如你的笔记本电脑快速实现一个智能天气决策助手。它不联网仅基于你预先存储或手动输入的天气数据以及你的个人日历来给出建议。4.1 场景设定与数据准备假设我们有一个SQLite数据库文件qis_weather.db。我们需要先往里面灌入一些基础数据。1. 模拟天气数据表我们创建一个表来存储可以是手动输入或从某个本地文件导入的天气预报。-- 在 qis_weather.db 中执行 CREATE TABLE weather_forecast ( date TEXT PRIMARY KEY, -- ‘2023-10-27‘ outlook TEXT, -- ‘sunny‘, ‘rainy‘, ‘cloudy‘, ‘snowy‘ high_temp INTEGER, -- 最高温 low_temp INTEGER, -- 最低温 precipitation_prob INTEGER -- 降水概率百分比 ); INSERT INTO weather_forecast VALUES (‘2023-10-27‘, ‘rainy‘, 18, 12, 80); INSERT INTO weather_forecast VALUES (‘2023-10-28‘, ‘cloudy‘, 20, 15, 30);2. 模拟个人日历表CREATE TABLE personal_calendar ( id INTEGER PRIMARY KEY, event_date TEXT, event_time TEXT, description TEXT, location TEXT, is_outdoor INTEGER DEFAULT 0 ); INSERT INTO personal_calendar VALUES (1, ‘2023-10-27‘, ‘09:30‘, ‘团队晨会‘, ‘公司会议室‘, 0); INSERT INTO personal_calendar VALUES (2, ‘2023-10-27‘, ‘14:00‘, ‘客户拜访‘, ‘市中心咖啡厅‘, 0); INSERT INTO personal_calendar VALUES (3, ‘2023-10-28‘, ‘11:00‘, ‘公园跑步‘, ‘中央公园‘, 1);4.2 定义决策规则现在我们向decision_rules等表插入我们的“智能”规则。这些规则定义了助手的行为逻辑。规则1早晨天气检查-- 规则定义每天早上7点触发检查今日天气 INSERT INTO decision_rules (rule_name, description, trigger_event, trigger_condition_sql) VALUES ( ‘morning_weather_check‘, ‘早晨检查天气并根据天气情况给出初步建议‘, ‘scheduled_morning‘, -- 假设有一个外部调度器在7:00发送此事件 NULL -- 无条件触发 ); -- 因子需要今日天气展望和降水概率 INSERT INTO rule_factors (rule_id, factor_key, data_source, retrieval_sql) VALUES (last_insert_rowid(), ‘today_outlook‘, ‘table:weather_forecast‘, “SELECT outlook FROM weather_forecast WHERE date date(‘now‘)“), (last_insert_rowid(), ‘precip_prob‘, ‘table:weather_forecast‘, “SELECT precipitation_prob FROM weather_forecast WHERE date date(‘now‘)“); -- 逻辑与动作 INSERT INTO rule_logic_actions (rule_id, logic_expression, action_type, action_value, route_target_rule_name, priority) VALUES (last_insert_rowid(), ‘?1 “rainy“ AND ?2 50‘, ‘route_to‘, ‘high_rain_alert‘, ‘rainy_day_planning‘, 0), (last_insert_rowid(), ‘?1 “snowy“‘, ‘set_output‘, ‘今日有雪建议穿戴防滑靴和厚外套。‘, NULL, 1), (last_insert_rowid(), ‘?1 “sunny“‘, ‘set_output‘, ‘今日晴好适合户外活动。‘, NULL, 2), (last_insert_rowid(), ‘1 1‘, ‘set_output‘, ‘今日天气一般请按常规划。‘, NULL, 3); -- 默认规则规则2雨天行程规划-- 接上一条规则的路由目标 -- 首先插入规则定义 INSERT INTO decision_rules (rule_name, description, trigger_event) VALUES (‘rainy_day_planning‘, ‘针对雨天的详细行程规划‘, ‘route_trigger‘); -- 注意trigger_event这里可以是任意标识由路由触发 -- 因子需要今日是否有户外活动 INSERT INTO rule_factors (rule_id, factor_key, data_source, retrieval_sql) VALUES (last_insert_rowid(), ‘has_outdoor_event‘, ‘table:personal_calendar‘, “SELECT 1 FROM personal_calendar WHERE event_date date(‘now‘) AND is_outdoor 1 LIMIT 1“); -- 逻辑与动作 INSERT INTO rule_logic_actions (rule_id, logic_expression, action_type, action_value, priority) VALUES (last_insert_rowid(), ‘?1 1‘, ‘set_output‘, ‘今天有户外活动且会下雨建议取消或移至室内并务必携带雨具。‘, 0), (last_insert_rowid(), ‘1 1‘, ‘set_output‘, ‘今天有雨但无户外行程。出门请带伞注意通勤安全。‘, 1);4.3 运行与测试现在我们用Python脚本连接这个数据库并触发规则执行。def main(): # 初始化引擎连接到我们的数据库文件 engine QIS_SQLiteEngine(‘qis_weather.db‘) # 模拟早晨7点的定时触发 print(“ 早晨7点天气决策助手启动 “) results engine.execute_on_trigger(‘scheduled_morning‘) for output in results: if output[‘action_type‘] ‘set_output‘: print(f“建议{output[‘value‘]}“) elif output[‘action_type‘] ‘route_to‘: print(f“触发进一步分析{output[‘next_rule‘]}“) # 引擎内部已递归执行这里打印路由结果 # 在实际日志或返回结果中能看到深层输出 if __name__ ‘__main__‘: main()运行这个脚本假设今天是2023-10-27你会看到类似这样的输出 早晨7点天气决策助手启动 触发进一步分析rainy_day_planning 建议今天有雨但无户外行程。出门请带伞注意通勤安全。系统首先触发了morning_weather_check规则发现天气是雨天且降水概率高于是路由到rainy_day_planning规则。该规则检查日历发现今天没有户外活动我们的日历里10月27日都是室内活动于是给出了最终的出行建议。整个过程中没有任何网络请求所有数据查询和逻辑判断都在本地的SQLite文件中完成响应是毫秒级的。你可以随时修改personal_calendar表添加一个户外事件然后重新运行脚本建议就会变成“建议取消或移至室内”。5. 性能优化、扩展与安全考量在单机设备上运行一个完整的规则引擎性能和资源占用是需要仔细权衡的。同时如何让这个系统更强大、更安全也是实际应用必须面对的。5.1 性能优化策略规则与因子缓存在引擎初始化时一次性将活跃的decision_rules、rule_factors、rule_logic_actions加载到内存中。SQLite的IO对于频繁的规则查询会成为瓶颈。可以使用字典或类来缓存这些元数据仅当规则发生变化时通过监听某个版本号或更新时间戳才刷新缓存。因子值批量预取对于一条规则下的多个因子如果它们的data_source相同例如都来自table:weather_forecast可以尝试合并retrieval_sql或使用更高效的JOIN查询一次性获取而不是执行多次独立的SQL查询。连接池与读写分离如果应用并发量稍高可以考虑使用像sqlite3的连接池如sqlite3.Connection配合线程局部存储或者采用主数据库文件只读、每个线程/进程使用自己的内存数据库临时拷贝的模式来缓解写竞争。索引优化确保decision_rules(trigger_event, is_active)、rule_factors(rule_id)、rule_logic_actions(rule_id, priority)等关键查询字段上有索引。SQLite创建索引非常简单CREATE INDEX idx_rules_trigger ON decision_rules(trigger_event, is_active);。表达式预编译对于logic_expression可以将其解析成抽象语法树AST或可执行的函数对象并缓存起来。每次执行时只需替换参数值并执行编译好的函数这比每次都要解析字符串并调用eval或其安全替代品快得多。5.2 系统能力扩展支持更复杂的事件触发目前的trigger_event是简单的字符串匹配。可以扩展为支持复合事件或模式匹配。例如在decision_rules表中增加一个trigger_pattern字段存储类似“sensor.temperature 30 AND time.between(14:00, 16:00)”的表达式由引擎统一解析和监听。集成外部动作执行action_type目前有set_output、route_to、call_function。可以扩展call_function的能力使其能调用预注册的Python函数、执行系统命令需极其谨慎、发送本地通知如桌面弹窗、甚至通过HTTP请求与本地其他服务交互在确保安全的前提下。引入概率与模糊逻辑当前的逻辑是布尔判断。可以扩展rule_logic_actions表增加confidence置信度字段。逻辑表达式可以返回一个概率值系统综合多个规则的置信度给出一个概率化的建议这更贴近现实世界的“智能”。规则版本管理与A/B测试为规则增加version和effective_time/expire_time字段。可以同时部署多个版本的规则通过上下文如用户ID哈希来决定启用哪一套实现无感的规则A/B测试和灰度发布。可视化规则编辑器既然所有逻辑都数据化了那么为其开发一个简单的Web或桌面GUI来编辑这些表数据就变得非常可行。通过下拉框选择数据源、拖拽生成逻辑树最终生成对应的SQL和表达式存入数据库大大降低了业务人员定义规则的门槛。5.3 安全与可靠性加固彻底弃用不安全的eval如前所述这是头等大事。必须使用安全的表达式求值库如simpleeval。配置simpleeval允许的操作符和函数白名单彻底杜绝代码注入。import simpleeval def safe_eval(expr, names): evaluator simpleeval.SimpleEval() # 只允许最基本的操作 evaluator.operators {cls: op for cls, op in evaluator.operators.items() if cls in (simpleeval.Operator.add, ...)} # 自定义白名单 evaluator.functions {} # 不允许任何函数或只允许安全的数学函数 evaluator.names names try: return evaluator.eval(expr) except Exception: return False严格审计retrieval_sql和trigger_condition_sql这些字段直接执行SQL。必须确保它们来自完全可信的来源如经过审核的后台管理系统。可以考虑使用参数化查询设计一种模板语法让retrieval_sql变成“SELECT value FROM sensor WHERE type {sensor_type}”然后在引擎内部用参数安全替换{sensor_type}。白名单表与列维护一个允许访问的表和列的白名单在执行动态SQL前进行解析和校验。最小权限连接如果可能使用只读模式打开SQLite数据库文件uriTrue模式下使用modero参数限制动态SQL的破坏能力。完善的日志与回滚execution_logs表应详细记录每次规则触发的输入、输出、耗时、是否出错。对于涉及状态修改的动作如更新某个表应考虑在数据库事务中执行一旦某个规则链中的动作失败可以回滚整个链保持状态一致性。规则循环检测规则之间相互路由可能产生死循环A-B-C-A。在执行引擎中需要维护一个“已执行规则栈”或“已访问规则集合”当检测到循环时立即终止并记录错误。6. 适用场景与局限性分析经过上面的拆解这个“SQLite上的二次智能路由”方案的轮廓已经非常清晰了。它不是一个万能AI但在特定场景下它能提供一种极其轻量、可控、私密的自动化决策能力。6.1 理想的应用场景边缘设备与物联网IoT智能家居网关、工业边缘计算盒子。设备本地根据传感器数据温度、湿度、运动触发规则控制继电器、发送本地警报无需永远在线和云端交互响应更快隐私无忧。离线优先的移动应用旅行规划APP、户外运动助手。用户提前下载目的地天气、地图兴趣点数据到本地SQLite。APP根据用户的实时GPS位置、时间、本地存储的POI信息离线给出游览建议、风险提示。个人自动化与效率工具类似于本地版的“IFTTT”或“快捷指令”。你可以定义这样的规则“如果当前时间在晚上10点后且手机连接家庭Wi-Fi且静音模式未开启则自动将手机设为静音”。所有逻辑和状态存储在本地完全由你控制。游戏或模拟器的AI逻辑一些单机游戏中的NPC行为树或状态机可以用这套系统来驱动。将NPC的感知看见玩家、血量低、行为攻击、逃跑、巡逻定义为规则和路由修改游戏难度只需修改数据库里的几个参数无需重新编译代码。原型验证与业务规则管理在开发一个复杂的业务系统前可以用这个框架快速将产品经理写的业务规则“如果用户是新注册且来自XX渠道则发放Y类优惠券”实现成一个可运行的原型方便验证逻辑正确性。后期甚至可以直接将这套规则引擎集成到生产系统中。6.2 固有的局限性无法进行真正的“学习”这里的“智能”是静态的、基于预设规则的。它不会从数据中自我进化也无法处理训练数据中没有出现过的新模式。它本质是一个高度可配置的、声明式的自动化系统而非机器学习模型。复杂逻辑的表达成本虽然用SQL和表达式能描述很多逻辑但对于极其复杂、多变的决策流例如涉及大量循环、递归、动态规划用关系表来定义可能会变得非常冗长和难以维护。这时候传统的编程语言仍然是更合适的工具。性能天花板对于每秒需要处理成千上万次事件、规则库包含数万条规则的超高性能场景纯SQLite和Python解释器可能成为瓶颈。虽然优化能提升很多但其设计初衷仍是轻量级、中等负载的场景。分布式与状态同步这套系统天然是单机版的。如果你需要在多个设备间同步规则和状态需要自己构建一套同步机制例如通过SQLite的WAL文件同步或定义一套上游的规则发布服务。这增加了复杂性。6.3 我的实操心得与避坑指南在几个小项目中实际应用这个模式后我积累了一些“踩坑”得来的经验规则调试是痛点当规则链不按预期运行时定位问题很麻烦。务必在早期就实现一个详细的、可追溯的执行日志系统。不仅要记录最终结果还要记录每一步的因子取值、逻辑表达式求值过程、路由路径。最好能有一个“回放”功能输入一个事件能一步步可视化看到规则的执行流。版本控制你的规则库decision_rules这些表就是你的代码。一定要用版本控制系统如Git来管理它们的变更。可以定期将整个数据库导出为SQL dump文件或者设计一个将规则导出为JSON/YAML格式的脚本方便diff和回滚。从简单开始渐进复杂不要一开始就试图设计一个包罗万象的规则系统。从一个非常具体的、高价值的场景开始比如“电脑锁屏时自动静音”实现它、跑通它然后再慢慢添加更多的规则和更复杂的路由。这有助于你早期就发现框架设计上的缺陷。为“未知”留出出口总会有一些情况是你的规则集覆盖不到的。一定要设计一个“默认规则”或“降级动作”。当所有规则都不匹配时系统应该有一个明确的、安全的默认行为而不是静默失败或抛出异常。性能测试要趁早即使你觉得规则不多也应在开发中期就进行压力测试。模拟高频事件触发看看引擎的响应时间和内存占用。你可能会发现某些retrieval_sql写得不好没有用上索引或者某个logic_expression解析成了性能黑洞。早发现早优化。这个项目的魅力在于它用最简单、最普遍的技术SQLite实现了一个听起来很“智能”的概念。它把决策的权力从云端还给了设备从黑盒变成了白盒。当你看到那些冰冷的数据库记录在你的引擎驱动下像活过来一样串联、判断、给出建议时你会感受到一种不同于调用云API的、实实在在的掌控感。这或许就是“二次智能”的精髓不在于有多深奥的算法而在于将确定的逻辑以灵活、可管理的方式部署在每一个需要它的角落。
