Chunk API Reference¶

Complete API documentation for the Chunk dataclass.

Class Definition¶

@dataclass(frozen=True)
class Chunk(Generic[C]):
    """
    Wrapper for intermediate results from async generator nodes.

    A frozen (immutable) dataclass that wraps yielded values with metadata.

    Type Parameters:
        C: The type of the wrapped output value

    Attributes:
        uuid: UUID of the source node that yielded this chunk
        output: The yielded value
    """

Attributes¶

uuid¶

uuid: str

The UUID of the node that produced this chunk.

Example:

async for item in executor.yielding():
    if isinstance(item, Chunk):
        print(f"Chunk from node: {item.uuid}")

output¶

output: C

The yielded value from the node's async generator.

Example:

async for item in executor.yielding():
    if isinstance(item, Chunk):
        value = item.output
        print(f"Value: {value}")

Usage¶

Creating Chunks¶

Chunks are created automatically by Grafo when nodes yield values. You typically don't create them manually:

async def yielding_task():
    for i in range(5):
        yield f"item_{i}"  # Automatically wrapped in Chunk

node = Node(coroutine=yielding_task, uuid="producer")

Receiving Chunks¶

Chunks are received when using TreeExecutor.yielding():

executor = TreeExecutor(roots=[node])

async for item in executor.yielding():
    if isinstance(item, Chunk):
        # item is a Chunk object
        source = item.uuid
        value = item.output

Immutability¶

Immutability

Chunk is a frozen dataclass and cannot be modified.

async for item in executor.yielding():
    if isinstance(item, Chunk):
        try:
            item.output = "new value" # This will raise an error!
        except AttributeError:
            print("Chunks are immutable")