Prolog in Python

Stellt hier eure Projekte vor.
Internetseiten, Skripte, und alles andere bzgl. Python.
Antworten
Benutzeravatar
pillmuncher
User
Beiträge: 1533
Registriert: Samstag 21. März 2009, 22:59
Wohnort: Pfaffenwinkel

Über die letzten Wochen habe ich Hornet komplett überarbeitet:

https://github.com/pillmuncher/hornet

Hornet ist ein Prolog-in-Python, d.h. eine eingebettete Domänen-spezifische Sprache zur Logik-Programierung in Python. Es fußt auf drei grundlegenden Ideen:
  • die DSL selbst, basierend auf Expresison Trees
  • ein Mechanismus zur Ausführung darin geschriebener Logik-Programme, in Form einer dreiläufigen Continuation Monade
  • Tail-Call Elimination in Form eines Decorators und einer Driver-Funktion, um Rekursion verwenden zu können ohne den Ausführungs-Stack zu sprengen
Die DSL verwendet (ähnlich wie SymPy) Expression Trees, hier zur Modellierung von Horn-Klauseln. Daher der Name: HornET. Horn-Klauseln sind ein Konzept aus der Logik, bei der Quantifizierung implizit ist. Horn-Klauseln sind deswegen eine vollständige Form der Prädikaten-Logik erster Stufe (First Order Predicate Logic, FOL).

Der Ausführungsmechanismus beruht auf der Continuation Monade, hier erweitert auf drei Continuations: eine für den Erfolgsfall, eine für das Scheitern um Backtracking zu starten, und eine um Suchbäume zu beschneiden. Jede Stelle im Code, die alternative Lösungswege ausführen kann, wird Choice Point genannt. Bei speziellen Choice Points, den sog. Prunable Choice Points, kann eine Continuation eingeführt werden, die die Alternativen beschneidet, d.h. dass nicht alle Alternativen ausgewertet werden, sofern im Kontext des derzeitigen Ausführungsastes ein Cut steht (in Prolog `!`, hier 'cut').

Warum sollte man Prolog in Python verwenden wollen? Hier einige Beispiele:

Das vordefinierte Standard-Prädikat 'append/3' (d.h. eine logische Prozedur mit drei Argumenten) fügt zwei Listen zu einer dritten zusammen:

Code: Alles auswählen

append([A | B], C, [A | D]).when(
    append(B, C, D),
)
append([], A, A)
Verwendet werden kann es so:

Code: Alles auswählen

from hornet import database
from hornet.symbols import append, L, L1, L2
db = database()
for subst in db.ask(append([1, 2, 3], [4, 5], L)):
    print(subst[L])
das erzeugt das erwartete Ergebnis:

Code: Alles auswählen

[1, 2, 3, 4, 5]
Man kann das dasselbe Programm aber auch "rückwärts" laufen lassen und fragen: Welche Listen ergeben zusammengefügt die Liste '[1, 2, 3, 4, 5]'?

Code: Alles auswählen

for i, subst in enumerate(db.ask(append(L1, L2, [1, 2, 3, 4, 5]))):
    print(f"{i}:", subst[L1], "+", subst[L2])
Ergebnis:

Code: Alles auswählen

0: [1, 2, 3, 4, 5] + []
1: [1, 2, 3, 4] + [5]
2: [1, 2, 3] + [4, 5]
3: [1, 2] + [3, 4, 5]
4: [1] + [2, 3, 4, 5]
5: [] + [1, 2, 3, 4, 5]
Logik-Programmierung kann überall eingesetzt werden, wo symbolische Berechnung das Ziel ist, zB. zur symbolischen Differenzierung:

Code: Alles auswählen

db.tell(
    diff(X, Y, Z).when(d(X, Y, Z)),
    d(X, X, 1).when(
        cut,
    ),
    d(C, X, 0).when(
        is_atomic(C),
        cut,
    ),
    s(X + 0, Y).when(
        cut,
        s(X, Y),
    ),
    s(0 + X, Y).when(
        cut,
        s(X, Y),
    ),
    s(X - 0, Y).when(
        cut,
        s(X, Y),
    ),
    s(0 - X, -Y).when(
        cut,
        s(X, Y),
    ),
    d(U + V, X, A + B).when(
        cut,
        d(U, X, A),
        d(V, X, B),
    ),
    d(U - V, X, A - B).when(
        cut,
        d(U, X, A),
        d(V, X, B),
    ),
    d(0 * U, X, 0).when(
        cut,
    ),
    d(U * 0, X, 0).when(
        cut,
    ),
    d(1 * U, X, V).when(
        cut,
        d(U, X, V),
    ),
    d(U * 1, X, V).when(
        cut,
        d(U, X, V),
    ),
    d(C * U, X, C * A).when(
        is_atomic(C),
        ~equal(C, X), 
        cut, 
        d(U, X, A),
    ),
    d(U * V, X, B * U + A * V).when(
        cut,
        d(U, X, A),
        d(V, X, B),
    ),
    d(0 / U, X, 0).when(
        cut,
    ),
    d(U / V, X, (A * V - B * U) / (V * V)).when(
        cut,
        d(U, X, A),
        d(V, X, B),
    ),
    d(U**C, X, C * A * U ** (C - 1)).when(
        is_atomic(C),
        ~equal(C, X),
        cut,
        d(U, X, A),
    ),
    d(U**-C, X, -C * A * U ** (-C - 1)).when(
        cut,
        d(U, X, A),
    ),
    d(U**C, X, C * A * U ** (C - 1)).when(
        equal(C, -C1),
        is_atomic(C1),
        ~equal(C1, X),
        cut,
        d(U, X, A),
    ),
    d(sin(W), X, Z * cos(W)).when(
        cut,
        d(W, X, Z),
    ),
    d(exp(W), X, Z * exp(W)).when(
        cut,
        d(W, X, Z),
    ),
    d(log(W), X, Z / W).when(
        cut,
        d(W, X, Z),
    ),
    d(cos(W), X, -(Z * sin(W))).when(
        cut,
        d(W, X, Z),
    ),
    # Kürzung
    simp(X, Y).when(s(X, Y)),
    s(A, A).when(
        is_atomic(A),
        cut,
    ),
    # Addition
    s(0 + A, B).when(
        cut,
        s(A, B),
    ),
    s(A + 0, B).when(
        cut,
        s(A, B),
    ),
    ...
)
Beispiel:

Code: Alles auswählen

from hornet.symbols import x, diff, simp, A, B, C
def diff_test(db):
    formula = x**-3 + 2 * x**2 + 7 * (x + 9)
    for subst in db.ask(simp(formula, A), diff(A, x, B), simp(B, C)):
        print(f"{subst[A] = !s}")
        print(f"{subst[B] = !s}")
        print(f"{subst[C] = !s}")
Ergebnis:

Code: Alles auswählen

subst[A] = x ** -3 + 2 * x ** 2 + 7 * (x + 9)
subst[B] = -3 * 1 * x ** (-3 - 1) + 2 * 2 * 1 * x ** (2 - 1) + 7 * (1 + 0)
subst[C] = -3 * x ** -4 + 4 * x + 7
Ein anderes Beispiel ist FizzBuzz:

Code: Alles auswählen

from toolz import take

from hornet import DCGs, database
from hornet.symbols import M, N, R, S, V, Ws, _, cut, equal, fb, fizzbuzz
from hornet.symbols import ifelse, inline, join, let, phrase, word, words

def main(db):
    db.tell(
        fizzbuzz(R).when(
            fb(1, R),
        ),
        fb(N, R).when(
            phrase(words(N), Ws),
            join(Ws, S),
            ifelse(
                equal(S, ""),
                equal(N, R),
                equal(S, R),
            ),
        ),
        fb(N, R).when(
            let(M, N + 1),
            fb(M, R),
        ),
        *DCGs(
            words(N).when(
                word(3, N),
                word(5, N),
                word(7, N),
                inline(cut),
            ),
            word(3, N).when(inline(let(M, N % 3), equal(M, 0)), ["fizz"]),
            word(5, N).when(inline(let(M, N % 5), equal(M, 0)), ["buzz"]),
            word(7, N).when(inline(let(M, N % 7), equal(M, 0)), ["quux"]),
            word(_, _),
        ),
    )

    for s in take(1111, db.ask(fizzbuzz(V))):
        print(s[V])


if __name__ == "__main__":
    main(database())
Hier wird eine DCG, eine Definite Clause Grammar, verwendet um die Ausgabe-Strings zu berechnen. Ergebnis:

Code: Alles auswählen

1
2
fizz
4
buzz
fizz
quux
8
fizz
buzz
11
fizz
13
quux
fizzbuzz
16
...
104
fizzbuzzquux
106
...
Dieses Beispiel zeigt auch, wie die jeweils nächste Zeile mittels Backtracking erzeugt wird.

Warum habe ich das gebaut? Weil ich wissen wollte, wie Prolog funktioniert. Hornet ist nicht fertig, es fehlen zB. noch viele vordefinierte System-Prädikate, Unit Tests, und ich kann auch nicht garantieren, dass es keine Bugs gibt. Als Proof-of-Concept allerdings funktioniert es und zeigt, dass es nicht nur möglich, sondern auch schnell genug ist für einige Anwendungsfälle. Einfache Parser etwa sind oft schnell genug und können sehr leicht gebaut werden, insbesondere Dank der integrierten DCG-Expansion. Ich habe auch großen Wert auf Interoperabilität mit Python gelegt, so dass Prädikate leicht in Python selbst implementiert werden können, etwa so:

Code: Alles auswählen

@db.tell
@predicate(univ(C, L))
def _univ(db: Database, subst: Subst) -> Step:
    cin = subst[C]
    lin = subst[L]
    assert isinstance(cin, Functor | Atom | Variable)
    match cin:
        case Atom(name=name):
            return _unify(lin, promote([cin]))(db, subst.map)
        case Functor(name=name, args=args):
            return _unify(lin, promote([Atom(name), *args]))(db, subst.map)
        case Variable():
            assert isinstance(lin, Cons | Empty)
            head, *tail = to_python_list(lin)
            assert isinstance(head, Atom)
            return _unify(cin, Functor(head.name, *tail))(db, subst.map)
Über Rückmeldungen würde ich mich freuen.
In specifications, Murphy's Law supersedes Ohm's.
Benutzeravatar
pillmuncher
User
Beiträge: 1533
Registriert: Samstag 21. März 2009, 22:59
Wohnort: Pfaffenwinkel

Es gibt zwei neue Features: Ein Ereigniskalkül und eine modallogische Erweiterung. Damit lassen sich Fragen formulieren wie: "Zeigte Alice, unser CFO, schuldhaftes Verhalten in Bezug auf die Verletzung der Regel 'reg31' durch Transaktion `tx17`? Beides hier zusammen in einem Beispiel:

Code: Alles auswählen

from hornet import database
from hornet.clauses import Database, Environment, Subst, predicate, resolve
from hornet.combinators import Step
from hornet.modalities import deontic_worlds, epistemic_worlds, exists, forall, modal
from hornet.symbols import (
    TX, Agent, Amount, Fact, Limit, Regulation, Report, Role, T, T_report, Tmax,
    _, accessible, after, appointed, deemed_known, enacted, greater, happens_at,
    holds_at, in_force, initiates, mentions, obligation, published, review, tenure,
    threshold, transaction, violated, violation,
)


def compliance(db: Database) -> Database:
    child = modal(db.new_child())

    @child.tell
    @predicate(deemed_known(Agent, Fact, T))
    def _(db: Database, subst: Subst) -> Step[Database, Environment]:
        return forall(
            deontic_worlds(subst[Agent], subst[T]),
            exists(
                epistemic_worlds(subst[Agent], subst[T]),
                resolve(accessible(subst[Agent], subst[Fact], subst[T])),
            ),
        )(db, subst.env)

    return child


def main() -> None:
    db = compliance(database())

    # using Robert Kowalski's Event Calculus here:
    db.tell(
        # Regulation reg31 came into force at time 0.
        happens_at(enacted('reg31'), 0),
        # Alice was appointed CFO at time 1.
        happens_at(appointed('alice', 'cfo'), 1),
        # Bob transferred 250,000 at time 2.
        happens_at(transaction('tx17', 'bob', 250_000), 2),
        # Report rep42 was published at time 3.
        happens_at(published('rep42'), 3),
        # Alice did not review report rep42.
        # happens_at(performed('alice', review('rep42'), _))
        # Regulation reg31 sets a transaction threshold of 100,000.
        threshold('reg31', 100_000),
        # Report rep42 mentions transaction tx17.
        mentions('rep42', 'tx17'),
        # Enacting a regulation causes it to be in force.
        initiates(enacted(Regulation), in_force(Regulation)),
        # Appointing an agent to a role causes them to hold that role.
        initiates(appointed(Agent, Role), tenure(Agent, Role)),
        # A transaction violates a regulation if it exceeds the
        # threshold while the regulation is in force.
        violation(TX, Regulation).when(
            happens_at(transaction(TX, _, Amount), T),
            holds_at(in_force(Regulation), T),
            threshold(Regulation, Limit),
            greater(Amount, Limit),
        ),
        # An agent can access a violation if they can access the
        # underlying transaction.
        accessible(Agent, violated(TX, Regulation), Tmax).when(
            accessible(Agent, transaction(TX, Amount), Tmax),
            violation(TX, Regulation),
        ),
        # An agent can access a transaction if a report mentioning it
        # was published while they held the CFO role.
        accessible(Agent, transaction(TX, Amount), Tmax).when(
            mentions(Report, TX),
            happens_at(published(Report), T_report),
            ~after(T_report, Tmax),
            holds_at(tenure(Agent, 'cfo'), Tmax),
        ),
        # An agent is obligated to review a report if it was published
        # while they held the CFO role.
        obligation(Agent, review(Report), T_report).when(
            happens_at(published(Report), T_report),
            holds_at(tenure(Agent, 'cfo'), T_report),
        ),
    )

    # The Query: Is Alice legally "deemed to know" the violation?
    # Logic: In all worlds where she reviews the report (Deontic),
    # she gains access to the transaction data (Epistemic).
    query = deemed_known('alice', violated('tx17', 'reg31'), 3)
    result = any(db.ask(query))
    print(f'Liability Check: {query} -> {result}')


if __name__ == '__main__':
    main()
Ergebnis:

Code: Alles auswählen

❯ py -m examples.audit
Liability Check: deemed_known(alice, violated(tx17, reg31), 3) -> True
Das Ganze basiert auf einer konstruktivistischen Variante der Kripke-Semantik. Dabei werden die Operatoren notwendigerweise φ, möglicherweise φ und unmöglicherweise φ als Quantoren auf möglichen Welten definiert - Welten die der Fall sein könnten. Bei Kripke werden diese Welten als gegeben vorausgesetzt, bei mir werden sie konstruiert als Variationen über einer bestehenden tatsächlichen Welt, einer Datenbank, die nicht-modale Fakten und Regeln enthält. Epistemische Welten werden konstruiert, indem in abgeleiteten Welten bestimmte Fakten als unerreichbar markiert werden. Deontische Welten dagegen werden monoton erweitert - es werden nur Fakten und Regeln hinzugefügt. Diese definieren Obligationen, also das, was man tun und was man unterlassen muss. Daraus lässt sich dann ableiten: Agent A weiß dass φ, wenn Fakt φ in allen möglichen Welten für A erreichbar war, und war verpflichtet zu ψ wenn Obligation ψ für A in allen Welten gilt. Diese modalen Dimensionen lassen sich auch kombinieren, so dass man epistemische Obligationen formulieren kann: "Als CFO war Alice dazu verpflichtet zu wissen, dass φ, und somit war ihr tatsächliches Nichtwissen rechtlich schuldhaft".

Das Ereigniskalkül ist einfach: Jedes Ereignis (Event) tritt zu einem bestimmten Zeitpunkt ein, bestimmte Ereignisse initiieren sog. Fluents, die dann so lange gelten bis sie durch ein anderes Ereignis terminiert werden. Alice wird zu einem bestimmten Zeitpunkt CFO und bleibt es solange, bis sie selbst kündigt, ihr gekündigt wird, oder sie eine andere Rolle in der Firma übernimmt. Daraus lassen sich Kausalketten und logische Zustände zu bestimmten Zeitpunkten ableiten: Alice war verpflichtet den Report 'rep42` zu lesen, weil sie zum Zeitpunkt der Veröffentlichung CFO war und das zu den Pflichten des CFOs gehört.

Als nächstes plane ich probabilistische Logik, die beschreibt was wahrscheinlich ist, und abduktive Logik, die Hypothesen (counterfactuals) generieren kann, also Fakten die der Fall sein könnten, so dass man Fragen stellen kann wie "Unter welchen Voraussetzungen wäre die Transaktion 'tx17' kein Regelverstoß gewesen?", oder auch "Wie können wir den Regelverstoß rechtlich ausgleichen?". Und dann noch Gödel-Löb Logik, die es erlaubt, Fragen in Bezug auf Beweisbarkeit zu stellen. Das ist in einem konstruktivistischen System wie Hornet insofern interessant, als dass hier Wahrheit und Beweisbarkeit bereits zusammenfallen. Selbstbezügliche Fragen über Wahrheit erzeugen daher kein Paradox, wie in nicht-konstruktivistischen Systemen.

Weiterhin plane ich, einen Tracer einzubauen. Dieser kann dann den gesamten Resolutionssuchbaum aufzeichnen, inklusive aller Äste, die nicht zu einem Ergebnis geführt haben. Ein Tracer ist für Audit-Szenarien wie die oben gezeigten unablässlich, weil ein Auditor damit nicht nur Tatsachenurteile präsentiert bekommnt, sondern auch, wie diese zustandegekommen sind. Außerdem können derartige Informationen dazu verwendet werden, ein LLM zu füttern. Das wäre dann eine neuartige Basis für NeSy, neuro-symbolische KI. Aktuelle NeSy Systeme operieren nur auf simplen Propositionen wie "die Tasse steht auf dem Tisch". Mit modallogischen Erweiterungen könnte man ein deutlich komplexeres Set von Propositionen als Basis verwenden, das sogar Hypothesen und kontrafaktische Propositionen beinhaltet.

Hier das modallogische Framework:
https://github.com/pillmuncher/hornet/b ... alities.py

Und here die technische Doku:
https://github.com/pillmuncher/hornet/b ... l_logic.md
In specifications, Murphy's Law supersedes Ohm's.
Antworten