Repository: topoteretes/cognee
File: cognee/modules/graph/utils/resolve_edges_to_text.py (Python)

Code before:
    first_words = text.split()[:first_n_words]
    top_words = _get_top_n_frequent_words(text, top_n=top_n_words)
    return f"{' '.join(first_words)}... [{top_words}]"


def _extract_nodes_from_edges(retrieved_edges: List[Edge]) -> dict:
    """Creates a dictionary of nodes with their names and content."""

    logger.debug(
        "Extracting nodes from retrieved edges",
        extra={"edge_count": len(retrieved_edges)},
    )

    nodes = {}

    for edge in retrieved_edges:
        for node in (edge.node1, edge.node2):
            if node.id in nodes:
                continue

            text = node.attributes.get("text")
            if text:
                name = _create_title_from_text(text)
                content = text
            else:
                logger.warning(
                    "Node text missing, using fallback attributes",
                    extra={"node_id": str(node.id)},
                )
                name = node.attributes.get("name", "Unnamed Node")
                content = node.attributes.get("description", name)

            nodes[node.id] = {"node": node, "name": name, "content": content}

    return nodes


async def resolve_edges_to_text(retrieved_edges: List[Edge]) -> str:
    """Converts retrieved graph edges into a human-readable string format."""
    nodes = _extract_nodes_from_edges(retrieved_edges)

    node_section = "\n".join(
        f"Node: {info['name']}\n__node_content_start__\n{info['content']}\n__node_content_end__\n"
        for info in nodes.values()
    )

    connections = []

    logger.debug(
        "Resolving edges to text",
        extra={"edge_count": len(retrieved_edges)},

Review comment (bug):
Changing this to debug level, but should probably be removed entirely as it's not a fallback mechanism so no need to log this

Code after:
    first_words = text.split()[:first_n_words]
    top_words = _get_top_n_frequent_words(text, top_n=top_n_words)
    return f"{' '.join(first_words)}... [{top_words}]"


def _extract_nodes_from_edges(retrieved_edges: List[Edge]) -> dict:
    """Creates a dictionary of nodes with their names and content."""

    logger.debug(
        "Extracting nodes from retrieved edges",
        extra={"edge_count": len(retrieved_edges)},
    )

    nodes = {}

    for edge in retrieved_edges:
        for node in (edge.node1, edge.node2):
            if node.id in nodes:
                continue

            text = node.attributes.get("text")
            if text:
                name = _create_title_from_text(text)
                content = text
            else:
                logger.debug(
                    "Node text missing, using fallback attributes",
                    extra={"node_id": str(node.id)},
                )
                name = node.attributes.get("name", "Unnamed Node")
                content = node.attributes.get("description", name)

            nodes[node.id] = {"node": node, "name": name, "content": content}

    return nodes


async def resolve_edges_to_text(retrieved_edges: List[Edge]) -> str:
    """Converts retrieved graph edges into a human-readable string format."""
    nodes = _extract_nodes_from_edges(retrieved_edges)

    node_section = "\n".join(
        f"Node: {info['name']}\n__node_content_start__\n{info['content']}\n__node_content_end__\n"
        for info in nodes.values()
    )

    connections = []

    logger.debug(
        "Resolving edges to text",
        extra={"edge_count": len(retrieved_edges)},