lc_agents/executor/
mod.rs

1// lc-agents/src/executor/mod.rs
2//! Agent base traits and executor implementation.
3
4use crate::types::{AgentFinish, AgentOutput, AgentStep};
5use async_trait::async_trait;
6use lc_core::language_models::TokenUsage;
7use std::collections::HashMap;
8use std::future::Future;
9use std::pin::Pin;
10use std::sync::atomic::AtomicUsize;
11
12/// Cache-namespace counter for each `AgentExecutor` instance.
13///
14/// P2-1: the same `(inputs, intermediate_steps)` may be planned into different actions
15/// by different Agents across executors, so a shared cache would cross-contaminate
16/// results. Giving each instance a unique namespace isolates cache keys by construction,
17/// without hurting deterministic hits across multiple `invoke`s on the same instance.
18static CACHE_NS: AtomicUsize = AtomicUsize::new(0);
19
20/// Agent error types.
21#[derive(Debug, thiserror::Error)]
22#[non_exhaustive]
23pub enum AgentError {
24    /// Output parsing error.
25    #[error("Output parsing error: {0}")]
26    OutputParsingError(String),
27
28    /// Tool not found.
29    #[error("Tool not found: {0}")]
30    ToolNotFound(String),
31
32    /// Tool execution error.
33    #[error("Tool execution error: {0}")]
34    ToolExecutionError(String),
35
36    /// Max iterations reached.
37    #[error("Max iterations reached")]
38    MaxIterationsReached,
39
40    /// Budget exceeded (approval/budget gate §4.2). Callers catch it to distinguish a
41    /// "budget stop" from "the model did not converge".
42    #[error("Budget exceeded: {0:?}")]
43    BudgetExceeded(BudgetExceeded),
44
45    /// Cross-process resume (§4.2): checkpoint read / write / restore failed.
46    #[error("Resume error: {0}")]
47    Resume(String),
48
49    /// Other error.
50    #[error("Agent error: {0}")]
51    Other(String),
52}
53
54/// Base Agent trait.
55///
56/// Defines the core interface for agents. Agent is responsible for planning,
57/// not execution. Execution is handled by AgentExecutor.
58#[async_trait]
59pub trait BaseAgent: Send + Sync {
60    /// Plans the next action.
61    ///
62    /// # Arguments
63    /// * `intermediate_steps` - History of executed steps.
64    /// * `inputs` - User input.
65    ///
66    /// # Returns
67    /// * `AgentOutput::Action` - Action to execute.
68    /// * `AgentOutput::Finish` - Final answer.
69    async fn plan(
70        &self,
71        intermediate_steps: &[AgentStep],
72        inputs: &HashMap<String, String>,
73    ) -> Result<AgentOutput, AgentError>;
74
75    /// Plans the next action, streaming any model text through `on_token` as it
76    /// is produced.
77    ///
78    /// # Arguments
79    /// * `intermediate_steps` - History of executed steps.
80    /// * `inputs` - User input.
81    /// * `on_token` - Called with each chunk of model text as it becomes
82    ///   available, taking **ownership** of the chunk (so the returned future
83    ///   never borrows the token and stays `'static`). Streaming-capable agents
84    ///   (e.g. ReAct, function-calling) emit free text per token; steps that
85    ///   produce no free text (e.g. a function-calling step invoking a tool)
86    ///   emit nothing. Non-streaming agents deliver the whole answer in one
87    ///   call.
88    ///
89    /// # Returns
90    /// * `AgentOutput::Action` - Action to execute.
91    /// * `AgentOutput::Finish` - Final answer.
92    ///
93    /// The default implementation delegates to [`BaseAgent::plan`] and forwards
94    /// the whole final-answer text as a single chunk — behaviorally identical
95    /// to calling `plan()` directly. Streaming-capable agents (e.g. ReAct)
96    /// override this to emit per-token chunks from the model's streaming chat
97    /// API; callers must still call [`BaseAgent::plan`] for the non-streaming
98    /// path (e.g. `invoke`) so that path is unaffected.
99    async fn plan_stream(
100        &self,
101        intermediate_steps: &[AgentStep],
102        inputs: &HashMap<String, String>,
103        on_token: &mut (dyn FnMut(String) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send),
104    ) -> Result<AgentOutput, AgentError> {
105        let output = self.plan(intermediate_steps, inputs).await?;
106        if let AgentOutput::Finish(finish) = &output {
107            on_token(finish.output().unwrap_or("").to_string()).await;
108        }
109        Ok(output)
110    }
111
112    /// Returns input keys.
113    fn input_keys(&self) -> Vec<&str> {
114        vec!["input"]
115    }
116
117    /// Returns allowed tools list.
118    fn get_allowed_tools(&self) -> Option<Vec<&str>> {
119        None
120    }
121
122    /// Returns stopped response when max iterations reached.
123    fn return_stopped_response(&self, _intermediate_steps: &[AgentStep]) -> AgentFinish {
124        AgentFinish::new(
125            "Agent stopped due to iteration limit or time limit.".to_string(),
126            String::new(),
127        )
128    }
129
130    /// Returns the token usage from the most recent `plan()` call, if available.
131    ///
132    /// Agents that make LLM calls inside `plan()` may override this to report
133    /// cost metrics to `AgentExecutor` (P1-5). Defaults to `None`.
134    fn last_token_usage(&self) -> Option<TokenUsage> {
135        None
136    }
137}
138
139/// Minimum allowed `max_iterations`.
140const MIN_MAX_ITERATIONS: usize = 1;
141
142/// Upper bound for `max_iterations` — guards against runaway loops.
143const MAX_MAX_ITERATIONS: usize = 100;
144
145/// Default number of tools executed concurrently.
146const DEFAULT_MAX_CONCURRENCY: usize = 8;
147
148mod agent_loop;
149mod budget;
150mod compaction;
151mod engine;
152mod hooks;
153#[cfg(test)]
154mod tests;
155mod tools;
156
157pub use budget::{BudgetConfig, BudgetExceeded};
158pub use compaction::{
159    estimate_step_tokens, CompactionConfig, CompactionStrategy, CompactionTrigger,
160};
161pub use engine::AgentExecutor;
lc_agents/executor/mod.rs

lc_agents/executor/
mod.rs
