Function run_model_step_foreach

app/workflow/workflow.cpp:955–1071

}

std::string parse_foreach_source(const engine::io::json::Value & step) {
    std::string value = workflow_string(step, "foreach");
    if (value.size() >= 3 && value.front() == '$' && value[1] == '{' && value.back() == '}') {
        value = value.substr(2, value.size() - 3);
    }
    const std::string suffix = ".items";
    if (value.size() <= suffix.size() ||
        value.compare(value.size() - suffix.size(), suffix.size(), suffix) != 0) {
        throw std::runtime_error("workflow foreach must reference ${step_id.items}: " + value);
    }
    return value.substr(0, value.size() - suffix.size());
}

void run_model_step_foreach(
    const engine::runtime::ModelRegistry & registry,
    const engine::io::json::Value & step,
    const WorkflowRunOptions & options,
    WorkflowContext & context) {
    const std::string id = workflow_string(step, "id");
    const std::string source_id = parse_foreach_source(step);
    if (source_id == id) {
        throw std::runtime_error("workflow foreach cannot reference the current step: " + id);
    }
    const auto source_it = context.batches.find(source_id);
    if (source_it == context.batches.end()) {
        throw std::runtime_error("workflow foreach source is not a batch: " + source_id);
    }
    const auto source_items = source_it->second;
    const auto * request_value = step.find("request");
    if (request_value == nullptr || request_value->is_null()) {
        throw std::runtime_error("model step requires request object: " + id);
    }

    engine::runtime::ModelLoadRequest load_request;
    load_request.model_path = resolve_workflow_path(workflow_string(step, "model"), context);
    load_request.model_spec_override = options.model_spec_override;
    if (const auto family = workflow_optional_string(step, "family")) {
        load_request.family_hint = *family;
    }
    if (const auto config = workflow_optional_string(step, "config")) {
        load_request.config_id = *config;
    }
    if (const auto weight = workflow_optional_string(step, "weight")) {
        load_request.weight_id = *weight;
    }
    load_request.options = merged_options(options.load_options, step.find("load_options"), context);

    engine::runtime::SessionOptions session_options;
    session_options.backend = parse_step_backend(step, options.backend);
    session_options.options = merged_options(options.session_options, step.find("session_options"), context);

    const engine::runtime::TaskSpec task_spec{
        engine::runtime::parse_voice_task_kind(workflow_string(step, "task")),
        engine::runtime::parse_run_mode(workflow_string_or(step, "mode", "offline")),
    };
    if (task_spec.mode != engine::runtime::RunMode::Offline) {
        throw std::runtime_error("workflow model steps currently require offline mode: " + id);
    }