diff --git a/trustgraph-flow/trustgraph/gateway/agent.py b/trustgraph-flow/trustgraph/gateway/agent.py index c86b5f02..b6307a62 100644 --- a/trustgraph-flow/trustgraph/gateway/agent.py +++ b/trustgraph-flow/trustgraph/gateway/agent.py @@ -24,16 +24,26 @@ class AgentRequestor(MultiResponseServiceRequestor): ) def from_response(self, message): + resp = { + } + if message.answer: - return { "answer": message.answer }, True - else: - return {}, False + resp["answer"] = message.answer + + if message.thought: + resp["thought"] = message.thought + + if message.observation: + resp["observation"] = message.observation + + # The 2nd boolean expression indicates whether we're done responding + return resp, (message.answer is not None) class AgentEndpoint(MultiResponseServiceEndpoint): def __init__(self, pulsar_host, timeout, auth): super(AgentEndpoint, self).__init__( - endpoint_path="/api/v1/text-completion", + endpoint_path="/api/v1/agent", auth=auth, requestor = AgentRequestor( pulsar_host=pulsar_host, timeout=timeout, auth=auth diff --git a/trustgraph-flow/trustgraph/gateway/endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint.py index 40371cc9..8e844533 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint.py @@ -68,20 +68,13 @@ class MultiResponseServiceEndpoint(ServiceEndpoint): async def handle(self, request): - id = str(uuid.uuid4()) - try: data = await request.json() - - q = self.sub.subscribe(id) - - await asyncio.to_thread( - self.pub.send, id, self.to_request(data) - ) + print(data) def responder(x): - print("Resp:", x) + print(x) resp = await self.requestor.process(data, responder) @@ -94,5 +87,3 @@ class MultiResponseServiceEndpoint(ServiceEndpoint): { "error": str(e) } ) - finally: - self.sub.unsubscribe(id) diff --git a/trustgraph-flow/trustgraph/gateway/requestor.py b/trustgraph-flow/trustgraph/gateway/requestor.py index 985ff2dd..ad6cf185 100644 --- a/trustgraph-flow/trustgraph/gateway/requestor.py +++ b/trustgraph-flow/trustgraph/gateway/requestor.py @@ -48,25 +48,20 @@ class ServiceRequestor: async def process(self, request): id = str(uuid.uuid4()) - print(id) try: q = self.sub.subscribe(id) - print("Pub...") await asyncio.to_thread( self.pub.send, id, self.to_request(request) ) - print("Pubd") try: resp = await asyncio.to_thread(q.get, timeout=self.timeout) except Exception as e: raise RuntimeError("Timeout") - print("Recvd...") - if resp.error: return { "error": resp.error.message }