| 1537 | } |
| 1538 | |
| 1539 | async fn complete_streaming_structured( |
| 1540 | &self, |
| 1541 | _messages: &[Message], |
| 1542 | _system: Option<&str>, |
| 1543 | tools: &[ToolDefinition], |
| 1544 | directive: &StructuredDirective, |
| 1545 | _cancel_token: CancellationToken, |
| 1546 | ) -> anyhow::Result<mpsc::Receiver<StreamEvent>> { |
| 1547 | self.structured_calls |
| 1548 | .fetch_add(1, std::sync::atomic::Ordering::SeqCst); |
| 1549 | *self.last_directive.lock().unwrap() = Some(directive.clone()); |
| 1550 | *self.last_tool_names.lock().unwrap() = tools.iter().map(|t| t.name.clone()).collect(); |
| 1551 | let response = self.pop()?; |
| 1552 | let (tx, rx) = mpsc::channel(10); |
| 1553 | tokio::spawn(async move { |
| 1554 | for block in &response.message.content { |
| 1555 | if let ContentBlock::ToolUse { name, input, .. } = block { |
| 1556 | tx.send(StreamEvent::ToolUseStart { |
| 1557 | id: "call_001".to_string(), |
| 1558 | name: name.clone(), |
| 1559 | }) |
| 1560 | .await |
| 1561 | .ok(); |
| 1562 | let json_str = serde_json::to_string(input).unwrap(); |
| 1563 | for chunk in json_str.as_bytes().chunks(8) { |
| 1564 | tx.send(StreamEvent::ToolUseInputDelta( |
| 1565 | String::from_utf8_lossy(chunk).to_string(), |
| 1566 | )) |
| 1567 | .await |
| 1568 | .ok(); |
| 1569 | } |
| 1570 | } else if let ContentBlock::Text { text } = block { |
| 1571 | for chunk in text.as_bytes().chunks(8) { |
| 1572 | tx.send(StreamEvent::TextDelta( |
| 1573 | String::from_utf8_lossy(chunk).to_string(), |
| 1574 | )) |
| 1575 | .await |
| 1576 | .ok(); |
| 1577 | } |
| 1578 | } |
| 1579 | } |
| 1580 | tx.send(StreamEvent::Done(response)).await.ok(); |
| 1581 | }); |
| 1582 | Ok(rx) |
| 1583 | } |
| 1584 | } |
| 1585 | |
| 1586 | fn person_request(mode: StructuredMode) -> StructuredRequest { |