Struct RunnableParallel 

pub struct RunnableParallel<I: Send + Sync + 'static> { /* private fields */ }

A Runnable that runs multiple steps in parallel on the same input.

Each step is identified by a string key. The output is a HashMap<String, Value> where each key maps to the corresponding step’s output (serialized as serde_json::Value).

Example

let parallel = RunnableParallel::<String>::new()
    .with("length", RunnableLambda::new_sync(|s: String| s.len() as i64))
    .with("upper", RunnableLambda::new_sync(|s: String| s.to_uppercase()));

let result = parallel.invoke("hello".to_string(), None).await?;
// result = {"length": 5, "upper": "HELLO"}

Implementations

impl<I: Clone + Send + Sync + 'static> RunnableParallel<I>

pub fn new() -> Self

Create an empty parallel runnable.

pub fn with<O, R>(self, key: &str, runnable: R) -> Self

where O: Serialize + Send + Sync + 'static, R: Runnable<I, O> + Send + Sync + 'static, R::Error: Into<LcelError>,

Add a step with the given key.

The step’s output will be serialized to serde_json::Value and stored under the key in the output HashMap.

pub fn len(&self) -> usize

Number of parallel steps.

pub fn is_empty(&self) -> bool

Whether there are no steps.

pub fn assign<O, R>( self, key: &str, runnable: R, ) -> RunnableSequence<I, HashMap<String, Value>>

where I: 'static, O: Serialize + Send + Sync + 'static, R: Runnable<HashMap<String, Value>, O> + Send + Sync + 'static, R::Error: Into<LcelError>,

Add an assign step that injects a new key into the output HashMap.

This is the LCEL equivalent of Python’s RunnableParallel.assign(). It pipes the parallel output (a HashMap<String, Value>) through a RunnableAssign that runs the given runnable on the HashMap and merges the result under the specified key.

Example

let chain = RunnableParallel::<String>::new()
    .with("context", retriever.pipe(format_docs))
    .assign("question", RunnableLambda::new_sync(|m: HashMap<String, Value>| {
        m.get("context").map(|c| c.to_string()).unwrap_or_default()
    }))
    .pipe(prompt_template)
    .pipe(llm);

How it works

assign() returns self.pipe(RunnableAssign). The RunnableAssign receives the HashMap output from the parallel step, runs the provided runnable on it, and merges the result back.

Trait Implementations

impl<I: Send + Sync + 'static> Debug for RunnableParallel<I>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<I: Clone + Send + Sync + 'static> Default for RunnableParallel<I>

fn default() -> Self

Returns the “default value” for a type.

impl<I: Clone + Send + Sync + 'static> Runnable<I, HashMap<String, Value>> for RunnableParallel<I>

fn invoke<'life0, 'async_trait>( &'life0 self, input: I, config: Option<RunnableConfig>, ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, Value>, LcelError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Execute all steps in parallel using tokio tasks.

fn batch<'life0, 'async_trait>( &'life0 self, inputs: Vec<I>, config: Option<RunnableConfig>, ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, LcelError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Batch: each step processes all inputs independently.

fn stream<'life0, 'async_trait>( &'life0 self, input: I, config: Option<RunnableConfig>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<HashMap<String, Value>, LcelError>> + Send>>, LcelError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Stream: invoke and return single-element stream.

type Error = LcelError

Error type.

fn batch_as_completed<'life0, 'async_trait>( &'life0 self, inputs: Vec<Input>, config: Option<RunnableConfig>, ) -> Pin<Box<dyn Future<Output = Result<Vec<(usize, Output)>, Self::Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Batch processing that returns results in completion order.

fn transform<'life0, 'async_trait>( &'life0 self, input: Pin<Box<dyn Stream<Item = Result<Input, Self::Error>> + Send>>, config: Option<RunnableConfig>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<Output, Self::Error>> + Send + '_>>, Self::Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Stream-to-stream transformation - the core of LCEL streaming.