Data streams¶

Data streams are the typed channels through which agents exchange data. Every input and output of an Agent is a stream. When two agents connect, UNaIVERSE inspects their StreamType descriptors, finds compatible pairs, and wires them together automatically, with built-in type checking, optional buffering, and either point-to-point delivery or pub/sub broadcast. You write no networking code.

Why typed?

Two peers must agree on what they exchange before they exchange it. A StreamType is that contract, kind, shape, dtype. It's why you can connect a classifier and a generator with confidence that nobody sends the wrong thing.

The data types¶

The shorthand you'll use most. Pass these as plain strings to proc_inputs / proc_outputs, or build a full StreamType.

Value	Meaning
`"tensor"`	A `torch.Tensor`: images, audio, embeddings, signals.
`"img"`	A PIL `Image`.
`"text"`	A plain Python string.
`"file"`	Any file (a PDF, an audio clip, a zip, a CSV), carried as a `FileContainer`.
`"all"`	Wildcard, matches any of the others.

The file type, for whole files

Use "file" to move an actual file between agents, not just text or numbers. A file travels as a small FileContainer (from unaiverse.streams.dataprops import FileContainer) with three fields: content (the raw bytes), filename, and mime_type. When you declare a "file" input, your forward() receives a FileContainer. To send one out, return a path string, raw bytes, or a FileContainer. It is the right type for documents, media, datasets, or checkpoints that are not naturally a tensor, an image, or text.

# The two are equivalent for simple cases:
proc_inputs = ["text"]
proc_inputs = [StreamType(data_type="text")]

Defining I/O with `StreamType`¶

For anything beyond the shorthand, describe the stream precisely:

from unaiverse.streams.dataprops import StreamType

# Image input: batch × channels × height × width (None = dynamic)
image_input = StreamType(data_type="tensor",
                         tensor_shape=(None, 3, 224, 224),
                         tensor_dtype="torch.float32")

text_output = StreamType(data_type="text")            # text out
any_stream  = StreamType(data_type="all")             # wildcard in

UNaIVERSE uses these descriptors to validate incoming data, match compatible streams between peers, and apply transforms automatically.

`StreamType` properties¶

data_type · str · required: "tensor", "img", "text", "file", or "all".
tensor_shape · tuple: Expected shape for tensor streams; None marks dynamic dims, e.g. (None, 3, 224, 224).
tensor_dtype · str: Expected dtype, e.g. "torch.float32" or "torch.long".
tensor_labels · list[str]: Human-readable labels for flat tensor dimensions (handy for tabular data).
pubsub · bool · default: False: Broadcast writes to all subscribers via a topic, instead of point-to-point.
public_only · bool · default: False: Create only the public variant of the stream. By default UNaIVERSE builds both a private (in-world) and a public descriptor for every stream.
private_only · bool · default: False: Create only the private/world variant; never expose the stream publicly.

The \"all\" wildcard

An agent declaring StreamType(data_type="all") as input accepts connections from any stream. Use it for relay nodes, loggers, or any agent that should consume everything it can find.

Point-to-point or broadcast

By default a stream delivers point-to-point: it connects one writer to the readers wired to it, the classic request-and-response case. Set pubsub=True and the stream becomes a broadcast topic instead: one writer, any number of subscribers, and each subscriber receives every sample. Point-to-point fits a one-to-one exchange; a topic fits a shared feed many agents should all see (a sensor everyone reads, a chat room's messages). A reader joins a topic with the subscribe action.

An agent's own streams: `proc_input_N`, `proc_output_N`, and `std*`¶

Read any world example and you'll bump into names like proc_input_0, proc_output_0, and the proxies stdin / stdout. They look cryptic, but they are just fixed names for the slots where an agent's data sits. Once you know the convention you can read every example. This is the section the examples link back to.

The whole idea in one picture

Think of an agent as a little machine with labelled in-trays and out-trays. A florist's "is-it-dry?" agent has one in-tray (the sensor reading) and one out-tray (the decision). A chatbot has one of each (text in, text out). The trays always have the same names, so any action, and any other agent, knows where to look.

flowchart LR
    subgraph GIN[group: processor_in]
        I0[proc_input_0]
        I1[proc_input_1]
    end
    subgraph BRAIN[the brain]
        F["forward()"]
    end
    subgraph GOUT[group: processor]
        O0[proc_output_0]
        O1[proc_output_1]
    end
    I0 --> F
    I1 --> F
    F --> O0
    F --> O1

The everyday way: `get()` and `set()`¶

Inside an action you never touch stream objects by hand. You use four ready-made proxies, named after the Unix standard streams:

Proxy	Reads / writes	You use it to…
`self.stdin`	the agent's inputs	feed the brain its next input
`self.stdout`	the agent's outputs	read or publish what the brain produced
`self.stdtar`	the target slots	hand a model the correct answer while it learns
`self.stdext`	environmental slots	read data coming from outside the processor

Almost every agent has exactly one input and one output, so the two calls you reach for most need no slot name at all:

sample = self.stdin.get()     # read the input
self.stdout.set(result)       # write the output

get() with no argument returns the agent's input; set(value) with just a value writes the agent's output. For the common case, that is the entire API, learn these two before anything else.

The built-in process action is exactly "read stdin, run forward(), write stdout"; learn does the same but also reads stdtar (the correct answer) and runs a backward pass. So a custom action is usually just staging an input with self.stdin.set(...) so that a later process transition runs the model on it (see Chapter 5).

When an agent has more than one slot¶

Some agents read or write several things at once (an image and a question, a prediction and a confidence). Then you point at a specific slot, by position or by name:

image      = self.stdin.get(0)              # by position (zero-based)
self.stdout.set(0, prediction)

answer = self.stdin.get("proc_input_0")     # by name
self.stdout.set("proc_output_0", answer)

self.stdout.set([image_out, label_out])     # several outputs at once, in order

These are real patterns from the example worlds, not toy snippets:

Read by position (social_learning)Read by name or by type (turing)Stage an input by name (chat)

A student re-publishes exactly what process just consumed and produced:

image      = self.stdin.get(0, requested_by="teach")
prediction = self.stdout.get(0, requested_by="teach")

A guest pulls every queued chat message, then reads its text reply by data type instead of by slot position:

msgs  = self.stdin.get("chat", requested_by="get_msgs", all_uuids=True)
reply = self.stdout.get(requested_by="send_msgs", data_type="text")

A user drops an augmented prompt onto the first input slot so a later process step answers it:

self.stdin.set("proc_input_0", augmented_msg)

Where those slot names come from¶

The names proc_input_0, proc_output_0, … are not magic, they come straight from the lists you pass when you build the Agent:

agent = Agent(proc=my_model, proc_inputs=["text"], proc_outputs=["text"])

Each list position becomes a stream with a canonical name:

proc_inputs[0] becomes proc_input_0, proc_inputs[1] becomes proc_input_1, and so on.
proc_outputs[0] becomes proc_output_0, and so on.

So proc_inputs=["text"] gives a single input slot, proc_input_0; a two-input model (image and a question) would have proc_input_0 and proc_input_1. You never write these names when constructing the agent (the framework assigns them), but they are exactly what get("..."), set("..."), and send address. This is also why get() with no key works when there is only one slot: there is nothing to disambiguate.

Groups: many slots, one channel¶

Every input slot is automatically put in the group processor_in and every output slot in the group processor. A group simply carries several streams over a single network channel, so an agent with three inputs still opens one tidy connection, not three. You rarely touch groups directly; just know that "the processor group" means "this agent's outputs" and "the processor-in group" means "its inputs".

Environmental streams: data from outside the brain (`stdext`)¶

Not every stream is produced by a model. A camera, a microphone, a sensor, or a dataset sitting on disk is an environmental source: the data enters the world from the outside, untouched by anyone's forward(). These are environmental streams, and an agent reads them through the self.stdext proxy, exactly the way it reads self.stdin.

The most common publisher of environmental streams is a World itself. A world can own streams and offer them to every member with self.add_streams, usually built from a stream generator such as ImageFileStream or LabelStream. The animal_school world does this to be the environment its students learn from:

animal_school/src/world.py (trimmed)

self.add_streams([
    DataStream.create(group="albatross", stream=ImageFileStream(image_dir=data_path, ...)),
    DataStream.create(group="albatross", stream=LabelStream(label_dir=data_path, ...)),
])
# ...same for "cheetah", "giraffe", and a mixed "all" set...

Two things to read here:

Each add_streams call publishes a picture stream and its label stream together, bundled by the same group ("albatross"). Grouping is what keeps an image and its label travelling as one unit (the group idea from above).
The group "all" is just a name the world chose for the mixed set of every class, so do not confuse it with the data_type="all" wildcard from the type table; one is a stream group, the other is a type.

Because the world owns these streams, members address them by reference as <world>:albatross, <world>:all, and so on (owner:stream-name, where <world> is a wildcard the framework fills in with the world host). That is exactly what the animal_school teacher snapshots when it records its curriculum, and the full walk-through lives in Chapter 8. The takeaway: proc_* streams are a brain's own I/O; environmental streams are the raw feed of the world around it, and stdext is how you read them.

The same names travel between agents¶

When one agent sends data to another, it names the destination slot with the same convention. From the chat broadcaster relaying a message to everyone:

await self.send(target=other_users,
                data_samples={"proc_output_0": f"**{sender}:** {msg}"},
                num_steps=1)

The key "proc_output_0" says "deliver this payload onto the first output slot", so on every recipient it lands exactly where their check_messages action is already looking. The slot names are the shared vocabulary that lets a sensor, a chatbot, and a human all exchange data without anyone wiring streams by hand.

flowchart LR
    subgraph S[Sender agent]
        SO[proc_output_0<br/>the produced message]
    end
    subgraph R[Recipient agent]
        RI[proc_output_0<br/>where check_messages looks]
    end
    SO -->|"send(data_samples={'proc_output_0': msg})"| RI

Because both sides use the same slot name, the sender never needs to know how the recipient is built, it just addresses a named tray. The full mechanics of send and data_samples are in Chapter 6.

Transforms¶

A StreamType can pre-/post-process data as it moves between the network and your processor, so forward() always sees clean inputs.

Before forward()After forward()

import torchvision.transforms as T

image_input = StreamType(
    data_type="tensor", tensor_shape=(None, 3, 224, 224),
    stream_to_proc_transforms=T.Compose([
        T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
    ]),
)

prob_output = StreamType(
    data_type="tensor", tensor_shape=(None, 1000),
    proc_to_stream_transforms=lambda x: x.softmax(dim=-1),
)

stream_to_proc_transforms runs on incoming data before forward(); proc_to_stream_transforms runs on the output before it hits the stream. Both accept any callable, including torchvision.transforms.Compose.

Stream classes¶

Beyond single values, UNaIVERSE provides stream classes for common patterns:

Class	Description
`DataStream`	Base single-value stream, holds the most recent sample.
`BufferedDataStream`	Keeps a history of samples with indexed access and replay.
`ImageFileStream`	Streams images from a disk directory in sequence.
`LabelStream`	Streams classification labels (ints or one-hot tensors).
`TokensStream`	Streams tokenized text sequences.

Built-in generators¶

unaiverse.streams.streamlib ships ready-made BufferedStream subclasses for synthetic signals, great for testing pipelines or building signal-school worlds:

from unaiverse.streams.streamlib import Sin, Random, CombSin

noise     = Random(std=1.0, shape=(1,))                  # uniform noise in [0, std)
sine_wave = Sin(freq=0.1, phase=0.0, delta=0.1)          # a sine wave
comb      = CombSin(f_cap=[0.1, 0.5, 1.0],               # explicit frequencies
                    c_cap=[1.0, 1.0, 1.0],               # one coefficient each
                    order=3, delta=0.1)                  # summed sines

Where next¶

Agents, proc_inputs / proc_outputs in context.
Interactions, how streams move each tick.
DataProps API reference.