Fraud triage¶
The runtime's differentiating flow — suspension, human-in-the-loop approval, and the fail-closed timeout — on the workload the project names first. Two accounts each send one suspicious transaction through a single streaming pipeline:
- Account A: the agent triages the transaction via a scripted model call,
stages an approval request, and suspends. An analyst's approval re-enters
the pipeline on the same key before the deadline, and the resumed activation
emits the freeze decision:
b"freeze:acct-a". - Account B: the same suspension — but nobody answers. When the scripted
clock passes the 30-second deadline, the HITL timer fires the default deny
route, which emits its deterministic fallback output
b"__hitl_timeout__". A timeout is an explicit outcome on the main output, never a silent drop.
The agent¶
triage is the entire agent. On the first activation it calls the model,
stages an approval intent with ctx.request_approval(...), and returns
Suspend(timeout_ms=...); the runtime persists a continuation and arms the
fail-closed timer. On resume the human's decision is on ctx.resume_approval,
and the agent turns it into a freeze or a release. Nothing external has
happened at suspension time — the approval request leaves the pipeline as an
APPROVAL-kind ToolIntent on .intents.
The harness¶
Everything below scripted_stream() is scripted clockwork, not agent logic. A
TestStream plays both the events topic and the approvals topic, and drives
both clocks explicitly: the watermark carries event time, and the
advance_processing_time(60) step is what fires account B's real-time HITL
timer. No sleep(), no wall clock — the same discipline every timer test in
the repository follows.
Where the intent_id really comes from
The approval branch computes the pending intent's id with the runtime's
intent_id_for(entity_key, seq, step_index) — a pure function, which is
why a replayed activation mints byte-identical intents and the effector
can deduplicate them. That line exists only because this example has no
effector: in production the effector consumes the
approval intent, routes it to a human, and publishes the decision onto the
approvals topic already carrying the intent_id. You never compute it
yourself.
The fail-closed behavior demonstrated here — the timer fires the fallback
route, the continuation is cleared, and a late approval is refused as
orphaned_result — is pinned for every pipeline by the HITL semantics gates
(tests/semantics/test_hitl_fail_closed.py);
see also the errors output for the orphaned_result and
hitl_timeout records those paths produce.
Run it¶
Running it on Dataflow¶
The pipeline below is scripted: a TestStream plays the transaction and
approval topics so the example runs offline with no credentials. Swapping that
harness for real Pub/Sub topics — the same triage agent, with topics, the
provider and the approval deadline as launch parameters — is what
Fraud triage on Dataflow packages as a Flex
Template.
The whole program¶
The code below is included verbatim from
examples/fraud_triage.py
— the same file
tests/examples/test_fraud_triage.py
executes in CI, asserting the freeze output, the single approval intent for
account A, and the deny fallback for account B.
"""Fraud triage with human approval: suspension, resume, and the fail-closed timeout.
Two accounts each send one suspicious transaction. The agent triages via a
scripted model call, stages an approval request, and suspends with an explicit
deadline. Account A's approval arrives in time and the resumed activation
freezes the account; account B's approval never arrives, so when the scripted
clock passes the deadline the runtime's default deny route emits its
deterministic fallback output — a timeout is an explicit outcome, never a
silent drop.
The approvals topic is played by a `TestStream` branch here. In production the
effector consumes the approval intent off `.intents`, routes it to a human, and
publishes the decision back onto the pipeline's approvals topic *already
carrying* the pending `intent_id`; this module computes that id with the
runtime's `intent_id_for` only because there is no effector in an offline
example — the id is a pure function of `(entity_key, seq, step_index)`, which
is the whole effectively-once argument in one line.
Run it offline, with no credentials and no docker:
uv run python -m examples.fraud_triage
Every function a pipeline references is module-level so the DirectRunner can
pickle it by reference.
"""
from __future__ import annotations
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.testing.test_stream import TestStream
from apache_beam.transforms.window import TimestampedValue
from beam_agents import AgentConfig, RunAgent, RunAgentOutputs
from beam_agents._protos import AgentEnvelope
from beam_agents.core.agent import Complete, Suspend, intent_id_for
from beam_agents.core.context import ActivationContext
from beam_agents.model.client import LlmRequest
from beam_agents.model.fake import FakeLLM, match_any, match_contains, respond_with
ACCOUNT_A = b"acct-a"
ACCOUNT_B = b"acct-b"
# How long the agent waits for a human decision before failing closed.
APPROVAL_TIMEOUT_MS = 30_000
# Account A suspends on its first activation (seq=0); the triage model call
# consumes step 0, so the approval intent is minted at step 1. In production
# the effector publishes the decision with this id already attached — computing
# it here just demonstrates that the id is deterministic.
APPROVAL_INTENT_ID = intent_id_for(ACCOUNT_A, seq=0, step_index=1)
def make_provider() -> FakeLLM:
"""Scripted triage: wire transfers read as suspicious, everything else clears."""
return FakeLLM(
[
(match_contains("wire-transfer"), respond_with(b"suspicious")),
(match_any(), respond_with(b"ok")),
]
)
async def triage(ctx: ActivationContext) -> Complete | Suspend:
"""Triage a transaction; suspend for approval when the model flags it.
On resume the human's decision is on `ctx.resume_approval`, delivered on
the same key the suspension committed under.
"""
if ctx.is_resume:
approval = ctx.resume_approval
if approval is not None and approval.approved:
return Complete(output=b"freeze:" + ctx.entity_key)
return Complete(output=b"release:" + ctx.entity_key)
verdict = await ctx.call_model(
LlmRequest(
model_id="fake-triage-model",
messages=[ctx.single_event.decode()],
tools_schema=None,
sampling_params=None,
)
)
if verdict.response != b"suspicious":
return Complete(output=b"cleared:" + ctx.entity_key)
# Stage the approval request (an APPROVAL-kind ToolIntent on `.intents`)
# and suspend. The runtime persists a continuation and arms the fail-closed
# HITL timer; nothing external has happened yet.
ctx.request_approval('{"action":"freeze","reason":"suspicious wire transfer"}')
return Suspend(snapshot=b"awaiting-approval", timeout_ms=APPROVAL_TIMEOUT_MS)
def _transaction(key: bytes, payload: bytes, t_ms: int) -> TimestampedValue[AgentEnvelope]:
envelope = AgentEnvelope(entity_key=key, event_time_ms=t_ms, external_event=payload)
return TimestampedValue(envelope, t_ms / 1000)
def _approval(key: bytes, intent_id: str, t_ms: int) -> TimestampedValue[AgentEnvelope]:
"""A human decision re-entering on the account's key — the approvals topic's job."""
envelope = AgentEnvelope(entity_key=key, event_time_ms=t_ms)
envelope.approval.intent_id = intent_id
envelope.approval.approved = True
envelope.approval.approver = "analyst@example.test"
envelope.approval.decided_at_ms = t_ms
return TimestampedValue(envelope, t_ms / 1000)
def scripted_stream() -> TestStream:
"""The harness: two transactions, one in-time approval, one elapsed deadline.
`TestStream` scripts both clocks. The watermark carries event time; the
processing-time advance is what fires account B's real-time HITL timer —
no `sleep()`, no wall clock.
"""
return (
TestStream()
.advance_watermark_to(0)
.add_elements([_transaction(ACCOUNT_A, b'{"kind":"wire-transfer","amount":9500}', 1_000)])
.add_elements([_transaction(ACCOUNT_B, b'{"kind":"wire-transfer","amount":8200}', 2_000)])
# The analyst approves account A's freeze well inside the deadline.
.add_elements([_approval(ACCOUNT_A, APPROVAL_INTENT_ID, t_ms=5_000)])
# Nobody answers for account B: advance the scripted clock past its
# 30s deadline so the HITL timer fires the default deny route.
.advance_processing_time(60)
.advance_watermark_to_infinity()
)
def build(pipeline: beam.Pipeline) -> RunAgentOutputs:
"""Wire the scripted stream, keyed by account, into `RunAgent`."""
keyed = (
pipeline
| "Transactions" >> scripted_stream()
| "KeyByAccount"
>> beam.WithKeys(lambda env: env.entity_key).with_output_types(tuple[bytes, AgentEnvelope])
)
return keyed | RunAgent(triage, config=AgentConfig(provider_factory=make_provider))
def main() -> None:
"""Run the example pipeline, printing each decision to stdout."""
options = PipelineOptions()
options.view_as(StandardOptions).streaming = True
with beam.Pipeline(options=options) as pipeline:
outputs = build(pipeline)
outputs.output | "Print" >> beam.Map(print)
if __name__ == "__main__":
main()