open_agent/client/query.rs
1/// A pinned, boxed stream of events from the model.
2///
3/// This type alias represents an asynchronous stream that yields [`StreamEvent`] items.
4/// Each item is wrapped in a `Result` to handle potential errors during streaming.
5///
6/// The stream is:
7/// - **Pinned** (`Pin<Box<...>>`): Required for safe async operations and self-referential types
8/// - **Boxed**: Allows dynamic dispatch and hides the concrete stream implementation
9/// - **Send**: Can be safely transferred between threads
10///
11/// # Events
12///
13/// - [`StreamEvent::Block`]: a completed [`ContentBlock`] — assistant text, or a fully
14/// assembled tool call
15/// - [`StreamEvent::Reasoning`]: chain-of-thought text, only when
16/// [`AgentOptions::include_reasoning`] is enabled
17/// - [`StreamEvent::Finish`]: exactly once, as the final item, carrying the
18/// [`FinishReason`]
19///
20/// # Migrating from `ContentStream` (0.7.x and earlier)
21///
22/// The stream used to yield bare `ContentBlock`s. Wrap the old match in
23/// [`StreamEvent::into_block`] to get the previous behaviour, then handle
24/// [`StreamEvent::Finish`] where the distinction between a clean stop and a truncated
25/// response matters.
26///
27/// # Error Handling
28///
29/// Errors in the stream indicate issues like:
30/// - Network failures or timeouts
31/// - Malformed SSE events
32/// - JSON parsing errors
33/// - API errors from the model provider
34///
35/// When an error occurs, the stream typically terminates. It's the caller's responsibility
36/// to handle errors appropriately.
37///
38/// # Examples
39///
40/// ```rust,no_run
41/// use open_agent::{query, AgentOptions, ContentBlock, FinishReason, StreamEvent};
42/// use futures::StreamExt;
43///
44/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
45/// let options = AgentOptions::builder()
46/// .model("gpt-4")
47/// .api_key("sk-...")
48/// .build()?;
49///
50/// let mut stream = query("Hello!", &options).await?;
51///
52/// while let Some(result) = stream.next().await {
53/// match result? {
54/// StreamEvent::Block(ContentBlock::Text(text)) => print!("{}", text.text),
55/// StreamEvent::Finish(reason) => println!("\nstopped: {reason}"),
56/// _ => {}
57/// }
58/// }
59/// # Ok(())
60/// # }
61/// ```
62pub type EventStream = Pin<Box<dyn Stream<Item = Result<StreamEvent>> + Send>>;
63
64/// Simple query function for single-turn interactions without conversation history.
65///
66/// This is a stateless convenience function for simple queries that don't require
67/// multi-turn conversations. It creates a temporary HTTP client, sends a single
68/// prompt, and returns a stream of events.
69///
70/// For multi-turn conversations or more control over the interaction, use [`Client`] instead.
71///
72/// # Parameters
73///
74/// - `prompt`: The user's message to send to the model
75/// - `options`: Configuration including model, API key, tools, etc.
76///
77/// # Returns
78///
79/// Returns an [`EventStream`] that yields events as they arrive from the model. The stream
80/// must be polled to completion to receive all content and the terminating
81/// [`StreamEvent::Finish`].
82///
83/// # Behavior
84///
85/// 1. Creates a temporary HTTP client with configured timeout
86/// 2. Builds message array (system prompt + user prompt)
87/// 3. Converts tools to the wire format if provided
88/// 4. Makes an HTTP POST request to the path the configured
89/// [`ApiProtocol`](crate::ApiProtocol) selects
90/// 5. Parses Server-Sent Events (SSE) response stream
91/// 6. Aggregates chunks into complete content blocks
92/// 7. Returns stream that yields events as they complete, ending with `Finish`
93///
94/// # Error Handling
95///
96/// This function can return errors for:
97/// - HTTP client creation failures
98/// - Network errors during the request
99/// - API errors (authentication, invalid model, rate limits, etc.)
100/// - SSE parsing errors
101/// - JSON deserialization errors
102///
103/// # Performance Notes
104///
105/// - Creates a new HTTP client for each call (consider using `Client` for repeated queries)
106/// - Timeout is configurable via `AgentOptions::timeout` (default: 120 seconds)
107/// - Streaming begins immediately; no buffering of the full response
108///
109/// # Examples
110///
111/// ## Basic Usage
112///
113/// ```rust,no_run
114/// use open_agent::{query, AgentOptions, ContentBlock, FinishReason, StreamEvent};
115/// use futures::StreamExt;
116///
117/// #[tokio::main]
118/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
119/// let options = AgentOptions::builder()
120/// .system_prompt("You are a helpful assistant")
121/// .model("gpt-4")
122/// .api_key("sk-...")
123/// .build()?;
124///
125/// let mut stream = query("What's the capital of France?", &options).await?;
126///
127/// while let Some(event) = stream.next().await {
128/// match event? {
129/// StreamEvent::Block(ContentBlock::Text(text)) => print!("{}", text.text),
130/// StreamEvent::Finish(FinishReason::Length) => {
131/// eprintln!("response truncated at the token cap");
132/// }
133/// _ => {}
134/// }
135/// }
136///
137/// Ok(())
138/// }
139/// ```
140///
141/// ## With Tools
142///
143/// ```rust,no_run
144/// use open_agent::{query, AgentOptions, Tool, ContentBlock, StreamEvent};
145/// use futures::StreamExt;
146/// use serde_json::json;
147///
148/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
149/// let calculator = Tool::new(
150/// "calculator",
151/// "Performs calculations",
152/// json!({"type": "object"}),
153/// |input| Box::pin(async move { Ok(json!({"result": 42})) })
154/// );
155///
156/// let options = AgentOptions::builder()
157/// .model("gpt-4")
158/// .api_key("sk-...")
159/// .tools(vec![calculator])
160/// .build()?;
161///
162/// let mut stream = query("Calculate 2+2", &options).await?;
163///
164/// while let Some(event) = stream.next().await {
165/// match event?.into_block() {
166/// Some(ContentBlock::ToolUse(tool_use)) => {
167/// println!("Model wants to use: {}", tool_use.name());
168/// // Note: You'll need to manually execute tools and continue
169/// // the conversation. For automatic execution, use Client.
170/// }
171/// Some(ContentBlock::Text(text)) => print!("{}", text.text),
172/// _ => {}
173/// }
174/// }
175/// # Ok(())
176/// # }
177/// ```
178///
179/// ## Error Handling
180///
181/// ```rust,no_run
182/// use open_agent::{query, AgentOptions};
183/// use futures::StreamExt;
184///
185/// # async fn example() {
186/// let options = AgentOptions::builder()
187/// .model("gpt-4")
188/// .api_key("invalid-key")
189/// .build()
190/// .unwrap();
191///
192/// match query("Hello", &options).await {
193/// Ok(mut stream) => {
194/// while let Some(result) = stream.next().await {
195/// match result {
196/// Ok(event) => println!("Event: {:?}", event),
197/// Err(e) => {
198/// eprintln!("Stream error: {}", e);
199/// break;
200/// }
201/// }
202/// }
203/// }
204/// Err(e) => eprintln!("Query failed: {}", e),
205/// }
206/// # }
207/// ```
208pub async fn query(prompt: &str, options: &AgentOptions) -> Result<EventStream> {
209 // Create HTTP client with configured timeout
210 // The timeout applies to the entire request, not individual chunks
211 let client = reqwest::Client::builder()
212 .timeout(Duration::from_secs(options.timeout()))
213 .build()
214 .map_err(Error::Http)?;
215
216 // Build messages array for the API request
217 // OpenAI format expects an array of message objects with role and content
218 let mut messages = Vec::new();
219
220 // Add system prompt if provided
221 // System prompts set the assistant's behavior and context
222 if !options.system_prompt().is_empty() {
223 messages.push(OpenAIMessage {
224 role: "system".to_string(),
225 content: Some(OpenAIContent::Text(options.system_prompt().to_string())),
226 tool_calls: None,
227 tool_call_id: None,
228 });
229 }
230
231 // Add user prompt
232 // This is the actual query from the user
233 messages.push(OpenAIMessage {
234 role: "user".to_string(),
235 content: Some(OpenAIContent::Text(prompt.to_string())),
236 tool_calls: None,
237 tool_call_id: None,
238 });
239
240 // Convert tools to OpenAI format if any are provided
241 // Tools are described using JSON Schema for parameter validation
242 let tools = if !options.tools().is_empty() {
243 Some(
244 options
245 .tools()
246 .iter()
247 .map(|t| t.to_openai_format())
248 .collect(),
249 )
250 } else {
251 None
252 };
253
254 // Build the OpenAI-compatible request payload
255 // stream=true enables Server-Sent Events for incremental responses
256 let request = OpenAIRequest {
257 model: options.model().to_string(),
258 messages,
259 stream: true, // Critical: enables SSE streaming
260 max_tokens: options.max_tokens(),
261 temperature: options.temperature(),
262 tools,
263 };
264
265 stream_request(&client, options, &request).await
266}