From 053fc5511b38a059e40ce977a8ffc6eac46e51ac Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 13 Aug 2024 23:23:19 +0100 Subject: [PATCH] Neo4j writing roughly works --- Makefile | 2 +- docker-compose-openai-neo4j.yaml | 255 ++++++++++++++++++ scripts/triples-query-neo4j | 6 + scripts/triples-write-neo4j | 6 + templates/components/images.jsonnet | 1 + templates/components/neo4j.jsonnet | 22 ++ templates/docker-compose-openai-neo4j.jsonnet | 68 +++++ trustgraph/query/triples/neo4j/__init__.py | 3 + trustgraph/query/triples/neo4j/__main__.py | 7 + trustgraph/query/triples/neo4j/service.py | 153 +++++++++++ trustgraph/storage/triples/neo4j/__init__.py | 3 + trustgraph/storage/triples/neo4j/__main__.py | 7 + trustgraph/storage/triples/neo4j/write.py | 155 +++++++++++ 13 files changed, 687 insertions(+), 1 deletion(-) create mode 100644 docker-compose-openai-neo4j.yaml create mode 100755 scripts/triples-query-neo4j create mode 100755 scripts/triples-write-neo4j create mode 100644 templates/components/neo4j.jsonnet create mode 100644 templates/docker-compose-openai-neo4j.jsonnet create mode 100644 trustgraph/query/triples/neo4j/__init__.py create mode 100755 trustgraph/query/triples/neo4j/__main__.py create mode 100755 trustgraph/query/triples/neo4j/service.py create mode 100644 trustgraph/storage/triples/neo4j/__init__.py create mode 100755 trustgraph/storage/triples/neo4j/__main__.py create mode 100755 trustgraph/storage/triples/neo4j/write.py diff --git a/Makefile b/Makefile index 25af5451..75ae1308 100644 --- a/Makefile +++ b/Makefile @@ -33,7 +33,7 @@ set-version: # sed -i 's/trustgraph-flow:[0-9]*\.[0-9]*\.[0-9]*/trustgraph-flow:'${VERSION}'/' docker-compose*.yaml echo '"${VERSION}"' > templates/version.jsonnet -TEMPLATES=azure bedrock claude cohere mix ollama openai vertexai +TEMPLATES=azure bedrock claude cohere mix ollama openai vertexai openai-neo4j DCS=$(foreach template,${TEMPLATES},${template:%=docker-compose-%.yaml}) update-templates: set-version ${DCS} diff --git a/docker-compose-openai-neo4j.yaml b/docker-compose-openai-neo4j.yaml new file mode 100644 index 00000000..9f2459c7 --- /dev/null +++ b/docker-compose-openai-neo4j.yaml @@ -0,0 +1,255 @@ +"services": + "chunker": + "command": + - "chunker-recursive" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "embeddings": + "command": + - "embeddings-hf" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "etcd": + "command": + - "etcd" + - "-advertise-client-urls=http://127.0.0.1:2379" + - "-listen-client-urls" + - "http://0.0.0.0:2379" + - "--data-dir" + - "/etcd" + "environment": + "ETCD_AUTO_COMPACTION_MODE": "revision" + "ETCD_AUTO_COMPACTION_RETENTION": "1000" + "ETCD_QUOTA_BACKEND_BYTES": "4294967296" + "ETCD_SNAPSHOT_COUNT": "50000" + "image": "quay.io/coreos/etcd:v3.5.5" + "ports": + - "2379:2379" + "restart": "on-failure:100" + "volumes": + - "etcd:/etcd" + "grafana": + "environment": + "GF_ORG_NAME": "trustgraph.ai" + "image": "docker.io/grafana/grafana:10.0.0" + "ports": + - "3000:3000" + "restart": "on-failure:100" + "volumes": + - "grafana-storage:/var/lib/grafana" + - "./grafana/dashboard.yml:/etc/grafana/provisioning/dashboards/dashboard.yml" + - "./grafana/datasource.yml:/etc/grafana/provisioning/datasources/datasource.yml" + - "./grafana/dashboard.json:/var/lib/grafana/dashboards/dashboard.json" + "graph-rag": + "command": + - "graph-rag" + - "-p" + - "pulsar://pulsar:6650" + - "--prompt-request-queue" + - "non-persistent://tg/request/prompt-rag" + - "--prompt-response-queue" + - "non-persistent://tg/response/prompt-rag-response" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "init-pulsar": + "command": + - "sh" + - "-c" + - "pulsar-admin --admin-url http://pulsar:8080 tenants create tg && pulsar-admin --admin-url http://pulsar:8080 namespaces create tg/flow && pulsar-admin --admin-url http://pulsar:8080 namespaces create tg/request && pulsar-admin --admin-url http://pulsar:8080 namespaces create tg/response && pulsar-admin --admin-url http://pulsar:8080 namespaces set-retention --size -1 --time 3m tg/response" + "depends_on": + "pulsar": + "condition": "service_started" + "image": "docker.io/apachepulsar/pulsar:3.3.0" + "restart": "on-failure:100" + "kg-extract-definitions": + "command": + - "kg-extract-definitions" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "kg-extract-relationships": + "command": + - "kg-extract-relationships" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "milvus": + "command": + - "milvus" + - "run" + - "standalone" + "environment": + "ETCD_ENDPOINTS": "etcd:2379" + "MINIO_ADDRESS": "minio:9000" + "image": "docker.io/milvusdb/milvus:v2.4.5" + "ports": + - "9091:9091" + - "19530:19530" + "restart": "on-failure:100" + "volumes": + - "milvus:/var/lib/milvus" + "minio": + "command": + - "minio" + - "server" + - "/minio_data" + - "--console-address" + - ":9001" + "environment": + "MINIO_ROOT_PASSWORD": "minioadmin" + "MINIO_ROOT_USER": "minioadmin" + "image": "docker.io/minio/minio:RELEASE.2024-07-04T14-25-45Z" + "ports": + - "9001:9001" + "restart": "on-failure:100" + "volumes": + - "minio-data:/minio_data" + "neo4j": + "environment": + "NEO4J_AUTH": "neo4j/password" + "image": "docker.io/neo4j:5.22.0-community-bullseye" + "ports": + - "7474:7474" + - "7687:7687" + "restart": "on-failure:100" + "volumes": + - "neo4j:/data" + "pdf-decoder": + "command": + - "pdf-decoder" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "prometheus": + "image": "docker.io/prom/prometheus:v2.53.1" + "ports": + - "9090:9090" + "restart": "on-failure:100" + "volumes": + - "./prometheus:/etc/prometheus" + - "prometheus-data:/prometheus" + "prompt": + "command": + - "prompt-generic" + - "-p" + - "pulsar://pulsar:6650" + - "--text-completion-request-queue" + - "non-persistent://tg/request/text-completion" + - "--text-completion-response-queue" + - "non-persistent://tg/response/text-completion-response" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "prompt-rag": + "command": + - "prompt-generic" + - "-p" + - "pulsar://pulsar:6650" + - "-i" + - "non-persistent://tg/request/prompt-rag" + - "-o" + - "non-persistent://tg/response/prompt-rag-response" + - "--text-completion-request-queue" + - "non-persistent://tg/request/text-completion-rag" + - "--text-completion-response-queue" + - "non-persistent://tg/response/text-completion-rag-response" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "pulsar": + "command": "bin/pulsar standalone" + "image": "docker.io/apachepulsar/pulsar:3.3.0" + "ports": + - "6650:6650" + - "8080:8080" + "restart": "on-failure:100" + "volumes": + - "pulsar-conf:/pulsar/conf" + - "pulsar-data:/pulsar/data" + "pulsar-manager": + "environment": + "SPRING_CONFIGURATION_FILE": "/pulsar-manager/pulsar-manager/application.properties" + "image": "docker.io/apachepulsar/pulsar-manager:v0.3.0" + "ports": + - "9527:9527" + - "7750:7750" + "restart": "on-failure:100" + "query-graph-embeddings": + "command": + - "ge-query-milvus" + - "-p" + - "pulsar://pulsar:6650" + - "-t" + - "http://milvus:19530" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "query-triples": + "command": + - "triples-query-neo4j" + - "-p" + - "pulsar://pulsar:6650" + - "-g" + - "http://neo4j:7474" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "store-graph-embeddings": + "command": + - "ge-write-milvus" + - "-p" + - "pulsar://pulsar:6650" + - "-t" + - "http://milvus:19530" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "store-triples": + "command": + - "triples-write-neo4j" + - "-p" + - "pulsar://pulsar:6650" + - "-g" + - "http://neo4j:7474" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "text-completion": + "command": + - "text-completion-openai" + - "-p" + - "pulsar://pulsar:6650" + - "-k" + - "${OPENAI_KEY}" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "text-completion-rag": + "command": + - "text-completion-openai" + - "-p" + - "pulsar://pulsar:6650" + - "-k" + - "${OPENAI_KEY}" + - "-i" + - "non-persistent://tg/request/text-completion-rag" + - "-o" + - "non-persistent://tg/response/text-completion-rag-response" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" + "vectorize": + "command": + - "embeddings-vectorize" + - "-p" + - "pulsar://pulsar:6650" + "image": "docker.io/trustgraph/trustgraph-flow:0.6.0" + "restart": "on-failure:100" +"volumes": + "etcd": {} + "grafana-storage": {} + "milvus": {} + "minio-data": {} + "neo4j": {} + "prometheus-data": {} + "pulsar-conf": {} + "pulsar-data": {} diff --git a/scripts/triples-query-neo4j b/scripts/triples-query-neo4j new file mode 100755 index 00000000..05d97b10 --- /dev/null +++ b/scripts/triples-query-neo4j @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +from trustgraph.query.triples.neo4j import run + +run() + diff --git a/scripts/triples-write-neo4j b/scripts/triples-write-neo4j new file mode 100755 index 00000000..58786d44 --- /dev/null +++ b/scripts/triples-write-neo4j @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +from trustgraph.storage.triples.neo4j import run + +run() + diff --git a/templates/components/images.jsonnet b/templates/components/images.jsonnet index f5f94983..33b7b65d 100644 --- a/templates/components/images.jsonnet +++ b/templates/components/images.jsonnet @@ -1,6 +1,7 @@ local version = import "version.jsonnet"; { cassandra: "docker.io/cassandra:4.1.5", + neo4j: "docker.io/neo4j:5.22.0-community-bullseye", pulsar: "docker.io/apachepulsar/pulsar:3.3.0", pulsar_manager: "docker.io/apachepulsar/pulsar-manager:v0.3.0", etcd: "quay.io/coreos/etcd:v3.5.5", diff --git a/templates/components/neo4j.jsonnet b/templates/components/neo4j.jsonnet new file mode 100644 index 00000000..47f23989 --- /dev/null +++ b/templates/components/neo4j.jsonnet @@ -0,0 +1,22 @@ +local base = import "base.jsonnet"; +local images = import "images.jsonnet"; +{ + volumes +: { + neo4j: {}, + }, + services +: { + neo4j: base + { + image: images.neo4j, + ports: [ + "7474:7474", + "7687:7687", + ], + environment: { + NEO4J_AUTH: "neo4j/password", + }, + volumes: [ + "neo4j:/data" + ], + } + }, +} diff --git a/templates/docker-compose-openai-neo4j.jsonnet b/templates/docker-compose-openai-neo4j.jsonnet new file mode 100644 index 00000000..db8c4725 --- /dev/null +++ b/templates/docker-compose-openai-neo4j.jsonnet @@ -0,0 +1,68 @@ + +local base = import "components/base.jsonnet"; +local url = import "components/url.jsonnet"; +local images = import "components/images.jsonnet"; + +local neo4j = import "components/neo4j.jsonnet"; +local pulsar = import "components/pulsar.jsonnet"; +local milvus = import "components/milvus.jsonnet"; +local grafana = import "components/grafana.jsonnet"; +local trustgraph = import "components/trustgraph.jsonnet"; + +local config = neo4j + pulsar + milvus + grafana + trustgraph + { + + services: std.mergePatch(super.services, { + + "text-completion": base + { + image: images.trustgraph, + command: [ + "text-completion-openai", + "-p", + url.pulsar, + "-k", + "${OPENAI_KEY}", + ], + }, + + "text-completion-rag": base + { + image: images.trustgraph, + command: [ + "text-completion-openai", + "-p", + url.pulsar, + "-k", + "${OPENAI_KEY}", + "-i", + "non-persistent://tg/request/text-completion-rag", + "-o", + "non-persistent://tg/response/text-completion-rag-response", + ], + }, + + "query-triples": { + image: images.trustgraph, + command: [ + "triples-query-neo4j", + "-p", + url.pulsar, + "-g", + "http://neo4j:7474", + ], + }, + + "store-triples": { + image: images.trustgraph, + command: [ + "triples-write-neo4j", + "-p", + url.pulsar, + "-g", + "http://neo4j:7474", + ], + } + + }) +}; + +std.manifestYamlDoc(config) + diff --git a/trustgraph/query/triples/neo4j/__init__.py b/trustgraph/query/triples/neo4j/__init__.py new file mode 100644 index 00000000..ba844705 --- /dev/null +++ b/trustgraph/query/triples/neo4j/__init__.py @@ -0,0 +1,3 @@ + +from . service import * + diff --git a/trustgraph/query/triples/neo4j/__main__.py b/trustgraph/query/triples/neo4j/__main__.py new file mode 100755 index 00000000..89684e3e --- /dev/null +++ b/trustgraph/query/triples/neo4j/__main__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from . hf import run + +if __name__ == '__main__': + run() + diff --git a/trustgraph/query/triples/neo4j/service.py b/trustgraph/query/triples/neo4j/service.py new file mode 100755 index 00000000..c40a50fc --- /dev/null +++ b/trustgraph/query/triples/neo4j/service.py @@ -0,0 +1,153 @@ + +""" +Triples query service. Input is a (s, p, o) triple, some values may be +null. Output is a list of triples. +""" + +from .... direct.cassandra import TrustGraph +from .... schema import TriplesQueryRequest, TriplesQueryResponse +from .... schema import Value, Triple +from .... schema import triples_request_queue +from .... schema import triples_response_queue +from .... base import ConsumerProducer + +module = ".".join(__name__.split(".")[1:-1]) + +default_input_queue = triples_request_queue +default_output_queue = triples_response_queue +default_subscriber = module +default_graph_host='localhost' + +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) + graph_host = params.get("graph_host", default_graph_host) + + super(Processor, self).__init__( + **params | { + "input_queue": input_queue, + "output_queue": output_queue, + "subscriber": subscriber, + "input_schema": TriplesQueryRequest, + "output_schema": TriplesQueryResponse, + "graph_host": graph_host, + } + ) + + self.tg = TrustGraph([graph_host]) + + def create_value(self, ent): + if ent.startswith("http://") or ent.startswith("https://"): + return Value(value=ent, is_uri=True) + else: + return Value(value=ent, is_uri=False) + + def handle(self, msg): + + v = msg.value() + + # Sender-produced ID + id = msg.properties()["id"] + + print(f"Handling input {id}...", flush=True) + + triples = [] + + if v.s is not None: + if v.p is not None: + if v.o is not None: + resp = self.tg.get_spo( + v.s.value, v.p.value, v.o.value, + limit=v.limit + ) + triples.append((v.s.value, v.p.value, v.o.value)) + else: + resp = self.tg.get_sp( + v.s.value, v.p.value, + limit=v.limit + ) + for t in resp: + triples.append((v.s.value, v.p.value, t.o)) + else: + if v.o is not None: + resp = self.tg.get_os( + v.o.value, v.s.value, + limit=v.limit + ) + for t in resp: + triples.append((v.s.value, t.p, v.o.value)) + else: + resp = self.tg.get_s( + v.s.value, + limit=v.limit + ) + for t in resp: + triples.append((v.s.value, t.p, t.o)) + else: + if v.p is not None: + if v.o is not None: + resp = self.tg.get_po( + v.p.value, v.o.value, + limit=v.limit + ) + for t in resp: + triples.append((t.s, v.p.value, v.o.value)) + else: + resp = self.tg.get_p( + v.p.value, + limit=v.limit + ) + for t in resp: + triples.append((t.s, v.p.value, t.o)) + else: + if v.o is not None: + resp = self.tg.get_o( + v.o.value, + limit=v.limit + ) + for t in resp: + triples.append((t.s, t.p, v.o.value)) + else: + resp = self.tg.get_all( + limit=v.limit + ) + for t in resp: + triples.append((t.s, t.p, t.o)) + + triples = [ + Triple( + s=self.create_value(t[0]), + p=self.create_value(t[1]), + o=self.create_value(t[2]) + ) + for t in triples + ] + + print("Send response...", flush=True) + r = TriplesQueryResponse(triples=triples) + 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( + '-g', '--graph-host', + default="localhost", + help=f'Graph host (default: localhost)' + ) + +def run(): + + Processor.start(module, __doc__) + diff --git a/trustgraph/storage/triples/neo4j/__init__.py b/trustgraph/storage/triples/neo4j/__init__.py new file mode 100644 index 00000000..d891d55f --- /dev/null +++ b/trustgraph/storage/triples/neo4j/__init__.py @@ -0,0 +1,3 @@ + +from . write import * + diff --git a/trustgraph/storage/triples/neo4j/__main__.py b/trustgraph/storage/triples/neo4j/__main__.py new file mode 100755 index 00000000..c05d8c6d --- /dev/null +++ b/trustgraph/storage/triples/neo4j/__main__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from . write import run + +if __name__ == '__main__': + run() + diff --git a/trustgraph/storage/triples/neo4j/write.py b/trustgraph/storage/triples/neo4j/write.py new file mode 100755 index 00000000..b5f4d049 --- /dev/null +++ b/trustgraph/storage/triples/neo4j/write.py @@ -0,0 +1,155 @@ + +""" +Graph writer. Input is graph edge. Writes edges to Cassandra graph. +""" + +import pulsar +import base64 +import os +import argparse +import time + +from neo4j import GraphDatabase + +from .... schema import Triple +from .... schema import triples_store_queue +from .... log_level import LogLevel +from .... base import Consumer + +module = ".".join(__name__.split(".")[1:-1]) + +default_input_queue = triples_store_queue +default_subscriber = module + +default_graph_host = 'bolt://localhost:7687' +default_username = 'neo4j' +default_password = 'password' + +class Processor(Consumer): + + def __init__(self, **params): + + input_queue = params.get("input_queue", default_input_queue) + subscriber = params.get("subscriber", default_subscriber) + graph_host = params.get("graph_host", default_graph_host) + username = params.get("username", default_username) + password = params.get("passowrd", default_password) + + super(Processor, self).__init__( + **params | { + "input_queue": input_queue, + "subscriber": subscriber, + "input_schema": Triple, + "graph_host": graph_host, + } + ) + + self.db = "neo4j" + + self.io = GraphDatabase.driver(graph_host, auth=(username, password)) + + def create_node(self, uri): + + print("Create node", uri) + + summary = self.io.execute_query( + "MERGE (n:Node {uri: $uri})", + uri=uri, + database_=self.db, + ).summary + + print("Created {nodes_created} nodes in {time} ms.".format( + nodes_created=summary.counters.nodes_created, + time=summary.result_available_after + )) + + def create_literal(self, value): + + print("Create literal", value) + + summary = self.io.execute_query( + "MERGE (n:Literal {value: $value})", + value=value, + database_=self.db, + ).summary + + print("Created {nodes_created} nodes in {time} ms.".format( + nodes_created=summary.counters.nodes_created, + time=summary.result_available_after + )) + + def relate_node(self, src, uri, dest): + + print("Create node rel", src, uri, dest) + + summary = self.io.execute_query( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Node {uri: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + src=src, dest=dest, uri=uri, + database_=self.db, + ).summary + + print("Created {nodes_created} nodes in {time} ms.".format( + nodes_created=summary.counters.nodes_created, + time=summary.result_available_after + )) + + def relate_literal(self, src, uri, dest): + + print("Create literal rel", src, uri, dest) + + summary = self.io.execute_query( + "MATCH (src:Node {uri: $src}) " + "MATCH (dest:Literal {value: $dest}) " + "MERGE (src)-[:Rel {uri: $uri}]->(dest)", + src=src, dest=dest, uri=uri, + database_=self.db, + ).summary + + print("Created {nodes_created} nodes in {time} ms.".format( + nodes_created=summary.counters.nodes_created, + time=summary.result_available_after + )) + def handle(self, msg): + + v = msg.value() + + self.create_node(v.s.value) + + if v.o.is_uri: + self.create_node(v.o.value) + self.relate_node(v.s.value, v.p.value, v.o.value) + else: + self.create_literal(v.o.value) + self.relate_literal(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( + '-g', '--graph_host', + default=default_graph_host, + help=f'Graph host (default: {default_graph_host})' + ) + + parser.add_argument( + '--username', + default=default_username, + help=f'Neo4j username (default: {default_username})' + ) + + parser.add_argument( + '--password', + default=default_password, + help=f'Neo4j password (default: {default_password})' + ) + +def run(): + + Processor.start(module, __doc__) +