lc_core/runnables/
config.rs1use serde_json::Value;
5use std::collections::HashMap;
6use std::sync::Arc;
7use uuid::Uuid;
8
9use lc_callbacks::{CallbackManager, RunTree, RunType};
10
11use super::cancellation::CancellationToken;
12
13pub const RUN_META_PARENT_RUN_ID: &str = "__lc_parent_run_id";
20
21pub const RUN_META_TRACE_ID: &str = "__lc_trace_id";
25
26pub fn run_tree_from_config(
40 name: impl Into<String>,
41 run_type: RunType,
42 inputs: serde_json::Value,
43 config: Option<&RunnableConfig>,
44) -> RunTree {
45 let mut run = RunTree::new(name, run_type, inputs);
46 let Some(cfg) = config else {
47 return run;
48 };
49
50 for tag in &cfg.tags {
51 run = run.with_tag(tag.clone());
52 }
53 for (key, value) in &cfg.metadata {
54 if key == RUN_META_PARENT_RUN_ID || key == RUN_META_TRACE_ID {
55 continue;
56 }
57 run = run.with_metadata(key.clone(), value.clone());
58 }
59
60 if let Some(parent) = cfg
61 .metadata
62 .get(RUN_META_PARENT_RUN_ID)
63 .and_then(|v| v.as_str())
64 {
65 match Uuid::parse_str(parent) {
66 Ok(id) => run.parent_run_id = Some(id),
67 Err(_) => {
68 log::warn!("ignoring invalid {RUN_META_PARENT_RUN_ID} '{parent}' in RunnableConfig")
69 }
70 }
71 }
72 if let Some(trace) = cfg.metadata.get(RUN_META_TRACE_ID).and_then(|v| v.as_str()) {
73 match Uuid::parse_str(trace) {
74 Ok(id) => run.trace_id = Some(id),
75 Err(_) => {
76 log::warn!("ignoring invalid {RUN_META_TRACE_ID} '{trace}' in RunnableConfig")
77 }
78 }
79 }
80
81 run
82}
83
84#[derive(Debug, Clone, Default)]
86pub struct RunnableConfig {
87 pub tags: Vec<String>,
89
90 pub metadata: HashMap<String, Value>,
92
93 pub max_concurrency: Option<usize>,
95
96 pub run_id: Option<Uuid>,
98
99 pub run_name: Option<String>,
101
102 pub callbacks: Option<Arc<CallbackManager>>,
104
105 pub cancellation_token: Option<CancellationToken>,
107
108 pub temperature: Option<f32>,
114
115 pub max_tokens: Option<usize>,
120
121 pub configurable: HashMap<String, Value>,
128}
129
130impl RunnableConfig {
131 pub fn new() -> Self {
133 Self::default()
134 }
135
136 pub fn with_tag(mut self, tag: impl Into<String>) -> Self {
138 self.tags.push(tag.into());
139 self
140 }
141
142 pub fn with_metadata(mut self, key: impl Into<String>, value: Value) -> Self {
144 self.metadata.insert(key.into(), value);
145 self
146 }
147
148 pub fn with_max_concurrency(mut self, max: usize) -> Self {
150 self.max_concurrency = Some(max);
151 self
152 }
153
154 pub fn with_run_id(mut self, id: Uuid) -> Self {
156 self.run_id = Some(id);
157 self
158 }
159
160 pub fn with_run_name(mut self, name: impl Into<String>) -> Self {
162 self.run_name = Some(name.into());
163 self
164 }
165
166 pub fn with_callbacks(mut self, callbacks: Arc<CallbackManager>) -> Self {
168 self.callbacks = Some(callbacks);
169 self
170 }
171
172 pub fn with_cancellation_token(mut self, token: CancellationToken) -> Self {
174 self.cancellation_token = Some(token);
175 self
176 }
177
178 pub fn with_temperature(mut self, temperature: f32) -> Self {
180 self.temperature = Some(temperature);
181 self
182 }
183
184 pub fn with_max_tokens(mut self, max_tokens: usize) -> Self {
186 self.max_tokens = Some(max_tokens);
187 self
188 }
189
190 pub fn with_configurable(mut self, key: impl Into<String>, value: Value) -> Self {
200 self.configurable.insert(key.into(), value);
201 self
202 }
203
204 pub fn configurable_value(&self, key: &str) -> Option<&Value> {
206 self.configurable.get(key)
207 }
208
209 pub fn is_cancelled(&self) -> bool {
211 self.cancellation_token
212 .as_ref()
213 .is_some_and(|t| t.is_cancelled())
214 }
215
216 pub fn merge(mut self, other: RunnableConfig) -> Self {
218 for tag in other.tags {
220 if !self.tags.iter().any(|existing| existing == &tag) {
221 self.tags.push(tag);
222 }
223 }
224
225 self.metadata.extend(other.metadata);
227
228 self.configurable.extend(other.configurable);
230
231 if other.max_concurrency.is_some() {
233 self.max_concurrency = other.max_concurrency;
234 }
235 if other.run_id.is_some() {
236 self.run_id = other.run_id;
237 }
238 if other.run_name.is_some() {
239 self.run_name = other.run_name;
240 }
241 if other.cancellation_token.is_some() {
242 self.cancellation_token = other.cancellation_token;
243 }
244 if other.temperature.is_some() {
245 self.temperature = other.temperature;
246 }
247 if other.max_tokens.is_some() {
248 self.max_tokens = other.max_tokens;
249 }
250
251 if let (Some(self_cb), Some(other_cb)) = (&self.callbacks, &other.callbacks) {
255 self.callbacks = Some(Arc::new(self_cb.merge_with(other_cb)));
256 } else if other.callbacks.is_some() {
257 self.callbacks = other.callbacks;
258 }
259
260 self
261 }
262}
263
264#[cfg(test)]
265mod tests {
266 use super::*;
267 use serde_json::json;
268
269 #[test]
270 fn configurable_roundtrip() {
271 let cfg = RunnableConfig::new().with_configurable("session_id", json!("s1"));
272 assert_eq!(cfg.configurable_value("session_id"), Some(&json!("s1")));
273 assert_eq!(cfg.configurable_value("missing"), None);
274 }
275
276 #[test]
277 fn configurable_merge_overrides() {
278 let base = RunnableConfig::new().with_configurable("which", json!("a"));
279 let other = RunnableConfig::new().with_configurable("which", json!("b"));
280 let merged = base.merge(other);
281 assert_eq!(merged.configurable_value("which"), Some(&json!("b")));
282 assert_eq!(merged.configurable_value("none"), None);
284 }
285}