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

Flow management

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-22 22:43:07 +01:00
parent a9197d11ee
commit 009878141f
21 changed files with 856 additions and 291 deletions
Download patch file Download diff file Expand all files Collapse all files

78
test-api/test-library-add-doc Executable file
View file

@ -0,0 +1,78 @@
#!/usr/bin/env python3
import requests
import json
import sys
import base64
url = "http://localhost:8088/api/v1/"
############################################################################
id = "http://trustgraph.ai/doc/12345678"
with open("docs/README.cats") as f:
doc = base64.b64encode(f.read().encode("utf-8")).decode("utf-8")
input = {
"operation": "add",
"document": {
"id": id,
"metadata": [
{
"s": {
"v": id,
"e": True,
},
"p": {
"v": "http://www.w3.org/2000/01/rdf-schema#label",
"e": True,
},
"o": {
"v": "Mark's pets", "e": False,
},
},
{
"s": {
"v": id,
"e": True,
},
"p": {
"v": 'https://schema.org/keywords',
"e": True,
},
"o": {
"v": "cats", "e": False,
},
},
],
"document": doc,
"kind": "text/plain",
"user": "trustgraph",
"collection": "default",
"title": "Mark's cats",
"comments": "Test doc taken from the TrustGraph repo",
}
}
resp = requests.post(
f"{url}librarian",
json=input,
)
print(resp.text)
resp = resp.json()
print(resp)
if "error" in resp:
print(f"Error: {resp['error']}")
sys.exit(1)
# print(resp["response"])
print(resp)
sys.exit(0)
############################################################################

90
test-api/test-library-add-doc2 Executable file
View file

@ -0,0 +1,90 @@
#!/usr/bin/env python3
import requests
import json
import sys
import base64
url = "http://localhost:8088/api/v1/"
############################################################################
id = "http://trustgraph.ai/doc/12345678"
source = "../sources/20160001634.pdf"
with open(source, "rb") as f:
doc = base64.b64encode(f.read()).decode("utf-8")
input = {
"operation": "add",
"id": id,
"document": {
"metadata": [
{
"s": {
"v": id,
"e": True,
},
"p": {
"v": "http://www.w3.org/2000/01/rdf-schema#label",
"e": True,
},
"o": {
"v": "Challenger report volume 1", "e": False,
},
},
{
"s": {
"v": id,
"e": True,
},
"p": {
"v": 'https://schema.org/keywords',
"e": True,
},
"o": {
"v": "space shuttle", "e": False,
},
},
{
"s": {
"v": id,
"e": True,
},
"p": {
"v": 'https://schema.org/keywords',
"e": True,
},
"o": {
"v": "nasa", "e": False,
},
},
],
"document": doc,
"kind": "application/pdf",
"user": "trustgraph",
"collection": "default",
}
}
resp = requests.post(
f"{url}librarian",
json=input,
)
print(resp.text)
resp = resp.json()
print(resp)
if "error" in resp:
print(f"Error: {resp['error']}")
sys.exit(1)
print(resp)
sys.exit(0)
############################################################################

39
test-api/test-library-list Executable file
View file

@ -0,0 +1,39 @@
#!/usr/bin/env python3
import requests
import json
import sys
import base64
url = "http://localhost:8088/api/v1/"
############################################################################
user = "trustgraph"
input = {
"operation": "list",
"user": user,
}
resp = requests.post(
f"{url}librarian",
json=input,
)
print(resp.text)
resp = resp.json()
print(resp)
if "error" in resp:
print(f"Error: {resp['error']}")
sys.exit(1)
# print(resp["response"])
print(resp)
sys.exit(0)
############################################################################

2
tests/test-config Normal file
View file

@ -0,0 +1,2 @@
#!/usr/bin/env python3

92
tests/test-flow Executable file
View file

@ -0,0 +1,92 @@
#!/usr/bin/env python3
import requests
url = "http://localhost:8088/"
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "list-classes",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "get-class",
"class-name": "default",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "put-class",
"class-name": "bunch",
"class-definition": "{}",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "get-class",
"class-name": "bunch",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "list-classes",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "delete-class",
"class-name": "bunch",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "list-classes",
}
)
print(resp)
print(resp.text)
resp = requests.post(
f"{url}/api/v1/flow",
json={
"operation": "list-flows",
}
)
print(resp)
print(resp.text)

36
tests/test-load-pdf Executable file
View file

@ -0,0 +1,36 @@
#!/usr/bin/env python3
import pulsar
from pulsar.schema import JsonSchema
import base64
from trustgraph.schema import Document, Metadata
client = pulsar.Client("pulsar://localhost:6650", listener_name="localhost")
prod = client.create_producer(
topic="persistent://tg/flow/document-load:0000",
schema=JsonSchema(Document),
chunking_enabled=True,
)
path = "../sources/Challenger-Report-Vol1.pdf"
with open(path, "rb") as f:
blob = base64.b64encode(f.read()).decode("utf-8")
message = Document(
metadata = Metadata(
id = "00001",
metadata = [],
user="trustgraph",
collection="default",
),
data=blob
)
prod.send(message)
prod.close()
client.close()

37
tests/test-load-text Executable file
View file

@ -0,0 +1,37 @@
#!/usr/bin/env python3
import pulsar
from pulsar.schema import JsonSchema
import base64
from trustgraph.schema import TextDocument, Metadata
client = pulsar.Client("pulsar://localhost:6650", listener_name="localhost")
prod = client.create_producer(
topic="persistent://tg/flow/text-document-load:0000",
schema=JsonSchema(TextDocument),
chunking_enabled=True,
)
path = "docs/README.cats"
with open(path, "r") as f:
# blob = base64.b64encode(f.read()).decode("utf-8")
blob = f.read()
message = TextDocument(
metadata = Metadata(
id = "00001",
metadata = [],
user="trustgraph",
collection="default",
),
text=blob
)
prod.send(message)
prod.close()
client.close()

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

@ -234,7 +234,7 @@ class AsyncProcessor:
PulsarClient.add_args(parser)
parser.add_argument(
'--config-push-queue',
'--config-queue',
default=default_config_queue,
help=f'Config push queue {default_config_queue}',
)

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

@ -12,5 +12,5 @@ from . agent import *
from . lookup import *
from . library import *
from . config import *
from . flows import *

8
trustgraph-base/trustgraph/schema/flows.py
View file

@ -20,14 +20,14 @@ from . types import Error
# Prompt services, abstract the prompt generation
class FlowRequest(Record):
operation = String() # list_classes, get_class, put_class, delete_class
# list_flows, get_flow, start_flow, stop_flow
operation = String() # list-classes, get-class, put-class, delete-class
# list-flows, get-flow, start-flow, stop-flow
# get_class, put_class, delete_class, start_flow
class_name = String()
# put_class
class = String()
class_definition = String()
# start_flow
description = String()
@ -44,7 +44,7 @@ class FlowResponse(Record):
flow_ids = Array(String())
# get_class
class = String()
class_definition = String()
# get_flow
flow = String()

19
trustgraph-base/trustgraph/schema/graph.py
View file

@ -18,8 +18,6 @@ class EntityContexts(Record):
metadata = Metadata()
entities = Array(EntityContext())
entity_contexts_ingest_queue = topic('entity-contexts-load')
############################################################################
# Graph embeddings are embeddings associated with a graph entity
@ -33,8 +31,6 @@ class GraphEmbeddings(Record):
metadata = Metadata()
entities = Array(EntityEmbeddings())
graph_embeddings_store_queue = topic('graph-embeddings-store')
############################################################################
# Graph embeddings query
@ -49,13 +45,6 @@ class GraphEmbeddingsResponse(Record):
error = Error()
entities = Array(Value())
graph_embeddings_request_queue = topic(
'graph-embeddings', kind='non-persistent', namespace='request'
)
graph_embeddings_response_queue = topic(
'graph-embeddings', kind='non-persistent', namespace='response'
)
############################################################################
# Graph triples
@ -64,8 +53,6 @@ class Triples(Record):
metadata = Metadata()
triples = Array(Triple())
triples_store_queue = topic('triples-store')
############################################################################
# Triples query
@ -82,9 +69,3 @@ class TriplesQueryResponse(Record):
error = Error()
triples = Array(Triple())
triples_request_queue = topic(
'triples', kind='non-persistent', namespace='request'
)
triples_response_queue = topic(
'triples', kind='non-persistent', namespace='response'
)

21
trustgraph-base/trustgraph/schema/lookup.py
View file

@ -17,26 +17,5 @@ class LookupResponse(Record):
text = String()
error = Error()
encyclopedia_lookup_request_queue = topic(
'encyclopedia', kind='non-persistent', namespace='request'
)
encyclopedia_lookup_response_queue = topic(
'encyclopedia', kind='non-persistent', namespace='response',
)
dbpedia_lookup_request_queue = topic(
'dbpedia', kind='non-persistent', namespace='request'
)
dbpedia_lookup_response_queue = topic(
'dbpedia', kind='non-persistent', namespace='response',
)
internet_search_request_queue = topic(
'internet-search', kind='non-persistent', namespace='request'
)
internet_search_response_queue = topic(
'internet-search', kind='non-persistent', namespace='response',
)
############################################################################

13
trustgraph-base/trustgraph/schema/models.py
View file

@ -19,13 +19,6 @@ class TextCompletionResponse(Record):
out_token = Integer()
model = String()
text_completion_request_queue = topic(
'text-completion', kind='non-persistent', namespace='request'
)
text_completion_response_queue = topic(
'text-completion', kind='non-persistent', namespace='response'
)
############################################################################
# Embeddings
@ -37,9 +30,3 @@ class EmbeddingsResponse(Record):
error = Error()
vectors = Array(Array(Double()))
embeddings_request_queue = topic(
'embeddings', kind='non-persistent', namespace='request'
)
embeddings_response_queue = topic(
'embeddings', kind='non-persistent', namespace='response'
)

4
trustgraph-base/trustgraph/schema/object.py
View file

@ -18,8 +18,6 @@ class ObjectEmbeddings(Record):
key_name = String()
id = String()
object_embeddings_store_queue = topic('object-embeddings-store')
############################################################################
# Stores rows of information
@ -29,5 +27,5 @@ class Rows(Record):
row_schema = RowSchema()
rows = Array(Map(String()))
rows_store_queue = topic('rows-store')

7
trustgraph-base/trustgraph/schema/prompt.py
View file

@ -55,12 +55,5 @@ class PromptResponse(Record):
# JSON encoded
object = String()
prompt_request_queue = topic(
'prompt', kind='non-persistent', namespace='request'
)
prompt_response_queue = topic(
'prompt', kind='non-persistent', namespace='response'
)
############################################################################

13
trustgraph-base/trustgraph/schema/retrieval.py
View file

@ -20,13 +20,6 @@ class GraphRagResponse(Record):
error = Error()
response = String()
graph_rag_request_queue = topic(
'graph-rag', kind='non-persistent', namespace='request'
)
graph_rag_response_queue = topic(
'graph-rag', kind='non-persistent', namespace='response'
)
############################################################################
# Document RAG text retrieval
@ -41,9 +34,3 @@ class DocumentRagResponse(Record):
error = Error()
response = String()
document_rag_request_queue = topic(
'doc-rag', kind='non-persistent', namespace='request'
)
document_rag_response_queue = topic(
'doc-rag', kind='non-persistent', namespace='response'
)

111
trustgraph-flow/trustgraph/config/service/flow.py Normal file
View file

@ -0,0 +1,111 @@
from trustgraph.schema import FlowResponse, Error
class FlowConfig:
def __init__(self, config):
self.config = config
async def handle_list_classes(self, msg):
names = list(self.config["flow-classes"].keys())
return FlowResponse(
error = None,
class_names = names,
)
async def handle_get_class(self, msg):
return FlowResponse(
error = None,
class_definition = self.config["flow-classes"][msg.class_name],
)
async def handle_put_class(self, msg):
self.config["flow-classes"][msg.class_name] = msg.class_definition
return FlowResponse(
error = None,
)
async def handle_delete_class(self, msg):
del self.config["flow-classes"][msg.class_name]
return FlowResponse(
error = None,
)
async def handle_list_flows(self, msg):
names = list(self.config["flows"].keys())
return FlowResponse(
error = None,
flow_ids = names,
)
async def handle_get_flow(self, msg):
flow = self.config["flows"][msg.flow_id]
return FlowResponse(
error = None,
flow = flow,
)
async def handle_start_flow(self, msg):
cls = self.config["flow-classes"][msg.class_name]
print(cls)
return FlowResponse(
error = None,
)
async def handle_stop_flow(self, msg):
flow = self.config["flows"][msg.flow_id]
return FlowResponse(
error = None,
)
async def handle(self, msg):
print("Handle message ", msg.operation)
if msg.operation == "list-classes":
resp = await self.handle_list_classes(msg)
elif msg.operation == "get-class":
resp = await self.handle_get_class(msg)
elif msg.operation == "put-class":
resp = await self.handle_put_class(msg)
elif msg.operation == "delete-class":
resp = await self.handle_delete_class(msg)
elif msg.operation == "list-flows":
resp = await self.handle_list_flows(msg)
elif msg.operation == "get-flow":
resp = await self.handle_get_flow(msg)
elif msg.operation == "start-flow":
resp = await self.handle_start_flow(msg)
elif msg.operation == "stop-flow":
resp = await self.handle_stop_flow(msg)
else:
resp = FlowResponse(
value=None,
directory=None,
values=None,
error=Error(
type = "bad-operation",
message = "Bad operation"
)
)
return resp

191
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -5,81 +5,129 @@ Config service. Manages system global configuration state
from pulsar.schema import JsonSchema
from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush
from trustgraph.schema import Error
from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush
from trustgraph.schema import config_request_queue, config_response_queue
from trustgraph.schema import config_push_queue
from trustgraph.schema import FlowRequest, FlowResponse
from trustgraph.schema import flow_request_queue, flow_response_queue
from trustgraph.log_level import LogLevel
from trustgraph.base import AsyncProcessor, Consumer, Producer
from . config import Configuration
from . flow import FlowConfig
from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
from ... base import Consumer, Producer
default_ident = "config-svc"
default_request_queue = config_request_queue
default_response_queue = config_response_queue
default_push_queue = config_push_queue
default_config_request_queue = config_request_queue
default_config_response_queue = config_response_queue
default_config_push_queue = config_push_queue
default_flow_request_queue = flow_request_queue
default_flow_response_queue = flow_response_queue
class Processor(AsyncProcessor):
def __init__(self, **params):
request_queue = params.get("request_queue", default_request_queue)
response_queue = params.get("response_queue", default_response_queue)
push_queue = params.get("push_queue", default_push_queue)
config_request_queue = params.get(
"config_request_queue", default_config_request_queue
)
config_response_queue = params.get(
"config_response_queue", default_config_response_queue
)
config_push_queue = params.get(
"config_push_queue", default_config_push_queue
)
flow_request_queue = params.get(
"flow_request_queue", default_flow_request_queue
)
flow_response_queue = params.get(
"flow_response_queue", default_flow_response_queue
)
id = params.get("id")
request_schema = ConfigRequest
response_schema = ConfigResponse
push_schema = ConfigResponse
flow_request_schema = FlowRequest
flow_response_schema = FlowResponse
super(Processor, self).__init__(
**params | {
"request_schema": request_schema.__name__,
"response_schema": response_schema.__name__,
"push_schema": push_schema.__name__,
"config_request_schema": ConfigRequest.__name__,
"config_response_schema": ConfigResponse.__name__,
"config_push_schema": ConfigPush.__name__,
"flow_request_schema": FlowRequest.__name__,
"flow_response_schema": FlowResponse.__name__,
}
)
request_metrics = ConsumerMetrics(id + "-request")
response_metrics = ProducerMetrics(id + "-response")
push_metrics = ProducerMetrics(id + "-push")
config_request_metrics = ConsumerMetrics(id + "-config-request")
config_response_metrics = ProducerMetrics(id + "-config-response")
config_push_metrics = ProducerMetrics(id + "-config-push")
self.push_pub = Producer(
client = self.client,
topic = push_queue,
schema = ConfigPush,
metrics = push_metrics,
)
flow_request_metrics = ConsumerMetrics(id + "-flow-request")
flow_response_metrics = ProducerMetrics(id + "-flow-response")
self.response_pub = Producer(
client = self.client,
topic = response_queue,
schema = ConfigResponse,
metrics = response_metrics,
)
self.subs = Consumer(
self.config_request_consumer = Consumer(
taskgroup = self.taskgroup,
client = self.client,
flow = None,
topic = request_queue,
topic = config_request_queue,
subscriber = id,
schema = request_schema,
handler = self.on_message,
metrics = request_metrics,
schema = ConfigRequest,
handler = self.on_config_request,
metrics = config_request_metrics,
)
self.config_response_producer = Producer(
client = self.client,
topic = config_response_queue,
schema = ConfigResponse,
metrics = config_response_metrics,
)
self.config_push_producer = Producer(
client = self.client,
topic = config_push_queue,
schema = ConfigPush,
metrics = config_push_metrics,
)
self.flow_request_consumer = Consumer(
taskgroup = self.taskgroup,
client = self.client,
flow = None,
topic = flow_request_queue,
subscriber = id,
schema = FlowRequest,
handler = self.on_flow_request,
metrics = flow_request_metrics,
)
self.flow_response_producer = Producer(
client = self.client,
topic = flow_response_queue,
schema = FlowResponse,
metrics = flow_response_metrics,
)
self.config = Configuration(self.push)
self.flow = FlowConfig(self.config)
print("Service initialised.")
async def start(self):
await self.push()
await self.subs.start()
await self.config_request_consumer.start()
await self.flow_request_consumer.start()
async def push(self):
@ -92,11 +140,11 @@ class Processor(AsyncProcessor):
error = None,
)
await self.push_pub.send(resp)
await self.config_push_producer.send(resp)
print("Pushed version ", self.config.version)
async def on_message(self, msg, consumer, flow):
async def on_config_request(self, msg, consumer, flow):
try:
@ -109,19 +157,54 @@ class Processor(AsyncProcessor):
resp = await self.config.handle(v)
await self.response_pub.send(resp, properties={"id": id})
await self.config_response_producer.send(
resp, properties={"id": id}
)
except Exception as e:
resp = ConfigResponse(
error=Error(
type = "unexpected-error",
type = "config-error",
message = str(e),
),
text=None,
)
await self.response_pub.send(resp, properties={"id": id})
await self.config_response_producer.send(
resp, properties={"id": id}
)
async def on_flow_request(self, msg, consumer, flow):
try:
v = msg.value()
# Sender-produced ID
id = msg.properties()["id"]
print(f"Handling {id}...", flush=True)
resp = await self.flow.handle(v)
await self.flow_response_producer.send(
resp, properties={"id": id}
)
except Exception as e:
resp = FlowResponse(
error=Error(
type = "flow-error",
message = str(e),
),
text=None,
)
await self.flow_response_producer.send(
resp, properties={"id": id}
)
@staticmethod
def add_args(parser):
@ -129,21 +212,33 @@ class Processor(AsyncProcessor):
AsyncProcessor.add_args(parser)
parser.add_argument(
'-q', '--request-queue',
default=default_request_queue,
help=f'Request queue (default: {default_request_queue})'
'--config-request-queue',
default=default_config_request_queue,
help=f'Config request queue (default: {default_config_request_queue})'
)
parser.add_argument(
'-r', '--response-queue',
default=default_response_queue,
help=f'Response queue {default_response_queue}',
'--config-response-queue',
default=default_config_response_queue,
help=f'Config response queue {default_config_response_queue}',
)
parser.add_argument(
'--push-queue',
default=default_push_queue,
help=f'Config push queue (default: {default_push_queue})'
default=default_config_push_queue,
help=f'Config push queue (default: {default_config_push_queue})'
)
parser.add_argument(
'--flow-request-queue',
default=default_flow_request_queue,
help=f'Flow request queue (default: {default_flow_request_queue})'
)
parser.add_argument(
'--flow-response-queue',
default=default_flow_response_queue,
help=f'Flow response queue {default_flow_response_queue}',
)
def run():

51
trustgraph-flow/trustgraph/gateway/flow.py Normal file
View file

@ -0,0 +1,51 @@
from .. schema import FlowRequest, FlowResponse, ConfigKey, ConfigValue
from .. schema import flow_request_queue
from .. schema import flow_response_queue
from . endpoint import ServiceEndpoint
from . requestor import ServiceRequestor
class FlowRequestor(ServiceRequestor):
def __init__(self, pulsar_client, timeout, auth):
super(FlowRequestor, self).__init__(
pulsar_client=pulsar_client,
request_queue=flow_request_queue,
response_queue=flow_response_queue,
request_schema=FlowRequest,
response_schema=FlowResponse,
timeout=timeout,
)
def to_request(self, body):
return FlowRequest(
operation = body.get("operation", None),
class_name = body.get("class-name", None),
class_definition = body.get("class-definition", None),
description = body.get("description", None),
flow_id = body.get("flow-id", None),
)
def from_response(self, message):
response = { }
if message.class_names is not None:
response["class-names"] = message.class_names
if message.flow_ids is not None:
response["flow-ids"] = message.flow_ids
if message.class_definition is not None:
response["class-definition"] = message.class_definition
if message.flow is not None:
response["flow"] = message.flow
if message.description is not None:
response["description"] = message.description
return response, True

1
trustgraph-flow/trustgraph/gateway/mux.py
View file

@ -5,7 +5,6 @@ import uuid
from aiohttp import web, WSMsgType
from . socket import SocketEndpoint
from . text_completion import TextCompletionRequestor
MAX_OUTSTANDING_REQUESTS = 15
WORKER_CLOSE_WAIT = 0.01

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

@ -25,28 +25,30 @@ from .. log_level import LogLevel
from . serialize import to_subgraph
from . running import Running
from . text_completion import TextCompletionRequestor
from . prompt import PromptRequestor
from . graph_rag import GraphRagRequestor
from . document_rag import DocumentRagRequestor
from . triples_query import TriplesQueryRequestor
from . graph_embeddings_query import GraphEmbeddingsQueryRequestor
from . embeddings import EmbeddingsRequestor
from . encyclopedia import EncyclopediaRequestor
from . agent import AgentRequestor
from . dbpedia import DbpediaRequestor
from . internet_search import InternetSearchRequestor
from . librarian import LibrarianRequestor
#from . text_completion import TextCompletionRequestor
#from . prompt import PromptRequestor
#from . graph_rag import GraphRagRequestor
#from . document_rag import DocumentRagRequestor
#from . triples_query import TriplesQueryRequestor
#from . graph_embeddings_query import GraphEmbeddingsQueryRequestor
#from . embeddings import EmbeddingsRequestor
#from . encyclopedia import EncyclopediaRequestor
#from . agent import AgentRequestor
#from . dbpedia import DbpediaRequestor
#from . internet_search import InternetSearchRequestor
#from . librarian import LibrarianRequestor
from . config import ConfigRequestor
from . triples_stream import TriplesStreamEndpoint
from . graph_embeddings_stream import GraphEmbeddingsStreamEndpoint
from . document_embeddings_stream import DocumentEmbeddingsStreamEndpoint
from . triples_load import TriplesLoadEndpoint
from . graph_embeddings_load import GraphEmbeddingsLoadEndpoint
from . document_embeddings_load import DocumentEmbeddingsLoadEndpoint
from . flow import FlowRequestor
#from . triples_stream import TriplesStreamEndpoint
#from . graph_embeddings_stream import GraphEmbeddingsStreamEndpoint
#from . document_embeddings_stream import DocumentEmbeddingsStreamEndpoint
#from . triples_load import TriplesLoadEndpoint
#from . graph_embeddings_load import GraphEmbeddingsLoadEndpoint
#from . document_embeddings_load import DocumentEmbeddingsLoadEndpoint
from . mux import MuxEndpoint
from . document_load import DocumentLoadSender
from . text_load import TextLoadSender
#from . document_load import DocumentLoadSender
#from . text_load import TextLoadSender
from . metrics import MetricsEndpoint
from . endpoint import ServiceEndpoint
@ -105,162 +107,170 @@ class Api:
self.auth = Authenticator(allow_all=True)
self.services = {
"text-completion": TextCompletionRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"prompt": PromptRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"graph-rag": GraphRagRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"document-rag": DocumentRagRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"triples-query": TriplesQueryRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"graph-embeddings-query": GraphEmbeddingsQueryRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"embeddings": EmbeddingsRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"agent": AgentRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"librarian": LibrarianRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
# "text-completion": TextCompletionRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "prompt": PromptRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "graph-rag": GraphRagRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "document-rag": DocumentRagRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "triples-query": TriplesQueryRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "graph-embeddings-query": GraphEmbeddingsQueryRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "embeddings": EmbeddingsRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "agent": AgentRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "librarian": LibrarianRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
"config": ConfigRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"encyclopedia": EncyclopediaRequestor(
"flow": FlowRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"dbpedia": DbpediaRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"internet-search": InternetSearchRequestor(
pulsar_client=self.pulsar_client, timeout=self.timeout,
auth = self.auth,
),
"document-load": DocumentLoadSender(
pulsar_client=self.pulsar_client,
),
"text-load": TextLoadSender(
pulsar_client=self.pulsar_client,
),
# "encyclopedia": EncyclopediaRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "dbpedia": DbpediaRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "internet-search": InternetSearchRequestor(
# pulsar_client=self.pulsar_client, timeout=self.timeout,
# auth = self.auth,
# ),
# "document-load": DocumentLoadSender(
# pulsar_client=self.pulsar_client,
# ),
# "text-load": TextLoadSender(
# pulsar_client=self.pulsar_client,
# ),
}
self.endpoints = [
ServiceEndpoint(
endpoint_path = "/api/v1/text-completion", auth=self.auth,
requestor = self.services["text-completion"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/prompt", auth=self.auth,
requestor = self.services["prompt"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/graph-rag", auth=self.auth,
requestor = self.services["graph-rag"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/document-rag", auth=self.auth,
requestor = self.services["document-rag"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/triples-query", auth=self.auth,
requestor = self.services["triples-query"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/graph-embeddings-query",
auth=self.auth,
requestor = self.services["graph-embeddings-query"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/embeddings", auth=self.auth,
requestor = self.services["embeddings"],
),
ServiceEndpoint(
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/text-completion", auth=self.auth,
# requestor = self.services["text-completion"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/prompt", auth=self.auth,
# requestor = self.services["prompt"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/graph-rag", auth=self.auth,
# requestor = self.services["graph-rag"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/document-rag", auth=self.auth,
# requestor = self.services["document-rag"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/triples-query", auth=self.auth,
# requestor = self.services["triples-query"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/graph-embeddings-query",
# auth=self.auth,
# requestor = self.services["graph-embeddings-query"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/embeddings", auth=self.auth,
# requestor = self.services["embeddings"],
# ),
# ServiceEndpoint(
# 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/config", auth=self.auth,
requestor = self.services["config"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/encyclopedia", auth=self.auth,
requestor = self.services["encyclopedia"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/dbpedia", auth=self.auth,
requestor = self.services["dbpedia"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/internet-search", auth=self.auth,
requestor = self.services["internet-search"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/load/document", auth=self.auth,
requestor = self.services["document-load"],
),
ServiceEndpoint(
endpoint_path = "/api/v1/load/text", auth=self.auth,
requestor = self.services["text-load"],
),
TriplesStreamEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
GraphEmbeddingsStreamEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
DocumentEmbeddingsStreamEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
TriplesLoadEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
GraphEmbeddingsLoadEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
DocumentEmbeddingsLoadEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
),
MuxEndpoint(
pulsar_client=self.pulsar_client,
auth = self.auth,
services = self.services,
),
MetricsEndpoint(
endpoint_path = "/api/v1/metrics",
prometheus_url = self.prometheus_url,
auth = self.auth,
endpoint_path = "/api/v1/flow", auth=self.auth,
requestor = self.services["flow"],
),
# ServiceEndpoint(
# endpoint_path = "/api/v1/encyclopedia", auth=self.auth,
# requestor = self.services["encyclopedia"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/dbpedia", auth=self.auth,
# requestor = self.services["dbpedia"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/internet-search", auth=self.auth,
# requestor = self.services["internet-search"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/load/document", auth=self.auth,
# requestor = self.services["document-load"],
# ),
# ServiceEndpoint(
# endpoint_path = "/api/v1/load/text", auth=self.auth,
# requestor = self.services["text-load"],
# ),
# TriplesStreamEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# GraphEmbeddingsStreamEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# DocumentEmbeddingsStreamEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# TriplesLoadEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# GraphEmbeddingsLoadEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# DocumentEmbeddingsLoadEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# ),
# MuxEndpoint(
# pulsar_client=self.pulsar_client,
# auth = self.auth,
# services = self.services,
# ),
# MetricsEndpoint(
# endpoint_path = "/api/v1/metrics",
# prometheus_url = self.prometheus_url,
# auth = self.auth,
# ),
]
for ep in self.endpoints: