semantic/examples/prov/main.go

// Command prov records where every claim came from, in W3C PROV-O terms: an
// Entity was generated by an Activity, which was run by an Agent. Keeping that
// vocabulary means the record joins other PROV data without translation.
//
// The package declares the three types and the Log they are written to, and
// stops there — where a log is kept is the caller's decision, so the log below
// is forty lines of this program. What the run shows is a derived claim traced
// back through reasoning to two claims, two documents of different formats, and
// the SHA-256 of the bytes each was read from.
//
//	go run ./examples/prov
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/hanzoai/semantic"
	"github.com/hanzoai/semantic/extract"
	"github.com/hanzoai/semantic/ingest"
	"github.com/hanzoai/semantic/kg"
	"github.com/hanzoai/semantic/normalize"
	"github.com/hanzoai/semantic/parse"
	"github.com/hanzoai/semantic/prov"
	"github.com/hanzoai/semantic/reason"
	"github.com/hanzoai/semantic/split"
)

// Two sources of different kinds. One is prose an extractor reads; the other
// is a record whose fields are already the assertion. Both end up as claims,
// and the derived fact below rests on one of each.
const (
	notes    = "Ada Lovelace works for Babbage Engines.\n"
	registry = `[{"child":"Babbage Engines","relation":"part_of","parent":"Engines Group"}]`
)

func main() {
	ctx := context.Background()
	l := &journal{}

	// Whoever bears responsibility. A program and the person who ran it are
	// both agents, and PROV keeps them apart.
	must(l.Agent(prov.Agent{ID: "semantic/pipeline", Kind: "software"}))
	must(l.Agent(prov.Agent{ID: "[email protected]", Kind: "person"}))

	dir := write()
	prose := fromProse(ctx, l, filepath.Join(dir, "notes.txt"))
	record := fromRecord(ctx, l, filepath.Join(dir, "registry.json"))
	target := derive(ctx, l, append(prose, record...))

	trace(l, target)
	fmt.Printf("\nthe log holds %d entities, %d activities, %d agents\n",
		len(l.entities), len(l.activities), len(l.agents))
}

// fromProse is the whole reading pipeline with a PROV entry per stage. The
// origin a reader writes into Doc.Meta is already the provenance of the
// document — reference, size, modification time, content hash — so the entity
// naming the document names its hash, and the activity names what it read.
func fromProse(ctx context.Context, l *journal, path string) []semantic.Triple {
	start := time.Now()
	docs, err := ingest.File{}.Ingest(ctx, path)
	if err != nil {
		log.Fatal(err)
	}
	d := docs[0]
	o := ingest.OriginOf(d)
	docID := "doc:" + o.Hash[:16]

	must(l.Entity(prov.Entity{ID: "file:" + filepath.Base(o.Ref)}))
	must(l.Entity(prov.Entity{ID: docID, GeneratedBy: "act:ingest/prose",
		DerivedFrom: []string{"file:" + filepath.Base(o.Ref)}}))
	must(l.Activity(prov.Activity{ID: "act:ingest/prose", Kind: "ingest", Agent: "semantic/pipeline",
		Start: start, End: time.Now(), Used: []string{"file:" + filepath.Base(o.Ref)}}))
	fmt.Printf("ingest    %-14s → %s   %d bytes\n", filepath.Base(o.Ref), docID, o.Size)

	if d, err = (parse.Any{}).Parse(ctx, d); err != nil {
		log.Fatal(err)
	}
	d.Text = normalize.Text(d.Text)

	start = time.Now()
	chunks, err := split.Sentences{Max: 1}.Split(ctx, d)
	if err != nil {
		log.Fatal(err)
	}
	for _, c := range chunks {
		must(l.Entity(prov.Entity{ID: chunkID(o.Hash, c.Index),
			GeneratedBy: "act:split", DerivedFrom: []string{docID}}))
	}
	must(l.Activity(prov.Activity{ID: "act:split", Kind: "split", Agent: "semantic/pipeline",
		Start: start, End: time.Now(), Used: []string{docID}}))
	fmt.Printf("split     %s → %d chunk\n", docID, len(chunks))

	rules := extract.Rules{Names: map[string]string{
		"Ada Lovelace":    "PERSON",
		"Babbage Engines": "ORG",
	}}
	start = time.Now()
	var facts []semantic.Triple
	var used []string
	for _, c := range chunks {
		ts, err := rules.Extract(ctx, c)
		if err != nil {
			log.Fatal(err)
		}
		for _, t := range ts {
			from := chunkID(o.Hash, t.From.Index)
			used = append(used, from)
			claim(l, t, "act:extract", from)
		}
		facts = append(facts, ts...)
	}
	must(l.Activity(prov.Activity{ID: "act:extract", Kind: "extract", Agent: "semantic/pipeline",
		Start: start, End: time.Now(), Used: used}))
	return facts
}

// fromRecord is the other kind of source. A JSON array is one document per
// element, and the record's own fields are the assertion, so there is nothing
// between the bytes and the claim but a field mapping — which is still an
// activity, and is still recorded as one.
func fromRecord(ctx context.Context, l *journal, path string) []semantic.Triple {
	start := time.Now()
	docs, err := ingest.File{}.Ingest(ctx, path)
	if err != nil {
		log.Fatal(err)
	}
	d := docs[0]
	o := ingest.OriginOf(d)
	docID := "doc:" + o.Hash[:16]

	must(l.Entity(prov.Entity{ID: "file:" + filepath.Base(o.Ref)}))
	must(l.Entity(prov.Entity{ID: docID, GeneratedBy: "act:ingest/record",
		DerivedFrom: []string{"file:" + filepath.Base(o.Ref)}}))
	must(l.Activity(prov.Activity{ID: "act:ingest/record", Kind: "ingest", Agent: "semantic/pipeline",
		Start: start, End: time.Now(), Used: []string{"file:" + filepath.Base(o.Ref)}}))
	fmt.Printf("ingest    %-14s → %s   %d bytes, %d record\n",
		filepath.Base(o.Ref), docID, o.Size, len(docs))

	r := ingest.Record(d)
	t := semantic.Triple{
		Subject:   fmt.Sprint(r["child"]),
		Predicate: fmt.Sprint(r["relation"]),
		Object:    fmt.Sprint(r["parent"]),
		Score:     1,
	}
	claim(l, t, "act:map", docID)
	must(l.Activity(prov.Activity{ID: "act:map", Kind: "map fields", Agent: "semantic/pipeline",
		Start: start, End: time.Now(), Used: []string{docID}}))
	return []semantic.Triple{t}
}

// derive closes the two claims under one rule. Model.Prov renders the closure
// in PROV terms itself — every fact an entity named by its key, every derived
// fact generated by an activity whose kind is the rule that fired, using the
// premises — so the reasoner's record joins the log above with no translation,
// as long as both name a claim the same way. That is why the claims written
// above are named by reason.Key.
func derive(ctx context.Context, l *journal, facts []semantic.Triple) string {
	m, err := reason.Engine{Rules: []reason.Rule{
		must1(reason.Parse("works_for(?p, ?g) :- works_for(?p, ?c), part_of(?c, ?g).")),
	}}.Run(ctx, facts)
	if err != nil {
		log.Fatal(err)
	}

	es, as := m.Prov("semantic/reasoner")
	for _, e := range es {
		if e.GeneratedBy != "" { // given facts were not derived by this run
			must(l.Entity(e))
		}
	}
	for _, a := range as {
		must(l.Activity(a))
	}
	must(l.Agent(prov.Agent{ID: "semantic/reasoner", Kind: "software"}))

	fmt.Printf("\nreason    %d facts given, %d derived\n", len(m.Given()), len(m.Derived()))
	for _, f := range m.Derived() {
		fmt.Printf("          %s\n", f)
	}
	return m.Derived()[0].Key()
}

// claim writes one assertion as a PROV entity. Two ids name the same claim
// here and they are not the same string: reason.Key is what a derivation
// points at, kg.Hash is what the graph calls it. The log is keyed by the
// first, so the reasoner's own record joins it.
func claim(l *journal, t semantic.Triple, by, from string) {
	must(l.Entity(prov.Entity{ID: reason.Key(t), GeneratedBy: by, DerivedFrom: []string{from}}))
	fmt.Printf("claim     %-16s %-10s %-16s  kg id %s\n",
		t.Subject, t.Predicate, t.Object, kg.Hash(t.Subject, t.Predicate, t.Object))
}

// trace walks back from a claim to the bytes it came from. Nothing on the way
// is a stored explanation: each step is an entity naming the activity that
// generated it and the entities that activity used.
func trace(l *journal, target string) {
	fmt.Printf("\nwhere did %s come from?\n", show(target))
	seen := map[string]bool{}
	var walk func(id string, depth int)
	walk = func(id string, depth int) {
		if seen[id] {
			return
		}
		seen[id] = true
		e, ok := l.entities[id]
		if !ok {
			return
		}
		pad := strings.Repeat("   ", depth+1)
		if e.GeneratedBy == "" {
			fmt.Printf("%s%s\n", pad, show(id))
			return
		}
		a := l.activities[e.GeneratedBy]
		fmt.Printf("%s%s\n%s   ↑ %s, by %s\n", pad, show(id), pad, a.Kind, a.Agent)
		for _, up := range e.DerivedFrom {
			walk(up, depth+1)
		}
	}
	walk(target, 0)

	acts, err := l.Trace(target)
	if err != nil {
		log.Fatal(err)
	}
	var kinds []string
	for _, a := range acts {
		kinds = append(kinds, a.Kind)
	}
	fmt.Printf("  every activity behind it, nearest first: %v\n", kinds)
}

// journal is a prov.Log kept in memory. Append-only: an entity written twice
// is the same claim, not a second one.
type journal struct {
	entities   map[string]prov.Entity
	activities map[string]prov.Activity
	agents     map[string]prov.Agent
}

func (j *journal) Entity(e prov.Entity) error {
	if j.entities == nil {
		j.entities = map[string]prov.Entity{}
	}
	if _, ok := j.entities[e.ID]; !ok {
		j.entities[e.ID] = e
	}
	return nil
}

func (j *journal) Activity(a prov.Activity) error {
	if j.activities == nil {
		j.activities = map[string]prov.Activity{}
	}
	j.activities[a.ID] = a
	return nil
}

func (j *journal) Agent(a prov.Agent) error {
	if j.agents == nil {
		j.agents = map[string]prov.Agent{}
	}
	j.agents[a.ID] = a
	return nil
}

// Trace is every activity behind an entity, nearest first: the one that
// generated it, then the ones behind everything it was derived from.
func (j *journal) Trace(id string) ([]prov.Activity, error) {
	var out []prov.Activity
	seen := map[string]bool{}
	queue := []string{id}
	for len(queue) > 0 {
		cur := queue[0]
		queue = queue[1:]
		if seen[cur] {
			continue
		}
		seen[cur] = true
		e, ok := j.entities[cur]
		if !ok || e.GeneratedBy == "" {
			continue
		}
		if a, ok := j.activities[e.GeneratedBy]; ok {
			out = append(out, a)
		}
		queue = append(queue, e.DerivedFrom...)
	}
	return out, nil
}

var _ prov.Log = (*journal)(nil)

// chunkID names one chunk by the document it came from and its place in it.
func chunkID(hash string, index int) string {
	return fmt.Sprintf("chunk:%s/%d", hash[:8], index)
}

// show renders a claim key readably: reason separates the three terms with a
// character no term contains, which is right for a key and wrong for a line of
// output.
func show(id string) string {
	if found := strings.Contains(id, "\x1f"); found {
		parts := strings.Split(id, "\x1f")
		return fmt.Sprintf("%s %s %s", parts[0], parts[1], parts[2])
	}
	return id
}

func write() string {
	dir, err := os.MkdirTemp("", "semantic-prov")
	if err != nil {
		log.Fatal(err)
	}
	for name, body := range map[string]string{"notes.txt": notes, "registry.json": registry} {
		if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil {
			log.Fatal(err)
		}
	}
	return dir
}

func must(err error) {
	if err != nil {
		log.Fatal(err)
	}
}

func must1[T any](v T, err error) T {
	if err != nil {
		log.Fatal(err)
	}
	return v
}