More package hacking

trustgraph-flow/scripts/chunker-recursive

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.chunking.recursive import run
+
+run()
+

trustgraph-flow/scripts/chunker-token

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.chunking.token import run
+
+run()
+

trustgraph-flow/scripts/concat-parquet

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+
+"""
+Concatenates multiple parquet files into a single parquet output
+"""
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pandas as pd
+import sys
+import argparse
+
+parser = argparse.ArgumentParser(
+    prog="combine-parquet",
+    description=__doc__
+)
+
+parser.add_argument(
+    '-i', '--input',
+    nargs='*',
+    help=f'Input files'
+)
+
+parser.add_argument(
+    '-o', '--output',
+    help=f'Output files'
+)
+
+args = parser.parse_args()
+
+df = None
+
+for file in args.input:
+
+    part = pq.read_table(file).to_pandas()
+
+    if df is None:
+        df = part
+    else:
+        df = pd.concat([df, part], ignore_index=True)
+
+if df is not None:
+
+    table = pa.Table.from_pandas(df)
+    pq.write_table(table, args.output)

trustgraph-flow/scripts/de-query-milvus

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.doc_embeddings.milvus import run
+
+run()
+

trustgraph-flow/scripts/de-query-qdrant

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.doc_embeddings.qdrant import run
+
+run()
+

trustgraph-flow/scripts/de-write-milvus

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.doc_embeddings.milvus import run
+
+run()
+

trustgraph-flow/scripts/de-write-qdrant

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.doc_embeddings.qdrant import run
+
+run()
+

trustgraph-flow/scripts/document-rag

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.retrieval.document_rag import run
+
+run()
+

trustgraph-flow/scripts/dump-parquet

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import pyarrow as pa
+import pyarrow.csv as pc
+import pyarrow.parquet as pq
+import pandas as pd
+import sys
+
+df = None
+
+for file in sys.argv[1:]:
+
+    part = pq.read_table(file).to_pandas()
+
+    if df is None:
+        df = part
+    else:
+        df = pd.concat([df, part], ignore_index=True)
+
+if df is not None:
+
+    table = pa.Table.from_pandas(df)
+    pc.write_csv(table, sys.stdout.buffer)
+

trustgraph-flow/scripts/embeddings-ollama

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.embeddings.ollama import run
+
+run()
+

trustgraph-flow/scripts/embeddings-vectorize

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.embeddings.vectorize import run
+
+run()
+

trustgraph-flow/scripts/ge-dump-parquet

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.dump.graph_embeddings.parquet import run
+
+run()
+

trustgraph-flow/scripts/ge-query-milvus

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.graph_embeddings.milvus import run
+
+run()
+

trustgraph-flow/scripts/ge-query-qdrant

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.graph_embeddings.qdrant import run
+
+run()
+

trustgraph-flow/scripts/ge-write-milvus

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.graph_embeddings.milvus import run
+
+run()
+

trustgraph-flow/scripts/ge-write-qdrant

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.graph_embeddings.qdrant import run
+
+run()
+

trustgraph-flow/scripts/graph-rag

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.retrieval.graph_rag import run
+
+run()
+

trustgraph-flow/scripts/graph-show

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+"""
+Connects to the graph query service and dumps all graph edges.
+"""
+
+import argparse
+import os
+from trustgraph.clients.triples_query_client import TriplesQueryClient
+
+default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+
+def show_graph(pulsar):
+
+    tq = TriplesQueryClient(pulsar_host=pulsar)
+
+    rows = tq.request(None, None, None, limit=10_000_000)
+
+    for row in rows:
+        print(row.s.value, row.p.value, row.o.value)
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='graph-show',
+        description=__doc__,
+    )
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    args = parser.parse_args()
+
+    try:
+
+        show_graph(args.pulsar_host)
+
+    except Exception as e:
+
+        print("Exception:", e, flush=True)
+
+main()
+

trustgraph-flow/scripts/graph-to-turtle

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+
+"""
+Connects to the graph query service and dumps all graph edges.
+"""
+
+import argparse
+import os
+from trustgraph.clients.triples_query_client import TriplesQueryClient
+import rdflib
+import io
+import sys
+
+default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+
+def show_graph(pulsar):
+
+    tq = TriplesQueryClient(pulsar_host=pulsar)
+
+    rows = tq.request(None, None, None, limit=10_000_000)
+
+    g = rdflib.Graph()
+
+    for row in rows:
+
+        sv = rdflib.term.URIRef(row.s.value)
+        pv = rdflib.term.URIRef(row.p.value)
+
+        if row.o.is_uri:
+
+            # Skip malformed URLs with spaces in
+            if " " in row.o.value:
+                continue
+
+            ov = rdflib.term.URIRef(row.o.value)
+        else:
+            ov = rdflib.term.Literal(row.o.value)
+
+        g.add((sv, pv, ov))
+
+    g.serialize(destination="output.ttl", format="turtle")
+
+    buf = io.BytesIO()
+
+    g.serialize(destination=buf, format="turtle")
+
+    sys.stdout.write(buf.getvalue().decode("utf-8"))
+
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='graph-show',
+        description=__doc__,
+    )
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    args = parser.parse_args()
+
+    try:
+
+        show_graph(args.pulsar_host)
+
+    except Exception as e:
+
+        print("Exception:", e, flush=True)
+
+main()
+

trustgraph-flow/scripts/init-pulsar-manager

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+
+CSRF_TOKEN=$(curl http://localhost:7750/pulsar-manager/csrf-token)
+
+curl \
+    -H "X-XSRF-TOKEN: $CSRF_TOKEN" \
+    -H "Cookie: XSRF-TOKEN=$CSRF_TOKEN;" \
+    -H 'Content-Type: application/json' \
+    -X PUT \
+    http://localhost:7750/pulsar-manager/users/superuser \
+    -d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "username@test.org"}'

trustgraph-flow/scripts/kg-extract-definitions

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.extract.kg.definitions import run
+
+run()
+

trustgraph-flow/scripts/kg-extract-relationships

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.extract.kg.relationships import run
+
+run()
+

trustgraph-flow/scripts/kg-extract-topics

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.extract.kg.topics import run
+
+run()
+

trustgraph-flow/scripts/load-graph-embeddings

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+
+"""
+Loads Graph embeddings into TrustGraph processing.
+"""
+
+import pulsar
+from pulsar.schema import JsonSchema
+from trustgraph.schema import GraphEmbeddings, Value
+from trustgraph.schema import graph_embeddings_store_queue
+import argparse
+import os
+import time
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from trustgraph.log_level import LogLevel
+
+class Loader:
+
+    def __init__(
+            self,
+            pulsar_host,
+            output_queue,
+            log_level,
+            file,
+    ):
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level.to_pulsar())
+        )
+
+        self.producer = self.client.create_producer(
+            topic=output_queue,
+            schema=JsonSchema(GraphEmbeddings),
+            chunking_enabled=True,
+        )
+
+        self.file = file
+
+    def run(self):
+
+        try:
+
+            path = self.file
+
+            print("Reading file...")
+            table = pq.read_table(path)
+            print("Loaded.")
+
+            names = set(table.column_names)
+
+            if "embeddings" not in names:
+                print("No 'embeddings' column")
+
+            if "entity" not in names:
+                print("No 'entity' column")
+
+            embc = table.column("embeddings")
+            entc = table.column("entity")
+
+            for emb, ent in zip(embc, entc):
+
+                b = emb.as_py()
+                n = ent.as_py()
+
+                r = GraphEmbeddings(
+                    vectors=b,
+                    entity=Value(
+                        value=n,
+                        is_uri=n.startswith("https:")
+                    )
+                )
+
+                self.producer.send(r)
+
+        except Exception as e:
+            print(e, flush=True)
+            
+    def __del__(self):
+        self.client.close()
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='loader',
+        description=__doc__,
+    )
+
+    default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+    default_output_queue = graph_embeddings_store_queue
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-o', '--output-queue',
+        default=default_output_queue,
+        help=f'Output queue (default: {default_output_queue})'
+    )
+
+    parser.add_argument(
+        '-l', '--log-level',
+        type=LogLevel,
+        default=LogLevel.ERROR,
+        choices=list(LogLevel),
+        help=f'Output queue (default: info)'
+    )
+
+    parser.add_argument(
+        '-f', '--file',
+        required=True,
+        help=f'File to load'
+    )
+
+    args = parser.parse_args()
+
+    while True:
+
+        try:
+            p = Loader(
+                pulsar_host=args.pulsar_host,
+                output_queue=args.output_queue,
+                log_level=args.log_level,
+                file=args.file,
+            )
+
+            p.run()
+
+            print("File loaded.")
+            break
+
+        except Exception as e:
+
+            print("Exception:", e, flush=True)
+            print("Will retry...", flush=True)
+
+        time.sleep(10)
+
+main()
+

trustgraph-flow/scripts/load-pdf

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+
+"""
+Loads a PDF document into TrustGraph processing.
+"""
+
+import pulsar
+from pulsar.schema import JsonSchema
+from trustgraph.schema import Document, Source, document_ingest_queue
+import base64
+import hashlib
+import argparse
+import os
+import time
+
+from trustgraph.log_level import LogLevel
+
+class Loader:
+
+    def __init__(
+            self,
+            pulsar_host,
+            output_queue,
+            log_level,
+            file,
+    ):
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level.to_pulsar())
+        )
+
+        self.producer = self.client.create_producer(
+            topic=output_queue,
+            schema=JsonSchema(Document),
+            chunking_enabled=True,
+        )
+
+        self.file = file
+
+    def run(self):
+
+        try:
+
+            path = self.file
+            data = open(path, "rb").read()
+
+            id = hashlib.sha256(path.encode("utf-8")).hexdigest()[0:8]
+
+            r = Document(
+                source=Source(
+                    source=path,
+                    title=path,
+                    id=id,
+                ),
+                data=base64.b64encode(data),
+            )
+
+            self.producer.send(r)
+
+        except Exception as e:
+            print(e, flush=True)
+            
+    def __del__(self):
+        self.client.close()
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='loader',
+        description=__doc__,
+    )
+
+    default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+    default_output_queue = document_ingest_queue
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-o', '--output-queue',
+        default=default_output_queue,
+        help=f'Output queue (default: {default_output_queue})'
+    )
+
+    parser.add_argument(
+        '-l', '--log-level',
+        type=LogLevel,
+        default=LogLevel.ERROR,
+        choices=list(LogLevel),
+        help=f'Output queue (default: info)'
+    )
+
+    parser.add_argument(
+        '-f', '--file',
+        required=True,
+        help=f'File to load'
+    )
+
+    args = parser.parse_args()
+
+    while True:
+
+        try:
+            p = Loader(
+                pulsar_host=args.pulsar_host,
+                output_queue=args.output_queue,
+                log_level=args.log_level,
+                file=args.file,
+            )
+
+            p.run()
+
+            print("File loaded.")
+            break
+
+        except Exception as e:
+
+            print("Exception:", e, flush=True)
+            print("Will retry...", flush=True)
+
+        time.sleep(10)
+
+main()
+

trustgraph-flow/scripts/load-text

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+
+"""
+Loads a text document into TrustGraph processing.
+"""
+
+import pulsar
+from pulsar.schema import JsonSchema
+from trustgraph.schema import TextDocument, Source, text_ingest_queue
+import base64
+import hashlib
+import argparse
+import os
+import time
+
+from trustgraph.log_level import LogLevel
+
+class Loader:
+
+    def __init__(
+            self,
+            pulsar_host,
+            output_queue,
+            log_level,
+            file,
+    ):
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level.to_pulsar())
+        )
+
+        self.producer = self.client.create_producer(
+            topic=output_queue,
+            schema=JsonSchema(TextDocument),
+            chunking_enabled=True,
+        )
+
+        self.file = file
+
+    def run(self):
+
+        try:
+
+            path = self.file
+            data = open(path, "rb").read()
+
+            id = hashlib.sha256(path.encode("utf-8")).hexdigest()[0:8]
+
+            r = TextDocument(
+                source=Source(
+                    source=path,
+                    title=path,
+                    id=id,
+                ),
+                text=data,
+            )
+
+            self.producer.send(r)
+
+        except Exception as e:
+            print(e, flush=True)
+            
+    def __del__(self):
+        self.client.close()
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='loader',
+        description=__doc__,
+    )
+
+    default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+    default_output_queue = text_ingest_queue
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-o', '--output-queue',
+        default=default_output_queue,
+        help=f'Output queue (default: {default_output_queue})'
+    )
+
+    parser.add_argument(
+        '-l', '--log-level',
+        type=LogLevel,
+        default=LogLevel.ERROR,
+        choices=list(LogLevel),
+        help=f'Output queue (default: info)'
+    )
+
+    parser.add_argument(
+        '-f', '--file',
+        required=True,
+        help=f'File to load'
+    )
+
+    args = parser.parse_args()
+
+    while True:
+
+        try:
+            p = Loader(
+                pulsar_host=args.pulsar_host,
+                output_queue=args.output_queue,
+                log_level=args.log_level,
+                file=args.file,
+            )
+
+            p.run()
+
+            print("File loaded.")
+            break
+
+        except Exception as e:
+
+            print("Exception:", e, flush=True)
+            print("Will retry...", flush=True)
+
+        time.sleep(10)
+
+main()
+

trustgraph-flow/scripts/load-triples

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+
+"""
+Loads Graph embeddings into TrustGraph processing.
+"""
+
+import pulsar
+from pulsar.schema import JsonSchema
+from trustgraph.schema import Triple, Value
+from trustgraph.schema import triples_store_queue
+import argparse
+import os
+import time
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from trustgraph.log_level import LogLevel
+
+class Loader:
+
+    def __init__(
+            self,
+            pulsar_host,
+            output_queue,
+            log_level,
+            file,
+    ):
+
+        self.client = pulsar.Client(
+            pulsar_host,
+            logger=pulsar.ConsoleLogger(log_level.to_pulsar())
+        )
+
+        self.producer = self.client.create_producer(
+            topic=output_queue,
+            schema=JsonSchema(Triple),
+            chunking_enabled=True,
+        )
+
+        self.file = file
+
+    def run(self):
+
+        try:
+
+            path = self.file
+
+            print("Reading file...")
+            table = pq.read_table(path)
+            print("Loaded.")
+
+            names = set(table.column_names)
+
+            if "s" not in names:
+                print("No 's' column")
+
+            if "p" not in names:
+                print("No 'p' column")
+
+            if "o" not in names:
+                print("No 'o' column")
+
+            sc = table.column("s")
+            pc = table.column("p")
+            oc = table.column("o")
+
+            for s, p, o in zip(sc, pc, oc):
+
+                r = Triple(
+                    s=Value(value=s.as_py(), is_uri=True),
+                    p=Value(value=p.as_py(), is_uri=True),
+                    o=Value(value=o.as_py(), is_uri=o.as_py().startswith("https:"))
+                )
+
+                self.producer.send(r)
+
+        except Exception as e:
+            print(e, flush=True)
+            
+    def __del__(self):
+        self.client.close()
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='loader',
+        description=__doc__,
+    )
+
+    default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+    default_output_queue = triples_store_queue
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-o', '--output-queue',
+        default=default_output_queue,
+        help=f'Output queue (default: {default_output_queue})'
+    )
+
+    parser.add_argument(
+        '-l', '--log-level',
+        type=LogLevel,
+        default=LogLevel.ERROR,
+        choices=list(LogLevel),
+        help=f'Output queue (default: info)'
+    )
+
+    parser.add_argument(
+        '-f', '--file',
+        required=True,
+        help=f'File to load'
+    )
+
+    args = parser.parse_args()
+
+    while True:
+
+        try:
+            p = Loader(
+                pulsar_host=args.pulsar_host,
+                output_queue=args.output_queue,
+                log_level=args.log_level,
+                file=args.file,
+            )
+
+            p.run()
+
+            print("File loaded.")
+            break
+
+        except Exception as e:
+
+            print("Exception:", e, flush=True)
+            print("Will retry...", flush=True)
+
+        time.sleep(10)
+
+main()
+

trustgraph-flow/scripts/metering

@@ -0,0 +1,5 @@
+#!/usr/bin/env python3
+
+from trustgraph.metering import run
+
+run()

trustgraph-flow/scripts/object-extract-row

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.extract.object.row import run
+
+run()
+

trustgraph-flow/scripts/oe-write-milvus

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.object_embeddings.milvus import run
+
+run()
+

trustgraph-flow/scripts/pdf-decoder

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.decoding.pdf import run
+
+run()
+

trustgraph-flow/scripts/prompt-generic

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.prompt.generic import run
+
+run()
+

trustgraph-flow/scripts/prompt-template

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.prompt.template import run
+
+run()
+

trustgraph-flow/scripts/query-document-rag

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+
+"""
+Uses the Document RAG service to answer a query
+"""
+
+import argparse
+import os
+from trustgraph.clients.document_rag_client import DocumentRagClient
+
+default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+
+def query(pulsar, query):
+
+    rag = DocumentRagClient(pulsar_host=pulsar)
+    resp = rag.request(query)
+    print(resp)
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='graph-show',
+        description=__doc__,
+    )
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-q', '--query',
+        required=True,
+        help=f'Query to execute',
+    )
+
+    args = parser.parse_args()
+
+    try:
+
+        query(args.pulsar_host, args.query)
+
+    except Exception as e:
+
+        print("Exception:", e, flush=True)
+
+main()
+

trustgraph-flow/scripts/query-graph-rag

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+
+"""
+Uses the GraphRAG service to answer a query
+"""
+
+import argparse
+import os
+from trustgraph.clients.graph_rag_client import GraphRagClient
+
+default_pulsar_host = os.getenv("PULSAR_HOST", 'pulsar://localhost:6650')
+
+def query(pulsar, query):
+
+    rag = GraphRagClient(pulsar_host=pulsar)
+    resp = rag.request(query)
+    print(resp)
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='graph-show',
+        description=__doc__,
+    )
+
+    parser.add_argument(
+        '-p', '--pulsar-host',
+        default=default_pulsar_host,
+        help=f'Pulsar host (default: {default_pulsar_host})',
+    )
+
+    parser.add_argument(
+        '-q', '--query',
+        required=True,
+        help=f'Query to execute',
+    )
+
+    args = parser.parse_args()
+
+    try:
+
+        query(args.pulsar_host, args.query)
+
+    except Exception as e:
+
+        print("Exception:", e, flush=True)
+
+main()
+

trustgraph-flow/scripts/rows-write-cassandra

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.rows.cassandra import run
+
+run()
+

trustgraph-flow/scripts/run-processing

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.processing import run
+
+run()
+

trustgraph-flow/scripts/text-completion-azure

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.azure import run
+
+run()
+

trustgraph-flow/scripts/text-completion-bedrock

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.bedrock import run
+
+run()
+

trustgraph-flow/scripts/text-completion-claude

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.claude import run
+
+run()
+

trustgraph-flow/scripts/text-completion-cohere

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.cohere import run
+
+run()
+

trustgraph-flow/scripts/text-completion-llamafile

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.llamafile import run
+
+run()
+

trustgraph-flow/scripts/text-completion-ollama

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.ollama import run
+
+run()
+

trustgraph-flow/scripts/text-completion-openai

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.openai import run
+
+run()
+

trustgraph-flow/scripts/text-completion-vertexai

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.model.text_completion.vertexai import run
+
+run()
+

trustgraph-flow/scripts/tg-init-pulsar

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+
+"""
+Initialises Pulsar with Trustgraph tenant / namespaces & policy
+"""
+
+import requests
+import time
+import argparse
+
+default_pulsar_admin_url = "http://pulsar:8080"
+
+def get_clusters(url):
+
+    print("Get clusters...", flush=True)
+
+    resp = requests.get(f"{url}/admin/v2/clusters")
+
+    if resp.status_code != 200: raise RuntimeError("Could not fetch clusters")
+
+    return resp.json()
+
+def ensure_tenant(url, tenant, clusters):
+
+    resp = requests.get(f"{url}/admin/v2/tenants/{tenant}")
+
+    if resp.status_code == 200:
+        print(f"Tenant {tenant} already exists.", flush=True)
+        return
+
+    resp = requests.put(
+        f"{url}/admin/v2/tenants/{tenant}",
+        json={
+            "adminRoles": [],
+            "allowedClusters": clusters,
+        }
+    )
+
+    if resp.status_code != 204:
+        print(resp.text, flush=True)
+        raise RuntimeError("Tenant creation failed.")
+
+    print(f"Tenant {tenant} created.", flush=True)
+
+def ensure_namespace(url, tenant, namespace, config):
+
+    resp = requests.get(f"{url}/admin/v2/namespaces/{tenant}/{namespace}")
+
+    if resp.status_code == 200:
+        print(f"Namespace {tenant}/{namespace} already exists.", flush=True)
+        return
+
+    resp = requests.put(
+        f"{url}/admin/v2/namespaces/{tenant}/{namespace}",
+        json=config,
+    )
+
+    if resp.status_code != 204:
+        print(resp.status_code, flush=True)
+        print(resp.text, flush=True)
+        raise RuntimeError(f"Namespace {tenant}/{namespace} creation failed.")
+
+    print(f"Namespace {tenant}/{namespace} created.", flush=True)
+
+def init(url, tenant="tg"):
+
+    clusters = get_clusters(url)
+
+    ensure_tenant(url, tenant, clusters)
+
+    ensure_namespace(url, tenant, "flow", {})
+
+    ensure_namespace(url, tenant, "request", {})
+
+    ensure_namespace(url, tenant, "response", {
+        "retention_policies": {
+            "retentionSizeInMB": -1,
+            "retentionTimeInMinutes": 3,
+        }
+    })
+
+def main():
+
+    parser = argparse.ArgumentParser(
+        prog='tg-init-pulsar',
+        description=__doc__,
+    )
+
+    parser.add_argument(
+        '-p', '--pulsar-admin-url',
+        default=default_pulsar_admin_url,
+        help=f'Pulsar admin URL (default: {default_pulsar_admin_url})',
+    )
+
+    args = parser.parse_args()
+
+    while True:
+
+        try:
+
+            print(flush=True)
+            print(
+                f"Initialising with Pulsar {args.pulsar_admin_url}...",
+                flush=True
+            )
+            init(args.pulsar_admin_url, "tg")
+            print("Initialisation complete.", flush=True)
+            break
+
+        except Exception as e:
+
+            print("Exception:", e, flush=True)
+
+            print("Sleeping...", flush=True)
+            time.sleep(2)
+            print("Will retry...", flush=True)
+
+main()
+

trustgraph-flow/scripts/tg-processor-state

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import requests
+import tabulate
+
+url = 'http://localhost:9090/api/v1/query?query=processor_state%7Bprocessor_state%3D%22running%22%7D'
+
+resp = requests.get(url)
+
+obj = resp.json()
+
+tbl = [
+    [
+        m["metric"]["job"],
+        "running" if int(m["value"][1]) > 0 else "down"
+    ]
+    for m in obj["data"]["result"]
+]
+
+print(tabulate.tabulate(
+    tbl, tablefmt="pretty", headers=["processor", "state"],
+    stralign="left"
+))
+

trustgraph-flow/scripts/triples-dump-parquet

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.dump.triples.parquet import run
+
+run()
+

trustgraph-flow/scripts/triples-query-cassandra

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.triples.cassandra import run
+
+run()
+

trustgraph-flow/scripts/triples-query-neo4j

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.query.triples.neo4j import run
+
+run()
+

trustgraph-flow/scripts/triples-write-cassandra

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.triples.cassandra import run
+
+run()
+

trustgraph-flow/scripts/triples-write-neo4j

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from trustgraph.storage.triples.neo4j import run
+
+run()
+

trustgraph-flow/setup.py

@@ -0,0 +1,116 @@
+import setuptools
+import os
+import importlib
+
+with open("README.md", "r") as fh:
+    long_description = fh.read()
+
+# Load a version number module
+spec = importlib.util.spec_from_file_location(
+    'version', 'trustgraph/flow_version.py'
+)
+version_module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(version_module)
+
+version = version_module.__version__
+
+setuptools.setup(
+    name="trustgraph-flow",
+    version=version,
+    author="trustgraph.ai",
+    author_email="security@trustgraph.ai",
+    description="TrustGraph provides a means to run a pipeline of flexible AI processing components in a flexible means to achieve a processing pipeline.",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/trustgraph-ai/trustgraph",
+    packages=setuptools.find_namespace_packages(
+        where='./',
+    ),
+    classifiers=[ 
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
+        "Operating System :: OS Independent",
+    ],
+    python_requires='>=3.8',
+    download_url = "https://github.com/trustgraph-ai/trustgraph/archive/refs/tags/v" + version + ".tar.gz",
+    install_requires=[
+        "trustgraph-base",
+        "urllib3",
+        "rdflib",
+        "pymilvus",
+        "langchain",
+        "langchain-core",
+        "langchain-text-splitters",
+        "langchain-community",
+        "requests",
+        "cassandra-driver",
+        "pulsar-client",
+        "pypdf",
+        "qdrant-client",
+        "tabulate",
+        "anthropic",
+        "google-cloud-aiplatform",
+        "pyyaml",
+        "prometheus-client",
+        "pyarrow",
+        "cohere",
+        "boto3",
+        "openai",
+        "neo4j",
+        "tiktoken",
+    ],
+    scripts=[
+        "scripts/chunker-recursive",
+        "scripts/chunker-token",
+        "scripts/concat-parquet",
+        "scripts/de-query-milvus",
+        "scripts/de-query-qdrant",
+        "scripts/de-write-milvus",
+        "scripts/de-write-qdrant",
+        "scripts/document-rag",
+        "scripts/dump-parquet",
+        "scripts/embeddings-ollama",
+        "scripts/embeddings-vectorize",
+        "scripts/ge-dump-parquet",
+        "scripts/ge-query-milvus",
+        "scripts/ge-query-qdrant",
+        "scripts/ge-write-milvus",
+        "scripts/ge-write-qdrant",
+        "scripts/graph-rag",
+        "scripts/graph-show",
+        "scripts/graph-to-turtle",
+        "scripts/init-pulsar-manager",
+        "scripts/kg-extract-definitions",
+        "scripts/kg-extract-topics",
+        "scripts/kg-extract-relationships",
+        "scripts/load-graph-embeddings",
+        "scripts/load-pdf",
+        "scripts/load-text",
+        "scripts/load-triples",
+        "scripts/metering",
+        "scripts/object-extract-row",
+        "scripts/oe-write-milvus",
+        "scripts/pdf-decoder",
+        "scripts/prompt-generic",
+        "scripts/prompt-template",
+        "scripts/query-document-rag",
+        "scripts/query-graph-rag",
+        "scripts/rows-write-cassandra",
+        "scripts/run-processing",
+        "scripts/text-completion-azure",
+        "scripts/text-completion-bedrock",
+        "scripts/text-completion-claude",
+        "scripts/text-completion-cohere",
+        "scripts/text-completion-llamafile",
+        "scripts/text-completion-ollama",
+        "scripts/text-completion-openai",
+        "scripts/text-completion-vertexai",
+        "scripts/tg-init-pulsar",
+        "scripts/tg-processor-state",
+        "scripts/triples-dump-parquet",
+        "scripts/triples-query-cassandra",
+        "scripts/triples-query-neo4j",
+        "scripts/triples-write-cassandra",
+        "scripts/triples-write-neo4j",
+    ]
+)

trustgraph-flow/trustgraph/chunking/recursive/__init__.py

@@ -0,0 +1,3 @@
+
+from . chunker import *
+

trustgraph-flow/trustgraph/chunking/recursive/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . chunker import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/chunking/recursive/chunker.py

@@ -0,0 +1,108 @@
+
+"""
+Simple decoder, accepts text documents on input, outputs chunks from the
+as text as separate output objects.
+"""
+
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from prometheus_client import Histogram
+
+from ... schema import TextDocument, Chunk, Source
+from ... schema import text_ingest_queue, chunk_ingest_queue
+from ... log_level import LogLevel
+from ... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = text_ingest_queue
+default_output_queue = chunk_ingest_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        chunk_size = params.get("chunk_size", 2000)
+        chunk_overlap = params.get("chunk_overlap", 100)
+        
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": TextDocument,
+                "output_schema": Chunk,
+            }
+        )
+
+        if not hasattr(__class__, "chunk_metric"):
+            __class__.chunk_metric = Histogram(
+                'chunk_size', 'Chunk size',
+                buckets=[100, 160, 250, 400, 650, 1000, 1600,
+                         2500, 4000, 6400, 10000, 16000]
+            )
+
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=chunk_size,
+            chunk_overlap=chunk_overlap,
+            length_function=len,
+            is_separator_regex=False,
+        )
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Chunking {v.source.id}...", flush=True)
+
+        texts = self.text_splitter.create_documents(
+            [v.text.decode("utf-8")]
+        )
+
+        for ix, chunk in enumerate(texts):
+
+            id = v.source.id + "-c" + str(ix)
+
+            r = Chunk(
+                source=Source(
+                    source=v.source.source,
+                    id=id,
+                    title=v.source.title
+                ),
+                chunk=chunk.page_content.encode("utf-8"),
+            )
+
+            __class__.chunk_metric.observe(len(chunk.page_content))
+
+            self.send(r)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '-z', '--chunk-size',
+            type=int,
+            default=2000,
+            help=f'Chunk size (default: 2000)'
+        )
+
+        parser.add_argument(
+            '-v', '--chunk-overlap',
+            type=int,
+            default=100,
+            help=f'Chunk overlap (default: 100)'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/chunking/token/__init__.py

@@ -0,0 +1,3 @@
+
+from . chunker import *
+

trustgraph-flow/trustgraph/chunking/token/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . chunker import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/chunking/token/chunker.py

@@ -0,0 +1,107 @@
+
+"""
+Simple decoder, accepts text documents on input, outputs chunks from the
+as text as separate output objects.
+"""
+
+from langchain_text_splitters import TokenTextSplitter
+from prometheus_client import Histogram
+
+from ... schema import TextDocument, Chunk, Source
+from ... schema import text_ingest_queue, chunk_ingest_queue
+from ... log_level import LogLevel
+from ... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = text_ingest_queue
+default_output_queue = chunk_ingest_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        chunk_size = params.get("chunk_size", 250)
+        chunk_overlap = params.get("chunk_overlap", 15)
+        
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": TextDocument,
+                "output_schema": Chunk,
+            }
+        )
+
+        if not hasattr(__class__, "chunk_metric"):
+            __class__.chunk_metric = Histogram(
+                'chunk_size', 'Chunk size',
+                buckets=[100, 160, 250, 400, 650, 1000, 1600,
+                         2500, 4000, 6400, 10000, 16000]
+            )
+
+        self.text_splitter = TokenTextSplitter(
+            encoding_name="cl100k_base",
+            chunk_size=chunk_size,
+            chunk_overlap=chunk_overlap,
+        )
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Chunking {v.source.id}...", flush=True)
+
+        texts = self.text_splitter.create_documents(
+            [v.text.decode("utf-8")]
+        )
+
+        for ix, chunk in enumerate(texts):
+
+            id = v.source.id + "-c" + str(ix)
+
+            r = Chunk(
+                source=Source(
+                    source=v.source.source,
+                    id=id,
+                    title=v.source.title
+                ),
+                chunk=chunk.page_content.encode("utf-8"),
+            )
+
+            __class__.chunk_metric.observe(len(chunk.page_content))
+
+            self.send(r)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '-z', '--chunk-size',
+            type=int,
+            default=250,
+            help=f'Chunk size (default: 250)'
+        )
+
+        parser.add_argument(
+            '-v', '--chunk-overlap',
+            type=int,
+            default=15,
+            help=f'Chunk overlap (default: 15)'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/decoding/pdf/__init__.py

@@ -0,0 +1,3 @@
+
+from . pdf_decoder import *
+

trustgraph-flow/trustgraph/decoding/pdf/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . pdf_decoder import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py

@@ -0,0 +1,87 @@
+
+"""
+Simple decoder, accepts PDF documents on input, outputs pages from the
+PDF document as text as separate output objects.
+"""
+
+import tempfile
+import base64
+from langchain_community.document_loaders import PyPDFLoader
+
+from ... schema import Document, TextDocument, Source
+from ... schema import document_ingest_queue, text_ingest_queue
+from ... log_level import LogLevel
+from ... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = document_ingest_queue
+default_output_queue = text_ingest_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": Document,
+                "output_schema": TextDocument,
+            }
+        )
+
+        print("PDF inited")
+
+    def handle(self, msg):
+
+        print("PDF message received")
+
+        v = msg.value()
+
+        print(f"Decoding {v.source.id}...", flush=True)
+
+        with tempfile.NamedTemporaryFile(delete_on_close=False) as fp:
+
+            fp.write(base64.b64decode(v.data))
+            fp.close()
+
+            with open(fp.name, mode='rb') as f:
+
+                loader = PyPDFLoader(fp.name)
+                pages = loader.load()
+
+                for ix, page in enumerate(pages):
+
+                    id = v.source.id + "-p" + str(ix)
+                    r = TextDocument(
+                        source=Source(
+                            source=v.source.source,
+                            title=v.source.title,
+                            id=id,
+                        ),
+                        text=page.page_content.encode("utf-8"),
+                    )
+
+                    self.send(r)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/direct/cassandra.py

@@ -0,0 +1,108 @@
+
+from cassandra.cluster import Cluster
+from cassandra.auth import PlainTextAuthProvider
+
+class TrustGraph:
+
+    def __init__(self, hosts=None):
+
+        if hosts is None:
+            hosts = ["localhost"]
+            
+        self.cluster = Cluster(hosts)
+        self.session = self.cluster.connect()
+
+        self.init()
+
+    def clear(self):
+
+        self.session.execute("""
+            drop keyspace if exists trustgraph;
+        """);
+
+        self.init()
+
+    def init(self):
+
+        self.session.execute("""
+            create keyspace if not exists trustgraph
+                with replication = { 
+                   'class' : 'SimpleStrategy', 
+                   'replication_factor' : 1 
+                };
+        """);
+
+        self.session.set_keyspace('trustgraph')
+
+        self.session.execute("""
+            create table if not exists triples (
+                s text,
+                p text,
+                o text,
+                PRIMARY KEY (s, p, o)
+            );
+        """);
+
+        self.session.execute("""
+            create index if not exists triples_p
+                ON triples (p);
+        """);
+
+        self.session.execute("""
+            create index if not exists triples_o
+                ON triples (o);
+        """);
+
+    def insert(self, s, p, o):
+    
+        self.session.execute(
+            "insert into triples (s, p, o) values (%s, %s, %s)",
+            (s, p, o)
+        )
+
+    def get_all(self, limit=50):
+        return self.session.execute(
+            f"select s, p, o from triples limit {limit}"
+        )
+
+    def get_s(self, s, limit=10):
+        return self.session.execute(
+            f"select p, o from triples where s = %s limit {limit}",
+            (s,)
+        )
+
+    def get_p(self, p, limit=10):
+        return self.session.execute(
+            f"select s, o from triples where p = %s limit {limit}",
+            (p,)
+        )
+
+    def get_o(self, o, limit=10):
+        return self.session.execute(
+            f"select s, p from triples where o = %s limit {limit}",
+            (o,)
+        )
+
+    def get_sp(self, s, p, limit=10):
+        return self.session.execute(
+            f"select o from triples where s = %s and p = %s limit {limit}",
+            (s, p)
+        )
+
+    def get_po(self, p, o, limit=10):
+        return self.session.execute(
+            f"select s from triples where p = %s and o = %s allow filtering limit {limit}",
+            (p, o)
+        )
+
+    def get_os(self, o, s, limit=10):
+        return self.session.execute(
+            f"select p from triples where o = %s and s = %s limit {limit}",
+            (o, s)
+        )
+
+    def get_spo(self, s, p, o, limit=10):
+        return self.session.execute(
+            f"""select s as x from triples where s = %s and p = %s and o = %s limit {limit}""",
+            (s, p, o)
+        )

trustgraph-flow/trustgraph/direct/milvus_doc_embeddings.py

@@ -0,0 +1,138 @@
+
+from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType
+import time
+
+class DocVectors:
+
+    def __init__(self, uri="http://localhost:19530", prefix='doc'):
+
+        self.client = MilvusClient(uri=uri)
+
+        # Strategy is to create collections per dimension.  Probably only
+        # going to be using 1 anyway, but that means we don't need to
+        # hard-code the dimension anywhere, and no big deal if more than
+        # one are created.
+        self.collections = {}
+
+        self.prefix = prefix
+
+        # Time between reloads
+        self.reload_time = 90
+
+        # Next time to reload - this forces a reload at next window
+        self.next_reload = time.time() + self.reload_time
+        print("Reload at", self.next_reload)
+
+    def init_collection(self, dimension):
+
+        collection_name = self.prefix + "_" + str(dimension)
+
+        pkey_field = FieldSchema(
+            name="id",
+            dtype=DataType.INT64,
+            is_primary=True,
+            auto_id=True,
+        )
+
+        vec_field = FieldSchema(
+            name="vector",
+            dtype=DataType.FLOAT_VECTOR,
+            dim=dimension,
+        )
+
+        doc_field = FieldSchema(
+            name="doc",
+            dtype=DataType.VARCHAR,
+            max_length=65535,
+        )
+
+        schema = CollectionSchema(
+            fields = [pkey_field, vec_field, doc_field],
+            description = "Document embedding schema",
+        )
+
+        self.client.create_collection(
+            collection_name=collection_name,
+            schema=schema,
+            metric_type="COSINE",
+        )
+
+        index_params = MilvusClient.prepare_index_params()
+
+        index_params.add_index(
+            field_name="vector",
+            metric_type="COSINE",
+            index_type="IVF_SQ8",
+            index_name="vector_index",
+            params={ "nlist": 128 }
+        )
+
+        self.client.create_index(
+            collection_name=collection_name,
+            index_params=index_params
+        )
+
+        self.collections[dimension] = collection_name
+
+    def insert(self, embeds, doc):
+
+        dim = len(embeds)
+
+        if dim not in self.collections:
+            self.init_collection(dim)
+    
+        data = [
+            {
+                "vector": embeds,
+                "doc": doc,
+            }
+        ]
+
+        self.client.insert(
+            collection_name=self.collections[dim],
+            data=data
+        )
+
+    def search(self, embeds, fields=["doc"], limit=10):
+
+        dim = len(embeds)
+
+        if dim not in self.collections:
+            self.init_collection(dim)
+
+        coll = self.collections[dim]
+
+        search_params = {
+            "metric_type": "COSINE",
+            "params": {
+                "radius": 0.1,
+                "range_filter": 0.8
+            }
+        }
+
+        print("Loading...")
+        self.client.load_collection(
+            collection_name=coll,
+        )
+
+        print("Searching...")
+
+        res = self.client.search(
+            collection_name=coll,
+            data=[embeds],
+            limit=limit,
+            output_fields=fields,
+            search_params=search_params,
+        )[0]
+
+
+        # If reload time has passed, unload collection
+        if time.time() > self.next_reload:
+            print("Unloading, reload at", self.next_reload)
+            self.client.release_collection(
+                collection_name=coll,
+            )
+            self.next_reload = time.time() + self.reload_time
+
+        return res
+

trustgraph-flow/trustgraph/direct/milvus_graph_embeddings.py

@@ -0,0 +1,138 @@
+
+from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType
+import time
+
+class EntityVectors:
+
+    def __init__(self, uri="http://localhost:19530", prefix='entity'):
+
+        self.client = MilvusClient(uri=uri)
+
+        # Strategy is to create collections per dimension.  Probably only
+        # going to be using 1 anyway, but that means we don't need to
+        # hard-code the dimension anywhere, and no big deal if more than
+        # one are created.
+        self.collections = {}
+
+        self.prefix = prefix
+
+        # Time between reloads
+        self.reload_time = 90
+
+        # Next time to reload - this forces a reload at next window
+        self.next_reload = time.time() + self.reload_time
+        print("Reload at", self.next_reload)
+
+    def init_collection(self, dimension):
+
+        collection_name = self.prefix + "_" + str(dimension)
+
+        pkey_field = FieldSchema(
+            name="id",
+            dtype=DataType.INT64,
+            is_primary=True,
+            auto_id=True,
+        )
+
+        vec_field = FieldSchema(
+            name="vector",
+            dtype=DataType.FLOAT_VECTOR,
+            dim=dimension,
+        )
+
+        entity_field = FieldSchema(
+            name="entity",
+            dtype=DataType.VARCHAR,
+            max_length=65535,
+        )
+
+        schema = CollectionSchema(
+            fields = [pkey_field, vec_field, entity_field],
+            description = "Graph embedding schema",
+        )
+
+        self.client.create_collection(
+            collection_name=collection_name,
+            schema=schema,
+            metric_type="COSINE",
+        )
+
+        index_params = MilvusClient.prepare_index_params()
+
+        index_params.add_index(
+            field_name="vector",
+            metric_type="COSINE",
+            index_type="IVF_SQ8",
+            index_name="vector_index",
+            params={ "nlist": 128 }
+        )
+
+        self.client.create_index(
+            collection_name=collection_name,
+            index_params=index_params
+        )
+
+        self.collections[dimension] = collection_name
+
+    def insert(self, embeds, entity):
+
+        dim = len(embeds)
+
+        if dim not in self.collections:
+            self.init_collection(dim)
+    
+        data = [
+            {
+                "vector": embeds,
+                "entity": entity,
+            }
+        ]
+
+        self.client.insert(
+            collection_name=self.collections[dim],
+            data=data
+        )
+
+    def search(self, embeds, fields=["entity"], limit=10):
+
+        dim = len(embeds)
+
+        if dim not in self.collections:
+            self.init_collection(dim)
+
+        coll = self.collections[dim]
+
+        search_params = {
+            "metric_type": "COSINE",
+            "params": {
+                "radius": 0.1,
+                "range_filter": 0.8
+            }
+        }
+
+        print("Loading...")
+        self.client.load_collection(
+            collection_name=coll,
+        )
+
+        print("Searching...")
+
+        res = self.client.search(
+            collection_name=coll,
+            data=[embeds],
+            limit=limit,
+            output_fields=fields,
+            search_params=search_params,
+        )[0]
+
+
+        # If reload time has passed, unload collection
+        if time.time() > self.next_reload:
+            print("Unloading, reload at", self.next_reload)
+            self.client.release_collection(
+                collection_name=coll,
+            )
+            self.next_reload = time.time() + self.reload_time
+
+        return res
+

trustgraph-flow/trustgraph/direct/milvus_object_embeddings.py

@@ -0,0 +1,154 @@
+
+from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType
+import time
+
+class ObjectVectors:
+
+    def __init__(self, uri="http://localhost:19530", prefix='obj'):
+
+        self.client = MilvusClient(uri=uri)
+
+        # Strategy is to create collections per dimension.  Probably only
+        # going to be using 1 anyway, but that means we don't need to
+        # hard-code the dimension anywhere, and no big deal if more than
+        # one are created.
+        self.collections = {}
+
+        self.prefix = prefix
+
+        # Time between reloads
+        self.reload_time = 90
+
+        # Next time to reload - this forces a reload at next window
+        self.next_reload = time.time() + self.reload_time
+        print("Reload at", self.next_reload)
+
+    def init_collection(self, dimension, name):
+
+        collection_name = self.prefix + "_" + name + "_" + str(dimension)
+
+        pkey_field = FieldSchema(
+            name="id",
+            dtype=DataType.INT64,
+            is_primary=True,
+            auto_id=True,
+        )
+
+        vec_field = FieldSchema(
+            name="vector",
+            dtype=DataType.FLOAT_VECTOR,
+            dim=dimension,
+        )
+
+        name_field = FieldSchema(
+            name="name",
+            dtype=DataType.VARCHAR,
+            max_length=65535,
+        )
+
+        key_name_field = FieldSchema(
+            name="key_name",
+            dtype=DataType.VARCHAR,
+            max_length=65535,
+        )
+
+        key_field = FieldSchema(
+            name="key",
+            dtype=DataType.VARCHAR,
+            max_length=65535,
+        )
+
+        schema = CollectionSchema(
+            fields = [
+                pkey_field, vec_field, name_field, key_name_field, key_field
+            ],
+            description = "Object embedding schema",
+        )
+
+        self.client.create_collection(
+            collection_name=collection_name,
+            schema=schema,
+            metric_type="COSINE",
+        )
+
+        index_params = MilvusClient.prepare_index_params()
+
+        index_params.add_index(
+            field_name="vector",
+            metric_type="COSINE",
+            index_type="IVF_SQ8",
+            index_name="vector_index",
+            params={ "nlist": 128 }
+        )
+
+        self.client.create_index(
+            collection_name=collection_name,
+            index_params=index_params
+        )
+
+        self.collections[(dimension, name)] = collection_name
+
+    def insert(self, embeds, name, key_name, key):
+
+        dim = len(embeds)
+
+        if (dim, name) not in self.collections:
+            self.init_collection(dim, name)
+    
+        data = [
+            {
+                "vector": embeds,
+                "name": name,
+                "key_name": key_name,
+                "key": key,
+            }
+        ]
+
+        self.client.insert(
+            collection_name=self.collections[(dim, name)],
+            data=data
+        )
+
+    def search(self, embeds, name, fields=["key_name", "name"], limit=10):
+
+        dim = len(embeds)
+
+        if dim not in self.collections:
+            self.init_collection(dim, name)
+
+        coll = self.collections[(dim, name)]
+
+        search_params = {
+            "metric_type": "COSINE",
+            "params": {
+                "radius": 0.1,
+                "range_filter": 0.8
+            }
+        }
+
+        print("Loading...")
+        self.client.load_collection(
+            collection_name=coll,
+        )
+
+        print("Searching...")
+
+        res = self.client.search(
+            collection_name=coll,
+            data=[embeds],
+            limit=limit,
+            output_fields=fields,
+            search_params=search_params,
+        )[0]
+
+
+        # If reload time has passed, unload collection
+        if time.time() > self.next_reload:
+            print("Unloading, reload at", self.next_reload)
+            self.client.release_collection(
+                collection_name=coll,
+            )
+            self.next_reload = time.time() + self.reload_time
+
+        return res
+

trustgraph-flow/trustgraph/document_rag.py

@@ -0,0 +1,132 @@
+
+from . clients.document_embeddings_client import DocumentEmbeddingsClient
+from . clients.triples_query_client import TriplesQueryClient
+from . clients.embeddings_client import EmbeddingsClient
+from . clients.prompt_client import PromptClient
+
+from . schema import DocumentEmbeddingsRequest, DocumentEmbeddingsResponse
+from . schema import TriplesQueryRequest, TriplesQueryResponse
+from . schema import prompt_request_queue
+from . schema import prompt_response_queue
+from . schema import embeddings_request_queue
+from . schema import embeddings_response_queue
+from . schema import document_embeddings_request_queue
+from . schema import document_embeddings_response_queue
+
+LABEL="http://www.w3.org/2000/01/rdf-schema#label"
+DEFINITION="http://www.w3.org/2004/02/skos/core#definition"
+
+class DocumentRag:
+
+    def __init__(
+            self,
+            pulsar_host="pulsar://pulsar:6650",
+            pr_request_queue=None,
+            pr_response_queue=None,
+            emb_request_queue=None,
+            emb_response_queue=None,
+            de_request_queue=None,
+            de_response_queue=None,
+            verbose=False,
+            module="test",
+    ):
+
+        self.verbose=verbose
+
+        if pr_request_queue is None:
+            pr_request_queue = prompt_request_queue
+
+        if pr_response_queue is None:
+            pr_response_queue = prompt_response_queue
+
+        if emb_request_queue is None:
+            emb_request_queue = embeddings_request_queue
+
+        if emb_response_queue is None:
+            emb_response_queue = embeddings_response_queue
+
+        if de_request_queue is None:
+            de_request_queue = document_embeddings_request_queue
+
+        if de_response_queue is None:
+            de_response_queue = document_embeddings_response_queue
+
+        if self.verbose:
+            print("Initialising...", flush=True)
+
+        # FIXME: Configurable
+        self.entity_limit = 20
+
+        self.de_client = DocumentEmbeddingsClient(
+            pulsar_host=pulsar_host,
+            subscriber=module + "-de",
+            input_queue=de_request_queue,
+            output_queue=de_response_queue,
+        )            
+
+        self.embeddings = EmbeddingsClient(
+            pulsar_host=pulsar_host,
+            input_queue=emb_request_queue,
+            output_queue=emb_response_queue,
+            subscriber=module + "-emb",
+        )
+
+        self.lang = PromptClient(
+            pulsar_host=pulsar_host,
+            input_queue=pr_request_queue,
+            output_queue=pr_response_queue,
+            subscriber=module + "-de-prompt",
+        )
+
+        if self.verbose:
+            print("Initialised", flush=True)
+
+    def get_vector(self, query):
+
+        if self.verbose:
+            print("Compute embeddings...", flush=True)
+
+        qembeds = self.embeddings.request(query)
+
+        if self.verbose:
+            print("Done.", flush=True)
+
+        return qembeds
+
+    def get_docs(self, query):
+
+        vectors = self.get_vector(query)
+
+        if self.verbose:
+            print("Get entities...", flush=True)
+
+        docs = self.de_client.request(
+            vectors, self.entity_limit
+        )
+
+        if self.verbose:
+            print("Docs:", flush=True)
+            for doc in docs:
+                print(doc, flush=True)
+
+        return docs
+
+    def query(self, query):
+
+        if self.verbose:
+            print("Construct prompt...", flush=True)
+
+        docs = self.get_docs(query)
+
+        if self.verbose:
+            print("Invoke LLM...", flush=True)
+            print(docs)
+            print(query)
+
+        resp = self.lang.request_document_prompt(query, docs)
+
+        if self.verbose:
+            print("Done", flush=True)
+
+        return resp
+

trustgraph-flow/trustgraph/dump/graph_embeddings/parquet/__init__.py

@@ -0,0 +1,3 @@
+
+from . processor import *
+

trustgraph-flow/trustgraph/dump/graph_embeddings/parquet/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . write import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/dump/graph_embeddings/parquet/processor.py

@@ -0,0 +1,85 @@
+
+"""
+Write graph embeddings to parquet files in a directory.
+"""
+
+import pulsar
+import base64
+import os
+import argparse
+import time
+
+from .... schema import GraphEmbeddings
+from .... schema import graph_embeddings_store_queue
+from .... base import Consumer
+
+from . writer import ParquetWriter
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = graph_embeddings_store_queue
+default_subscriber = module
+default_graph_host='localhost'
+default_directory = "."
+default_file_template = "graph-embeds-{id}.parquet"
+default_rotation_time = 60
+        
+class Processor(Consumer):
+
+    def __init__(self, **params):
+        
+        input_queue = params.get("input_queue", default_input_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        directory = params.get("directory", default_directory)
+        file_template = params.get("file_template", default_file_template)
+        rotation_time = params.get("rotation_time", default_rotation_time)
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "subscriber": subscriber,
+                "input_schema": GraphEmbeddings,
+            }
+        )
+
+        self.writer = ParquetWriter(directory, file_template, rotation_time)
+
+    def __del__(self):
+        if hasattr(self, "writer"):
+            del self.writer
+
+    def handle(self, msg):
+
+        v = msg.value()
+        self.writer.write(v.vectors, v.entity.value)
+
+    @staticmethod
+    def add_args(parser):
+
+        Consumer.add_args(
+            parser, default_input_queue, default_subscriber,
+        )
+
+        parser.add_argument(
+            '-d', '--directory',
+            default=default_directory,
+            help=f'Directory to write to (default: {default_directory})'
+        )
+
+        parser.add_argument(
+            '-f', '--file-template',
+            default=default_file_template,
+            help=f'Directory to write to (default: {default_file_template})'
+        )
+
+        parser.add_argument(
+            '-t', '--rotation-time',
+            type=int,
+            default=default_rotation_time,
+            help=f'Rotation time / seconds (default: {default_rotation_time})'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/dump/graph_embeddings/parquet/writer.py

@@ -0,0 +1,94 @@
+
+import threading
+import queue
+import time
+import uuid
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+class ParquetWriter:
+
+    def __init__(self, directory, file_template, rotation_time):
+        self.directory = directory
+        self.file_template = file_template
+        self.rotation_time = rotation_time
+
+        self.q = queue.Queue()
+
+        self.running = True
+
+        self.thread = threading.Thread(target=(self.writer_thread))
+        self.thread.start()
+
+    def writer_thread(self):
+
+        items = []
+
+        timeout = None
+
+        while self.running:
+
+            try:
+
+                item = self.q.get(timeout=1)
+
+                if timeout == None:
+                    timeout = time.time() + self.rotation_time
+
+                items.append(item)
+
+            except queue.Empty:
+                pass
+
+            if timeout:
+                if time.time() > timeout:
+
+                    self.write_file(items)
+                    timeout = None
+                    items = []
+
+    def write_file(self, items):
+
+        try:
+
+            schema = pa.schema([
+                pa.field('embeddings', pa.list_(pa.list_(pa.float64()))),
+                pa.field('entity', pa.string()),
+            ])
+
+            fname = self.file_template.format(id=str(uuid.uuid4()))
+            path = f"{self.directory}/{fname}"
+
+            writer = pq.ParquetWriter(path, schema)
+
+            batch = pa.record_batch(
+                [
+                    [i[0] for i in items],
+                    [i[1] for i in items],
+                ],
+                names=['embeddings', 'entity']
+            )
+
+            writer.write_batch(batch)
+
+            writer.close()
+
+            print(f"Wrote {path}.")
+
+        except Exception as e:
+
+            print("Parquet write:", e)
+
+    def write(self, embeds, ent):
+        self.q.put((embeds, ent))
+
+    def __del__(self):
+
+        self.running = False
+
+        if hasattr(self, "q"):
+            self.thread.join()
+
+            
+
+        

trustgraph-flow/trustgraph/dump/triples/parquet/__init__.py

@@ -0,0 +1,3 @@
+
+from . processor import *
+

trustgraph-flow/trustgraph/dump/triples/parquet/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . write import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/dump/triples/parquet/processor.py

@@ -0,0 +1,85 @@
+
+"""
+Write graphs triples to parquet files in a directory.
+"""
+
+import pulsar
+import base64
+import os
+import argparse
+import time
+
+from .... schema import Triple
+from .... schema import triples_store_queue
+from .... base import Consumer
+
+from . writer import ParquetWriter
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = triples_store_queue
+default_subscriber = module
+default_graph_host='localhost'
+default_directory = "."
+default_file_template = "triples-{id}.parquet"
+default_rotation_time = 60
+        
+class Processor(Consumer):
+
+    def __init__(self, **params):
+        
+        input_queue = params.get("input_queue", default_input_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        directory = params.get("directory", default_directory)
+        file_template = params.get("file_template", default_file_template)
+        rotation_time = params.get("rotation_time", default_rotation_time)
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "subscriber": subscriber,
+                "input_schema": Triple,
+            }
+        )
+
+        self.writer = ParquetWriter(directory, file_template, rotation_time)
+
+    def __del__(self):
+        if hasattr(self, "writer"):
+            del self.writer
+
+    def handle(self, msg):
+
+        v = msg.value()
+        self.writer.write(v.s.value, v.p.value, v.o.value)
+
+    @staticmethod
+    def add_args(parser):
+
+        Consumer.add_args(
+            parser, default_input_queue, default_subscriber,
+        )
+
+        parser.add_argument(
+            '-d', '--directory',
+            default=default_directory,
+            help=f'Directory to write to (default: {default_directory})'
+        )
+
+        parser.add_argument(
+            '-f', '--file-template',
+            default=default_file_template,
+            help=f'Directory to write to (default: {default_file_template})'
+        )
+
+        parser.add_argument(
+            '-t', '--rotation-time',
+            type=int,
+            default=default_rotation_time,
+            help=f'Rotation time / seconds (default: {default_rotation_time})'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/dump/triples/parquet/writer.py

@@ -0,0 +1,96 @@
+
+import threading
+import queue
+import time
+import uuid
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+class ParquetWriter:
+
+    def __init__(self, directory, file_template, rotation_time):
+        self.directory = directory
+        self.file_template = file_template
+        self.rotation_time = rotation_time
+
+        self.q = queue.Queue()
+
+        self.running = True
+
+        self.thread = threading.Thread(target=(self.writer_thread))
+        self.thread.start()
+
+    def writer_thread(self):
+
+        triples = []
+
+        timeout = None
+
+        while self.running:
+
+            try:
+
+                item = self.q.get(timeout=1)
+
+                if timeout == None:
+                    timeout = time.time() + self.rotation_time
+
+                triples.append(item)
+
+            except queue.Empty:
+                pass
+
+            if timeout:
+                if time.time() > timeout:
+
+                    self.write_file(triples)
+                    timeout = None
+                    triples = []
+
+    def write_file(self, triples):
+
+        try:
+
+            schema = pa.schema([
+                pa.field('s', pa.string()),
+                pa.field('p', pa.string()),
+                pa.field('o', pa.string()),
+            ])
+
+            fname = self.file_template.format(id=str(uuid.uuid4()))
+            path = f"{self.directory}/{fname}"
+
+            writer = pq.ParquetWriter(path, schema)
+
+            batch = pa.record_batch(
+                [
+                    [tpl[0] for tpl in triples],
+                    [tpl[1] for tpl in triples],
+                    [tpl[2] for tpl in triples],
+                ],
+                names=['s', 'p', 'o']
+            )
+
+            writer.write_batch(batch)
+
+            writer.close()
+
+            print(f"Wrote {path}.")
+
+        except Exception as e:
+
+            print("Parquet write:", e)
+
+    def write(self, s, p, o):
+        self.q.put((s, p, o))
+
+    def __del__(self):
+
+        self.running = False
+
+        if hasattr(self, "q"):
+            self.thread.join()
+
+            
+
+        

trustgraph-flow/trustgraph/embeddings/ollama/__init__.py

@@ -0,0 +1,3 @@
+
+from . processor import *
+

trustgraph-flow/trustgraph/embeddings/ollama/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . processor import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/embeddings/ollama/processor.py

@@ -0,0 +1,84 @@
+
+"""
+Embeddings service, applies an embeddings model selected from HuggingFace.
+Input is text, output is embeddings vector.
+"""
+from langchain_community.embeddings import OllamaEmbeddings
+
+from ... schema import EmbeddingsRequest, EmbeddingsResponse
+from ... schema import embeddings_request_queue, embeddings_response_queue
+from ... log_level import LogLevel
+from ... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = embeddings_request_queue
+default_output_queue = embeddings_response_queue
+default_subscriber = module
+default_model="mxbai-embed-large"
+default_ollama = 'http://localhost:11434'
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": EmbeddingsRequest,
+                "output_schema": EmbeddingsResponse,
+            }
+        )
+
+        self.embeddings = OllamaEmbeddings(base_url=ollama, model=model)
+
+    def handle(self, msg):
+
+        v = msg.value()
+
+        # Sender-produced ID
+
+        id = msg.properties()["id"]
+
+        print(f"Handling input {id}...", flush=True)
+
+        text = v.text
+        embeds = self.embeddings.embed_query([text])
+
+        print("Send response...", flush=True)
+        r = EmbeddingsResponse(vectors=[embeds])
+
+        self.producer.send(r, properties={"id": id})
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '-m', '--model',
+            default=default_model,
+            help=f'Embeddings model (default: {default_model})'
+        )
+
+        parser.add_argument(
+            '-r', '--ollama',
+            default=default_ollama,
+            help=f'ollama (default: {default_ollama})'
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/embeddings/vectorize/__init__.py

@@ -0,0 +1,3 @@
+
+from . vectorize import *
+

trustgraph-flow/trustgraph/embeddings/vectorize/__main__.py

@@ -0,0 +1,6 @@
+
+from . vectorize import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/embeddings/vectorize/vectorize.py

@@ -0,0 +1,103 @@
+
+"""
+Vectorizer, calls the embeddings service to get embeddings for a chunk.
+Input is text chunk, output is chunk and vectors.
+"""
+
+from ... schema import Chunk, ChunkEmbeddings
+from ... schema import chunk_ingest_queue, chunk_embeddings_ingest_queue
+from ... schema import embeddings_request_queue, embeddings_response_queue
+from ... clients.embeddings_client import EmbeddingsClient
+from ... log_level import LogLevel
+from ... base import ConsumerProducer
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = chunk_ingest_queue
+default_output_queue = chunk_embeddings_ingest_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        emb_request_queue = params.get(
+            "embeddings_request_queue", embeddings_request_queue
+        )
+        emb_response_queue = params.get(
+            "embeddings_response_queue", embeddings_response_queue
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "embeddings_request_queue": emb_request_queue,
+                "embeddings_response_queue": emb_response_queue,
+                "subscriber": subscriber,
+                "input_schema": Chunk,
+                "output_schema": ChunkEmbeddings,
+            }
+        )
+
+        self.embeddings = EmbeddingsClient(
+            pulsar_host=self.pulsar_host,
+            input_queue=emb_request_queue,
+            output_queue=emb_response_queue,
+            subscriber=module + "-emb",
+        )
+
+    def emit(self, source, chunk, vectors):
+
+        r = ChunkEmbeddings(source=source, chunk=chunk, vectors=vectors)
+        self.producer.send(r)
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Indexing {v.source.id}...", flush=True)
+
+        chunk = v.chunk.decode("utf-8")
+
+        try:
+
+            vectors = self.embeddings.request(chunk)
+
+            self.emit(
+                source=v.source,
+                chunk=chunk.encode("utf-8"),
+                vectors=vectors
+            )
+
+        except Exception as e:
+            print("Exception:", e, flush=True)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '--embeddings-request-queue',
+            default=embeddings_request_queue,
+            help=f'Embeddings request queue (default: {embeddings_request_queue})',
+        )
+
+        parser.add_argument(
+            '--embeddings-response-queue',
+            default=embeddings_response_queue,
+            help=f'Embeddings request queue (default: {embeddings_response_queue})',
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/extract/kg/definitions/__init__.py

@@ -0,0 +1,3 @@
+
+from . extract import *
+

trustgraph-flow/trustgraph/extract/kg/definitions/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . extract import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/extract/kg/definitions/extract.py

@@ -0,0 +1,134 @@
+
+"""
+Simple decoder, accepts embeddings+text chunks input, applies entity analysis to
+get entity definitions which are output as graph edges.
+"""
+
+import urllib.parse
+import json
+
+from .... schema import ChunkEmbeddings, Triple, Source, Value
+from .... schema import chunk_embeddings_ingest_queue, triples_store_queue
+from .... schema import prompt_request_queue
+from .... schema import prompt_response_queue
+from .... log_level import LogLevel
+from .... clients.prompt_client import PromptClient
+from .... rdf import TRUSTGRAPH_ENTITIES, DEFINITION
+from .... base import ConsumerProducer
+
+DEFINITION_VALUE = Value(value=DEFINITION, is_uri=True)
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = chunk_embeddings_ingest_queue
+default_output_queue = triples_store_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        pr_request_queue = params.get(
+            "prompt_request_queue", prompt_request_queue
+        )
+        pr_response_queue = params.get(
+            "prompt_response_queue", prompt_response_queue
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": ChunkEmbeddings,
+                "output_schema": Triple,
+                "prompt_request_queue": pr_request_queue,
+                "prompt_response_queue": pr_response_queue,
+            }
+        )
+
+        self.prompt = PromptClient(
+            pulsar_host=self.pulsar_host,
+            input_queue=pr_request_queue,
+            output_queue=pr_response_queue,
+            subscriber = module + "-prompt",
+        )
+
+    def to_uri(self, text):
+
+        part = text.replace(" ", "-").lower().encode("utf-8")
+        quoted = urllib.parse.quote(part)
+        uri = TRUSTGRAPH_ENTITIES + quoted
+
+        return uri
+
+    def get_definitions(self, chunk):
+
+        return self.prompt.request_definitions(chunk)
+
+    def emit_edge(self, s, p, o):
+
+        t = Triple(s=s, p=p, o=o)
+        self.producer.send(t)
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Indexing {v.source.id}...", flush=True)
+
+        chunk = v.chunk.decode("utf-8")
+
+        try:
+
+            defs = self.get_definitions(chunk)
+
+            for defn in defs:
+
+                s = defn.name
+                o = defn.definition
+
+                if s == "": continue
+                if o == "": continue
+
+                if s is None: continue
+                if o is None: continue
+
+                s_uri = self.to_uri(s)
+
+                s_value = Value(value=str(s_uri), is_uri=True)
+                o_value = Value(value=str(o), is_uri=False)
+
+                self.emit_edge(s_value, DEFINITION_VALUE, o_value)
+
+        except Exception as e:
+            print("Exception: ", e, flush=True)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '--prompt-request-queue',
+            default=prompt_request_queue,
+            help=f'Prompt request queue (default: {prompt_request_queue})',
+        )
+
+        parser.add_argument(
+            '--prompt-completion-response-queue',
+            default=prompt_response_queue,
+            help=f'Prompt response queue (default: {prompt_response_queue})',
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/extract/kg/relationships/__init__.py

@@ -0,0 +1,3 @@
+
+from . extract import *
+

trustgraph-flow/trustgraph/extract/kg/relationships/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . extract import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/extract/kg/relationships/extract.py

@@ -0,0 +1,208 @@
+
+"""
+Simple decoder, accepts vector+text chunks input, applies entity
+relationship analysis to get entity relationship edges which are output as
+graph edges.
+"""
+
+import urllib.parse
+import os
+from pulsar.schema import JsonSchema
+
+from .... schema import ChunkEmbeddings, Triple, GraphEmbeddings, Source, Value
+from .... schema import chunk_embeddings_ingest_queue, triples_store_queue
+from .... schema import graph_embeddings_store_queue
+from .... schema import prompt_request_queue
+from .... schema import prompt_response_queue
+from .... log_level import LogLevel
+from .... clients.prompt_client import PromptClient
+from .... rdf import RDF_LABEL, TRUSTGRAPH_ENTITIES
+from .... base import ConsumerProducer
+
+RDF_LABEL_VALUE = Value(value=RDF_LABEL, is_uri=True)
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = chunk_embeddings_ingest_queue
+default_output_queue = triples_store_queue
+default_vector_queue = graph_embeddings_store_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        vector_queue = params.get("vector_queue", default_vector_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        pr_request_queue = params.get(
+            "prompt_request_queue", prompt_request_queue
+        )
+        pr_response_queue = params.get(
+            "prompt_response_queue", prompt_response_queue
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": ChunkEmbeddings,
+                "output_schema": Triple,
+                "prompt_request_queue": pr_request_queue,
+                "prompt_response_queue": pr_response_queue,
+            }
+        )
+
+        self.vec_prod = self.client.create_producer(
+            topic=vector_queue,
+            schema=JsonSchema(GraphEmbeddings),
+        )
+
+        __class__.pubsub_metric.info({
+            "input_queue": input_queue,
+            "output_queue": output_queue,
+            "vector_queue": vector_queue,
+            "prompt_request_queue": pr_request_queue,
+            "prompt_response_queue": pr_response_queue,
+            "subscriber": subscriber,
+            "input_schema": ChunkEmbeddings.__name__,
+            "output_schema": Triple.__name__,
+            "vector_schema": GraphEmbeddings.__name__,
+        })
+
+        self.prompt = PromptClient(
+            pulsar_host=self.pulsar_host,
+            input_queue=pr_request_queue,
+            output_queue=pr_response_queue,
+            subscriber = module + "-prompt",
+        )
+
+    def to_uri(self, text):
+
+        part = text.replace(" ", "-").lower().encode("utf-8")
+        quoted = urllib.parse.quote(part)
+        uri = TRUSTGRAPH_ENTITIES + quoted
+
+        return uri
+
+    def get_relationships(self, chunk):
+
+        return self.prompt.request_relationships(chunk)
+
+    def emit_edge(self, s, p, o):
+
+        t = Triple(s=s, p=p, o=o)
+        self.producer.send(t)
+
+    def emit_vec(self, ent, vec):
+
+        r = GraphEmbeddings(entity=ent, vectors=vec)
+        self.vec_prod.send(r)
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Indexing {v.source.id}...", flush=True)
+
+        chunk = v.chunk.decode("utf-8")
+
+        try:
+
+            rels = self.get_relationships(chunk)
+
+            for rel in rels:
+
+                s = rel.s
+                p = rel.p
+                o = rel.o
+
+                if s == "": continue
+                if p == "": continue
+                if o == "": continue
+
+                if s is None: continue
+                if p is None: continue
+                if o is None: continue
+
+                s_uri = self.to_uri(s)
+                s_value = Value(value=str(s_uri), is_uri=True)
+
+                p_uri = self.to_uri(p)
+                p_value = Value(value=str(p_uri), is_uri=True)
+
+                if rel.o_entity: 
+                    o_uri = self.to_uri(o)
+                    o_value = Value(value=str(o_uri), is_uri=True)
+                else:
+                    o_value = Value(value=str(o), is_uri=False)
+
+                self.emit_edge(
+                    s_value,
+                    p_value,
+                    o_value
+                )
+
+                # Label for s
+                self.emit_edge(
+                    s_value,
+                    RDF_LABEL_VALUE,
+                    Value(value=str(s), is_uri=False)
+                )
+
+                # Label for p
+                self.emit_edge(
+                    p_value,
+                    RDF_LABEL_VALUE,
+                    Value(value=str(p), is_uri=False)
+                )
+
+                if rel.o_entity:
+                    # Label for o
+                    self.emit_edge(
+                        o_value,
+                        RDF_LABEL_VALUE,
+                        Value(value=str(o), is_uri=False)
+                    )
+
+                self.emit_vec(s_value, v.vectors)
+                self.emit_vec(p_value, v.vectors)
+                if rel.o_entity:
+                    self.emit_vec(o_value, v.vectors)
+
+        except Exception as e:
+            print("Exception: ", e, flush=True)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '-c', '--vector-queue',
+            default=default_vector_queue,
+            help=f'Vector output queue (default: {default_vector_queue})'
+        )
+
+        parser.add_argument(
+            '--prompt-request-queue',
+            default=prompt_request_queue,
+            help=f'Prompt request queue (default: {prompt_request_queue})',
+        )
+
+        parser.add_argument(
+            '--prompt-response-queue',
+            default=prompt_response_queue,
+            help=f'Prompt response queue (default: {prompt_response_queue})',
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+

trustgraph-flow/trustgraph/extract/kg/topics/__init__.py

@@ -0,0 +1,3 @@
+
+from . extract import *
+

trustgraph-flow/trustgraph/extract/kg/topics/__main__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from . extract import run
+
+if __name__ == '__main__':
+    run()
+

trustgraph-flow/trustgraph/extract/kg/topics/extract.py

@@ -0,0 +1,134 @@
+
+"""
+Simple decoder, accepts embeddings+text chunks input, applies entity analysis to
+get entity definitions which are output as graph edges.
+"""
+
+import urllib.parse
+import json
+
+from .... schema import ChunkEmbeddings, Triple, Source, Value
+from .... schema import chunk_embeddings_ingest_queue, triples_store_queue
+from .... schema import prompt_request_queue
+from .... schema import prompt_response_queue
+from .... log_level import LogLevel
+from .... clients.prompt_client import PromptClient
+from .... rdf import TRUSTGRAPH_ENTITIES, DEFINITION
+from .... base import ConsumerProducer
+
+DEFINITION_VALUE = Value(value=DEFINITION, is_uri=True)
+
+module = ".".join(__name__.split(".")[1:-1])
+
+default_input_queue = chunk_embeddings_ingest_queue
+default_output_queue = triples_store_queue
+default_subscriber = module
+
+class Processor(ConsumerProducer):
+
+    def __init__(self, **params):
+
+        input_queue = params.get("input_queue", default_input_queue)
+        output_queue = params.get("output_queue", default_output_queue)
+        subscriber = params.get("subscriber", default_subscriber)
+        pr_request_queue = params.get(
+            "prompt_request_queue", prompt_request_queue
+        )
+        pr_response_queue = params.get(
+            "prompt_response_queue", prompt_response_queue
+        )
+
+        super(Processor, self).__init__(
+            **params | {
+                "input_queue": input_queue,
+                "output_queue": output_queue,
+                "subscriber": subscriber,
+                "input_schema": ChunkEmbeddings,
+                "output_schema": Triple,
+                "prompt_request_queue": pr_request_queue,
+                "prompt_response_queue": pr_response_queue,
+            }
+        )
+
+        self.prompt = PromptClient(
+            pulsar_host=self.pulsar_host,
+            input_queue=pr_request_queue,
+            output_queue=pr_response_queue,
+            subscriber = module + "-prompt",
+        )
+
+    def to_uri(self, text):
+
+        part = text.replace(" ", "-").lower().encode("utf-8")
+        quoted = urllib.parse.quote(part)
+        uri = TRUSTGRAPH_ENTITIES + quoted
+
+        return uri
+
+    def get_topics(self, chunk):
+
+        return self.prompt.request_topics(chunk)
+
+    def emit_edge(self, s, p, o):
+
+        t = Triple(s=s, p=p, o=o)
+        self.producer.send(t)
+
+    def handle(self, msg):
+
+        v = msg.value()
+        print(f"Indexing {v.source.id}...", flush=True)
+
+        chunk = v.chunk.decode("utf-8")
+
+        try:
+
+            defs = self.get_topics(chunk)
+
+            for defn in defs:
+
+                s = defn.name
+                o = defn.definition
+
+                if s == "": continue
+                if o == "": continue
+
+                if s is None: continue
+                if o is None: continue
+
+                s_uri = self.to_uri(s)
+
+                s_value = Value(value=str(s_uri), is_uri=True)
+                o_value = Value(value=str(o), is_uri=False)
+
+                self.emit_edge(s_value, DEFINITION_VALUE, o_value)
+
+        except Exception as e:
+            print("Exception: ", e, flush=True)
+
+        print("Done.", flush=True)
+
+    @staticmethod
+    def add_args(parser):
+
+        ConsumerProducer.add_args(
+            parser, default_input_queue, default_subscriber,
+            default_output_queue,
+        )
+
+        parser.add_argument(
+            '--prompt-request-queue',
+            default=prompt_request_queue,
+            help=f'Prompt request queue (default: {prompt_request_queue})',
+        )
+
+        parser.add_argument(
+            '--prompt-completion-response-queue',
+            default=prompt_response_queue,
+            help=f'Prompt response queue (default: {prompt_response_queue})',
+        )
+
+def run():
+
+    Processor.start(module, __doc__)
+