lellm_agent/runtime/react/
llm_node.rs

1//! LLM 调用节点 — 执行单次 LLM 调用。
2//!
3//! **职责单一：** 只负责"调用 LLM + 收集流式响应 + emit Effects"。
4//! 不感知 Budget、Compaction、Iteration Limit 等运行时策略。
5//!
6//! # 分层
7//!
8//! ```text
9//! LLMNode  — 只负责 State ↔ Request ↔ Effects
10//!    │
11//! LlmInvoker — 只负责防御策略（retry, fallback, stream state machine）
12//!    │
13//! LlmProvider — 只负责 protocol adapter（stateless）
14//! ```
15
16use std::sync::Arc;
17
18use async_trait::async_trait;
19use futures_util::StreamExt;
20
21use lellm_core::{ChatResponse, ContentBlock, Message, TextBlock, ThinkingBlock};
22use lellm_graph::{GraphError, LeafContext, LeafNode, TerminalError};
23
24use super::super::config::{ToolUseConfig, build_request_inner_with_round};
25use super::super::context::{estimate_reasoning_block, estimate_text};
26use super::super::invoker::LlmInvoker;
27use super::super::runtime::ResolvedRound;
28use super::super::stream_translation::{TranslationResult, translate_provider_event};
29use super::super::tools::ToolExecutor;
30use super::super::typed_state::{AgentMutation, AgentState};
31
32/// 分离 output / reasoning token
33fn split_output_tokens(content: &[lellm_core::ContentBlock]) -> (usize, usize) {
34    let mut output_tokens: usize = 0;
35    let mut reasoning_tokens: usize = 0;
36    for b in content {
37        match b {
38            lellm_core::ContentBlock::Text(t) => output_tokens += estimate_text(&t.text),
39            lellm_core::ContentBlock::Thinking(th) => {
40                reasoning_tokens += estimate_reasoning_block(th)
41            }
42            lellm_core::ContentBlock::Image { .. } | lellm_core::ContentBlock::ToolCall(_) => {}
43        }
44    }
45    (output_tokens, reasoning_tokens)
46}
47
48/// LLM 调用节点。
49///
50/// # Typed State
51///
52/// 从 ctx 获取 `AgentState`，直接操作 typed 字段，写回 ctx。
53#[derive(Clone)]
54pub struct LLMNode {
55    pub name: String,
56    /// LLM 调用器 — 封装 retry/fallback/stream state machine
57    pub invoker: Arc<LlmInvoker>,
58    /// 工具执行器 — 用于获取工具定义
59    pub executor: ToolExecutor,
60    /// 配置 — 用于构建请求
61    pub config: ToolUseConfig,
62}
63
64impl LLMNode {
65    pub fn new(
66        name: impl Into<String>,
67        invoker: Arc<LlmInvoker>,
68        executor: ToolExecutor,
69        config: ToolUseConfig,
70    ) -> Self {
71        Self {
72            name: name.into(),
73            invoker,
74            executor,
75            config,
76        }
77    }
78}
79
80#[async_trait]
81impl LeafNode<AgentState> for LLMNode {
82    async fn execute(&self, ctx: &mut LeafContext<'_, AgentState>) -> Result<(), GraphError> {
83        // 1. 获取 AgentState
84        let state = ctx.state().clone();
85
86        // 2. Emit 迭代递增 Mutation
87        ctx.record(AgentMutation::IncrementIteration);
88
89        // 3. 获取工具定义 & 构建 LLM 请求
90        let round = ResolvedRound::new(self.executor.snapshot().await);
91
92        let req = build_request_inner_with_round(
93            self.invoker.model(),
94            &state.messages,
95            self.config.max_output_tokens,
96            &self.config.request_options,
97            state.iterations + 1,
98            &round.definitions,
99        );
100
101        // 4. 通过 LlmInvoker 执行流式调用（自动处理 retry/fallback）
102        let mut stream = self
103            .invoker
104            .invoke_stream(&req, &state.messages, state.iterations)
105            .await
106            .map_err(|e| {
107                GraphError::Terminal(TerminalError::NodeExecutionFailed {
108                    node: self.name.clone(),
109                    source: e.into(),
110                })
111            })?;
112
113        // 收集流式事件，构建完整的 ChatResponse
114        let mut content_blocks: Vec<ContentBlock> = Vec::new();
115        let mut current_text = String::new();
116        let mut current_thinking = String::new();
117        let mut tool_calls_count: usize = 0;
118        let mut usage: Option<lellm_core::TokenUsage> = None;
119
120        while let Some(event) = stream.next().await {
121            match event {
122                Ok(provider_event) => {
123                    // ResponseComplete 需要提取 tool_calls（所有权转移），单独处理
124                    if let lellm_provider::ProviderEvent::ResponseComplete {
125                        tool_calls,
126                        usage: u,
127                    } = &provider_event
128                    {
129                        for tc in tool_calls {
130                            content_blocks.push(ContentBlock::ToolCall(tc.clone()));
131                        }
132                        tool_calls_count = content_blocks
133                            .iter()
134                            .filter(|b| matches!(b, ContentBlock::ToolCall(_)))
135                            .count();
136                        usage = u.clone();
137                        continue;
138                    }
139
140                    match translate_provider_event(&provider_event) {
141                        TranslationResult::EmitWithText { chunk, delta } => {
142                            ctx.emit(chunk);
143                            current_text.push_str(&delta);
144                        }
145                        TranslationResult::EmitWithThinking { chunk, delta, .. } => {
146                            ctx.emit(chunk);
147                            current_thinking.push_str(&delta);
148                        }
149                        TranslationResult::Emit(chunk) => ctx.emit(chunk),
150                        _ => {}
151                    }
152                }
153                Err(e) => {
154                    return Err(GraphError::Terminal(TerminalError::NodeExecutionFailed {
155                        node: self.name.clone(),
156                        source: e.into(),
157                    }));
158                }
159            }
160        }
161
162        // 构建完整的 ChatResponse
163        if !current_thinking.is_empty() {
164            content_blocks.push(ContentBlock::Thinking(ThinkingBlock {
165                thinking: current_thinking,
166                redacted: None,
167            }));
168        }
169        if !current_text.is_empty() {
170            content_blocks.push(ContentBlock::Text(TextBlock {
171                text: current_text,
172                cache_control: None,
173            }));
174        }
175
176        let response = ChatResponse {
177            content: content_blocks,
178            usage: usage.unwrap_or_default(),
179            raw: serde_json::json!(null),
180        };
181
182        // 5. 分离 output / reasoning token，Emit Token Effects
183        let (output_tokens, reasoning_tokens) = split_output_tokens(&response.content);
184        ctx.record(AgentMutation::AddOutputTokens(output_tokens));
185        ctx.record(AgentMutation::AddReasoningTokens(reasoning_tokens));
186
187        // 6. Emit 消息追加 Mutation
188        let content = response.content.clone();
189        let msg = Message::Assistant { content };
190        ctx.record(AgentMutation::AppendMessage(msg));
191
192        // 7. 记录 tool_calls
193        let has_tools = response.has_tool_calls();
194        if has_tools {
195            ctx.record(AgentMutation::AddToolCalls(tool_calls_count));
196        }
197
198        // 8. Emit LastResponse（供 PostLLMGuard 检查）
199        ctx.record(AgentMutation::SetLastResponse(response));
200
201        // 路由决策全部交给 PostLLMGuard，此处不调用 ctx.goto()/ctx.end()
202        tracing::debug!(
203            iteration = state.iterations + 1,
204            has_tool_calls = has_tools,
205            "LLM call completed"
206        );
207
208        Ok(())
209    }
210}
lellm_agent/runtime/react/llm_node.rs

lellm_agent/runtime/react/
llm_node.rs
