|
from langchain_community.graphs.graph_document import ( |
|
Node as BaseNode, |
|
Relationship as BaseRelationship, |
|
) |
|
from models import Node, Relationship |
|
|
|
def format_property_key(s: str) -> str: |
|
words = s.split() |
|
if not words: |
|
return s |
|
first_word = words[0].lower() |
|
capitalized_words = [word.capitalize() for word in words[1:]] |
|
return "".join([first_word] + capitalized_words) |
|
|
|
def props_to_dict(props) -> dict: |
|
"""Convert properties to a dictionary.""" |
|
properties = {} |
|
if not props: |
|
return properties |
|
for p in props: |
|
properties[format_property_key(p.key)] = p.value |
|
return properties |
|
|
|
def map_to_base_node(node: Node) -> BaseNode: |
|
"""Map the KnowledgeGraph Node to the base Node.""" |
|
properties = props_to_dict(node.properties) if node.properties else {} |
|
properties["name"] = node.id.title() |
|
return BaseNode( |
|
id=node.id.title(), type=node.type.capitalize(), properties=properties |
|
) |
|
|
|
def map_to_base_relationship(rel: Relationship) -> BaseRelationship: |
|
"""Map the KnowledgeGraph Relationship to the base Relationship.""" |
|
source = map_to_base_node(rel.source) |
|
target = map_to_base_node(rel.target) |
|
properties = props_to_dict(rel.properties) if rel.properties else {} |
|
return BaseRelationship( |
|
source=source, target=target, type=rel.type, properties=properties |
|
) |
|
|