Fixing query stuff

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)
+    if x.is_uri: return Uri(x.value)
-    return Literal(x.v)
+    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)
 

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)

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)
+    if x.is_uri: return Uri(x.value)
-    return Literal(x.v)
+    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
         ]
 

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)

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

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 = []
 

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}")

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(