| 267 | } |
| 268 | |
| 269 | async fn stream( |
| 270 | &self, |
| 271 | request: LlmRequest, |
| 272 | ) -> GraphBitResult<Box<dyn Stream<Item = GraphBitResult<LlmResponse>> + Unpin + Send>> { |
| 273 | let url = format!("{}/chat/completions", self.base_url); |
| 274 | |
| 275 | let messages: Vec<PerplexityMessage> = |
| 276 | request.messages.iter().map(Self::convert_message).collect(); |
| 277 | |
| 278 | let tools: Option<Vec<PerplexityTool>> = if request.tools.is_empty() { |
| 279 | None |
| 280 | } else { |
| 281 | Some(request.tools.iter().map(Self::convert_tool).collect()) |
| 282 | }; |
| 283 | |
| 284 | let body = PerplexityStreamRequest { |
| 285 | model: self.model.clone(), |
| 286 | messages, |
| 287 | max_tokens: request.max_tokens, |
| 288 | temperature: request.temperature, |
| 289 | top_p: request.top_p, |
| 290 | tools: tools.clone(), |
| 291 | tool_choice: if tools.is_some() { |
| 292 | Some("auto".to_string()) |
| 293 | } else { |
| 294 | None |
| 295 | }, |
| 296 | stream: Some(true), // Enable streaming |
| 297 | }; |
| 298 | |
| 299 | // Add extra parameters |
| 300 | let mut request_json = serde_json::to_value(&body)?; |
| 301 | if let serde_json::Value::Object(ref mut map) = request_json { |
| 302 | for (key, value) in request.extra_params { |
| 303 | map.insert(key, value); |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // Timeout constants for different phases of the request |
| 308 | // These values balance responsiveness with network variability: |
| 309 | // - CONNECTION_TIMEOUT: Generous time for initial TLS handshake and HTTP connection |
| 310 | // (Perplexity API can be slow to establish connections under load) |
| 311 | // - ERROR_BODY_TIMEOUT: Short timeout since error responses are typically small JSON |
| 312 | // (don't want to wait long if API is unresponsive) |
| 313 | // - CHUNK_TIMEOUT: Moderate timeout for each SSE chunk; Perplexity streams can have |
| 314 | // pauses between tokens but should not hang indefinitely |
| 315 | const CONNECTION_TIMEOUT: Duration = Duration::from_secs(60); |
| 316 | const ERROR_BODY_TIMEOUT: Duration = Duration::from_secs(10); |
| 317 | const CHUNK_TIMEOUT: Duration = Duration::from_secs(30); |
| 318 | |
| 319 | // Apply timeout to initial connection |
| 320 | let response = timeout( |
| 321 | CONNECTION_TIMEOUT, |
| 322 | self.client |
| 323 | .post(&url) |
| 324 | .header("Authorization", format!("Bearer {}", self.api_key)) |
| 325 | .header("Content-Type", "application/json") |
| 326 | .json(&request_json) |