Node, Edge, and Graph¶

Overview¶

This documentation covers the graph structures in LionAGI, comprising individual Node objects, Edge objects that link them, and the overarching Graph class that manages these relationships. Also included is an EdgeCondition mechanism for conditional traversal.

Contents¶

EdgeCondition¶

class lionagi.protocols.graph.edge.EdgeCondition¶

Inherits from: pydantic.BaseModel, Condition

A condition that can be attached to an Edge to control or constrain its traversability. This class also integrates with Pydantic for data validation and serialization.

Attributes¶

sourceAny: The source object or data used for evaluating the condition.

Example¶

from lionagi.protocols.graph.edge import EdgeCondition

class CustomCondition(EdgeCondition):
    async def apply(self, *args, **kwargs) -> bool:
        # implement custom logic here
        return True

Edge¶

class lionagi.protocols.graph.edge.Edge(Element)¶

Inherits from: Element

Represents a directed connection from a head node to a tail node in a LionAGI graph. An optional condition determines if traversal is allowed. Additional metadata (like label) or properties can be stored in properties.

Attributes¶

headIDType: The ID of the head node.
tailIDType: The ID of the tail node.
propertiesdict[str, Any]: A dictionary holding additional properties, such as labels or an EdgeCondition.

Properties¶

label: list[str] | None¶: Read or set a list of labels describing this edge.

condition: EdgeCondition | None¶: A condition that determines if this edge is traversable. If None, traversal is unrestricted.

Methods¶

async check_condition(*args, **kwargs) → bool¶: Evaluates the condition (if any). Returns True if no condition is assigned or if the condition passes.

update_property(key: str, value: Any) → None¶: Adds or updates a custom property in properties.

update_condition_source(source: Any) → None¶: Updates the source field in the assigned condition without replacing the entire condition object.

Example¶

from lionagi.protocols.graph.edge import Edge, EdgeCondition

edge_condition = EdgeCondition(source="some_source")
edge = Edge(head=my_head_node, tail=my_tail_node, condition=edge_condition, label=["friendship"])
# edge now connects two nodes with a condition and a descriptive label.

Node¶

class lionagi.protocols.graph.node.Node(Element, Relational)¶

Inherits from: Element, Relational

A graph node that can store arbitrary content, an optional numeric embedding, and metadata in metadata. Nodes integrate with the LionAGI adapter system, enabling easy import/export to JSON, pandas Series, etc.

Attributes¶

contentAny: Arbitrary payload or data associated with this node.
embeddinglist[float] | None: An optional vector representation (e.g., for search or similarity).

Methods¶

classmethod adapt_to(obj_key: str, /, many: bool = False, **kwargs) → Any¶: Converts this node to a specified format using a registered adapter.

classmethod adapt_from(obj: Any, obj_key: str, /, many: bool = False, **kwargs) → Node¶: Constructs a node from an external representation, possibly returning a subclass if lion_class data is present.

classmethod list_adapters() → list[str]¶: Lists all available adapter keys for node conversions.

Example¶

from lionagi.protocols.graph.node import Node

class PersonNode(Node):
    pass

person = PersonNode(content={"name": "Alice"}, embedding=[0.1, 0.2, 0.3])
print(person.content)   # => {'name': 'Alice'}
print(person.embedding) # => [0.1, 0.2, 0.3]

Graph¶

class lionagi.protocols.graph.graph.Graph(Element, Relational)¶

Inherits from: Element, Relational

Represents an entire directed graph of Node and Edge objects. Internally, it uses two Pile instances (one for nodes, one for edges) and a node_edge_mapping for quick lookups.

Attributes¶

internal_nodesPile[Node]: Stores all nodes in the graph.
internal_edgesPile[Edge]: Stores all edges in the graph.
node_edge_mappingdict: Maps each node ID to its incoming and outgoing edges for quick access.

Methods¶

add_node(node: Relational) → None¶: Inserts a node into the graph. Raises RelationError if invalid or already present.

add_edge(edge: Edge) → None¶: Inserts an edge into the graph, linking two existing nodes. Raises RelationError if invalid or if referenced nodes are not found.

remove_node(node: ID.Ref) → None¶: Removes a node and all associated edges from the graph.

remove_edge(edge: Edge | str) → None¶: Removes a specific edge by object or ID.

find_node_edge(node: Any, direction: Literal['both', 'in', 'out'] = 'both') → list[Edge]¶: Finds edges connected to the given node in a specified direction.

get_heads() → Pile[Node]¶: Returns all nodes with no incoming edges (i.e., “head” nodes of subgraphs).

get_predecessors(node: Node) → Pile[Node]¶: Retrieves nodes that point to the given node.

get_successors(node: Node) → Pile[Node]¶: Retrieves nodes that are pointed to by the given node.

to_networkx(**kwargs) → Any¶: Converts this graph to a networkx.DiGraph object, including node and edge properties.

display(node_label='lion_class', edge_label='label', draw_kwargs={}, **kwargs) → None¶: Visualizes the graph using networkx and matplotlib (if installed).

is_acyclic() → bool¶: Checks whether the graph has any cycles. Returns True if acyclic.

Dunder Methods¶

__contains__(item: object) → bool¶: Checks if a given node or edge is in the graph.

Example¶

from lionagi.protocols.graph.node import Node
from lionagi.protocols.graph.edge import Edge
from lionagi.protocols.graph.graph import Graph

g = Graph()
node_a, node_b = Node(), Node()
g.add_node(node_a)
g.add_node(node_b)

e = Edge(head=node_a, tail=node_b, label=["relation"])
g.add_edge(e)

print(g.get_heads())  # node_a is a "head" if no inbound edges
print(g.is_acyclic()) # likely True unless you create a cycle

File Locations¶

EdgeCondition and Edge: lionagi/protocols/graph/edge.py
Node: lionagi/protocols/graph/node.py
Graph: lionagi/protocols/graph/graph.py