lc_core/runnables/
runnable_trait.rs

1// src/core/runnables/runnable_trait.rs
2//! Runnable trait - foundation of LCEL (LangChain Expression Language).
3//!
4//! Every LangChain component implements Runnable, enabling
5//! chaining, composition, and interoperability.
6
7use super::RunnableConfig;
8use async_trait::async_trait;
9use futures_util::Stream;
10use std::pin::Pin;
11
12/// Base trait for all LangChain components.
13///
14/// This trait defines the core interface every component must implement:
15/// - Single execution via `invoke`
16/// - Batch processing via `batch`
17/// - Streaming output via `stream`
18///
19/// # Example
20/// ```rust
21/// use langchainrust::core::runnables::Runnable;
22/// use langchainrust::RunnableConfig;
23/// use async_trait::async_trait;
24///
25/// // Define a simple Runnable: add one
26/// struct AddOne;
27///
28/// #[async_trait]
29/// impl Runnable<i32, i32> for AddOne {
30///     type Error = std::convert::Infallible;
31///
32///     async fn invoke(&self, input: i32, _config: Option<RunnableConfig>) -> Result<i32, Self::Error> {
33///         Ok(input + 1)
34///     }
35/// }
36/// ```
37#[async_trait]
38pub trait Runnable<Input: Send + Sync + 'static, Output: Send + Sync + 'static>:
39    Send + Sync
40{
41    /// Error type.
42    type Error: std::error::Error + Send + Sync + 'static;
43
44    /// Transforms single input to output.
45    ///
46    /// This is the primary method for single execution.
47    ///
48    /// # Arguments
49    /// * `input` - Input to process.
50    /// * `config` - Optional execution configuration.
51    ///
52    /// # Returns
53    /// Execution result.
54    async fn invoke(
55        &self,
56        input: Input,
57        config: Option<RunnableConfig>,
58    ) -> Result<Output, Self::Error>;
59
60    /// Batch processing - transforms multiple inputs to outputs.
61    ///
62    /// Default implementation processes inputs sequentially.
63    /// Override for concurrent execution or batch optimization.
64    ///
65    /// # Arguments
66    /// * `inputs` - Input vector.
67    /// * `config` - Optional batch configuration.
68    ///
69    /// # Returns
70    /// Result vector.
71    async fn batch(
72        &self,
73        inputs: Vec<Input>,
74        config: Option<RunnableConfig>,
75    ) -> Result<Vec<Output>, Self::Error> {
76        let mut results = Vec::with_capacity(inputs.len());
77
78        // Process each input sequentially
79        for input in inputs {
80            let result = self.invoke(input, config.clone()).await?;
81            results.push(result);
82        }
83
84        Ok(results)
85    }
86
87    /// Streaming output - for real-time responses (LLM, etc).
88    ///
89    /// Enables real-time stream processing of output,
90    /// suitable for chat models, token generation, etc.
91    ///
92    /// # Arguments
93    /// * `input` - Input to process.
94    /// * `config` - Optional configuration.
95    ///
96    /// # Returns
97    /// Output stream.
98    ///
99    /// # Default Implementation
100    /// Wraps invoke result as single-element stream.
101    /// Types supporting true streaming should override.
102    async fn stream(
103        &self,
104        input: Input,
105        config: Option<RunnableConfig>,
106    ) -> Result<Pin<Box<dyn Stream<Item = Result<Output, Self::Error>> + Send>>, Self::Error> {
107        // Default: wrap invoke result as single-element stream
108        // All Runables automatically get stream capability
109        // Types with true streaming (like LLM) should override
110        let result = self.invoke(input, config).await?;
111        let stream = futures_util::stream::once(async move { Ok(result) });
112        Ok(Box::pin(stream))
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119    use futures_util::StreamExt;
120
121    struct TestRunnable;
122
123    #[async_trait]
124    impl Runnable<String, String> for TestRunnable {
125        type Error = std::convert::Infallible;
126
127        async fn invoke(
128            &self,
129            input: String,
130            _config: Option<RunnableConfig>,
131        ) -> Result<String, Self::Error> {
132            Ok(format!("processed: {}", input))
133        }
134    }
135
136    #[tokio::test]
137    async fn test_default_stream_returns_single_element() {
138        let runnable = TestRunnable;
139        let mut stream = runnable.stream("test".to_string(), None).await.unwrap();
140
141        let first = stream.next().await;
142        assert!(first.is_some());
143        assert_eq!(first.unwrap().unwrap(), "processed: test");
144
145        let second = stream.next().await;
146        assert!(second.is_none());
147    }
148
149    #[tokio::test]
150    async fn test_invoke_matches_stream_result() {
151        let runnable = TestRunnable;
152
153        let invoke_result = runnable.invoke("hello".to_string(), None).await.unwrap();
154        let mut stream = runnable.stream("hello".to_string(), None).await.unwrap();
155        let stream_result = stream.next().await.unwrap().unwrap();
156
157        assert_eq!(invoke_result, stream_result);
158    }
159}
lc_core/runnables/runnable_trait.rs

lc_core/runnables/
runnable_trait.rs
