apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-20 18:51:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/release/v0.21' into pulsar-api-support-0.21

Browse source
This commit is contained in:
Tyler O 2025-02-12 09:59:54 +00:00
commit a5814515c7
86 changed files with 1145 additions and 273 deletions
Download patch file Download diff file Expand all files Collapse all files

5
templates/components/trustgraph.jsonnet
View file

@ -1,6 +1,8 @@
local base = import "base/base.jsonnet";
local images = import "values/images.jsonnet";
local url = import "values/url.jsonnet";
local minio = import "stores/minio.jsonnet";
local cassandra = import "stores/cassandra.jsonnet";
{
@ -182,3 +184,6 @@ local url = import "values/url.jsonnet";
}
// Minio and Cassandra are used by the Librarian
+ minio + cassandra

5
templates/generate
View file

@ -50,10 +50,13 @@ class Generator:
self.templates.joinpath(filename),
self.resources.joinpath(dir, filename),
self.resources.joinpath(filename),
pathlib.Path(dir).joinpath(filename),
]
else:
candidates = [
self.templates.joinpath(filename),
pathlib.Path(dir).joinpath(filename),
pathlib.Path(filename),
]
try:
@ -86,7 +89,7 @@ class Packager:
def __init__(self):
self.templates = pathlib.Path("./templates")
self.resources = pathlib.Path("./resources")
self.resources = pathlib.Path("./")
def process(
self, config, version="0.0.0", platform="docker-compose",

44
templates/stores/milvus.jsonnet
View file

@ -1,7 +1,8 @@
local base = import "base/base.jsonnet";
local images = import "values/images.jsonnet";
local minio = import "stores/minio.jsonnet";
{
minio {
etcd +: {
@ -47,47 +48,6 @@ local images = import "values/images.jsonnet";
},
mino +: {
create:: function(engine)
local vol = engine.volume("minio-data").with_size("20G");
local container =
engine.container("minio")
.with_image(images.minio)
.with_command([
"minio",
"server",
"/minio_data",
"--console-address",
":9001",
])
.with_environment({
MINIO_ROOT_USER: "minioadmin",
MINIO_ROOT_PASSWORD: "minioadmin",
})
.with_limits("0.5", "128M")
.with_reservations("0.25", "128M")
.with_port(9001, 9001, "api")
.with_volume_mount(vol, "/minio_data");
local containerSet = engine.containers(
"etcd", [ container ]
);
local service =
engine.service(containerSet)
.with_port(9001, 9001, "api");
engine.resources([
vol,
containerSet,
service,
])
},
milvus +: {
create:: function(engine)

49
templates/stores/minio.jsonnet Normal file
View file

@ -0,0 +1,49 @@
local base = import "base/base.jsonnet";
local images = import "values/images.jsonnet";
{
minio +: {
create:: function(engine)
local vol = engine.volume("minio-data").with_size("20G");
local container =
engine.container("minio")
.with_image(images.minio)
.with_command([
"minio",
"server",
"/minio_data",
"--console-address",
":9001",
])
.with_environment({
MINIO_ROOT_USER: "minioadmin",
MINIO_ROOT_PASSWORD: "minioadmin",
})
.with_limits("0.5", "128M")
.with_reservations("0.25", "128M")
.with_port(9000, 9000, "api")
.with_port(9001, 9001, "console")
.with_volume_mount(vol, "/minio_data");
local containerSet = engine.containers(
"etcd", [ container ]
);
local service =
engine.service(containerSet)
.with_port(9000, 9000, "api")
.with_port(9001, 9001, "console");
engine.resources([
vol,
containerSet,
service,
])
},
}

2
templates/values/images.jsonnet
View file

@ -5,7 +5,7 @@ local version = import "version.jsonnet";
pulsar: "docker.io/apachepulsar/pulsar:3.3.1",
pulsar_manager: "docker.io/apachepulsar/pulsar-manager:v0.4.0",
etcd: "quay.io/coreos/etcd:v3.5.15",
minio: "docker.io/minio/minio:RELEASE.2024-08-17T01-24-54Z",
minio: "docker.io/minio/minio:RELEASE.2025-02-03T21-03-04Z",
milvus: "docker.io/milvusdb/milvus:v2.4.9",
prometheus: "docker.io/prom/prometheus:v2.53.2",
grafana: "docker.io/grafana/grafana:11.1.4",

2
trustgraph-base/trustgraph/base/__init__.py
View file

@ -3,4 +3,6 @@ from . base_processor import BaseProcessor
from . consumer import Consumer
from . producer import Producer
from . consumer_producer import ConsumerProducer
from . publisher import Publisher
from . subscriber import Subscriber

27
trustgraph-base/trustgraph/base/base_processor.py
View file

@ -1,4 +1,5 @@
import asyncio
import os
import argparse
import pulsar
@ -30,6 +31,7 @@ class BaseProcessor:
pulsar_host = params.get("pulsar_host", self.default_pulsar_host)
pulsar_api_key = params.get("pulsar_api_key", None)
pulsar_listener = params.get("pulsar_listener", None)
log_level = params.get("log_level", LogLevel.INFO)
self.pulsar_host = pulsar_host
@ -43,9 +45,12 @@ class BaseProcessor:
else:
self.client = pulsar.Client(
pulsar_host,
listener_name=pulsar_listener,
logger=pulsar.ConsoleLogger(log_level.to_pulsar())
)
self.pulsar_listener = pulsar_listener
def __del__(self):
if hasattr(self, "client"):
@ -67,6 +72,11 @@ class BaseProcessor:
help=f'Pulsar API key',
)
parser.add_argument(
'--pulsar-listener',
help=f'Pulsar listener (default: none)',
)
parser.add_argument(
'-l', '--log-level',
type=LogLevel,
@ -89,11 +99,20 @@ class BaseProcessor:
help=f'Pulsar host (default: 8000)',
)
def run(self):
async def start(self):
pass
async def run(self):
raise RuntimeError("Something should have implemented the run method")
@classmethod
def start(cls, prog, doc):
async def launch_async(cls, args):
p = cls(**args)
await p.start()
await p.run()
@classmethod
def launch(cls, prog, doc):
parser = argparse.ArgumentParser(
prog=prog,
@ -114,8 +133,7 @@ class BaseProcessor:
try:
p = cls(**args)
p.run()
asyncio.run(cls.launch_async(args))
except KeyboardInterrupt:
print("Keyboard interrupt.")
@ -133,3 +151,4 @@ class BaseProcessor:
print("Will retry...", flush=True)
time.sleep(4)

5
trustgraph-base/trustgraph/base/consumer.py
View file

@ -1,4 +1,5 @@
import asyncio
from pulsar.schema import JsonSchema
import pulsar
from prometheus_client import Histogram, Info, Counter, Enum
@ -75,7 +76,7 @@ class Consumer(BaseProcessor):
print("Initialised consumer.", flush=True)
def run(self):
async def run(self):
__class__.state_metric.state('running')
@ -104,7 +105,7 @@ class Consumer(BaseProcessor):
try:
with __class__.request_metric.time():
self.handle(msg)
await self.handle(msg)
# Acknowledge successful processing of the message
self.consumer.acknowledge(msg)

2
trustgraph-base/trustgraph/base/consumer_producer.py
View file

@ -42,7 +42,7 @@ class ConsumerProducer(Consumer):
print("Initialised consumer/producer.")
def send(self, msg, properties={}):
async def send(self, msg, properties={}):
self.producer.send(msg, properties)
__class__.output_metric.inc()

2
trustgraph-base/trustgraph/base/producer.py
View file

@ -37,7 +37,7 @@ class Producer(BaseProcessor):
chunking_enabled=True,
)
def send(self, msg, properties={}):
async def send(self, msg, properties={}):
self.producer.send(msg, properties)
__class__.output_metric.inc()

19
trustgraph-flow/trustgraph/gateway/publisher.py → trustgraph-base/trustgraph/base/publisher.py
View file

@ -15,14 +15,22 @@ class Publisher:
self.q = queue.Queue(maxsize=max_size)
self.chunking_enabled = chunking_enabled
self.listener_name = listener
self.running = True
def start(self):
self.task = threading.Thread(target=self.run)
self.task.start()
def stop(self):
self.running = False
def join(self):
self.stop()
self.task.join()
def run(self):
while True:
while self.running:
try:
@ -44,9 +52,12 @@ class Publisher:
chunking_enabled=self.chunking_enabled,
)
while True:
while self.running:
id, item = self.q.get()
try:
id, item = self.q.get(timeout=0.5)
except queue.Empty:
continue
if id:
producer.send(item, { "id": id })
@ -61,3 +72,5 @@ class Publisher:
def send(self, id, msg):
self.q.put((id, msg))

13
trustgraph-flow/trustgraph/gateway/subscriber.py → trustgraph-base/trustgraph/base/subscriber.py
View file

@ -19,14 +19,21 @@ class Subscriber:
self.max_size = max_size
self.lock = threading.Lock()
self.listener_name = listener
self.running = True
def start(self):
self.task = threading.Thread(target=self.run)
self.task.start()
def stop(self):
self.running = False
def join(self):
self.task.join()
def run(self):
while True:
while self.running:
try:
@ -50,7 +57,7 @@ class Subscriber:
schema=self.schema,
)
while True:
while self.running:
msg = consumer.receive()
@ -68,12 +75,14 @@ class Subscriber:
if id in self.q:
try:
# FIXME: Timeout means data goes missing
self.q[id].put(value, timeout=0.5)
except:
pass
for q in self.full.values():
try:
# FIXME: Timeout means data goes missing
q.put(value, timeout=0.5)
except:
pass

3
trustgraph-base/trustgraph/exceptions.py
View file

@ -8,3 +8,6 @@ class LlmError(Exception):
class ParseError(Exception):
pass
class RequestError(Exception):
pass

1
trustgraph-base/trustgraph/schema/__init__.py
View file

@ -10,5 +10,6 @@ from . retrieval import *
from . metadata import *
from . agent import *
from . lookup import *
from . library import *

56
trustgraph-base/trustgraph/schema/library.py Normal file
View file

@ -0,0 +1,56 @@
from pulsar.schema import Record, Bytes, String, Array
from . types import Triple
from . topic import topic
from . types import Error
from . metadata import Metadata
from . documents import Document, TextDocument
# add(Metadata, Bytes) : error?
# copy(id, user, collection)
# move(id, user, collection)
# delete(id)
# get(id) : Bytes
# reindex(id)
# list(user, collection) : id[]
# info(id[]) : DocumentInfo[]
# search(<key,op,value>[]) : id[]
class DocumentPackage(Record):
metadata = Array(Triple())
document = Bytes()
kind = String()
user = String()
collection = String()
class DocumentInfo(Record):
metadata = Array(Triple())
kind = String()
user = String()
collection = String()
class Criteria(Record):
key = String()
value = String()
operator = String()
class LibrarianRequest(Record):
operation = String()
id = String()
document = DocumentPackage()
user = String()
collection = String()
criteria = Array(Criteria())
class LibrarianResponse(Record):
error = Error()
document = DocumentPackage()
info = Array(DocumentInfo())
librarian_request_queue = topic(
'librarian', kind='non-persistent', namespace='request'
)
librarian_response_queue = topic(
'librarian', kind='non-persistent', namespace='response',
)

4
trustgraph-bedrock/trustgraph/model/text_completion/bedrock/llm.py
View file

@ -109,7 +109,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -341,5 +341,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-embeddings-hf/trustgraph/embeddings/hf/hf.py
View file

@ -40,7 +40,7 @@ class Processor(ConsumerProducer):
self.embeddings = HuggingFaceEmbeddings(model_name=model)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -96,5 +96,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/scripts/librarian Executable file
View file

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

4
trustgraph-flow/setup.py
View file

@ -49,6 +49,7 @@ setuptools.setup(
"langchain-community",
"langchain-core",
"langchain-text-splitters",
"minio",
"neo4j",
"ollama",
"openai",
@ -78,8 +79,8 @@ setuptools.setup(
"scripts/de-write-qdrant",
"scripts/document-embeddings",
"scripts/document-rag",
"scripts/embeddings-ollama",
"scripts/embeddings-fastembed",
"scripts/embeddings-ollama",
"scripts/ge-query-milvus",
"scripts/ge-query-pinecone",
"scripts/ge-query-qdrant",
@ -91,6 +92,7 @@ setuptools.setup(
"scripts/kg-extract-definitions",
"scripts/kg-extract-relationships",
"scripts/kg-extract-topics",
"scripts/librarian",
"scripts/metering",
"scripts/object-extract-row",
"scripts/oe-write-milvus",

14
trustgraph-flow/trustgraph/agent/react/service.py
View file

@ -191,7 +191,7 @@ class Processor(ConsumerProducer):
return json.loads(json_str)
def handle(self, msg):
async def handle(self, msg):
try:
@ -231,7 +231,7 @@ class Processor(ConsumerProducer):
observation=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def observe(x):
@ -244,7 +244,7 @@ class Processor(ConsumerProducer):
observation=x,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
act = self.agent.react(v.question, history, think, observe)
@ -260,7 +260,7 @@ class Processor(ConsumerProducer):
thought=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -283,7 +283,7 @@ class Processor(ConsumerProducer):
]
)
self.recursive_input.send(r, properties={"id": id})
await self.recursive_input.send(r, properties={"id": id})
print("Done.", flush=True)
@ -303,7 +303,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
@staticmethod
def add_args(parser):
@ -379,5 +379,5 @@ description.'''
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/chunking/recursive/chunker.py
View file

@ -52,7 +52,7 @@ class Processor(ConsumerProducer):
is_separator_regex=False,
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Chunking {v.metadata.id}...", flush=True)
@ -70,7 +70,7 @@ class Processor(ConsumerProducer):
__class__.chunk_metric.observe(len(chunk.page_content))
self.send(r)
await self.send(r)
print("Done.", flush=True)
@ -98,5 +98,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/chunking/token/chunker.py
View file

@ -51,7 +51,7 @@ class Processor(ConsumerProducer):
chunk_overlap=chunk_overlap,
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Chunking {v.metadata.id}...", flush=True)
@ -69,7 +69,7 @@ class Processor(ConsumerProducer):
__class__.chunk_metric.observe(len(chunk.page_content))
self.send(r)
await self.send(r)
print("Done.", flush=True)
@ -97,5 +97,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
View file

@ -39,7 +39,7 @@ class Processor(ConsumerProducer):
print("PDF inited")
def handle(self, msg):
async def handle(self, msg):
print("PDF message received")
@ -64,7 +64,7 @@ class Processor(ConsumerProducer):
text=page.page_content.encode("utf-8"),
)
self.send(r)
await self.send(r)
print("Done.", flush=True)
@ -78,5 +78,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/embeddings/document_embeddings/embeddings.py
View file

@ -53,7 +53,7 @@ class Processor(ConsumerProducer):
subscriber=module + "-emb",
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -74,7 +74,7 @@ class Processor(ConsumerProducer):
chunks=embeds,
)
self.producer.send(r)
await self.send(r)
except Exception as e:
print("Exception:", e, flush=True)
@ -106,5 +106,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/embeddings/fastembed/processor.py
View file

@ -41,7 +41,7 @@ class Processor(ConsumerProducer):
self.embeddings = TextEmbedding(model_name = model)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -65,7 +65,7 @@ class Processor(ConsumerProducer):
error=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -85,5 +85,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/embeddings/graph_embeddings/embeddings.py
View file

@ -52,7 +52,7 @@ class Processor(ConsumerProducer):
subscriber=module + "-emb",
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -77,7 +77,7 @@ class Processor(ConsumerProducer):
entities=entities,
)
self.producer.send(r)
await self.send(r)
except Exception as e:
print("Exception:", e, flush=True)
@ -109,5 +109,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/embeddings/ollama/processor.py
View file

@ -45,7 +45,7 @@ class Processor(ConsumerProducer):
self.client = Client(host=ollama)
self.model = model
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -67,7 +67,7 @@ class Processor(ConsumerProducer):
error=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -93,5 +93,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/external/wikipedia/service.py vendored
View file

@ -39,7 +39,7 @@ class Processor(ConsumerProducer):
self.url = url
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -60,7 +60,7 @@ class Processor(ConsumerProducer):
text=resp
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -75,7 +75,7 @@ class Processor(ConsumerProducer):
),
text=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -98,5 +98,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

14
trustgraph-flow/trustgraph/extract/kg/definitions/extract.py
View file

@ -97,15 +97,15 @@ class Processor(ConsumerProducer):
return self.prompt.request_definitions(chunk)
def emit_edges(self, metadata, triples):
async def emit_edges(self, metadata, triples):
t = Triples(
metadata=metadata,
triples=triples,
)
self.producer.send(t)
await self.send(t)
def emit_ecs(self, metadata, entities):
async def emit_ecs(self, metadata, entities):
t = EntityContexts(
metadata=metadata,
@ -113,7 +113,7 @@ class Processor(ConsumerProducer):
)
self.ec_prod.send(t)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -172,7 +172,7 @@ class Processor(ConsumerProducer):
entities.append(ec)
self.emit_edges(
await self.emit_edges(
Metadata(
id=v.metadata.id,
metadata=[],
@ -182,7 +182,7 @@ class Processor(ConsumerProducer):
triples
)
self.emit_ecs(
await self.emit_ecs(
Metadata(
id=v.metadata.id,
metadata=[],
@ -225,5 +225,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

10
trustgraph-flow/trustgraph/extract/kg/relationships/extract.py
View file

@ -72,15 +72,15 @@ class Processor(ConsumerProducer):
return self.prompt.request_relationships(chunk)
def emit_edges(self, metadata, triples):
async def emit_edges(self, metadata, triples):
t = Triples(
metadata=metadata,
triples=triples,
)
self.producer.send(t)
await self.send(t)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -167,7 +167,7 @@ class Processor(ConsumerProducer):
o=Value(value=v.metadata.id, is_uri=True)
))
self.emit_edges(
await self.emit_edges(
Metadata(
id=v.metadata.id,
metadata=[],
@ -204,5 +204,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

12
trustgraph-flow/trustgraph/extract/kg/topics/extract.py
View file

@ -70,15 +70,15 @@ class Processor(ConsumerProducer):
return self.prompt.request_topics(chunk)
def emit_edge(self, metadata, s, p, o):
async def emit_edge(self, metadata, s, p, o):
t = Triples(
metadata=metadata,
triples=[Triple(s=s, p=p, o=o)],
)
self.producer.send(t)
await self.send(t)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -105,7 +105,9 @@ class Processor(ConsumerProducer):
s_value = Value(value=str(s_uri), is_uri=True)
o_value = Value(value=str(o), is_uri=False)
self.emit_edge(v. metadata, s_value, DEFINITION_VALUE, o_value)
await self.emit_edge(
v.metadata, s_value, DEFINITION_VALUE, o_value
)
except Exception as e:
print("Exception: ", e, flush=True)
@ -134,5 +136,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/extract/object/row/extract.py
View file

@ -130,16 +130,16 @@ class Processor(ConsumerProducer):
t = Rows(
metadata=metadata, row_schema=self.row_schema, rows=rows
)
self.producer.send(t)
await self.producer.send(t)
def emit_vec(self, metadata, name, vec, key_name, key):
r = ObjectEmbeddings(
metadata=metadata, vectors=vec, name=name, key_name=key_name, id=key
)
self.vec_prod.send(r)
await self.vec_prod.send(r)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
print(f"Indexing {v.metadata.id}...", flush=True)
@ -217,5 +217,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/gateway/document_embeddings_load.py
View file

@ -7,8 +7,8 @@ from aiohttp import WSMsgType
from .. schema import Metadata
from .. schema import DocumentEmbeddings, ChunkEmbeddings
from .. schema import document_embeddings_store_queue
from .. base import Publisher
from . publisher import Publisher
from . socket import SocketEndpoint
from . serialize import to_subgraph
@ -61,6 +61,6 @@ class DocumentEmbeddingsLoadEndpoint(SocketEndpoint):
],
)
self.publisher.send(None, elt)
await self.publisher.send(None, elt)
running.stop()

2
trustgraph-flow/trustgraph/gateway/document_embeddings_stream.py
View file

@ -6,8 +6,8 @@ import uuid
from .. schema import DocumentEmbeddings
from .. schema import document_embeddings_store_queue
from .. base import Subscriber
from . subscriber import Subscriber
from . socket import SocketEndpoint
from . serialize import serialize_document_embeddings

4
trustgraph-flow/trustgraph/gateway/endpoint.py
View file

@ -5,8 +5,8 @@ from aiohttp import web
import uuid
import logging
from . publisher import Publisher
from . subscriber import Subscriber
from .. base import Publisher
from .. base import Subscriber
logger = logging.getLogger("endpoint")
logger.setLevel(logging.INFO)

4
trustgraph-flow/trustgraph/gateway/graph_embeddings_load.py
View file

@ -7,8 +7,8 @@ from aiohttp import WSMsgType
from .. schema import Metadata
from .. schema import GraphEmbeddings, EntityEmbeddings
from .. schema import graph_embeddings_store_queue
from .. base import Publisher
from . publisher import Publisher
from . socket import SocketEndpoint
from . serialize import to_subgraph, to_value
@ -61,7 +61,7 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint):
]
)
self.publisher.send(None, elt)
await self.publisher.send(None, elt)
running.stop()

2
trustgraph-flow/trustgraph/gateway/graph_embeddings_stream.py
View file

@ -6,8 +6,8 @@ import uuid
from .. schema import GraphEmbeddings
from .. schema import graph_embeddings_store_queue
from .. base import Subscriber
from . subscriber import Subscriber
from . socket import SocketEndpoint
from . serialize import serialize_graph_embeddings

57
trustgraph-flow/trustgraph/gateway/librarian.py Normal file
View file

@ -0,0 +1,57 @@
from .. schema import LibrarianRequest, LibrarianResponse, Triples
from .. schema import librarian_request_queue
from .. schema import librarian_response_queue
from . endpoint import ServiceEndpoint
from . requestor import ServiceRequestor
from . serialize import serialize_document_package, serialize_document_info
from . serialize import to_document_package, to_document_info, to_criteria
class LibrarianRequestor(ServiceRequestor):
def __init__(self, pulsar_host, timeout, auth):
super(LibrarianRequestor, self).__init__(
pulsar_host=pulsar_host,
request_queue=librarian_request_queue,
response_queue=librarian_response_queue,
request_schema=LibrarianRequest,
response_schema=LibrarianResponse,
timeout=timeout,
)
def to_request(self, body):
if "document" in body:
dp = to_document_package(body["document"])
else:
dp = None
if "criteria" in body:
criteria = to_criteria(body["criteria"])
else:
criteria = None
limit = int(body.get("limit", 10000))
return LibrarianRequest(
operation = body.get("operation", None),
id = body.get("id", None),
document = dp,
user = body.get("user", None),
collection = body.get("collection", None),
criteria = criteria,
)
def from_response(self, message):
response = {}
if message.document:
response["document"] = serialize_document_package(message.document)
if message.info:
response["info"] = serialize_document_info(message.info)
return response, True

14
trustgraph-flow/trustgraph/gateway/requestor.py
View file

@ -4,8 +4,8 @@ from pulsar.schema import JsonSchema
import uuid
import logging
from . publisher import Publisher
from . subscriber import Subscriber
from .. base import Publisher
from .. base import Subscriber
logger = logging.getLogger("requestor")
logger.setLevel(logging.INFO)
@ -71,7 +71,10 @@ class ServiceRequestor:
raise RuntimeError("Timeout")
if resp.error:
err = { "error": resp.error.message }
err = { "error": {
"type": resp.error.type,
"message": resp.error.message,
} }
if responder:
await responder(err, True)
return err
@ -90,7 +93,10 @@ class ServiceRequestor:
logging.error(f"Exception: {e}")
err = { "error": str(e) }
err = { "error": {
"type": "gateway-error",
"message": str(e),
} }
if responder:
await responder(err, True)
return err

2
trustgraph-flow/trustgraph/gateway/sender.py
View file

@ -6,7 +6,7 @@ from pulsar.schema import JsonSchema
import uuid
import logging
from . publisher import Publisher
from .. base import Publisher
logger = logging.getLogger("sender")
logger.setLevel(logging.INFO)

71
trustgraph-flow/trustgraph/gateway/serialize.py
View file

@ -1,4 +1,7 @@
from .. schema import Value, Triple
import base64
from .. schema import Value, Triple, DocumentPackage, DocumentInfo
def to_value(x):
return Value(value=x["v"], is_uri=x["e"])
@ -77,3 +80,69 @@ def serialize_document_embeddings(message):
],
}
def serialize_document_package(message):
ret = {}
if message.metadata:
ret["metadata"] = serialize_subgraph(message.metdata)
if message.document:
blob = base64.b64encode(
message.document.encode("utf-8")
).decode("utf-8")
ret["document"] = blob
if message.kind:
ret["kind"] = message.kind
if message.user:
ret["user"] = message.user
if message.collection:
ret["collection"] = message.collection
return ret
def serialize_document_info(message):
ret = {}
if message.metadata:
ret["metadata"] = serialize_subgraph(message.metdata)
if message.kind:
ret["kind"] = message.kind
if message.user:
ret["user"] = message.user
if message.collection:
ret["collection"] = message.collection
return ret
def to_document_package(x):
return DocumentPackage(
metadata = to_subgraph(x["metadata"]),
document = base64.b64decode(x["document"].encode("utf-8")),
kind = x.get("kind", None),
user = x.get("user", None),
collection = x.get("collection", None),
)
def to_document_info(x):
return DocumentInfo(
metadata = to_subgraph(x["metadata"]),
kind = x.get("kind", None),
user = x.get("user", None),
collection = x.get("collection", None),
)
def to_criteria(x):
return [
Critera(v["key"], v["value"], v["operator"])
for v in x
]

11
trustgraph-flow/trustgraph/gateway/service.py
View file

@ -26,8 +26,6 @@ from .. log_level import LogLevel
from . serialize import to_subgraph
from . running import Running
from . publisher import Publisher
from . subscriber import Subscriber
from . text_completion import TextCompletionRequestor
from . prompt import PromptRequestor
from . graph_rag import GraphRagRequestor
@ -39,6 +37,7 @@ from . encyclopedia import EncyclopediaRequestor
from . agent import AgentRequestor
from . dbpedia import DbpediaRequestor
from . internet_search import InternetSearchRequestor
from . librarian import LibrarianRequestor
from . triples_stream import TriplesStreamEndpoint
from . graph_embeddings_stream import GraphEmbeddingsStreamEndpoint
from . document_embeddings_stream import DocumentEmbeddingsStreamEndpoint
@ -125,6 +124,10 @@ class Api:
pulsar_host=self.pulsar_host, timeout=self.timeout,
auth = self.auth, pulsar_api_key=self.pulsar_api_key,
),
"librarian": LibrarianRequestor(
pulsar_host=self.pulsar_host, timeout=self.timeout,
auth = self.auth,
),
"encyclopedia": EncyclopediaRequestor(
pulsar_host=self.pulsar_host, timeout=self.timeout,
auth = self.auth, pulsar_api_key=self.pulsar_api_key,
@ -179,6 +182,10 @@ class Api:
endpoint_path = "/api/v1/agent", auth=self.auth,
requestor = self.services["agent"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/librarian", auth=self.auth,
requestor = self.services["librarian"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/encyclopedia", auth=self.auth,
requestor = self.services["encyclopedia"],

4
trustgraph-flow/trustgraph/gateway/triples_load.py
View file

@ -7,8 +7,8 @@ from aiohttp import WSMsgType
from .. schema import Metadata
from .. schema import Triples
from .. schema import triples_store_queue
from .. base import Publisher
from . publisher import Publisher
from . socket import SocketEndpoint
from . serialize import to_subgraph
@ -53,7 +53,7 @@ class TriplesLoadEndpoint(SocketEndpoint):
triples=to_subgraph(data["triples"]),
)
self.publisher.send(None, elt)
await self.publisher.send(None, elt)
running.stop()

2
trustgraph-flow/trustgraph/gateway/triples_stream.py
View file

@ -6,8 +6,8 @@ import uuid
from .. schema import Triples
from .. schema import triples_store_queue
from .. base import Subscriber
from . subscriber import Subscriber
from . socket import SocketEndpoint
from . serialize import serialize_triples

3
trustgraph-flow/trustgraph/librarian/__init__.py Normal file
View file

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

7
trustgraph-flow/trustgraph/librarian/__main__.py Executable file
View file

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

51
trustgraph-flow/trustgraph/librarian/blob_store.py Normal file
View file

@ -0,0 +1,51 @@
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. knowledge import hash
from .. exceptions import RequestError
from minio import Minio
import time
import io
class BlobStore:
def __init__(
self,
minio_host, minio_access_key, minio_secret_key, bucket_name,
):
self.minio = Minio(
minio_host,
access_key = minio_access_key,
secret_key = minio_secret_key,
secure = False,
)
self.bucket_name = bucket_name
print("Connected to minio", flush=True)
self.ensure_bucket()
def ensure_bucket(self):
# Make the bucket if it doesn't exist.
found = self.minio.bucket_exists(self.bucket_name)
if not found:
self.minio.make_bucket(self.bucket_name)
print("Created bucket", self.bucket_name, flush=True)
else:
print("Bucket", self.bucket_name, "already exists", flush=True)
def add(self, object_id, blob, kind):
# FIXME: Loop retry
self.minio.put_object(
bucket_name = self.bucket_name,
object_name = "doc/" + str(object_id),
length = len(blob),
data = io.BytesIO(blob),
content_type = kind,
)
print("Add blob complete", flush=True)

56
trustgraph-flow/trustgraph/librarian/librarian.py Normal file
View file

@ -0,0 +1,56 @@
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. knowledge import hash
from .. exceptions import RequestError
from . table_store import TableStore
from . blob_store import BlobStore
import uuid
class Librarian:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password,
minio_host, minio_access_key, minio_secret_key,
bucket_name, keyspace, load_document, load_text,
):
self.blob_store = BlobStore(
minio_host, minio_access_key, minio_secret_key, bucket_name
)
self.table_store = TableStore(
cassandra_host, cassandra_user, cassandra_password, keyspace
)
self.load_document = load_document
self.load_text = load_text
async def add(self, id, document):
if document.kind not in (
"text/plain", "application/pdf"
):
raise RequestError("Invalid document kind: " + document.kind)
# Create object ID as a hash of the document
object_id = uuid.UUID(hash(document.document))
self.blob_store.add(object_id, document.document, document.kind)
self.table_store.add(object_id, document)
if document.kind == "application/pdf":
await self.load_document(id, document)
elif document.kind == "text/plain":
await self.load_text(id, document)
print("Add complete", flush=True)
return LibrarianResponse(
error = None,
document = None,
info = None,
)

354
trustgraph-flow/trustgraph/librarian/service.py Executable file
View file

@ -0,0 +1,354 @@
"""
Librarian service, manages documents in collections
"""
from functools import partial
import asyncio
import threading
import queue
from pulsar.schema import JsonSchema
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. schema import librarian_request_queue, librarian_response_queue
from .. schema import GraphEmbeddings
from .. schema import graph_embeddings_store_queue
from .. schema import Triples
from .. schema import triples_store_queue
from .. schema import DocumentEmbeddings
from .. schema import document_embeddings_store_queue
from .. schema import Document, Metadata
from .. schema import document_ingest_queue
from .. schema import TextDocument, Metadata
from .. schema import text_ingest_queue
from .. base import Publisher
from .. base import Subscriber
from .. log_level import LogLevel
from .. base import ConsumerProducer
from .. exceptions import RequestError
from . librarian import Librarian
module = ".".join(__name__.split(".")[1:-1])
default_input_queue = librarian_request_queue
default_output_queue = librarian_response_queue
default_subscriber = module
default_minio_host = "minio:9000"
default_minio_access_key = "minioadmin"
default_minio_secret_key = "minioadmin"
default_cassandra_host = "cassandra"
bucket_name = "library"
# FIXME: How to ensure this doesn't conflict with other usage?
keyspace = "librarian"
class Processor(ConsumerProducer):
def __init__(self, **params):
self.running = True
input_queue = params.get("input_queue", default_input_queue)
output_queue = params.get("output_queue", default_output_queue)
subscriber = params.get("subscriber", default_subscriber)
minio_host = params.get("minio_host", default_minio_host)
minio_access_key = params.get(
"minio_access_key",
default_minio_access_key
)
minio_secret_key = params.get(
"minio_secret_key",
default_minio_secret_key
)
cassandra_host = params.get("cassandra_host", default_cassandra_host)
cassandra_user = params.get("cassandra_user")
cassandra_password = params.get("cassandra_password")
triples_queue = params.get("triples_queue")
graph_embeddings_queue = params.get("graph_embeddings_queue")
document_embeddings_queue = params.get("document_embeddings_queue")
document_load_queue = params.get("document_load_queue")
text_load_queue = params.get("text_load_queue")
super(Processor, self).__init__(
**params | {
"input_queue": input_queue,
"output_queue": output_queue,
"subscriber": subscriber,
"input_schema": LibrarianRequest,
"output_schema": LibrarianResponse,
"minio_host": minio_host,
"minio_access_key": minio_access_key,
"cassandra_host": cassandra_host,
"cassandra_user": cassandra_user,
}
)
self.document_load = Publisher(
self.pulsar_host, document_load_queue, JsonSchema(Document),
listener=self.pulsar_listener,
)
self.text_load = Publisher(
self.pulsar_host, text_load_queue, JsonSchema(TextDocument),
listener=self.pulsar_listener,
)
self.triples_load = Subscriber(
self.pulsar_host, triples_store_queue,
"librarian", "librarian",
schema=JsonSchema(Triples),
listener=self.pulsar_listener,
)
self.triples_reader = threading.Thread(target=self.receive_triples)
self.librarian = Librarian(
cassandra_host = cassandra_host.split(","),
cassandra_user = cassandra_user,
cassandra_password = cassandra_password,
minio_host = minio_host,
minio_access_key = minio_access_key,
minio_secret_key = minio_secret_key,
bucket_name = bucket_name,
keyspace = keyspace,
load_document = self.load_document,
load_text = self.load_text,
)
print("Initialised.", flush=True)
async def start(self):
self.document_load.start()
self.text_load.start()
self.triples_load.start()
self.triples_sub = self.triples_load.subscribe_all("x")
self.triples_reader.start()
def receive_triples(self):
print("Receive triples!")
while self.running:
try:
msg = self.triples_sub.get(timeout=1)
except queue.Empty:
print("Tick")
continue
print(msg)
print("BYE")
def __del__(self):
self.running = False
if hasattr(self, "triples_sub"):
self.triples_sub.unsubscribe_all("x")
if hasattr(self, "document_load"):
self.document_load.stop()
self.document_load.join()
if hasattr(self, "text_load"):
self.text_load.stop()
self.text_load.join()
if hasattr(self, "triples_load"):
self.triples_load.stop()
self.triples_load.join()
async def load_document(self, id, document):
doc = Document(
metadata = Metadata(
id = id,
metadata = document.metadata,
user = document.user,
collection = document.collection
),
data = document.document
)
self.document_load.send(None, doc)
async def load_text(self, id, document):
doc = TextDocument(
metadata = Metadata(
id = id,
metadata = document.metadata,
user = document.user,
collection = document.collection
),
text = document.document
)
self.text_load.send(None, doc)
def parse_request(self, v):
if v.operation is None:
raise RequestError("Null operation")
if v.operation == "add":
if (
v.id and v.document and v.document.metadata and
v.document.document and v.document.kind
):
return partial(
self.librarian.add,
id = v.id,
document = v.document,
)
else:
raise RequestError("Invalid call")
raise RequestError("Invalid operation: " + v.operation)
async def handle(self, msg):
v = msg.value()
# Sender-produced ID
id = msg.properties()["id"]
print(f"Handling input {id}...", flush=True)
try:
func = self.parse_request(v)
except RequestError as e:
resp = LibrarianResponse(
error = Error(
type = "request-error",
message = str(e),
)
)
await self.send(resp, properties={"id": id})
return
try:
resp = await func()
except RequestError as e:
resp = LibrarianResponse(
error = Error(
type = "request-error",
message = str(e),
)
)
await self.send(resp, properties={"id": id})
return
except Exception as e:
print("Exception:", e, flush=True)
resp = LibrarianResponse(
error = Error(
type = "processing-error",
message = "Unhandled error: " + str(e),
)
)
await self.send(resp, properties={"id": id})
return
print("Send response...", flush=True)
await self.send(resp, 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(
'--minio-host',
default=default_minio_host,
help=f'Minio hostname (default: {default_minio_host})',
)
parser.add_argument(
'--minio-access-key',
default='minioadmin',
help='Minio access key / username '
f'(default: {default_minio_access_key})',
)
parser.add_argument(
'--minio-secret-key',
default='minioadmin',
help='Minio secret key / password '
f'(default: {default_minio_access_key})',
)
parser.add_argument(
'--cassandra-host',
default="cassandra",
help=f'Graph host (default: cassandra)'
)
parser.add_argument(
'--cassandra-user',
default=None,
help=f'Cassandra user'
)
parser.add_argument(
'--cassandra-password',
default=None,
help=f'Cassandra password'
)
parser.add_argument(
'--triples-queue',
default=triples_store_queue,
help=f'Triples queue (default: {triples_store_queue})'
)
parser.add_argument(
'--graph-embeddings-queue',
default=graph_embeddings_store_queue,
help=f'Graph embeddings queue (default: {triples_store_queue})'
)
parser.add_argument(
'--document-embeddings-queue',
default=document_embeddings_store_queue,
help='Document embeddings queue '
f'(default: {document_embeddings_store_queue})'
)
parser.add_argument(
'--document-load-queue',
default=document_ingest_queue,
help='Document load queue '
f'(default: {document_ingest_queue})'
)
parser.add_argument(
'--text-load-queue',
default=text_ingest_queue,
help='Text ingest queue '
f'(default: {text_ingest_queue})'
)
def run():
Processor.launch(module, __doc__)

131
trustgraph-flow/trustgraph/librarian/table_store.py Normal file
View file

@ -0,0 +1,131 @@
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. knowledge import hash
from .. exceptions import RequestError
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.query import BatchStatement
import uuid
import time
class TableStore:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password, keyspace,
):
self.keyspace = keyspace
print("Connecting to Cassandra...", flush=True)
if cassandra_user and cassandra_password:
auth_provider = PlainTextAuthProvider(
username=cassandra_user, password=cassandra_password
)
self.cluster = Cluster(
cassandra_host,
auth_provider=auth_provider
)
else:
self.cluster = Cluster(cassandra_host)
self.cassandra = self.cluster.connect()
print("Connected.", flush=True)
self.ensure_cassandra_schema()
self.insert_document_stmt = self.cassandra.prepare("""
insert into document
(id, user, collection, kind, object_id, metadata)
values (?, ?, ?, ?, ?, ?)
""")
def ensure_cassandra_schema(self):
print("Ensure Cassandra schema...", flush=True)
print("Keyspace...", flush=True)
# FIXME: Replication factor should be configurable
self.cassandra.execute(f"""
create keyspace if not exists {self.keyspace}
with replication = {{
'class' : 'SimpleStrategy',
'replication_factor' : 1
}};
""");
self.cassandra.set_keyspace(self.keyspace)
print("document table...", flush=True)
self.cassandra.execute("""
create table if not exists document (
user text,
collection text,
id uuid,
kind text,
object_id uuid,
metadata list<tuple<
text, boolean, text, boolean, text, boolean
>>,
PRIMARY KEY (user, collection, id)
);
""");
print("object index...", flush=True)
self.cassandra.execute("""
create index if not exists document_object
on document ( object_id)
""");
print("Cassandra schema OK.", flush=True)
def add(self, object_id, document):
if document.kind not in (
"text/plain", "application/pdf"
):
raise RequestError("Invalid document kind: " + document.kind)
# Create random doc ID
doc_id = uuid.uuid4()
print("Adding", object_id, doc_id)
metadata = [
(
v.s.value, v.s.is_uri, v.p.value, v.p.is_uri,
v.o.value, v.o.is_uri
)
for v in document.metadata
]
while True:
try:
resp = self.cassandra.execute(
self.insert_document_stmt,
(
doc_id, document.user, document.collection,
document.kind, object_id, metadata
)
)
break
except Exception as e:
print("Exception:", type(e))
print(f"{e}, retry...", flush=True)
time.sleep(1)
print("Add complete", flush=True)

4
trustgraph-flow/trustgraph/metering/counter.py
View file

@ -57,7 +57,7 @@ class Processor(Consumer):
return model["input_price"], model["output_price"]
return None, None # Return None if model is not found
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
modelname = v.model
@ -98,4 +98,4 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

52
trustgraph-flow/trustgraph/model/prompt/generic/service.py
View file

@ -78,7 +78,7 @@ class Processor(ConsumerProducer):
return json.loads(json_str)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -92,32 +92,32 @@ class Processor(ConsumerProducer):
if kind == "extract-definitions":
self.handle_extract_definitions(id, v)
await self.handle_extract_definitions(id, v)
return
elif kind == "extract-topics":
self.handle_extract_topics(id, v)
await self.handle_extract_topics(id, v)
return
elif kind == "extract-relationships":
self.handle_extract_relationships(id, v)
await self.handle_extract_relationships(id, v)
return
elif kind == "extract-rows":
self.handle_extract_rows(id, v)
await self.handle_extract_rows(id, v)
return
elif kind == "kg-prompt":
self.handle_kg_prompt(id, v)
await self.handle_kg_prompt(id, v)
return
elif kind == "document-prompt":
self.handle_document_prompt(id, v)
await self.handle_document_prompt(id, v)
return
else:
@ -125,7 +125,7 @@ class Processor(ConsumerProducer):
print("Invalid kind.", flush=True)
return
def handle_extract_definitions(self, id, v):
async def handle_extract_definitions(self, id, v):
try:
@ -164,7 +164,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(definitions=output, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -182,9 +182,9 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def handle_extract_topics(self, id, v):
async def handle_extract_topics(self, id, v):
try:
@ -223,7 +223,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(topics=output, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -241,9 +241,9 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def handle_extract_relationships(self, id, v):
async def handle_extract_relationships(self, id, v):
try:
@ -295,7 +295,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(relationships=output, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -313,9 +313,9 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def handle_extract_rows(self, id, v):
async def handle_extract_rows(self, id, v):
try:
@ -366,7 +366,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(rows=output, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -384,9 +384,9 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def handle_kg_prompt(self, id, v):
async def handle_kg_prompt(self, id, v):
try:
@ -400,7 +400,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(answer=ans, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -418,9 +418,9 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
def handle_document_prompt(self, id, v):
async def handle_document_prompt(self, id, v):
try:
@ -437,7 +437,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = PromptResponse(answer=ans, error=None)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
print("Done.", flush=True)
@ -455,7 +455,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
@staticmethod
def add_args(parser):
@ -481,5 +481,5 @@ def run():
raise RuntimeError("NOT IMPLEMENTED")
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

12
trustgraph-flow/trustgraph/model/prompt/template/service.py
View file

@ -156,7 +156,7 @@ class Processor(ConsumerProducer):
config = prompt_configuration,
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -191,7 +191,7 @@ class Processor(ConsumerProducer):
error=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
return
@ -206,7 +206,7 @@ class Processor(ConsumerProducer):
error=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
return
@ -224,7 +224,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
except Exception as e:
@ -240,7 +240,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
@staticmethod
def add_args(parser):
@ -294,5 +294,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/azure/llm.py
View file

@ -123,7 +123,7 @@ class Processor(ConsumerProducer):
return result
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -154,7 +154,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TextCompletionResponse(response=resp, error=None, in_token=inputtokens, out_token=outputtokens, model=self.model)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
except TooManyRequests:
@ -182,7 +182,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -224,4 +224,4 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/azure_openai/llm.py
View file

@ -84,7 +84,7 @@ class Processor(ConsumerProducer):
azure_endpoint = endpoint,
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -133,7 +133,7 @@ class Processor(ConsumerProducer):
model=self.model
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
except RateLimitError:
@ -161,7 +161,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -212,4 +212,4 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

6
trustgraph-flow/trustgraph/model/text_completion/claude/llm.py
View file

@ -73,7 +73,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -150,7 +150,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -190,6 +190,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/cohere/llm.py
View file

@ -69,7 +69,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -106,7 +106,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TextCompletionResponse(response=resp, error=None, in_token=inputtokens, out_token=outputtokens, model=self.model)
self.send(r, properties={"id": id})
self.await send(r, properties={"id": id})
print("Done.", flush=True)
@ -136,7 +136,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.producer.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -169,6 +169,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/googleaistudio/llm.py
View file

@ -102,7 +102,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -146,7 +146,7 @@ class Processor(ConsumerProducer):
out_token=outputtokens,
model=self.model
)
self.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -177,7 +177,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -217,6 +217,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/llamafile/llm.py
View file

@ -74,7 +74,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -122,7 +122,7 @@ class Processor(ConsumerProducer):
out_token=outputtokens,
model="llama.cpp"
)
self.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -145,7 +145,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -185,6 +185,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/ollama/llm.py
View file

@ -71,7 +71,7 @@ class Processor(ConsumerProducer):
self.model = model
self.llm = Client(host=ollama)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -96,7 +96,7 @@ class Processor(ConsumerProducer):
r = TextCompletionResponse(response=response_text, error=None, in_token=inputtokens, out_token=outputtokens, model="ollama")
self.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -119,7 +119,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -145,6 +145,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/model/text_completion/openai/llm.py
View file

@ -73,7 +73,7 @@ class Processor(ConsumerProducer):
print("Initialised", flush=True)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -126,7 +126,7 @@ class Processor(ConsumerProducer):
out_token=outputtokens,
model=self.model
)
self.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -156,7 +156,7 @@ class Processor(ConsumerProducer):
model=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -196,6 +196,6 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/doc_embeddings/milvus/service.py
View file

@ -40,7 +40,7 @@ class Processor(ConsumerProducer):
self.vecstore = DocVectors(store_uri)
def handle(self, msg):
async def handle(self, msg):
try:
@ -64,7 +64,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = DocumentEmbeddingsResponse(documents=chunks, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -82,7 +82,7 @@ class Processor(ConsumerProducer):
documents=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -102,5 +102,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/doc_embeddings/pinecone/service.py
View file

@ -56,7 +56,7 @@ class Processor(ConsumerProducer):
}
)
def handle(self, msg):
async def handle(self, msg):
try:
@ -100,7 +100,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = DocumentEmbeddingsResponse(documents=chunks, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -118,7 +118,7 @@ class Processor(ConsumerProducer):
documents=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -143,5 +143,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/doc_embeddings/qdrant/service.py
View file

@ -47,7 +47,7 @@ class Processor(ConsumerProducer):
self.client = QdrantClient(url=store_uri, api_key=api_key)
def handle(self, msg):
async def handle(self, msg):
try:
@ -81,7 +81,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = DocumentEmbeddingsResponse(documents=chunks, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -99,7 +99,7 @@ class Processor(ConsumerProducer):
documents=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -125,5 +125,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/graph_embeddings/milvus/service.py
View file

@ -46,7 +46,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -79,7 +79,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = GraphEmbeddingsResponse(entities=entities, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -97,7 +97,7 @@ class Processor(ConsumerProducer):
entities=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -117,5 +117,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/graph_embeddings/pinecone/service.py
View file

@ -62,7 +62,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -120,7 +120,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = GraphEmbeddingsResponse(entities=entities, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -138,7 +138,7 @@ class Processor(ConsumerProducer):
entities=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -163,5 +163,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py
View file

@ -52,7 +52,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -106,7 +106,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = GraphEmbeddingsResponse(entities=entities, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -124,7 +124,7 @@ class Processor(ConsumerProducer):
entities=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -150,5 +150,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/triples/cassandra/service.py
View file

@ -53,7 +53,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -154,7 +154,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TriplesQueryResponse(triples=triples, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -172,7 +172,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -205,5 +205,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/triples/falkordb/service.py
View file

@ -54,7 +54,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -301,7 +301,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TriplesQueryResponse(triples=triples, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -319,7 +319,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -345,5 +345,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/triples/memgraph/service.py
View file

@ -58,7 +58,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -313,7 +313,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TriplesQueryResponse(triples=triples, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -331,7 +331,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -369,5 +369,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/query/triples/neo4j/service.py
View file

@ -58,7 +58,7 @@ class Processor(ConsumerProducer):
else:
return Value(value=ent, is_uri=False)
def handle(self, msg):
async def handle(self, msg):
try:
@ -297,7 +297,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = TriplesQueryResponse(triples=triples, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -315,7 +315,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -353,5 +353,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/retrieval/document_rag/rag.py
View file

@ -79,7 +79,7 @@ class Processor(ConsumerProducer):
module=module,
)
def handle(self, msg):
async def handle(self, msg):
try:
@ -94,7 +94,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = DocumentRagResponse(response = response, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -112,7 +112,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -162,5 +162,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

8
trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py
View file

@ -98,7 +98,7 @@ class Processor(ConsumerProducer):
module=module,
)
def handle(self, msg):
async def handle(self, msg):
try:
@ -115,7 +115,7 @@ class Processor(ConsumerProducer):
print("Send response...", flush=True)
r = GraphRagResponse(response = response, error=None)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
print("Done.", flush=True)
@ -133,7 +133,7 @@ class Processor(ConsumerProducer):
response=None,
)
self.producer.send(r, properties={"id": id})
await self.send(r, properties={"id": id})
self.consumer.acknowledge(msg)
@ -216,5 +216,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py
View file

@ -35,7 +35,7 @@ class Processor(Consumer):
self.vecstore = DocVectors(store_uri)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -65,5 +65,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py
View file

@ -61,7 +61,7 @@ class Processor(Consumer):
self.last_index_name = None
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -166,5 +166,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py
View file

@ -42,7 +42,7 @@ class Processor(Consumer):
self.client = QdrantClient(url=store_uri)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -110,5 +110,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py
View file

@ -34,7 +34,7 @@ class Processor(Consumer):
self.vecstore = EntityVectors(store_uri)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -59,5 +59,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py
View file

@ -88,7 +88,7 @@ class Processor(Consumer):
"Gave up waiting for index creation"
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -170,5 +170,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py
View file

@ -67,7 +67,7 @@ class Processor(Consumer):
return cname
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -117,5 +117,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/object_embeddings/milvus/write.py
View file

@ -34,7 +34,7 @@ class Processor(Consumer):
self.vecstore = ObjectVectors(store_uri)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -57,5 +57,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/rows/cassandra/write.py
View file

@ -62,7 +62,7 @@ class Processor(Consumer):
self.session.execute("use trustgraph");
def handle(self, msg):
async def handle(self, msg):
try:
@ -143,5 +143,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/triples/cassandra/write.py
View file

@ -47,7 +47,7 @@ class Processor(Consumer):
self.password = graph_password
self.table = None
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -110,5 +110,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/triples/falkordb/write.py
View file

@ -118,7 +118,7 @@ class Processor(Consumer):
time=res.run_time_ms
))
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -154,5 +154,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/triples/memgraph/write.py
View file

@ -205,7 +205,7 @@ class Processor(Consumer):
src=t.s.value, dest=t.o.value, uri=t.p.value,
)
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -256,5 +256,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-flow/trustgraph/storage/triples/neo4j/write.py
View file

@ -158,7 +158,7 @@ class Processor(Consumer):
time=summary.result_available_after
))
def handle(self, msg):
async def handle(self, msg):
v = msg.value()
@ -206,5 +206,5 @@ class Processor(Consumer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)

4
trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py
View file

@ -131,7 +131,7 @@ class Processor(ConsumerProducer):
print("Initialisation complete", flush=True)
def handle(self, msg):
async def handle(self, msg):
try:
@ -248,5 +248,5 @@ class Processor(ConsumerProducer):
def run():
Processor.start(module, __doc__)
Processor.launch(module, __doc__)