lc_core/runnables/
passthrough.rs

1// lc-core/src/runnables/passthrough.rs
2//! RunnablePassthrough - a Runnable that passes input through unchanged.
3//!
4//! Useful in LCEL pipelines where you need to forward the input
5//! without transformation, e.g. in `RunnableParallel` branches
6//! or as a no-op step.
7
8use super::config::RunnableConfig;
9use super::error::LcelError;
10use super::runnable_trait::Runnable;
11use async_trait::async_trait;
12use futures_util::Stream;
13use std::pin::Pin;
14
15/// A `Runnable` that passes its input through unchanged.
16///
17/// # Example
18///
19/// ```rust,ignore
20/// let passthrough = RunnablePassthrough::<i32>::new();
21/// let result = passthrough.invoke(42, None).await?; // 42
22/// ```
23pub struct RunnablePassthrough<I: Send + Sync + 'static> {
24    _marker: std::marker::PhantomData<I>,
25}
26
27impl<I: Send + Sync + 'static> std::fmt::Debug for RunnablePassthrough<I> {
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        f.debug_struct("RunnablePassthrough")
30            .field("type", &std::any::type_name::<I>())
31            .finish()
32    }
33}
34
35impl<I: Send + Sync + 'static> Default for RunnablePassthrough<I> {
36    fn default() -> Self {
37        Self::new()
38    }
39}
40
41impl<I: Send + Sync + 'static> RunnablePassthrough<I> {
42    /// Create a new passthrough runnable.
43    pub fn new() -> Self {
44        Self {
45            _marker: std::marker::PhantomData,
46        }
47    }
48}
49
50impl<I: Clone + Send + Sync + 'static> Clone for RunnablePassthrough<I> {
51    fn clone(&self) -> Self {
52        Self::new()
53    }
54}
55
56#[async_trait]
57impl<I: Clone + Send + Sync + 'static> Runnable<I, I> for RunnablePassthrough<I> {
58    type Error = LcelError;
59
60    async fn invoke(&self, input: I, _config: Option<RunnableConfig>) -> Result<I, LcelError> {
61        Ok(input)
62    }
63
64    /// Passthrough supports true streaming: each input item
65    /// is forwarded immediately without buffering.
66    async fn stream(
67        &self,
68        input: I,
69        _config: Option<RunnableConfig>,
70    ) -> Result<Pin<Box<dyn Stream<Item = Result<I, LcelError>> + Send>>, LcelError> {
71        Ok(Box::pin(futures_util::stream::once(
72            async move { Ok(input) },
73        )))
74    }
75
76    /// Passthrough supports true transform: each item in the
77    /// input stream is forwarded immediately.
78    async fn transform(
79        &self,
80        input: Pin<Box<dyn Stream<Item = Result<I, LcelError>> + Send>>,
81        _config: Option<RunnableConfig>,
82    ) -> Result<Pin<Box<dyn Stream<Item = Result<I, LcelError>> + Send + '_>>, LcelError> {
83        Ok(input)
84    }
85}
86
87#[cfg(test)]
88mod tests {
89    use super::*;
90    use futures_util::StreamExt;
91
92    #[tokio::test]
93    async fn invoke_passthrough() {
94        let passthrough = RunnablePassthrough::<i32>::new();
95        let result = passthrough.invoke(42, None).await.unwrap();
96        assert_eq!(result, 42);
97    }
98
99    #[tokio::test]
100    async fn stream_passthrough() {
101        let passthrough = RunnablePassthrough::<String>::new();
102        let mut stream = passthrough.stream("hello".to_string(), None).await.unwrap();
103        let result = stream.next().await.unwrap().unwrap();
104        assert_eq!(result, "hello");
105    }
106
107    #[tokio::test]
108    async fn transform_passthrough() {
109        let passthrough = RunnablePassthrough::<i32>::new();
110        let input = Box::pin(futures_util::stream::iter(vec![
111            Ok(1i32),
112            Ok(2i32),
113            Ok(3i32),
114        ])) as Pin<Box<dyn Stream<Item = Result<i32, LcelError>> + Send>>;
115
116        let mut output = passthrough.transform(input, None).await.unwrap();
117        assert_eq!(output.next().await.unwrap().unwrap(), 1);
118        assert_eq!(output.next().await.unwrap().unwrap(), 2);
119        assert_eq!(output.next().await.unwrap().unwrap(), 3);
120        assert!(output.next().await.is_none());
121    }
122
123    #[tokio::test]
124    async fn default_works() {
125        let passthrough = RunnablePassthrough::<i32>::default();
126        let result = passthrough.invoke(99, None).await.unwrap();
127        assert_eq!(result, 99);
128    }
129}
lc_core/runnables/passthrough.rs

lc_core/runnables/
passthrough.rs
