# Effect > Use pipedog from Effect with @pipedog/effect. Run a pipeline as a scoped resource, stream its frames, feed it from a Stream, and write frame processors as Effect functions with services from Layers. Source: https://polvallverdu.github.io/pipedog/docs/guides/effect/ [Effect](https://effect.website) apps can drive pipedog with `@pipedog/effect`, a small adapter that maps pipedog's asyncio-style tasks onto Effect's fibers and scopes. Pipelines, processors and services stay the same; the adapter covers starting and stopping a pipeline, passing frames in and out, and writing processors as Effects. It is pipedog-only: Pipecat has no counterpart. ## Installation [#installation] ```bash bun add @pipedog/effect effect ``` `effect` is a peer dependency (version 4). ## Runtimes [#runtimes] | Runtime | Works | What it needs | | ------------------ | ----- | ------------- | | Bun | ✅ yes | Nothing. | | Node.js | ✅ yes | Nothing. | | Deno | ✅ yes | Nothing. | | Cloudflare Workers | ✅ yes | Nothing. | Generated from `pipecat.runtimes` and `pipecat.compat` in `packages/effect/package.json`: ✅ means the compat smoke test passes on that runtime. The smoke test on every runtime runs a pipeline with an Effect processor from end to end, interrupts a running pipeline and checks that no task is left, and cancels a pipeline while a processor's Effect is running and checks that the Effect is interrupted. ## The API [#the-api] | Export | What it does | | --------------------------------- | ------------------------------------------------------------------------------------------------------- | | `startPipeline(worker, options?)` | Runs a `PipelineWorker` for the lifetime of the current `Scope` and returns a `PipelineHandle`. | | `runPipeline(worker, options?)` | `startPipeline` and `join` in a scope of their own. | | `makeProcessor(fn, options?)` | Builds a `FrameProcessor` from `(frame, direction, ctx) => Effect`. Its services come from the context. | | `PipelineError` | The tagged error a failed run fails with. | A `PipelineHandle` has: | Member | | | ----------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | | `join` | Waits for the pipeline to end. See [How a run ends](#how-a-run-ends). | | `queueFrame(frame, direction?)` | `worker.queueFrame` as an Effect. | | `queueFrames(framesOrStream, direction?)` | Queues an iterable, or every frame of a `Stream` until it ends. | | `frames(...types)` | A scoped subscription to the frames of those classes that reach the end of the pipeline, as a `Stream` that ends with the pipeline. | | `stopWhenDone` | Queues an `EndFrame`. | | `cancel(reason?)` | Cancels the pipeline. | | `worker` | The worker. | `frames()` subscribes when it runs, not when the stream is read, so subscribe before queueing the frames you want to see. Frames are buffered without bound until the stream reads them. ### Options [#options] | Option | Type | Required | Description | |---|---|---|---| | `runner` | `WorkerRunnerOptions \| undefined` | no | Options for the `WorkerRunner` that runs the worker. Signal handling is off by default (`handleSigint: false`): an Effect app handles signals with its runtime (e.g. `NodeRuntime.runMain`), whose interruption cancels the pipeline. | | `failOnError` | `boolean \| undefined` | no | Fail with a `PipelineError` on the first `ErrorFrame` that reaches the worker, cancelling the pipeline. Default false: errors are reported by the pipeline as usual and the run fails only if the pipeline stops because of an unusable processor. | | `cancelReason` | `string \| undefined` | no | Reason given to the runner when closing the scope cancels a running pipeline. | ## A voice agent with Effect [#a-voice-agent-with-effect] The example below is the [voice bot](https://polvallverdu.github.io/pipedog/docs/voice-bot.md) with one addition, written in Effect: every final transcription goes through a moderation service before it reaches the LLM. The service is an ordinary Effect service, with a live Layer that calls OpenAI's moderation endpoint: ```ts import * as Context from "effect/Context"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schedule from "effect/Schedule"; class ModerationError extends Data.TaggedError("ModerationError")<{ cause: unknown }> {} /** Decides whether a user's words may go to the LLM. */ class Moderation extends Context.Service< Moderation, { readonly isAllowed: (text: string) => Effect.Effect } >()("Moderation") {} /** OpenAI's moderation endpoint. */ const ModerationLive = Layer.succeed(Moderation, { isAllowed: (text) => Effect.tryPromise({ try: async (signal) => { const response = await fetch("https://api.openai.com/v1/moderations", { method: "POST", headers: { authorization: `Bearer ${process.env.OPENAI_API_KEY}`, "content-type": "application/json", }, body: JSON.stringify({ model: "omni-moderation-latest", input: text }), signal, }); const body = (await response.json()) as { results: { flagged: boolean }[] }; return !body.results[0]?.flagged; }, catch: (cause) => new ModerationError({ cause }), }), }); ``` The processor is an Effect function. It sees every frame, as `processFrame` does, and passes on what should go on with `ctx.push`. `makeProcessor` returns an Effect that needs `Moderation`, so the processor gets whichever implementation the program is given: ```ts import { TranscriptionFrame, TTSSpeakFrame } from "@pipedog/core/frames/frames"; import { FrameDirection } from "@pipedog/core/processors/frame_processor"; import { makeProcessor } from "@pipedog/effect"; /** Lets a final transcription through only if Moderation allows it. */ const moderateTranscriptions = makeProcessor((frame, direction, ctx) => Effect.gen(function* () { if (!(frame instanceof TranscriptionFrame) || direction !== FrameDirection.DOWNSTREAM) { return yield* ctx.push(frame, direction); } const moderation = yield* Moderation; const allowed = yield* moderation.isAllowed(frame.text).pipe( Effect.retry({ times: 2, schedule: Schedule.exponential("100 millis") }), Effect.timeout("2 seconds"), // Fail open: a slow or failing moderation call must not stall the conversation. Effect.orElseSucceed(() => true), ); if (allowed) { yield* ctx.push(frame); } else { yield* ctx.push(new TTSSpeakFrame({ text: "Sorry, I can't help with that." })); } }), ); ``` A session builds the pipeline and runs it for as long as its scope is open: ```ts import { startPipeline } from "@pipedog/effect"; /** One voice session on `transport`, for as long as the scope is open. */ const session = (transport: BaseTransport) => Effect.gen(function* () { const context = new LLMContext(); const aggregators = new LLMContextAggregatorPair(context, { userParams: new LLMUserAggregatorParams({ vadAnalyzer: new SileroVADAnalyzer() }), }); const worker = new PipelineWorker( new Pipeline([ transport.input(), new OpenAISTTService(), yield* moderateTranscriptions, aggregators.user(), new OpenAILLMService({ settings: new OpenAILLMSettings({ systemInstruction: "You are a helpful assistant." }), }), new OpenAITTSService({ sampleRate: 24000 }), transport.output(), aggregators.assistant(), ]), { cancelOnIdleTimeout: false, params: new PipelineParams({ audioInSampleRate: 16000, audioOutSampleRate: 24000 }), }, ); // A hang-up cancels the pipeline, which interrupts this fiber's `join`. transport.eventHandler("on_client_disconnected", async () => { await worker.cancel({ reason: "client disconnected" }); }); const pipeline = yield* startPipeline(worker); yield* Effect.logInfo(`session ${worker.name} started`); yield* pipeline.join; }).pipe(Effect.scoped); ``` Your server forks one fiber per connection, with the live services: ```ts import * as Fiber from "effect/Fiber"; /** Called by your server for each connected client. */ export function onConnection(transport: BaseTransport): () => void { const fiber = Effect.runFork(session(transport).pipe(Effect.provide(ModerationLive))); // Interrupting the fiber cancels the pipeline and waits until it has stopped. return () => void Effect.runFork(Fiber.interrupt(fiber)); } ``` In a test, swap the Layer and look at what reaches the end of the pipeline: ```ts import { EndFrame, TextFrame } from "@pipedog/core/frames/frames"; import * as Stream from "effect/Stream"; const BlockEverything = Layer.succeed(Moderation, { isAllowed: () => Effect.succeed(false) }); const moderationTest = Effect.gen(function* () { const worker = new PipelineWorker(new Pipeline([yield* moderateTranscriptions]), { cancelOnIdleTimeout: false, }); const pipeline = yield* startPipeline(worker); const spoken = yield* pipeline.frames(TTSSpeakFrame); const passed = yield* pipeline.frames(TextFrame); yield* pipeline.queueFrames([ new TranscriptionFrame({ text: "something awful", userId: "u1", timestamp: "" }), new EndFrame(), ]); yield* pipeline.join; return { spoken: yield* Stream.runCollect(spoken), passed: yield* Stream.runCollect(passed) }; }).pipe(Effect.scoped, Effect.provide(BlockEverything)); ``` ## How a run ends [#how-a-run-ends] | The pipeline… | `join` | | ------------------------------------------------------------------------------------------------------- | --------------------------------------------------- | | ends with an `EndFrame` or `StopFrame` | succeeds | | is cancelled: `cancel()`, a `CancelWorkerFrame`, a processor that cancels it, or the scope closing | is interrupted | | is cancelled after an error from a processor that can no longer work (`ProcessorUnusablePolicy.CANCEL`) | fails with `PipelineError`, its `frame` set | | gets any `ErrorFrame`, with `failOnError: true` | is cancelled, and `join` fails with `PipelineError` | | throws from the worker | fails with `PipelineError`, its `cause` set | Without `failOnError`, `ErrorFrame`s are handled as usual: logged, passed to `on_pipeline_error`, and the pipeline goes on. ## Cancellation [#cancellation] Effect interrupts fibers; pipedog cancels tasks, as asyncio does (see [Interruptions and cancellation](https://polvallverdu.github.io/pipedog/docs/concepts/interruptions-and-cancellation.md)). The adapter maps one onto the other in both directions, and never lets either side outlive the other: * **Interrupting the Effect cancels the pipeline.** `startPipeline` is an `acquireRelease`: closing the scope (because the fiber was interrupted, timed out, or simply finished) cancels a pipeline that is still running and waits until the runner has finished, its processors cleaned up and its tasks done. Release is uninterruptible, as in any Effect finalizer. * **Cancelling the pipeline interrupts the Effect.** `join` is interrupted when the pipeline is cancelled from its side. A processor's Effect is interrupted when its task is cancelled, which happens when the pipeline is cancelled and when an `InterruptionFrame` (a user barging in) cancels the processing of an interruptible frame. The processor waits for the Effect's finalizers before it lets the cancellation through. * **Every call into the pipeline runs in a task of its own.** `queueFrame`, `stopWhenDone`, `cancel` and the rest start an asyncio task and wait for it: interrupting the Effect cancels that task and waits for it; the task ending cancelled interrupts the Effect. * **Frames a processor's Effect pushes are pushed by the processor's own task**, in order, as a Python processor pushes from `process_frame`. `startPipeline` turns the runner's signal handling off: an Effect app handles `SIGINT` with its runtime (`NodeRuntime.runMain` and friends interrupt the main fiber), and that interruption cancels the pipeline. A failure or defect of a processor's Effect becomes `pushError` on the processor: an `ErrorFrame` upstream, with the error as its `exception`. An Effect that interrupts itself drops the frame.