lc_agents/streaming/
tool_call_stream.rs

1//! StreamingFunctionCallingAgent - streaming output agent
2
3use std::pin::Pin;
4use std::sync::Arc;
5
6use futures_util::{Stream, StreamExt};
7use tokio::sync::mpsc;
8use tokio_stream::wrappers::ReceiverStream;
9
10use lc_core::language_models::BaseChatModel;
11use lc_providers::ProviderError;
12use lc_schema::Message;
13
14use super::state::AgentStreamEvent;
15
16/// Streaming Function Calling Agent
17///
18/// Streams LLM text (token by token), then emits FinalAnswer at the end.
19/// Tool-call state is exposed via `AgentStreamEvent::ToolCall`.
20/// Works with any LLM provider implementing `BaseChatModel`.
21pub struct StreamingFunctionCallingAgent {
22    llm: Arc<dyn BaseChatModel<Error = ProviderError> + Send + Sync>,
23}
24
25impl StreamingFunctionCallingAgent {
26    /// Creates a new Streaming Function Calling Agent
27    ///
28    /// # Backward compatibility
29    /// Old code `StreamingFunctionCallingAgent::new(openai_chat)` still works.
30    pub fn new<L>(llm: L) -> Self
31    where
32        L: BaseChatModel + Send + Sync + 'static,
33        L::Error: Into<ProviderError>,
34    {
35        Self {
36            llm: lc_providers::wrap_chat_model(llm),
37        }
38    }
39
40    /// Creates an agent from a wrapped `Arc<dyn BaseChatModel>`
41    pub fn from_arc(llm: Arc<dyn BaseChatModel<Error = ProviderError> + Send + Sync>) -> Self {
42        Self { llm }
43    }
44
45    /// Streams execution: returns an event stream
46    pub async fn invoke_stream(
47        &self,
48        input: String,
49    ) -> Pin<Box<dyn Stream<Item = AgentStreamEvent> + Send>> {
50        let (tx, rx) = mpsc::channel(32);
51        let llm = self.llm.clone();
52        let messages = vec![Message::human(input)];
53
54        tokio::spawn(async move {
55            let mut stream = match llm.stream_chat(messages, None).await {
56                Ok(s) => s,
57                Err(e) => {
58                    let _ = tx
59                        .send(AgentStreamEvent::Error {
60                            message: format!("Stream initialization failed: {}", e),
61                        })
62                        .await;
63                    return;
64                }
65            };
66
67            let mut full = String::new();
68            while let Some(chunk) = stream.next().await {
69                match chunk {
70                    Ok(chunk) => {
71                        full.push_str(&chunk.text);
72                        if tx
73                            .send(AgentStreamEvent::Text {
74                                content: chunk.text,
75                            })
76                            .await
77                            .is_err()
78                        {
79                            break;
80                        }
81                    }
82                    Err(e) => {
83                        let _ = tx
84                            .send(AgentStreamEvent::Error {
85                                message: format!("Stream error: {}", e),
86                            })
87                            .await;
88                        break;
89                    }
90                }
91            }
92
93            let _ = tx
94                .send(AgentStreamEvent::FinalAnswer { content: full })
95                .await;
96        });
97
98        Box::pin(ReceiverStream::new(rx))
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use lc_providers::{OpenAIChat, OpenAIConfig};
106
107    #[test]
108    fn test_new() {
109        let llm = OpenAIChat::new(OpenAIConfig::default());
110        let _agent = StreamingFunctionCallingAgent::new(llm);
111    }
112}
lc_agents/streaming/tool_call_stream.rs

lc_agents/streaming/
tool_call_stream.rs
