如何实现支持交互式添加事实的Prolog/Datalog程序Web API

一、先修正你的祖先规则 #

你提供的Prolog祖先规则存在逻辑错误，第二个子句的变量绑定无法建立正确递归关系，正确规则应为：

ancestor(X, Y) :- parent(X, Y).
ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).

二、Prolog方案（含ichiban/prolog包装实现） #

1. 基础动态事实优化 #

你提到的:- dynamic parent/2 + assertz/1是标准方案，可通过模块隔离优化，避免全局命名空间污染：

:- module(family, [parent/2, ancestor/2, add_parent/2]).
:- dynamic parent/2.

ancestor(X, Y) :- parent(X, Y).
ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).

add_parent(X, Y) :- assertz(parent(X, Y)).

封装add_parent/2谓词，对外暴露安全的添加接口，降低直接调用assertz/1的注入风险。

2. 用ichiban/prolog（Go）实现Web API #

ichiban/prolog是Go语言的Prolog解释器，适合快速搭建Web服务：

步骤1：初始化Prolog环境

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"

	"github.com/ichiban/prolog"
)

var p *prolog.Interpreter

func init() {
	p = prolog.New(nil)
	// 加载家族规则
	_, err := p.Exec(context.Background(), `
		:- module(family, [parent/2, ancestor/2, add_parent/2]).
		:- dynamic parent/2.
		ancestor(X, Y) :- parent(X, Y).
		ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).
		add_parent(X, Y) :- assertz(parent(X, Y)).
	`)
	if err != nil {
		panic(err)
	}
}

步骤2：实现POST /facts接口

func addFactHandler(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Fact string `json:"fact"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid request body", http.StatusBadRequest)
		return
	}

	// 严格校验事实格式
	if !strings.HasPrefix(req.Fact, "parent(") || !strings.HasSuffix(req.Fact, ").") {
		http.Error(w, "only parent/2 facts are allowed", http.StatusBadRequest)
		return
	}

	// 执行添加操作
	factContent := strings.TrimSuffix(req.Fact, ".")
	_, err := p.Exec(context.Background(), fmt.Sprintf("family:%s", factContent))
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to add fact: %v", err), http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusCreated)
	fmt.Fprintf(w, "fact added successfully")
}

步骤3：实现GET /ancestor接口

func queryAncestorHandler(w http.ResponseWriter, r *http.Request) {
	x := r.URL.Query().Get("x")
	y := r.URL.Query().Get("y")
	if x == "" || y == "" {
		http.Error(w, "missing x or y parameter", http.StatusBadRequest)
		return
	}

	// 执行查询
	sols, err := p.Query(context.Background(), fmt.Sprintf("family:ancestor('%s', '%s')", x, y))
	if err != nil {
		http.Error(w, fmt.Sprintf("query failed: %v", err), http.StatusInternalServerError)
		return
	}
	defer sols.Close()

	// 返回查询结果
	if sols.Next() {
		w.WriteHeader(http.StatusOK)
		fmt.Fprintf(w, "true")
	} else {
		w.WriteHeader(http.StatusOK)
		fmt.Fprintf(w, "false")
	}
}

步骤4：启动服务

func main() {
	http.HandleFunc("/facts", addFactHandler)
	http.HandleFunc("/ancestor", queryAncestorHandler)
	fmt.Println("server starting on :8080")
	http.ListenAndServe(":8080", nil)
}

三、Datalog方案 #

Datalog是Prolog的子集，语义严谨，天生适合增量逻辑查询，非常匹配你的场景：

1. 核心Datalog规则 #

parent(X, Y).  % 事实模板
ancestor(X, Y) :- parent(X, Y).
ancestor(X, Y) :- parent(X, Z), ancestor(Z, Y).

2. 实现思路（以Python PyDatalog为例） #

步骤1：初始化Datalog环境

from pyDatalog import pyDatalog

pyDatalog.create_terms('parent, ancestor, X, Y, Z')

步骤2：封装事实与查询逻辑

def add_parent_fact(parent_name, child_name):
    parent(parent_name, child_name)

def is_ancestor(x, y):
    return bool(ancestor(x, y))

步骤3：用Flask封装Web API

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/facts', methods=['POST'])
def add_fact():
    data = request.get_json()
    fact = data.get('fact')
    if not fact.startswith('parent(') or not fact.endswith('.'):
        return jsonify({"error": "only parent/2 facts are allowed"}), 400
    
    # 解析事实参数
    args = fact[7:-2].split(', ')
    if len(args) != 2:
        return jsonify({"error": "invalid fact format"}), 400
    
    add_parent_fact(args[0].strip(), args[1].strip())
    return jsonify({"message": "fact added successfully"}), 201

@app.route('/ancestor', methods=['GET'])
def query_ancestor():
    x = request.args.get('x')
    y = request.args.get('y')
    if not x or not y:
        return jsonify({"error": "missing x or y parameter"}), 400
    
    result = is_ancestor(x, y)
    return jsonify({"result": result}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

四、方案对比与优化建议 #

通用优化点 #

• 输入校验：严格限制仅允许parent/2格式的事实，避免恶意代码注入；
• 持久化：将事实存储到数据库（如SQLite、PostgreSQL），重启服务后不丢失；
• 并发控制：Prolog需加锁保护动态事实，Datalog优先选择支持多上下文的引擎；
• 错误处理：添加更详细的错误反馈，帮助用户定位问题。

内容的提问来源于stack exchange，提问作者craigpastro