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

- Output doc metadata with other edges.

Browse source 
- Batch extracted edges into a single store write payload
This commit is contained in:
Cyber MacGeddon 2024-10-23 16:21:46 +01:00
parent 2575390d5a
commit 1c38963cf9
2 changed files with 61 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

25
trustgraph-flow/trustgraph/extract/kg/definitions/extract.py
View file

@ -69,11 +69,11 @@ class Processor(ConsumerProducer):
return self.prompt.request_definitions(chunk)
def emit_edge(self, metadata, s, p, o):
def emit_edges(self, metadata, triples):
t = Triples(
metadata=metadata,
triples=[Triple(s=s, p=p, o=o)],
triples=triples,
)
self.producer.send(t)
@ -88,6 +88,13 @@ class Processor(ConsumerProducer):
defs = self.get_definitions(chunk)
triples = []
# FIXME: Putting metadata into triples store is duplicated in
# relationships extractor too
for t in v.metadata.metadata:
triples.append(t)
for defn in defs:
s = defn.name
@ -104,7 +111,19 @@ class Processor(ConsumerProducer):
s_value = Value(value=str(s_uri), is_uri=True)
o_value = Value(value=str(o), is_uri=False)
self.emit_edge(v.metadata, s_value, DEFINITION_VALUE, o_value)
triples.append(Triple(
s=s_value, p=DEFINITION_VALUE, o=o_value
))
self.emit_edges(
Metadata(
id=v.metadata.id,
metadata=[],
user=v.metadata.user,
collection=v.metadata.collection,
),
triples
)
except Exception as e:
print("Exception: ", e, flush=True)

65
trustgraph-flow/trustgraph/extract/kg/relationships/extract.py
View file

@ -92,11 +92,11 @@ class Processor(ConsumerProducer):
return self.prompt.request_relationships(chunk)
def emit_edge(self, metadata, s, p, o):
def emit_edges(self, metadata, triples):
t = Triples(
metadata=metadata,
triples=[Triple(s=s, p=p, o=o)],
triples=triples,
)
self.producer.send(t)
@ -116,6 +116,13 @@ class Processor(ConsumerProducer):
rels = self.get_relationships(chunk)
triples = []
# FIXME: Putting metadata into triples store is duplicated in
# relationships extractor too
for t in v.metadata.metadata:
triples.append(t)
for rel in rels:
s = rel.s
@ -142,43 +149,49 @@ class Processor(ConsumerProducer):
else:
o_value = Value(value=str(o), is_uri=False)
self.emit_edge(
v.metadata,
s_value,
p_value,
o_value
)
triples.append(Triple(
s=s_value,
p=p_value,
o=o_value
))
# Label for s
self.emit_edge(
v.metadata,
s_value,
RDF_LABEL_VALUE,
Value(value=str(s), is_uri=False)
)
triples.append(Triple(
s=s_value,
p=RDF_LABEL_VALUE,
o=Value(value=str(s), is_uri=False)
))
# Label for p
self.emit_edge(
v.metadata,
p_value,
RDF_LABEL_VALUE,
Value(value=str(p), is_uri=False)
)
triples.append(Triple(
s=p_value,
p=RDF_LABEL_VALUE,
o=Value(value=str(p), is_uri=False)
))
if rel.o_entity:
# Label for o
self.emit_edge(
v.metadata,
o_value,
RDF_LABEL_VALUE,
Value(value=str(o), is_uri=False)
)
triples.append(Triple(
s=o_value,
p=RDF_LABEL_VALUE,
o=Value(value=str(o), is_uri=False)
))
self.emit_vec(v.metadata, s_value, v.vectors)
self.emit_vec(v.metadata, p_value, v.vectors)
if rel.o_entity:
self.emit_vec(v.metadata, o_value, v.vectors)
self.emit_edges(
Metadata(
id=v.metadata.id,
metadata=[],
user=v.metadata.user,
collection=v.metadata.collection,
),
triples
)
except Exception as e:
print("Exception: ", e, flush=True)