Getting started

This walks through the smallest complete pipeline. It runs offline, with no credentials and no docker. Everything below is lifted from a file the test suite executes on every change.

Install first — see install, which is a source install because no release exists yet.

1. Write the agent

An agent is an async function that takes an activation context and returns an outcome. Module-level, because the DoFn holding it is serialized for the runner and has to pickle by reference.

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))

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

2. Supply a model

AgentConfig takes a factory, not a client. The factory runs worker-side, so connection pools are created where they are used rather than pickled across the wire.

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"))])

FakeLLM matches requests against ordered rules and records every call. It is the default model in the repository's own tests, and it is what makes an example runnable with no network.

3. Key the input

RunAgent does not key elements. It requires PCollection[KV[bytes, AgentEnvelope]] and raises ValueError at pipeline-construction time otherwise — a stateful DoFn cannot accept unkeyed input, and failing at construction beats failing on a worker.

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)
)

4. Run it

uv run python website/examples/fast_path.py

It prints fast_path: ok. What it asserts on the way there is the interesting part:

assert_that(results, equal_to([b"escalate:1", b"escalate:2", b"escalate:1"]))

Three events, two keys. acct-1's two events see a ring one deep and then two deep, because per-key state persists across activations and per-key serialization orders them relative to each other. acct-2 starts fresh at one: separate key, separate memory. Nothing in the agent code arranged that — it is what the transform is.

The example is executed by the repository's offline test tier on every change, so if this page and the code ever disagreed, CI would say so before you did.

What just happened

One process() call per event, each one an activation:

1. The runtime read the key's MemoryBlob out of Beam state and handed the agent a Memory facade over it.
2. ctx.memory.append staged a ring write. Nothing was persisted yet.
3. ctx.call_model consulted the replay cache before the provider, so a retried bundle would not call the provider a second time, and staged a trace event for the call.
4. Complete(output=...) ended the activation, and the runtime committed the staged memory write, the sequence-counter bump, and the emitted element together with the bundle.

Had step 3 raised, step 4 would never have run: no memory write, no seq advance, and a record on .errors instead of on .output. That all-or-nothing boundary is the property the rest of the documentation keeps referring back to.

What to read next

Depending on what you are trying to do:

• Perform an external write → intents and resume, then the effector.
• Ask a human → human in the loop.
• Handle failures → the errors output.
• Understand the guarantees → correctness invariants.
• Bring an existing LangGraph graph → the LangGraph adapter.
• See all four outputs at once → the four outputs.
• Call a tool with no side effect → read-only tools.
• Look up a name → the API reference, or the glossary if it is a word rather than a symbol.