Fast path — beam-agents

The smallest pipeline that is still a real one. Three events arrive on two keys; each activation reads durable working memory, calls a model, writes memory back, and completes. Nothing suspends, nothing leaves the pipeline, and the whole run finishes offline on the DirectRunner.

Read this one first. Every other example on this site is this shape plus one complication.

What the program does

Four steps, in order:

Append the event to a bounded per-key ring — ctx.memory.append.
Ask the model what to do — ctx.call_model.
End the activation and emit bytes — Complete(output=...).
Assert the two keys never saw each other's memory — assert_that, in main.

If you are meeting the runtime for the first time, getting started walks the same file as a tutorial. This page is the annotated read of it.

The agent

An agent is an async function from an activation context to an outcome. There is no base class to inherit and no registration call.

async def triage(ctx: ActivationContext) -> Complete:
    """Decide what to do about one event for one entity.

    Module-level, not a closure: the DoFn holding the agent is serialized for
    the runner, so the agent has to pickle by reference.
    """
    ctx.memory.append("recent", ctx.event, max_items=32)
    seen = len(ctx.memory.ring("recent"))

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[f"event={ctx.event.decode()} seen={seen}"],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=b"%s:%d" % (response.response, seen))

website/examples/fast_path.py (region: agent) — executed by the repository’s offline test tier.

Three things in those few lines are load-bearing.

It is module-level, not a closure. RunAgent holds the function inside a stateful DoFn, and Beam serializes that DoFn to ship it to workers. A module-level function pickles by reference; a closure over local state does not pickle at all. This is the first thing that bites people, and it fails at submission time rather than in the agent.

Memory is read and written through ctx, not through a client. ctx.memory is a facade over one keyed protobuf blob. append(..., max_items=32) keeps a bounded ring — bounded because working memory carries a per-key cap, and an unbounded list eventually reaches it. State and memory has the caps and the collection rules.

The write is staged, not applied. Had this function raised on the line after the append, the append would not persist and the key's sequence counter would not advance. Every effect an activation produces is buffered in the context and committed atomically with the Beam bundle — correctness invariant 1.

Complete(output=...) ends the activation and emits on .output. The other outcome is Suspend, which is what intents and resume is about.

Supplying the model

AgentConfig takes a factory, not a client instance.

def make_provider() -> FakeLLM:
    """The model used throughout these examples.

    `FakeLLM` matches requests against ordered rules and records every call, so
    an example is deterministic and needs no credentials or network. Swapping
    in a real provider is a change to this factory alone.
    """
    return FakeLLM([(match_any(), respond_with(b"escalate"))])

website/examples/fast_path.py (region: provider) — executed by the repository’s offline test tier.

The factory runs worker-side, so connection pools are constructed where they are used rather than pickled across the wire. FakeLLM matches requests against ordered rules and records every call it receives, which is what lets an example be deterministic with no credentials and no network — see the fake LLM spec.

Swapping in a real provider is a change to this one function. Nothing in triage names a vendor.

Keying the input

events = pipeline | "Events" >> beam.Create(
    [
        AgentEnvelope(entity_key=b"acct-1", event_time_ms=1_000, external_event=b"login"),
        AgentEnvelope(entity_key=b"acct-1", event_time_ms=2_000, external_event=b"transfer"),
        AgentEnvelope(entity_key=b"acct-2", event_time_ms=1_500, external_event=b"login"),
    ]
)

# `RunAgent` does not key elements itself. It validates the input is
# KV-shaped at pipeline-construction time and raises ValueError otherwise,
# because a stateful DoFn cannot accept anything else.
keyed = events | "Key" >> beam.WithKeys(lambda e: e.entity_key).with_output_types(
    tuple[bytes, AgentEnvelope]
)

outputs = keyed | "Agent" >> RunAgent(
    triage, config=AgentConfig(provider_factory=make_provider)
)

website/examples/fast_path.py (region: pipeline) — executed by the repository’s offline test tier.

RunAgent does not key elements for you, and it does not guess. It checks at pipeline-construction time that its input is PCollection[KV[bytes, AgentEnvelope]] and raises ValueError otherwise. A stateful DoFn cannot accept anything else, and failing during submission beats failing on a worker part-way through a backfill.

The key you choose is the unit of isolation: one memory blob, one sequence counter, one serialized stream of activations per key. Pick the entity the agent reasons about — an account, a session, an incident — not a request id.

What the assertion proves

# Per-key serialization means acct-1's two events are ordered relative
# to each other; the ring is 1 then 2 deep. acct-2 is a separate key
# with its own memory, so it starts at 1.
assert_that(results, equal_to([b"escalate:1", b"escalate:2", b"escalate:1"]))

website/examples/fast_path.py (region: assertion) — executed by the repository’s offline test tier.

acct-1 sends two events and sees its ring one deep and then two deep. acct-2 sends one event and starts from an empty ring. Two facts fall out of that:

Memory is per key. acct-2 cannot read acct-1's ring, and there is no configuration that lets it.
Per-key ordering is not a race. A Beam stateful DoFn processes one element at a time per key, so acct-1's second activation observes the first one's committed write. That is correctness invariant 4, and it is why no lock appears anywhere in the agent. Parallelism comes from having many keys, not from interleaving one.

Run it

uv run python website/examples/fast_path.py

No credentials, no docker, no network. The repository runs this exact command from tests/docs/test_website_examples.py on every change, so the code above cannot quietly stop working — see testing and CI.

What the fast path leaves out

Everything an activation cannot do inside a single process() call:

External writes. An agent never performs one itself; it stages an intent and suspends. → intents and resume
Asking a person. The same mechanism, plus a timer that decides what an unanswered question means. → approvals and timeout fallback
Failure. This program has no failing path, so .errors stays empty and unexamined. → the four outputs
Tools. triage calls the model directly and looks nothing up. → read-only tools and side effects

The complete program

"""Fast path: one event in, one decision out, inside a single activation.

The fast path is the simple case — the agent runs to completion inside one
`process()` call. It reads the event, consults durable per-key working memory,
calls the model, writes memory back, and completes. No suspension, no external
side effect, no second element.

Everything the activation touches is staged and committed atomically with the
Beam bundle: if this function raised halfway through, the memory write below
would not persist and the key's sequence counter would not advance.

Run it:  python website/examples/fast_path.py
"""

from __future__ import annotations

import apache_beam as beam
from apache_beam.testing.util import assert_that, equal_to

from beam_agents import AgentConfig, RunAgent
from beam_agents._protos import AgentEnvelope
from beam_agents.core.agent import Complete
from beam_agents.core.context import ActivationContext
from beam_agents.model.client import LlmRequest
from beam_agents.model.fake import FakeLLM, match_any, respond_with


# region: provider
def make_provider() -> FakeLLM:
    """The model used throughout these examples.

    `FakeLLM` matches requests against ordered rules and records every call, so
    an example is deterministic and needs no credentials or network. Swapping
    in a real provider is a change to this factory alone.
    """
    return FakeLLM([(match_any(), respond_with(b"escalate"))])


# endregion: provider


# region: agent
async def triage(ctx: ActivationContext) -> Complete:
    """Decide what to do about one event for one entity.

    Module-level, not a closure: the DoFn holding the agent is serialized for
    the runner, so the agent has to pickle by reference.
    """
    ctx.memory.append("recent", ctx.event, max_items=32)
    seen = len(ctx.memory.ring("recent"))

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[f"event={ctx.event.decode()} seen={seen}"],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=b"%s:%d" % (response.response, seen))


# endregion: agent


def build_pipeline(pipeline: beam.Pipeline) -> beam.pvalue.PCollection:
    """Wire the transform. Input must be pre-keyed by `entity_key`."""
    # region: pipeline
    events = pipeline | "Events" >> beam.Create(
        [
            AgentEnvelope(entity_key=b"acct-1", event_time_ms=1_000, external_event=b"login"),
            AgentEnvelope(entity_key=b"acct-1", event_time_ms=2_000, external_event=b"transfer"),
            AgentEnvelope(entity_key=b"acct-2", event_time_ms=1_500, external_event=b"login"),
        ]
    )

    # `RunAgent` does not key elements itself. It validates the input is
    # KV-shaped at pipeline-construction time and raises ValueError otherwise,
    # because a stateful DoFn cannot accept anything else.
    keyed = events | "Key" >> beam.WithKeys(lambda e: e.entity_key).with_output_types(
        tuple[bytes, AgentEnvelope]
    )

    outputs = keyed | "Agent" >> RunAgent(
        triage, config=AgentConfig(provider_factory=make_provider)
    )
    # endregion: pipeline
    return outputs.output


def main() -> None:
    with beam.Pipeline() as pipeline:
        results = build_pipeline(pipeline)
        # region: assertion
        # Per-key serialization means acct-1's two events are ordered relative
        # to each other; the ring is 1 then 2 deep. acct-2 is a separate key
        # with its own memory, so it starts at 1.
        assert_that(results, equal_to([b"escalate:1", b"escalate:2", b"escalate:1"]))
        # endregion: assertion
    print("fast_path: ok")


if __name__ == "__main__":
    main()

website/examples/fast_path.py — executed by the repository’s offline test tier.

Architecture — where the fast path sits among the two execution paths through RunAgent.
Correctness invariants — atomic commit and per-key serialization, the two rules this example demonstrates.
State and memory — what a key's blob holds and how it stays bounded.
RunAgent and AgentConfig in the API reference.