From 8c38dc30ffb7873111cc455cfd5b979453b13ae2 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 21 Apr 2025 20:09:53 +0100 Subject: [PATCH] Fixing query stuff --- .../trustgraph/base/graph_embeddings_client.py | 6 ++++-- .../base/graph_embeddings_query_service.py | 2 +- .../trustgraph/base/triples_client.py | 8 ++++---- .../trustgraph/base/triples_query_service.py | 4 ++-- trustgraph-base/trustgraph/knowledge/defs.py | 4 ++++ .../query/graph_embeddings/qdrant/service.py | 17 +++++------------ .../query/triples/cassandra/service.py | 7 ------- .../trustgraph/retrieval/graph_rag/graph_rag.py | 12 +++++++----- 8 files changed, 27 insertions(+), 33 deletions(-) diff --git a/trustgraph-base/trustgraph/base/graph_embeddings_client.py b/trustgraph-base/trustgraph/base/graph_embeddings_client.py index 67b290fb..e89364f2 100644 --- a/trustgraph-base/trustgraph/base/graph_embeddings_client.py +++ b/trustgraph-base/trustgraph/base/graph_embeddings_client.py @@ -4,8 +4,8 @@ from .. schema import GraphEmbeddingsRequest, GraphEmbeddingsResponse from .. knowledge import Uri, Literal def to_value(x): - if x.e: return Uri(x.v) - return Literal(x.v) + if x.is_uri: return Uri(x.value) + return Literal(x.value) class GraphEmbeddingsClient(RequestResponse): async def query(self, vectors, limit=20, user="trustgraph", @@ -21,6 +21,8 @@ class GraphEmbeddingsClient(RequestResponse): timeout=timeout ) + print(resp, flush=True) + if resp.error: raise RuntimeError(resp.error.message) diff --git a/trustgraph-base/trustgraph/base/graph_embeddings_query_service.py b/trustgraph-base/trustgraph/base/graph_embeddings_query_service.py index 003ba664..fb2e8dc5 100755 --- a/trustgraph-base/trustgraph/base/graph_embeddings_query_service.py +++ b/trustgraph-base/trustgraph/base/graph_embeddings_query_service.py @@ -49,7 +49,7 @@ class GraphEmbeddingsQueryService(FlowProcessor): print(f"Handling input {id}...", flush=True) - entities = self.query_graph_embeddings(request) + entities = await self.query_graph_embeddings(request) print("Send response...", flush=True) r = GraphEmbeddingsResponse(entities=entities, error=None) diff --git a/trustgraph-base/trustgraph/base/triples_client.py b/trustgraph-base/trustgraph/base/triples_client.py index 17e53ca1..56996e91 100644 --- a/trustgraph-base/trustgraph/base/triples_client.py +++ b/trustgraph-base/trustgraph/base/triples_client.py @@ -1,11 +1,11 @@ from . request_response_spec import RequestResponse, RequestResponseSpec from .. schema import TriplesQueryRequest, TriplesQueryResponse, Value -from .. knowledge import Uri, Literal +from .. knowledge import Uri, Literal, Triple def to_value(x): - if x.e: return Uri(x.v) - return Literal(x.v) + if x.is_uri: return Uri(x.value) + return Literal(x.value) def from_value(x): if x is None: return None @@ -35,7 +35,7 @@ class TriplesClient(RequestResponse): raise RuntimeError(resp.error.message) return [ - to_value(v) + Triple(to_value(v.s), to_value(v.p), to_value(v.o)) for v in resp.triples ] diff --git a/trustgraph-base/trustgraph/base/triples_query_service.py b/trustgraph-base/trustgraph/base/triples_query_service.py index ef90081c..37acc622 100755 --- a/trustgraph-base/trustgraph/base/triples_query_service.py +++ b/trustgraph-base/trustgraph/base/triples_query_service.py @@ -19,7 +19,7 @@ class TriplesQueryService(FlowProcessor): id = params.get("id") - super(TriplesStoreService, self).__init__(**params | { "id": id }) + super(TriplesQueryService, self).__init__(**params | { "id": id }) self.register_specification( ConsumerSpec( @@ -47,7 +47,7 @@ class TriplesQueryService(FlowProcessor): print(f"Handling input {id}...", flush=True) - triples = self.query_triples(request) + triples = await self.query_triples(request) print("Send response...", flush=True) r = TriplesQueryResponse(triples=triples, error=None) diff --git a/trustgraph-base/trustgraph/knowledge/defs.py b/trustgraph-base/trustgraph/knowledge/defs.py index d6290930..c55f5148 100644 --- a/trustgraph-base/trustgraph/knowledge/defs.py +++ b/trustgraph-base/trustgraph/knowledge/defs.py @@ -23,6 +23,10 @@ URL = 'https://schema.org/url' IDENTIFIER = 'https://schema.org/identifier' KEYWORD = 'https://schema.org/keywords' +class Triple: + def __init__(self, s, p, o): + self.s, self.p, self.o = s, p, o + class Uri(str): def is_uri(self): return True def is_literal(self): return False diff --git a/trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py b/trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py index d6ab67d2..32da00e5 100755 --- a/trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py +++ b/trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py @@ -44,21 +44,14 @@ class Processor(GraphEmbeddingsQueryService): try: - v = msg.value() - - # Sender-produced ID - id = msg.properties()["id"] - - print(f"Handling input {id}...", flush=True) - entity_set = set() entities = [] - for vec in v.vectors: + for vec in msg.vectors: dim = len(vec) collection = ( - "t_" + v.user + "_" + v.collection + "_" + + "t_" + msg.user + "_" + msg.collection + "_" + str(dim) ) @@ -67,7 +60,7 @@ class Processor(GraphEmbeddingsQueryService): search_result = self.qdrant.query_points( collection_name=collection, query=vec, - limit=v.limit * 2, + limit=msg.limit * 2, with_payload=True, ).points @@ -80,10 +73,10 @@ class Processor(GraphEmbeddingsQueryService): entities.append(ent) # Keep adding entities until limit - if len(entity_set) >= v.limit: break + if len(entity_set) >= msg.limit: break # Keep adding entities until limit - if len(entity_set) >= v.limit: break + if len(entity_set) >= msg.limit: break ents2 = [] diff --git a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py index 48818dad..6fcf4a19 100755 --- a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py +++ b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py @@ -59,11 +59,6 @@ class Processor(TriplesQueryService): ) self.table = table - # Sender-produced ID - id = msg.properties()["id"] - - print(f"Handling input {id}...", flush=True) - triples = [] if query.s is not None: @@ -138,8 +133,6 @@ class Processor(TriplesQueryService): return triples - print("Done.", flush=True) - except Exception as e: print(f"Exception: {e}") diff --git a/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py b/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py index c5e60760..28d32b6d 100644 --- a/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py +++ b/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py @@ -39,6 +39,8 @@ class Query: if self.verbose: print("Get entities...", flush=True) + print("HERE>", query, flush=True) + entities = await self.rag.graph_embeddings_client.query( vectors=vectors, limit=self.entity_limit, user=self.user, collection=self.collection, @@ -47,7 +49,7 @@ class Query: print("ENT>", entities, flush=True) entities = [ - e.value + str(e) for e in entities ] @@ -72,7 +74,7 @@ class Query: self.rag.label_cache[e] = e return e - self.rag.label_cache[e] = res[0].o.value + self.rag.label_cache[e] = str(res[0].o) return self.rag.label_cache[e] async def follow_edges(self, ent, subgraph, path_length): @@ -93,10 +95,10 @@ class Query: for triple in res: subgraph.add( - (triple.s.value, triple.p.value, triple.o.value) + (str(triple.s), str(triple.p), str(triple.o)) ) if path_length > 1: - await self.follow_edges(triple.o.value, subgraph, path_length-1) + await self.follow_edges(str(triple.o), subgraph, path_length-1) res = await self.rag.triples_client.request( user=self.user, collection=self.collection, @@ -106,7 +108,7 @@ class Query: for triple in res: subgraph.add( - (triple.s.value, triple.p.value, triple.o.value) + (str(triple.s), str(triple.p), str(triple.o)) ) res = await self.rag.triples_client.request(