diff --git a/test-api/test-llm2-api b/test-api/test-llm2-api new file mode 100755 index 00000000..93d9d348 --- /dev/null +++ b/test-api/test-llm2-api @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 + +import requests +import json +import sys + +url = "http://localhost:8088/api/v1/" + +############################################################################ + +input = { + "system": "", + "prompt": "Add 2 and 3" +} + +resp = requests.post( + f"{url}text-completion", + json=input, +) + +if resp.status_code != 200: + raise RuntimeError(f"Status code: {resp.status_code}") + +resp = resp.json() + +if "error" in resp: + print(f"Error: {resp['error']}") + sys.exit(1) + +print(resp["response"]) + +############################################################################ + diff --git a/test-api/test-load-document b/test-api/test-load-document index 77a0dffd..03dfc787 100755 --- a/test-api/test-load-document +++ b/test-api/test-load-document @@ -5,7 +5,7 @@ import json import sys import base64 -url = "http://localhost:8088/api/v1/" +url = "http://localhost:8088/api/v1/flow/0000/document-load" ############################################################################ @@ -88,10 +88,7 @@ input = { } -resp = requests.post( - f"{url}load/document", - json=input, -) +resp = requests.post(url, json=input) resp = resp.json() diff --git a/test-api/test-load-text b/test-api/test-load-text index 6865bdd4..dd2824a7 100755 --- a/test-api/test-load-text +++ b/test-api/test-load-text @@ -5,7 +5,7 @@ import json import sys import base64 -url = "http://localhost:8088/api/v1/" +url = "http://localhost:8088/api/v1/flow/0000/text-load" ############################################################################ @@ -85,10 +85,7 @@ input = { } -resp = requests.post( - f"{url}load/text", - json=input, -) +resp = requests.post(url, json=input) resp = resp.json() diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/document_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/document_load.py index 78cd7930..f92fc34f 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/document_load.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/document_load.py @@ -1,19 +1,18 @@ import base64 -from .. schema import Document, Metadata -from .. schema import document_ingest_queue +from ... schema import Document, Metadata from . sender import ServiceSender from . serialize import to_subgraph -class DocumentLoadSender(ServiceSender): - def __init__(self, pulsar_client): +class DocumentLoad(ServiceSender): + def __init__(self, pulsar_client, queue): - super(DocumentLoadSender, self).__init__( - pulsar_client=pulsar_client, - request_queue=document_ingest_queue, - request_schema=Document, + super(DocumentLoad, self).__init__( + pulsar_client = pulsar_client, + queue = queue, + schema = Document, ) def to_request(self, body): diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py index 4b896718..34fc2627 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py @@ -16,6 +16,8 @@ from . triples_query import TriplesQueryRequestor from . embeddings import EmbeddingsRequestor from . graph_embeddings_query import GraphEmbeddingsQueryRequestor from . prompt import PromptRequestor +from . text_load import TextLoad +from . document_load import DocumentLoad from . triples_stream import TriplesStream from . graph_embeddings_stream import GraphEmbeddingsStream @@ -32,6 +34,11 @@ request_response_dispatchers = { "triples-query": TriplesQueryRequestor, } +sender_dispatchers = { + "text-load": TextLoad, + "document-load": DocumentLoad, +} + receive_dispatchers = { "triples": TriplesStream, "graph-embeddings": GraphEmbeddingsStream, @@ -141,9 +148,6 @@ class DispatcherManager: if flow not in self.flows: raise RuntimeError("Invalid flow") - if kind not in request_response_dispatchers: - raise RuntimeError("Invalid kind") - key = (flow, kind) if key in self.dispatchers: @@ -156,15 +160,23 @@ class DispatcherManager: qconfig = intf_defs[kind] - dispatcher = request_response_dispatchers[kind]( - pulsar_client = self.pulsar_client, - request_queue = qconfig["request"], - response_queue = qconfig["response"], - timeout = 120, - consumer = f"api-gateway-{flow}-{kind}-request", - subscriber = f"api-gateway-{flow}-{kind}-request", - ) - + if kind in request_response_dispatchers: + dispatcher = request_response_dispatchers[kind]( + pulsar_client = self.pulsar_client, + request_queue = qconfig["request"], + response_queue = qconfig["response"], + timeout = 120, + consumer = f"api-gateway-{flow}-{kind}-request", + subscriber = f"api-gateway-{flow}-{kind}-request", + ) + elif kind in sender_dispatchers: + dispatcher = sender_dispatchers[kind]( + pulsar_client = self.pulsar_client, + queue = qconfig, + ) + else: + raise RuntimeError("Invalid kind") + await dispatcher.start() self.dispatchers[key] = dispatcher diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/sender.py b/trustgraph-flow/trustgraph/gateway/dispatch/sender.py index 86181021..2435cdc1 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/sender.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/sender.py @@ -15,18 +15,20 @@ class ServiceSender: def __init__( self, pulsar_client, - request_queue, request_schema, + queue, schema, ): self.pub = Publisher( - pulsar_client, request_queue, - schema=request_schema, + pulsar_client, queue, + schema=schema, ) async def start(self): - await self.pub.start() + async def stop(self): + await self.pub.stop() + def to_request(self, request): raise RuntimeError("Not defined") @@ -39,6 +41,8 @@ class ServiceSender: if responder: await responder({}, True) + return {} + except Exception as e: logging.error(f"Exception: {e}") diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/text_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/text_load.py index cc432698..53ea7452 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/text_load.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/text_load.py @@ -1,19 +1,18 @@ import base64 -from .. schema import TextDocument, Metadata -from .. schema import text_ingest_queue +from ... schema import TextDocument, Metadata from . sender import ServiceSender from . serialize import to_subgraph -class TextLoadSender(ServiceSender): - def __init__(self, pulsar_client): +class TextLoad(ServiceSender): + def __init__(self, pulsar_client, queue): - super(TextLoadSender, self).__init__( - pulsar_client=pulsar_client, - request_queue=text_ingest_queue, - request_schema=TextDocument, + super(TextLoad, self).__init__( + pulsar_client = pulsar_client, + queue = queue, + schema = TextDocument, ) def to_request(self, body): diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/manager.py b/trustgraph-flow/trustgraph/gateway/endpoint/manager.py new file mode 100644 index 00000000..eef7059c --- /dev/null +++ b/trustgraph-flow/trustgraph/gateway/endpoint/manager.py @@ -0,0 +1,62 @@ + +import asyncio + +from aiohttp import web + +from . constant_endpoint import ConstantEndpoint +from . variable_endpoint import VariableEndpoint +from . socket import SocketEndpoint +from . metrics import MetricsEndpoint + +from .. dispatch.manager import DispatcherManager + +class EndpointManager: + + def __init__( + self, dispatcher_manager, auth, prometheus_url, timeout=600 + ): + + self.dispatcher_manager = dispatcher_manager + self.timeout = timeout + + self.services = { + } + + self.endpoints = [ + ConstantEndpoint( + endpoint_path = "/api/v1/librarian", auth = auth, + dispatcher = dispatcher_manager.dispatch_librarian(), + ), + ConstantEndpoint( + endpoint_path = "/api/v1/config", auth = auth, + dispatcher = dispatcher_manager.dispatch_config(), + ), + ConstantEndpoint( + endpoint_path = "/api/v1/flow", auth = auth, + dispatcher = dispatcher_manager.dispatch_flow(), + ), + MetricsEndpoint( + endpoint_path = "/api/v1/metrics", + prometheus_url = prometheus_url, + auth = auth, + ), + VariableEndpoint( + endpoint_path = "/api/v1/flow/{flow}/{kind}", + auth = auth, + dispatcher = dispatcher_manager.dispatch_flow_service(), + ), + SocketEndpoint( + endpoint_path = "/api/v1/flow/{flow}/receive/{kind}", + auth = auth, + dispatcher = dispatcher_manager.dispatch_flow_receive() + ), + ] + + def add_routes(self, app): + for ep in self.endpoints: + ep.add_routes(app) + + async def start(self): + for ep in self.endpoints: + await ep.start() +