(
&self,
response: reqwest::Response,
agent_run_id: &str,
token_sink: &S,
)
| 1271 | } |
| 1272 | |
| 1273 | async fn stream_openai_tool_sse<S: TokenSink + Sync>( |
| 1274 | &self, |
| 1275 | response: reqwest::Response, |
| 1276 | agent_run_id: &str, |
| 1277 | token_sink: &S, |
| 1278 | ) -> AppResult<LLMTurn> { |
| 1279 | let mut stream = response.bytes_stream(); |
| 1280 | let mut line_buffer = String::new(); |
| 1281 | let mut output = String::new(); |
| 1282 | let mut stop_reason: Option<String> = None; |
| 1283 | let mut pending_calls: HashMap<usize, StreamingToolCall> = HashMap::new(); |
| 1284 | |
| 1285 | while let Some(chunk) = stream.next().await { |
| 1286 | line_buffer.push_str(&String::from_utf8_lossy(&chunk?)); |
| 1287 | |
| 1288 | while let Some(pos) = line_buffer.find('\n') { |
| 1289 | let mut line = line_buffer[..pos].to_string(); |
| 1290 | line_buffer.drain(..=pos); |
| 1291 | if line.ends_with('\r') { |
| 1292 | let _ = line.pop(); |
| 1293 | } |
| 1294 | |
| 1295 | let should_stop = self.parse_openai_sse_line( |
| 1296 | &line, |
| 1297 | agent_run_id, |
| 1298 | token_sink, |
| 1299 | &mut output, |
| 1300 | &mut pending_calls, |
| 1301 | &mut stop_reason, |
| 1302 | )?; |
| 1303 | |
| 1304 | if should_stop { |
| 1305 | return finalize_llm_turn(output, pending_calls, stop_reason); |
| 1306 | } |
| 1307 | } |
| 1308 | } |
| 1309 | |
| 1310 | finalize_llm_turn(output, pending_calls, stop_reason) |
| 1311 | } |
| 1312 | |
| 1313 | #[allow(clippy::too_many_arguments)] |
| 1314 | fn parse_openai_sse_line<S: TokenSink + Sync>( |
no test coverage detected