AI Rule Engine

将自然语言规则描述通过 AI 转换为可执行 Python 函数的规则引擎。

功能特性

• 自然语言 → 代码：使用 LLM 将规则描述转换为 Python 函数
• 安全执行：RestrictedPython 沙箱隔离
• 快速校验：规则编译后存储，后续判断无需 LLM 调用
• 冲突检测：自动检测矛盾规则
• LLM 兜底：无匹配时自动调用 LLM 补充规则

技术栈

• Python 3.10+
• SQLite（规则存储）
• LLM API（OpenAI GPT-4o / Anthropic Claude）

快速开始

安装

pip install -e .

启动服务

python -m rule_engine.api
# 默认 http://0.0.0.0:8000

API 接口

创建规则

curl -X POST http://localhost:8000/api/rules \
  -H "Content-Type: application/json" \
  -d '{
    "name": "premium_discount",
    "description": "高级会员8折",
    "condition_template": "如果用户订阅类型为 premium 且年龄大于 18",
    "priority": 1
  }'

校验事实

curl -X POST http://localhost:8000/api/rules/evaluate \
  -H "Content-Type: application/json" \
  -d '{
    "facts": {
      "subscription": "premium",
      "age": 25
    }
  }'

列出规则

curl http://localhost:8000/api/rules

获取规则

curl http://localhost:8000/api/rules/{rule_id}

删除规则

curl -X DELETE http://localhost:8000/api/rules/{rule_id}

项目结构

src/rule_engine/
├── __init__.py
├── api.py         # REST API
├── callback.py     # LLM 兜底
├── compiler.py    # LLM 编译器
├── conflict.py    # 冲突检测
├── executor.py    # 沙箱执行器
├── matcher.py     # 规则匹配
├── models.py      # 数据模型
└── store.py       # SQLite 存储

配置

环境变量	说明
OPENAI_API_KEY	OpenAI API Key
ANTHROPIC_API_KEY	Anthropic API Key

开发

测试

# 运行所有测试
python3 -c "
import sys
sys.path.insert(0, 'src')
sys.path.insert(0, 'tests')

# 模型测试
from rule_engine.models import Rule, CreateRuleRequest
r = Rule(id='t1', name='test', condition_template='t', code='def r(f): pass')
assert r.id == 't1'
print('models: OK')

# 存储测试
import tempfile, os
from rule_engine.store import RuleStore
fd, path = tempfile.mkstemp(suffix='.db')
os.close(fd)
store = RuleStore(path)
rule = store.create_rule(name='t', condition_template='c', code='def r(f): pass')
assert store.get_rule(rule.id).name == 't'
os.unlink(path)
print('store: OK')

# 执行器测试
from rule_engine.executor import RuleExecutor
ex = RuleExecutor()
code = 'def rule(f): return {\"action\": \"ok\"}' if False else 'def rule(facts):\\n    if facts.get(\"x\") == 1: return {\"a\": 1}\\n    return None'
result = ex.execute_rule(code, {'x': 1})
assert result == {'a': 1}
print('executor: OK')

# 冲突检测测试
from rule_engine.conflict import ConflictDetector
det = ConflictDetector()
rules = [
    {'id': 'r1', 'name': 'a', 'code': 'def rule(f): return {\"action\": \"allow\"}', 'priority': 1, 'is_active': True},
    {'id': 'r2', 'name': 'b', 'code': 'def rule(f): return {\"action\": \"deny\"}', 'priority': 1, 'is_active': True},
]
conflicts = det.detect_conflicts(rules)
assert len(conflicts) == 1
print('conflict: OK')

print('\\nAll tests passed!')
"

启动服务（带 LLM 回调）

python3 -c "
import sys
sys.path.insert(0, 'src')
from rule_engine.api import create_app
from rule_engine.store import RuleStore

store = RuleStore('rules.db')
srv = create_app(store, enable_callback=True)
print('Server started on http://0.0.0.0:8000')
srv.serve_forever()
"

规则格式

规则描述模板

{
  "name": "rule_name",
  "condition_template": "如果用户是 premium 会员且年龄大于 18",
  "code": "def rule(facts):\n    if facts.get('subscription') == 'premium' and facts.get('age', 0) > 18:\n        return {'action': 'discount', 'rate': 0.8}\n    return None"
}

Facts 格式

{
  "user_id": "u123",
  "subscription": "premium",
  "age": 25
}

许可证

MIT