Add Node

Contents

Add Node#

Node Type#

EvoFabric’s graph engine provides four types of nodes, each supporting synchronous, asynchronous, and their corresponding streaming processing modes. Different nodes define processing logic by inheriting the corresponding base classes and implementing the __call__ method, nodes requiring streaming output can utilize StreamWriter to achieve streaming output.

Note

EvoFabric supports directly adding ordinary Python functions as nodes. The system automatically identifies its type based on the function signature and encapsulates it as a suitable node instance. For details, see: callable_to_node()

Node Type#

Node Type

Derived class

Method Definition

Processing Logic

Streaming output

synchronization node

SyncNode

def __call__(state: State) -> StateDelta

Implement the synchronous processing logic for nodes in the method

Not supported

Asynchronous Node

AsyncNode

async def __call__(state: State) -> StateDelta

Implement the node’s asynchronous processing logic in the method

Not supported

Synchronous Streaming Node

SyncStreamNode

def __call__(state: State, stream_writer: StreamWriter) -> StateDelta

Implement the synchronous processing logic for nodes in the method and output streaming messages through put()

support

Asynchronous Streaming Node

AsyncStreamNode

async def __call__(state: State, stream_writer: StreamWriter) -> StateDelta

Implement the asynchronous processing logic for nodes in the method and output streaming messages through put()

support

Add Node#

Add nodes to the graph using the add_node() method, specifying the node name, behavior mode, and the merge strategy for multiple inputs (optional).

Each node represents an independent computing unit or logical step and can be a synchronous, asynchronous, or streaming node.

from evofabric.core.graph import GraphBuilder, AsyncNode
from evofabric.core.typing import State, StateDelta

class AnalyzeNode(AsyncNode):
    async def __call__(self, state: State) -> StateDelta:
        # add your code here
        return {"result": f"analyzed: {state.text}"}

graph = GraphBuilder(state_schema=State)
graph.add_node(
    name="analyze",
    node=AnalyzeNode(),
    action_mode="any",
    multi_input_merge_strategy={"default": lambda states: states[0]}
)

Parameter Description:

  • name: Node name, must be unique. Cannot use system reserved names start and end.

  • node: node instance, which can be one of four node types (synchronous, asynchronous, synchronous streaming, asynchronous streaming), or an ordinary Python function (the system automatically wraps it into the corresponding node type).

  • action_mode: The node’s trigger mode, controlling when to execute after the predecessor node completes.

    • “any”: Triggers when any predecessor node completes execution.

    • “all”: Triggered after all predecessor nodes in the same group have completed execution.

  • multi_input_merge_strategy: Multi-predecessor state merging strategy. When a node has inputs from multiple groups, a merge function can be specified for different groups. This parameter is a dictionary where the key is the edge’s group name, and the value is a function of type Callable[[List[State]], State].

    If this parameter is not specified, the system uses the default state update mechanism to sequentially merge the input states.

    See also

    See Add Edge for the definition and usage of groups

    Usage Example:

    # When node has multiple predecessor
graph.add_node(
    name="merge_result",
    node=lambda s: {"final": s["a"] + s["b"]},
    action_mode="all",
    multi_input_merge_strategy={
        "group_a": lambda states: states[0],
        "group_b": lambda states: states[-1]
    }
)

Operating Mechanism:

Each node receives a complete State input, executes the computation logic, and returns a dictionary-type state delta (StateDelta). The engine automatically merges the delta into the global state and passes it down along the defined edge structure.

Contents