From 5177df23dee4482aa93b902a0a8b90bbfb5a0dd4 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Thu, 4 Sep 2025 15:10:07 +0100 Subject: [PATCH] Structure query service --- trustgraph-flow/pyproject.toml | 1 + .../retrieval/structured_query/__init__.py | 1 + .../retrieval/structured_query/__main__.py | 5 + .../retrieval/structured_query/service.py | 167 ++++++++++++++++++ 4 files changed, 174 insertions(+) create mode 100644 trustgraph-flow/trustgraph/retrieval/structured_query/__init__.py create mode 100644 trustgraph-flow/trustgraph/retrieval/structured_query/__main__.py create mode 100644 trustgraph-flow/trustgraph/retrieval/structured_query/service.py diff --git a/trustgraph-flow/pyproject.toml b/trustgraph-flow/pyproject.toml index 75428ff2..0e9b5b0d 100644 --- a/trustgraph-flow/pyproject.toml +++ b/trustgraph-flow/pyproject.toml @@ -96,6 +96,7 @@ pdf-ocr-mistral = "trustgraph.decoding.mistral_ocr:run" prompt-template = "trustgraph.prompt.template:run" rev-gateway = "trustgraph.rev_gateway:run" run-processing = "trustgraph.processing:run" +structured-query = "trustgraph.retrieval.structured_query:run" text-completion-azure = "trustgraph.model.text_completion.azure:run" text-completion-azure-openai = "trustgraph.model.text_completion.azure_openai:run" text-completion-claude = "trustgraph.model.text_completion.claude:run" diff --git a/trustgraph-flow/trustgraph/retrieval/structured_query/__init__.py b/trustgraph-flow/trustgraph/retrieval/structured_query/__init__.py new file mode 100644 index 00000000..974260f2 --- /dev/null +++ b/trustgraph-flow/trustgraph/retrieval/structured_query/__init__.py @@ -0,0 +1 @@ +from . service import * \ No newline at end of file diff --git a/trustgraph-flow/trustgraph/retrieval/structured_query/__main__.py b/trustgraph-flow/trustgraph/retrieval/structured_query/__main__.py new file mode 100644 index 00000000..0bec8f9d --- /dev/null +++ b/trustgraph-flow/trustgraph/retrieval/structured_query/__main__.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python3 + +from . service import run + +run() \ No newline at end of file diff --git a/trustgraph-flow/trustgraph/retrieval/structured_query/service.py b/trustgraph-flow/trustgraph/retrieval/structured_query/service.py new file mode 100644 index 00000000..d637819c --- /dev/null +++ b/trustgraph-flow/trustgraph/retrieval/structured_query/service.py @@ -0,0 +1,167 @@ +""" +Structured Query Service - orchestrates natural language question processing. +Takes a question, converts it to GraphQL via nlp-query, executes via objects-query, +and returns the results. +""" + +import json +import logging +from typing import Dict, Any, Optional + +from ...schema import StructuredQueryRequest, StructuredQueryResponse +from ...schema import QuestionToStructuredQueryRequest, QuestionToStructuredQueryResponse +from ...schema import ObjectsQueryRequest, ObjectsQueryResponse +from ...schema import Error + +from ...base import FlowProcessor, ConsumerSpec, ProducerSpec, ClientSpec + +# Module logger +logger = logging.getLogger(__name__) + +default_ident = "structured-query" + +class Processor(FlowProcessor): + + def __init__(self, **params): + + id = params.get("id", default_ident) + + super(Processor, self).__init__( + **params | { + "id": id, + } + ) + + self.register_specification( + ConsumerSpec( + name = "request", + schema = StructuredQueryRequest, + handler = self.on_message + ) + ) + + self.register_specification( + ProducerSpec( + name = "response", + schema = StructuredQueryResponse, + ) + ) + + # Client spec for calling NLP query service + self.register_specification( + ClientSpec( + request_name = "nlp-query-request", + response_name = "nlp-query-response", + request_schema = QuestionToStructuredQueryRequest, + response_schema = QuestionToStructuredQueryResponse + ) + ) + + # Client spec for calling objects query service + self.register_specification( + ClientSpec( + request_name = "objects-query-request", + response_name = "objects-query-response", + request_schema = ObjectsQueryRequest, + response_schema = ObjectsQueryResponse + ) + ) + + logger.info("Structured Query service initialized") + + async def on_message(self, msg, consumer, flow): + """Handle incoming structured query request""" + + try: + request = msg.value() + + # Sender-produced ID + id = msg.properties()["id"] + + logger.info(f"Handling structured query request {id}: {request.question[:100]}...") + + # Step 1: Convert question to GraphQL using NLP query service + logger.info("Step 1: Converting question to GraphQL") + nlp_request = QuestionToStructuredQueryRequest( + question=request.question, + max_results=100 # Default limit + ) + + nlp_response = await self.client("nlp-query-request").request(nlp_request) + + if nlp_response.error is not None: + raise Exception(f"NLP query service error: {nlp_response.error.message}") + + if not nlp_response.graphql_query: + raise Exception("NLP query service returned empty GraphQL query") + + logger.info(f"Generated GraphQL query: {nlp_response.graphql_query[:200]}...") + logger.info(f"Detected schemas: {nlp_response.detected_schemas}") + logger.info(f"Confidence: {nlp_response.confidence}") + + # Step 2: Execute GraphQL query using objects query service + logger.info("Step 2: Executing GraphQL query") + + # For now, we'll use default user/collection values + # In a real implementation, these would come from authentication/context + objects_request = ObjectsQueryRequest( + user="default", # TODO: Get from authentication context + collection="default", # TODO: Get from request context + query=nlp_response.graphql_query, + variables=nlp_response.variables, + operation_name=None + ) + + objects_response = await self.client("objects-query-request").request(objects_request) + + if objects_response.error is not None: + raise Exception(f"Objects query service error: {objects_response.error.message}") + + # Handle GraphQL errors from the objects query service + graphql_errors = [] + if objects_response.errors: + for gql_error in objects_response.errors: + graphql_errors.append(f"{gql_error.message} (path: {gql_error.path})") + + logger.info("Step 3: Returning results") + + # Create response + response = StructuredQueryResponse( + error=None, + data=objects_response.data or "null", # JSON string + errors=graphql_errors + ) + + logger.info("Sending structured query response...") + await flow("response").send(response, properties={"id": id}) + + logger.info("Structured query request completed") + + except Exception as e: + + logger.error(f"Exception in structured query service: {e}", exc_info=True) + + logger.info("Sending error response...") + + response = StructuredQueryResponse( + error = Error( + type = "structured-query-error", + message = str(e), + ), + data = "null", + errors = [] + ) + + await flow("response").send(response, properties={"id": id}) + + @staticmethod + def add_args(parser): + """Add command-line arguments""" + + FlowProcessor.add_args(parser) + + # No additional arguments needed for this orchestrator service + +def run(): + """Entry point for structured-query command""" + Processor.launch(default_ident, __doc__) \ No newline at end of file