From a243d93db6ab383be47aac7f84c3b42c2e51deae Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 6 Dec 2024 20:58:31 +0000 Subject: [PATCH] Service/endpoint separation --- trustgraph-flow/trustgraph/gateway/command.py | 13 +- trustgraph-flow/trustgraph/gateway/service.py | 114 ++++++++++++------ 2 files changed, 84 insertions(+), 43 deletions(-) diff --git a/trustgraph-flow/trustgraph/gateway/command.py b/trustgraph-flow/trustgraph/gateway/command.py index 026523d9..86969bf3 100644 --- a/trustgraph-flow/trustgraph/gateway/command.py +++ b/trustgraph-flow/trustgraph/gateway/command.py @@ -5,17 +5,15 @@ from pulsar.schema import JsonSchema import uuid from aiohttp import web, WSMsgType -from .. schema import GraphEmbeddings -from .. schema import graph_embeddings_store_queue - -from . subscriber import Subscriber from . socket import SocketEndpoint -from . serialize import serialize_graph_embeddings +from . text_completion import TextCompletionRequestor class CommandEndpoint(SocketEndpoint): def __init__( - self, pulsar_host, auth, path="/api/v1/command" + self, pulsar_host, auth, + services, + path="/api/v1/command", ): super(CommandEndpoint, self).__init__( @@ -24,6 +22,9 @@ class CommandEndpoint(SocketEndpoint): self.pulsar_host=pulsar_host +# self.text_completion = TextCompletionRequestor( +# ) + async def start(self): pass diff --git a/trustgraph-flow/trustgraph/gateway/service.py b/trustgraph-flow/trustgraph/gateway/service.py index 81f02197..b0ad2fd1 100755 --- a/trustgraph-flow/trustgraph/gateway/service.py +++ b/trustgraph-flow/trustgraph/gateway/service.py @@ -31,22 +31,22 @@ from . serialize import to_subgraph from . running import Running from . publisher import Publisher from . subscriber import Subscriber -from . endpoint import ServiceEndpoint, MultiResponseServiceEndpoint -from . text_completion import TextCompletionEndpoint -from . prompt import PromptEndpoint -from . graph_rag import GraphRagEndpoint -from . triples_query import TriplesQueryEndpoint -from . embeddings import EmbeddingsEndpoint -from . encyclopedia import EncyclopediaEndpoint -from . agent import AgentEndpoint -from . dbpedia import DbpediaEndpoint -from . internet_search import InternetSearchEndpoint +from . text_completion import TextCompletionRequestor +from . prompt import PromptRequestor +from . graph_rag import GraphRagRequestor +from . triples_query import TriplesQueryRequestor +from . embeddings import EmbeddingsRequestor +from . encyclopedia import EncyclopediaRequestor +from . agent import AgentRequestor +from . dbpedia import DbpediaRequestor +from . internet_search import InternetSearchRequestor from . triples_stream import TriplesStreamEndpoint from . graph_embeddings_stream import GraphEmbeddingsStreamEndpoint from . triples_load import TriplesLoadEndpoint from . graph_embeddings_load import GraphEmbeddingsLoadEndpoint from . command import CommandEndpoint +from . endpoint import ServiceEndpoint, MultiResponseServiceEndpoint from . auth import Authenticator logger = logging.getLogger("api") @@ -78,42 +78,81 @@ class Api: else: self.auth = Authenticator(allow_all=True) + self.services = { + "text-completion": TextCompletionRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "prompt": PromptRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "graph-rag": GraphRagRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "triples-query": TriplesQueryRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "embeddings": EmbeddingsRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "agent": AgentRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "encyclopedia": EncyclopediaRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "dbpedia": DbpediaRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + "internet-search": InternetSearchRequestor( + pulsar_host=self.pulsar_host, timeout=self.timeout, + auth = self.auth, + ), + } + self.endpoints = [ - TextCompletionEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/text-completion", auth=self.auth, + requestor = self.services["text-completion"], ), - PromptEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/prompt", auth=self.auth, + requestor = self.services["prompt"], ), - GraphRagEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/graph-rag", auth=self.auth, + requestor = self.services["graph-rag"], ), - TriplesQueryEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/triples-query", auth=self.auth, + requestor = self.services["triples-query"], ), - EmbeddingsEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/embeddings", auth=self.auth, + requestor = self.services["embeddings"], ), - AgentEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + MultiResponseServiceEndpoint( + endpoint_path = "/api/v1/agent", auth=self.auth, + requestor = self.services["agent"], ), - EncyclopediaEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/encyclopedia", auth=self.auth, + requestor = self.services["encyclopedia"], ), - DbpediaEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/dbpedia", auth=self.auth, + requestor = self.services["dbpedia"], ), - InternetSearchEndpoint( - pulsar_host=self.pulsar_host, timeout=self.timeout, - auth = self.auth, + ServiceEndpoint( + endpoint_path = "/api/v1/internet-search", auth=self.auth, + requestor = self.services["internet-search"], ), TriplesStreamEndpoint( pulsar_host=self.pulsar_host, @@ -134,6 +173,7 @@ class Api: CommandEndpoint( pulsar_host=self.pulsar_host, auth = self.auth, + services = self.services, ), ]