apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-22 11:41:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Store maybe working

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-19 14:42:00 +01:00
parent 6aaa751fb5
commit abc202ddde
2 changed files with 9 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trustgraph-base/trustgraph/base/__init__.py
View file

@ -18,4 +18,6 @@ from . embeddings_client import EmbeddingsClientSpec
from . text_completion_client import TextCompletionClientSpec from . text_completion_client import TextCompletionClientSpec
from . prompt_client import PromptClientSpec from . prompt_client import PromptClientSpec
from . triples_store_service import TriplesStoreService from . triples_store_service import TriplesStoreService
from . graph_embeddings_store_service import GraphEmbeddingsStoreService
from . document_embeddings_store_service import DocumentEmbeddingsStoreService

28
trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py
View file

@ -8,31 +8,21 @@ from qdrant_client.models import PointStruct
from qdrant_client.models import Distance, VectorParams from qdrant_client.models import Distance, VectorParams
import uuid import uuid
from .... schema import GraphEmbeddings from .... base import GraphEmbeddingsStoreService
from .... schema import graph_embeddings_store_queue
from .... log_level import LogLevel
from .... base import Consumer
module = "ge-write" default_ident = "ge-write"
default_input_queue = graph_embeddings_store_queue
default_subscriber = module
default_store_uri = 'http://localhost:6333' default_store_uri = 'http://localhost:6333'
class Processor(Consumer): class Processor(GraphEmbeddingsStoreService):
def __init__(self, **params): def __init__(self, **params):
input_queue = params.get("input_queue", default_input_queue)
subscriber = params.get("subscriber", default_subscriber)
store_uri = params.get("store_uri", default_store_uri) store_uri = params.get("store_uri", default_store_uri)
api_key = params.get("api_key", None) api_key = params.get("api_key", None)
super(Processor, self).__init__( super(Processor, self).__init__(
**params | { **params | {
"input_queue": input_queue,
"subscriber": subscriber,
"input_schema": GraphEmbeddings,
"store_uri": store_uri, "store_uri": store_uri,
"api_key": api_key, "api_key": api_key,
} }
@ -67,11 +57,9 @@ class Processor(Consumer):
return cname return cname
async def handle(self, msg): async def store_graph_embeddings(self, message):
v = msg.value() for entity in message.entities:
for entity in v.entities:
if entity.entity.value == "" or entity.entity.value is None: return if entity.entity.value == "" or entity.entity.value is None: return
@ -99,9 +87,7 @@ class Processor(Consumer):
@staticmethod @staticmethod
def add_args(parser): def add_args(parser):
Consumer.add_args( GraphEmbeddingsStoreService.add_args(parser)
parser, default_input_queue, default_subscriber,
)
parser.add_argument( parser.add_argument(
'-t', '--store-uri', '-t', '--store-uri',
@ -117,5 +103,5 @@ class Processor(Consumer):
def run(): def run():
Processor.launch(module, __doc__) Processor.launch(default_ident, __doc__)