Processors

A processor transforms all matching entities in one Daft DataFrame operation. Declare the component types it needs, then return a new DataFrame from process(). Archetype runs the processor once per matching archetype.

from daft import DataFrame, col

from archetype import AsyncProcessor


class Move(AsyncProcessor):
    components = (Position, Velocity)
    priority = 10

    async def process(self, df: DataFrame, **_) -> DataFrame:
        return df.with_columns(
            {
                "position__x": col("position__x") + col("velocity__dx"),
                "position__y": col("position__y") + col("velocity__dy"),
            }
        )

Move receives only archetypes whose entities have both Position and Velocity. It does not loop over Python objects; the DataFrame expression updates the matching population together.

Run order

Lower priority values run first. Use priorities when one processor consumes the rows produced by another.

class Integrate(AsyncProcessor):
    components = (Position, Velocity)
    priority = 10


class ResolveCollisions(AsyncProcessor):
    components = (Position, Collider)
    priority = 20

Keep one processor responsible for one transformation. It makes order and tests straightforward.

Add processors to a world

Pass processors when you create the handle for the usual script path:

world = runtime.world("demo", processors=[Move(), ResolveCollisions()])

You can change a live world through its gated methods:

await world.add_processor(Move())
await world.remove_processor(Move)

remove_processor() takes the processor type, not an instance.

Use shared resources

Resources hold shared configuration or services that do not belong to one entity. Processors receive them as keyword arguments when they declare them.

from dataclasses import dataclass


@dataclass
class Rules:
    max_speed: float = 5.0


class LimitSpeed(AsyncProcessor):
    components = (Velocity,)

    async def process(self, df: DataFrame, rules: Rules, **_) -> DataFrame:
        return df.with_columns(
            {"velocity__dx": col("velocity__dx").clip(-rules.max_speed, rules.max_speed)}
        )


world = runtime.world("demo", processors=[LimitSpeed()], resources=[Rules()])

See Resources for lifecycle and fork behavior.

Call an LLM for each row

Use Daft's prompt() function when a processor needs an LLM call. Daft runs the row work in parallel and returns a column you can persist like any other state.

from daft.functions import prompt


class Think(AsyncProcessor):
    components = (Agent,)

    async def process(self, df: DataFrame, **_) -> DataFrame:
        return df.with_column(
            "agent__last_thought",
            prompt(
                "You are " + col("agent__name") + ". What should you do next?",
                model="gpt-5-mini",
            ),
        )

The component field is part of your history, so keep prompts and outputs small enough for the storage and cost profile you want.

Test a processor

Test the transformation with a representative DataFrame, then use a small world-level test to verify component matching and priority. Do not mutate a DataFrame in place; always return the DataFrame that should become the next state.

For engine-level details, see system execution.