Function generate_streaming

core/src/llm/structured.rs:194–290

Generate a structured JSON object with streaming partial updates. Calls `on_partial` with progressively more complete partial objects as tokens arrive. Returns the final validated object. In streaming mode, `max_repair_attempts` defaults to 0 because a repair would reset the partial object stream (confusing for consumers).

(
    client: &dyn LlmClient,
    req: &StructuredRequest,
    on_partial: PartialObjectCallback,
)

/// In streaming mode, `max_repair_attempts` defaults to 0 because a repair
/// would reset the partial object stream (confusing for consumers).
pub async fn generate_streaming(
    client: &dyn LlmClient,
    req: &StructuredRequest,
    on_partial: PartialObjectCallback,
) -> Result<StructuredResult> {
    let mode = resolve_mode(req.mode, client.native_structured_support());
    let messages = build_initial_messages(req, mode);
    let system = build_system_prompt(req, mode);
    let tools = build_tools(req, mode);
    let directive = build_directive(req, mode);

    let cancel_token = CancellationToken::new();
    let mut rx = client
        .complete_streaming_structured(&messages, Some(&system), &tools, &directive, cancel_token)
        .await
        .context("LLM streaming call failed during structured generation")?;

    let mut json_buffer = String::new();
    let mut last_valid_partial: Option<Value> = None;
    let mut final_response: Option<super::LlmResponse> = None;
    let mut last_parse_len: usize = 0;
    // Minimum bytes of new data before attempting a partial parse (reduces CPU)
    const PARSE_THRESHOLD: usize = 8;

    while let Some(event) = rx.recv().await {
        match event {
            StreamEvent::ToolUseInputDelta(delta) if mode == StructuredMode::Tool => {
                if final_response.is_some() {
                    continue;
                }
                json_buffer.push_str(&delta);
                if json_buffer.len() - last_parse_len >= PARSE_THRESHOLD {
                    if let Some(partial) = try_parse_partial_json(&json_buffer) {
                        if last_valid_partial.as_ref() != Some(&partial) {
                            on_partial(&partial);
                            last_valid_partial = Some(partial);
                        }
                    }
                    last_parse_len = json_buffer.len();
                }
            }
            StreamEvent::TextDelta(delta) if mode != StructuredMode::Tool => {
                if final_response.is_some() {
                    continue;
                }
                json_buffer.push_str(&delta);
                if json_buffer.len() - last_parse_len >= PARSE_THRESHOLD {
                    if let Some(json_start) = find_json_start(&json_buffer) {
                        let candidate = &json_buffer[json_start..];
                        if let Some(partial) = try_parse_partial_json(candidate) {
                            if last_valid_partial.as_ref() != Some(&partial) {
                                on_partial(&partial);
                                last_valid_partial = Some(partial);
                            }
                        }
                    }
                    last_parse_len = json_buffer.len();
                }