lc_langgraph/compiled/
graph.rs

1// crates/lc-langgraph/src/compiled/graph.rs
2//! CompiledGraph struct definition, constructors, getters, and runtime injection
3
4use super::types::{DynamicTask, GraphExecution};
5use crate::checkpointer::Checkpointer;
6use crate::edge::{ConditionalEdge, GraphEdge};
7use crate::errors::{GraphError, GraphResult};
8use crate::node::GraphNode;
9use crate::state::{Reducer, StateSchema};
10use crate::subgraph::SubgraphNode;
11use std::collections::HashMap;
12use std::sync::Arc;
13use tokio::sync::Mutex;
14use tokio::sync::RwLock;
15
16/// CompiledGraph - Ready-to-execute graph
17///
18/// Created from StateGraph.compile(). Handles:
19/// - State management and updates
20/// - Edge routing (fixed and conditional)
21/// - Execution with recursion limits
22/// - Checkpointing for persistence
23#[allow(clippy::type_complexity)]
24#[derive(Clone)]
25pub struct CompiledGraph<S: StateSchema> {
26    pub(super) nodes: HashMap<String, Arc<dyn GraphNode<S>>>,
27    pub(super) edges: Vec<GraphEdge>,
28    pub(super) entry_point: String,
29    pub(super) default_reducer: Arc<dyn Reducer<S>>,
30    pub(super) conditional_routers: HashMap<String, Arc<dyn ConditionalEdge<S>>>,
31    pub(super) checkpointer: Option<Arc<Mutex<dyn Checkpointer<S> + Send>>>,
32    pub(super) recursion_limit: usize,
33    pub(super) interrupt_before: Vec<String>,
34    pub(super) interrupt_after: Vec<String>,
35
36    pub(super) runtime_nodes: Arc<RwLock<HashMap<String, Arc<dyn GraphNode<S>>>>>,
37    pub(super) runtime_edges: Arc<RwLock<Vec<GraphEdge>>>,
38    pub(super) runtime_conditional_routers:
39        Arc<RwLock<HashMap<String, Arc<dyn ConditionalEdge<S>>>>>,
40    pub(super) task_inbox: Arc<Mutex<Vec<DynamicTask>>>,
41}
42
43impl<S: StateSchema> std::fmt::Debug for CompiledGraph<S> {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        f.debug_struct("CompiledGraph")
46            .field("nodes", &self.nodes.keys().collect::<Vec<_>>())
47            .field("edges", &self.edges)
48            .field("entry_point", &self.entry_point)
49            .field("recursion_limit", &self.recursion_limit)
50            .field("interrupt_before", &self.interrupt_before)
51            .field("interrupt_after", &self.interrupt_after)
52            .finish()
53    }
54}
55
56impl<S: StateSchema> CompiledGraph<S> {
57    pub(crate) fn new(
58        nodes: HashMap<String, Arc<dyn GraphNode<S>>>,
59        edges: Vec<GraphEdge>,
60        entry_point: String,
61        default_reducer: Arc<dyn Reducer<S>>,
62    ) -> Self {
63        Self {
64            nodes,
65            edges,
66            entry_point,
67            default_reducer,
68            conditional_routers: HashMap::new(),
69            checkpointer: None,
70            recursion_limit: 25,
71            interrupt_before: Vec::new(),
72            interrupt_after: Vec::new(),
73            runtime_nodes: Arc::new(RwLock::new(HashMap::new())),
74            runtime_edges: Arc::new(RwLock::new(Vec::new())),
75            runtime_conditional_routers: Arc::new(RwLock::new(HashMap::new())),
76            task_inbox: Arc::new(Mutex::new(Vec::new())),
77        }
78    }
79
80    pub(crate) fn add_router(&mut self, name: String, router: Arc<dyn ConditionalEdge<S>>) {
81        self.conditional_routers.insert(name, router);
82    }
83
84    pub fn with_checkpointer<C: Checkpointer<S> + 'static>(mut self, checkpointer: C) -> Self {
85        self.checkpointer = Some(Arc::new(Mutex::new(checkpointer)));
86        self
87    }
88
89    pub fn with_recursion_limit(mut self, limit: usize) -> Self {
90        self.recursion_limit = limit;
91        self
92    }
93
94    pub fn with_interrupt_before(mut self, nodes: Vec<String>) -> Self {
95        self.interrupt_before = nodes;
96        self
97    }
98
99    pub fn with_interrupt_after(mut self, nodes: Vec<String>) -> Self {
100        self.interrupt_after = nodes;
101        self
102    }
103
104    pub fn node_names(&self) -> Vec<String> {
105        self.nodes.keys().cloned().collect()
106    }
107
108    pub fn get_edges(&self) -> &[GraphEdge] {
109        &self.edges
110    }
111
112    pub fn entry_point(&self) -> &str {
113        &self.entry_point
114    }
115
116    pub fn recursion_limit(&self) -> usize {
117        self.recursion_limit
118    }
119
120    pub fn interrupt_before(&self) -> &[String] {
121        &self.interrupt_before
122    }
123
124    pub fn interrupt_after(&self) -> &[String] {
125        &self.interrupt_after
126    }
127
128    /// Get the last checkpoint state (for interrupt recovery)
129    pub async fn last_checkpoint_state(&self) -> Option<S> {
130        if let Some(ref cp) = self.checkpointer {
131            let guard = cp.lock().await;
132            let ids = guard.list().await.ok()?;
133            let last_id = ids.last()?.clone();
134            guard.load(&last_id).await.ok()
135        } else {
136            None
137        }
138    }
139
140    /// Create a resume execution context (from the last checkpoint)
141    /// interrupted_node may be "node_name" or "after_node_name"
142    pub async fn create_resume_execution(
143        &self,
144        interrupted_node: &str,
145    ) -> Option<GraphExecution<S>> {
146        let state = self.last_checkpoint_state().await?;
147        let current = interrupted_node
148            .strip_prefix("after_")
149            .unwrap_or(interrupted_node);
150        Some(GraphExecution {
151            state,
152            current_node: current.to_string(),
153            steps: Vec::new(),
154            recursion_count: 0,
155            interrupted_at: interrupted_node.to_string(),
156        })
157    }
158
159    pub(super) async fn get_node(&self, name: &str) -> GraphResult<Arc<dyn GraphNode<S>>> {
160        if let Some(node) = self.nodes.get(name) {
161            return Ok(node.clone());
162        }
163        if let Some(node) = self.runtime_nodes.read().await.get(name) {
164            return Ok(node.clone());
165        }
166        Err(GraphError::ExecutionError(format!(
167            "Node '{}' not found",
168            name
169        )))
170    }
171
172    /// Submit a new task for dynamic planning mid-execution
173    pub fn submit_task(&self, description: String) {
174        if let Ok(mut inbox) = self.task_inbox.try_lock() {
175            inbox.push(DynamicTask {
176                id: uuid::Uuid::new_v4().to_string(),
177                description,
178            });
179        }
180    }
181
182    /// Inject a runtime node directly
183    pub async fn inject_node(&self, name: &str, node: Arc<dyn GraphNode<S>>) -> GraphResult<()> {
184        self.runtime_nodes
185            .write()
186            .await
187            .insert(name.to_string(), node);
188        Ok(())
189    }
190
191    /// Inject a runtime edge directly
192    pub async fn inject_edge(&self, source: &str, target: &str) -> GraphResult<()> {
193        self.runtime_edges
194            .write()
195            .await
196            .push(GraphEdge::fixed(source, target));
197        Ok(())
198    }
199
200    /// Inject a subgraph as a runtime node
201    pub async fn inject_subgraph<SubS: StateSchema + 'static>(
202        &self,
203        name: &str,
204        subgraph: CompiledGraph<SubS>,
205        input_mapper: impl Fn(&S) -> SubS + Send + Sync + 'static,
206        output_mapper: impl Fn(&SubS, &mut S) + Send + Sync + 'static,
207    ) -> GraphResult<()> {
208        let node = SubgraphNode::new(name, subgraph, input_mapper, output_mapper);
209        self.runtime_nodes
210            .write()
211            .await
212            .insert(name.to_string(), Arc::new(node));
213        Ok(())
214    }
215}
lc_langgraph/compiled/graph.rs

lc_langgraph/compiled/
graph.rs
