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::{CacheControl, 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            self.config.tool_cache_policy,
100        );
101
102        // 4. 通过 LlmInvoker 执行流式调用（自动处理 retry/fallback）
103        let mut stream = self
104            .invoker
105            .invoke_stream(&req, &state.messages, state.iterations)
106            .await
107            .map_err(|e| {
108                GraphError::Terminal(TerminalError::NodeExecutionFailed {
109                    node: self.name.clone(),
110                    source: e.into(),
111                })
112            })?;
113
114        // 收集流式事件，构建完整的 ChatResponse
115        let mut content_blocks: Vec<ContentBlock> = Vec::new();
116        let mut current_text = String::new();
117        let mut current_thinking = String::new();
118        let mut tool_calls_count: usize = 0;
119        let mut usage: Option<lellm_core::TokenUsage> = None;
120
121        while let Some(event) = stream.next().await {
122            match event {
123                Ok(provider_event) => {
124                    // ResponseComplete 需要提取 tool_calls（所有权转移），单独处理
125                    if let lellm_provider::ProviderEvent::ResponseComplete {
126                        tool_calls,
127                        usage: u,
128                    } = &provider_event
129                    {
130                        for tc in tool_calls {
131                            content_blocks.push(ContentBlock::ToolCall(tc.clone()));
132                        }
133                        tool_calls_count = content_blocks
134                            .iter()
135                            .filter(|b| matches!(b, ContentBlock::ToolCall(_)))
136                            .count();
137                        usage = u.clone();
138                        continue;
139                    }
140
141                    match translate_provider_event(&provider_event) {
142                        TranslationResult::EmitWithText { chunk, delta } => {
143                            ctx.emit(chunk);
144                            current_text.push_str(&delta);
145                        }
146                        TranslationResult::EmitWithThinking { chunk, delta, .. } => {
147                            ctx.emit(chunk);
148                            current_thinking.push_str(&delta);
149                        }
150                        TranslationResult::Emit(chunk) => ctx.emit(chunk),
151                        _ => {}
152                    }
153                }
154                Err(e) => {
155                    return Err(GraphError::Terminal(TerminalError::NodeExecutionFailed {
156                        node: self.name.clone(),
157                        source: e.into(),
158                    }));
159                }
160            }
161        }
162
163        // 构建完整的 ChatResponse
164        if !current_thinking.is_empty() {
165            content_blocks.push(ContentBlock::Thinking(ThinkingBlock {
166                thinking: current_thinking,
167                redacted: None,
168            }));
169        }
170        if !current_text.is_empty() {
171            content_blocks.push(ContentBlock::Text(TextBlock {
172                text: current_text,
173                // 为 Assistant 响应的最后一个 TextBlock 添加 Breakpoint，
174                // 使对话历史在后续轮次可被前缀缓存命中。
175                cache_control: Some(CacheControl::Breakpoint),
176            }));
177        }
178
179        let response = ChatResponse {
180            content: content_blocks,
181            usage: usage.unwrap_or_default(),
182            raw: serde_json::json!(null),
183        };
184
185        // 5. 分离 output / reasoning token，Emit Token Effects
186        let (output_tokens, reasoning_tokens) = split_output_tokens(&response.content);
187        ctx.record(AgentMutation::AddOutputTokens(output_tokens));
188        ctx.record(AgentMutation::AddReasoningTokens(reasoning_tokens));
189
190        // 6. Emit 消息追加 Mutation
191        let content = response.content.clone();
192        let msg = Message::Assistant { content };
193        ctx.record(AgentMutation::AppendMessage(msg));
194
195        // 7. 记录 tool_calls
196        let has_tools = response.has_tool_calls();
197        if has_tools {
198            ctx.record(AgentMutation::AddToolCalls(tool_calls_count));
199        }
200
201        // 8. Emit LastResponse（供 PostLLMGuard 检查）
202        ctx.record(AgentMutation::SetLastResponse(response));
203
204        // 路由决策全部交给 PostLLMGuard，此处不调用 ctx.goto()/ctx.end()
205        tracing::debug!(
206            iteration = state.iterations + 1,
207            has_tool_calls = has_tools,
208            "LLM call completed"
209        );
210
211        Ok(())
212    }
213}
lellm_agent/runtime/react/llm_node.rs

lellm_agent/runtime/react/
llm_node.rs
