How to Add a Custom Processor to Switchyard for Observing or Modifying Request State During Streaming

To add a custom Processor to Switchyard, implement the Processor<S> trait defined in crates/libsy/src/core/processor.rs, wrap your implementation in an Arc<dyn Processor<S>>, and attach it to any algorithm using the with_processor builder method before invoking run_stream.

Switchyard, NVIDIA’s Rust-based LLM routing framework, enables fine-grained control over streaming requests through a pipeline of Processor objects. Each processor receives routing events and can inspect or mutate the request-side state (S) before the next algorithmic turn executes. This guide demonstrates how to implement the trait, attach processors to algorithms, and leverage the pattern in both Rust and Python.

Understanding the Processor Trait Architecture

The processor system lives in the core libsy crate and operates entirely on the request side, ensuring zero interference with remote inference servers.

The Core Trait Definition

In crates/libsy/src/core/processor.rs, Switchyard defines the Processor trait as:

pub trait Processor<S = ()>: Send + Sync {
    fn process(&mut self, state: &mut S, event: Event);
}

The generic parameter S represents the state type carried through the stream. The process method receives a mutable reference to this state and an Event enum (e.g., Event::Prompt, Event::Decision), allowing your implementation to observe or rewrite any part of the request context.

Integration with Algorithms

Algorithms like FallThrough expose a with_processor method that appends processors to an internal list. When Algorithm::run_stream executes, it iterates over this list for every event, calling process(&mut state, event) in sequence. This design ensures that downstream processors observe any mutations made by upstream ones.

Steps to Implement a Custom Processor

Follow these five steps to integrate custom logic into the Switchyard streaming pipeline.

1. Define a struct to hold configuration or mutable counters.

2. Implement the Processor<S> trait, providing the process method where you inspect the event and mutate state as needed.

3. Wrap the processor in Arc<dyn Processor<S>> to satisfy thread-safety requirements for shared access across the streaming runtime.

4. Attach the processor to your algorithm using .with_processor(...) before building the final chain.

5. Run the algorithm via run_stream; your processor executes automatically on every turn.

Practical Code Examples

Below are runnable examples demonstrating read-only observation, state mutation, and Python integration.

Example 1: Logging Prompt Evolution

This processor logs every prompt sent to the model without modifying state. It targets TurnState as carried by the FallThrough algorithm.

use std::sync::Arc;
use libsy::core::{processor::{Processor, Event}, state::TurnState};

struct PromptLogger;

impl Processor<TurnState> for PromptLogger {
    fn process(&mut self, state: &mut TurnState, event: Event) {
        if let Event::Prompt { msg } = &event {
            println!("[PromptLogger] Prompt sent to model: {}", msg);
        }
    }
}

// Attach and run
let alg = FallThrough::new(/* ... */)
    .with_processor(Arc::new(PromptLogger));

let result = alg.run_stream(request).await?;

Key implementation details:

• The implementation is bound to TurnState via Processor<TurnState>.
• The method signature process(&mut self, state: &mut TurnState, event: Event) matches the trait exactly.
• No state mutation occurs; the processor only emits side-effects to stdout.

Example 2: Prepending a System Prompt

This processor modifies the request state by injecting a static system prompt before the first user message, ensuring downstream components see the enriched context.

use std::sync::Arc;
use libsy::core::{processor::{Processor, Event}, state::TurnState};

struct SystemPromptInjector {
    sys_prompt: String,
    injected: bool,
}

impl Processor<TurnState> for SystemPromptInjector {
    fn process(&mut self, state: &mut TurnState, event: Event) {
        if !self.injected {
            if let Event::Prompt { msg } = &event {
                let new_msg = format!("{} {}", self.sys_prompt, msg);
                state.set_prompt(new_msg);  // Mutates the request state
                self.injected = true;
            }
        }
    }
}

let injector = SystemPromptInjector {
    sys_prompt: "You are a helpful AI assistant.".into(),
    injected: false,
};

let alg = FallThrough::new(/* ... */)
    .with_processor(Arc::new(injector));

let result = alg.run_stream(request).await?;

Key implementation details:

• Mutable struct fields (injected) track whether the modification has occurred.
• state.set_prompt() rewrites the prompt field inside TurnState before the algorithm proceeds.
• Because the processor runs early in the chain, subsequent processors and the final router observe the prefixed prompt.

Example 3: Python Bindings Implementation

Switchyard exposes the Processor trait to Python via switchyard-py, allowing custom logic in Python.

from switchyard import Algorithm, Event, Processor, TurnState
import asyncio

class UppercaseProcessor(Processor):
    def process(self, state: TurnState, event: Event):
        if isinstance(event, Event.Prompt):
            event.msg = event.msg.upper()
            state.set_prompt(event.msg)  # Reflect the change in state

alg = Algorithm.fall_through().with_processor(UppercaseProcessor())

async def run():
    result = await alg.run_stream(request)
    print(result)

asyncio.run(run())

Key implementation details:

• Python processors inherit from the Processor class defined in the bindings.
• The process method receives the same state and event objects as Rust, enabling full parity for observability and modification.
• Use this pattern when your team prefers Python for rapid prototyping of request transformations.

Key Source Files and References

The following files define the trait, built-in implementations, and integration points:

• crates/libsy/src/core/processor.rs — Defines the Processor trait and the Event enum.
• crates/libsy/src/algorithms/fall_through.rs — Implements with_processor for attaching custom processors to the FallThrough algorithm.
• crates/libsy/src/algorithms/util/tool_signals.rs — Contains ToolSignalProcessor, a reference implementation showing side-effect emission.

Summary

• Implement Processor<S> from libsy to receive Event notifications and mutable access to request state S.
• Use with_processor on any Switchyard algorithm to register your implementation before streaming begins.
• Wrap processors in Arc<dyn Processor<S>> to satisfy Send + Sync requirements for concurrent execution.
• Modify state or emit side-effects inside the process method to enforce policies, log telemetry, or transform prompts without altering core routing logic.
• Leverage Python bindings (switchyard-py) to write processors in Python when preferred.

Frequently Asked Questions

What is the Processor trait in Switchyard?

The Processor trait is a request-side hook defined in crates/libsy/src/core/processor.rs that allows custom logic to observe and mutate the state S carried through a streaming algorithm. It requires implementing a single method, process(&mut self, state: &mut S, event: Event), which the algorithm calls on every event turn.

Can I chain multiple custom processors?

Yes. Algorithms accept multiple processors via sequential calls to with_processor. They execute in the order attached, forming a pipeline where each processor’s mutations to state are visible to the next. Place processors that modify state early in the chain to ensure downstream observers see the final values.

How do I access the request state inside a processor?

The process method provides &mut S (e.g., &mut TurnState), allowing direct mutation of fields like the prompt or metadata. The concrete type S is determined by the algorithm; check the algorithm’s documentation or source (such as FallThrough) to determine which state type it carries.

Are custom processors compatible with Switchyard's Python bindings?

Yes. The switchyard-py crate exposes the Processor trait to Python. You can subclass Processor in Python, implement the process method, and pass the instance to .with_processor() on Python-wrapped algorithms, achieving the same observability and modification capabilities as native Rust processors.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}