Skip to content
beam-agents
GitHub

v1.0.0a1 · pre-release, not yet on PyPI · Apache-2.0 · Python 3.11–3.12

An agent is a Beam transform.

beam-agents runs your agent as a keyed, stateful step inside an Apache Beam pipeline — durable per-key memory, effectively-once side effects, and event-time semantics wrapped around the agent you already wrote. It is a runtime, not a framework for authoring agents.

events | RunAgent(my_agent)

Read the conceptsInstall from sourceGitHub ↗

The shape of an activation

The loop closes through the message bus, not the DAG.

Beam graphs are acyclic, so an agent that calls a tool and acts on the result cannot loop inside the graph. It suspends, and the answer re-enters as a new element on the same key.

How an activation flows through a beam-agents pipelineEvents, tool results, and approvals are keyed by entity and flattened into the RunAgent transform. RunAgent emits four outputs: output, intents, traces, and errors. The intents output leaves the pipeline to an outbox topic, is executed by an external effector, and the result re-enters the pipeline on the same key, resuming the suspended activation.eventstool resultsapprovalsKAFKA / PUB-SUBWithKeysFLATTENRunAgentKEYED STATECONTINUATION.output.traces.errors.intentsoutbox topicDEDUP BY INTENT_IDeffectorSEPARATE SERVICE
  1. An event arrives on the events topic and is keyed by entity id.
  2. RunAgent activates for that key, reading working memory and the replay cache from keyed state.
  3. The agent needs an external write. It stages a ToolIntent, suspends, and the intent leaves for the outbox.
  4. The effector executes it — once per intent id — and publishes the result, which re-enters on the same key.
  5. The suspended activation resumes where it stopped and runs to completion, emitting its output.
  6. State and outputs commit atomically with the bundle, or not at all.

Four outputs

A complete pipeline consumes all four. Most forget the last one.

.output
Terminal agent output, as bytes.
.intents
Side-effect requests, bound for the outbox. The only stream that comes back.
.traces
Spans per activation, model call, and staged intent.
.errors
Dead letters. A record here means the activation committed nothing at all.

What the runtime holds

Four invariants, each traceable to the test that gates it.

  • Atomic commit

    Memory writes, intents, traces, and outputs are staged and applied only on success. A failed activation mutates nothing.

    tests/core/test_dofn_commit.py

  • Deterministic intent ids

    uuid5 over key, seq, and step index — never a clock. A replayed bundle mints byte-identical intents, so the effector can deduplicate on them.

    tests/semantics/test_effectively_once_e2e.py

  • Replay-cached model calls

    Every call is keyed on its content plus the activation position. A bundle retry costs zero additional provider calls and takes the same path.

    openspec/specs/llm-replay-cache/spec.md

  • Per-key serialization

    Beam processes one element at a time per key, so memory is race-free by construction. Parallelism comes from having many keys.

    tests/core/test_dofn_streaming.py

An activation, in full

Every line here is executed on every change.

This is read from a real file at build time, not retyped. It runs on the DirectRunner with no credentials and no network, and the repository’s required test tier runs it.

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.

All six examples →

Where the project actually is

Read the right-hand column before you plan around this.

Built and tested

  • The RunAgent transform, keyed state, timers, and the async bridge
  • Effectively-once side effects via the outbox and reference effector
  • Human-in-the-loop approvals with fail-closed timeouts at both layers
  • Anthropic, OpenAI-compatible, and vLLM providers, plus the replay cache
  • LangGraph, Google ADK, and Pydantic AI adapters, on the conformance matrix
  • Traces to OTLP or BigQuery; errors and intents to Kafka or Pub/Sub
  • Long-term memory stores: Bigtable, Redis, Firestore, and SQL
  • The YAML pipeline provider, and a pyperf benchmark harness with a gated baseline

Not built

  • Any published release — 1.0.0 is declared, but no v1.0.0 tag exists and nothing is on PyPI
  • A Vertex AI provider
  • Spark beyond the weekly conformance leg — no per-PR job exercises it

The roadmap, marked as roadmap →

About this documentation

Every page declares what backs it, and the build fails when that stops being true.

Pages carry typed claims — symbols, modules, specs, tests — and a verifier imports the package and resolves each one before this site can build. A page marked stable needs a spec and a test. A page marked planned fails the moment the code it describes starts to exist. Code samples are read from files the test suite executes, and the API reference is generated from the installed package, not written alongside it.

How changes get made here →