Turning a CSV of Unstructured Rows Into Clean Records on One Key: Chunk, Call, Validate, Merge

Why Bulk Extraction Needs a Pipeline

CSV files often contain unstructured or semi-structured text (e.g., free-form addresses, product descriptions, or chat logs) that you want to turn into clean, typed records. Doing this row-by-row with a single API call is slow, expensive, and error-prone. A better approach is a four-stage pipeline: chunk, call, validate, merge. Each stage can use a different model to balance cost and quality, and you can run everything with one API key on the aggregator.

Stage 1: Chunk Rows

CSV rows vary in size and complexity. Sending each row individually wastes tokens on prompt overhead. Instead, group rows into chunks that fit comfortably within the model's context window—but leave room for the output.

• Estimate token count per row using a simple character-based heuristic (e.g., characters ÷ 4) and sum as you go.
• Set a chunk size limit based on the model's context window and your desired output length. For extraction, you rarely need the full window.
• Preserve row order by assigning each chunk a sequential index and recording the original row numbers within the chunk.

Example pseudocode:

chunks = []
current_chunk = []
current_tokens = 0
max_tokens = 3000  # conservative for output headroom

for i, row in enumerate(rows):
    row_tokens = len(row) // 4
    if current_tokens + row_tokens > max_tokens and current_chunk:
        chunks.append(current_chunk)
        current_chunk = []
        current_tokens = 0
    current_chunk.append((i, row))
    current_tokens += row_tokens

if current_chunk:
    chunks.append(current_chunk)

Stage 2: Call a Cheap Model for the First Pass

For bulk extraction, use a low-cost model (e.g., DeepSeek, Qwen, or GLM) to process each chunk. These models handle straightforward parsing well and keep costs down—especially important since you pay official price × 1.3 on the aggregator.

• Craft a clear prompt that specifies the output format (e.g., JSON array of objects) and includes the chunk's rows as a numbered list.
• Request an array where each element corresponds to a row, with a field indicating the original row index to maintain order.
• Limit output length to avoid truncation; if truncation occurs, split the chunk further.

Example prompt snippet:

Extract the following fields from each row: name, email, company.
Return a JSON array with objects containing "row_index" (0-based within this chunk) and the extracted fields.
Rows:
0: John Doe, john@example.com, Acme Inc
1: Jane Smith, jane@, Initech
...

Stage 3: Validate and Fix with a Stronger Model

Cheap models may miss edge cases, produce malformed JSON, or misalign rows. Use a stronger model (e.g., Claude or GPT) to validate and correct the output—but only for chunks that fail validation.

• Parse the JSON from the cheap model's response. If parsing fails, mark the chunk for repair.
• Check for completeness: ensure the number of extracted objects matches the number of input rows.
• Spot-check fields against simple rules (e.g., email contains "@", non-empty name).
• For failed chunks, re-send to the strong model with a repair prompt that includes the original rows and the faulty output, asking for corrections.

This two-tier approach keeps costs low while ensuring quality. Note that strong model calls are also billed at official price × 1.3, but you only use them where needed.

Stage 4: Merge Results Without Losing Order

After processing all chunks, merge the extracted records back into a single dataset, preserving the original CSV row order.

• Collect results from each chunk, each with a row_index relative to its chunk.
• Map to global indices using the chunk's starting row number.
• Sort by original index or fill an array of the same length as the input CSV.
• Handle missing rows by inserting placeholders or flagging them for manual review.

Example merge logic:

all_records = [None] * len(rows)
for chunk_idx, chunk_results in enumerate(results_per_chunk):
    start_idx = chunk_start_indices[chunk_idx]
    for rec in chunk_results:
        global_idx = start_idx + rec['row_index']
        all_records[global_idx] = rec

Putting It All Together

With one API key on the aggregator, you can orchestrate this pipeline using any HTTP client. The key benefits:

• Cost efficiency: cheap models for the bulk, strong models only for validation.
• Speed: chunking reduces the number of API calls.
• Accuracy: validation catches errors before they propagate.
• Order preservation: index tracking ensures your output aligns with the input.

Remember that on the aggregator, you pay official price × 1.3 for all calls. If you contribute keys or resources, you earn credits at official price × 1.1 (or × 1.2 for premium models) in USDC—a way to offset costs.

Next Steps

• Experiment with chunk sizes and model choices for your specific data.
• Add logging to track which chunks required repair.
• Consider parallelizing chunk processing (respecting rate limits) to speed up large files.