Skip to content
beaconv0.1.0

Pipelines

Pipeline runs steps in order and feeds each step’s output to the next. run_traced also records a content hash of every step’s input and output, a compact fingerprint of the whole run that you can snapshot.

from beacon import Block, Pipeline, extract_json, template


class ExtractIntent(Block):
    def __init__(self, llm):
        self.llm = llm


    def run(self, message: str) -> dict:
        prompt = template("Classify as JSON: <msg>{m}</msg>", m=message)
        return extract_json(self.llm(prompt), schema=INTENT)


def route(intent: dict) -> str:
    return {"cancel": "retention", "refund": "billing"}.get(intent["intent"], "support")


pipeline = Pipeline([ExtractIntent(llm), route])
pipeline.run("I want to cancel")          # 'retention'

Rules

Section titled “Rules”
  • The first step receives the arguments given to run; each later step receives the previous output as its single argument.
  • Steps can be Block subclasses (override run) or plain callables, which are wrapped automatically.
  • Pipelines nest: a Pipeline is itself a Block.

Traces

Section titled “Traces”
>>> result = pipeline.run_traced("I want to cancel")
>>> result.output
'retention'
>>> result.trace()
[{'index': 0, 'block': 'ExtractIntent', 'input_hash': 'b4f6ee8b23e7441b', 'output_hash': '1f1bc8aeb9db2e0b'},
 {'index': 1, 'block': 'route', 'input_hash': '1f1bc8aeb9db2e0b', 'output_hash': '052f5628c6167d05'}]

Hashes come from hash_content, and are None for values that aren’t JSON-like. With a recorded model response, the trace is fully deterministic:

from beacon.snapshots import assert_snapshot


def test_cancel_flow():
    pipeline = Pipeline([ExtractIntent(llm=lambda p: RECORDED), route])
    result = pipeline.run_traced("I want to cancel")
    assert_snapshot("flows/cancel", {"output": result.output, "trace": result.trace()})