Online runs don't save responses that parse_func rejects, so re-runs pay for them again #744

raw_response = data.response_message
responses = self._process_response(data)
if not responses:
    return
...
async with aiofiles.open(filename, "a") as f:
    await f.write(json_string + "\n")

"""Minimal OpenAI-compatible stub on 127.0.0.1 (no API key, no network)."""
import json
import threading
from collections import Counter
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

calls = Counter()


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if not body["messages"]:  # curator's rate-limit probe
            self.send_response(400)
            self.end_headers()
            return
        prompt = body["messages"][-1]["content"]
        calls[prompt] += 1
        reply = {
            "id": "x", "object": "chat.completion", "created": 0, "model": body["model"],
            "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": f"answer to {prompt}"}}],
            "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
        }
        data = json.dumps(reply).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)


server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
BASE_URL = f"http://127.0.0.1:{server.server_address[1]}/v1"

import os
import tempfile

from stub_server import BASE_URL, calls

from bespokelabs import curator

work = tempfile.mkdtemp()
rows = [{"q": f"q{i}"} for i in range(10)]


def make(parse):
    class Answer(curator.LLM):
        def prompt(self, row):
            return row["q"]

    Answer.parse = parse
    return Answer(model_name="gpt-4o-mini", backend="openai", backend_params={"base_url": BASE_URL, "api_key": "sk-local", "max_retries": 0})


def buggy_parse(self, row, response):
    if row["q"] in ("q3", "q7"):
        raise KeyError("typo")
    return {"q": row["q"], "a": response}


def fixed_parse(self, row, response):
    return {"q": row["q"], "a": response}


try:
    make(buggy_parse)(rows, working_dir=work)
except ValueError as e:
    print("run 1:", e)
run_dir = next(d for d in os.listdir(work) if d != "metadata.db")
print("responses saved after run 1:", sum(1 for _ in open(os.path.join(work, run_dir, "responses_0.jsonl"))), "of", sum(calls.values()), "requests sent")
sent = sum(calls.values())
make(fixed_parse)(rows, working_dir=work)
print("run 2 (parse fixed): requests sent:", sum(calls.values()) - sent)

import tempfile

from stub_server import BASE_URL, calls

from bespokelabs import curator


class Answer(curator.LLM):
    def prompt(self, row):
        return row["q"]

    def parse(self, row, response):
        if row["q"] in ("q3", "q7"):
            return []  # drop these rows
        return {"q": row["q"], "a": response}


work = tempfile.mkdtemp()
for run in (1, 2, 3):
    sent = sum(calls.values())
    try:
        Answer(model_name="gpt-4o-mini", backend="openai", backend_params={"base_url": BASE_URL, "api_key": "sk-local", "max_retries": 0})(
            [{"q": f"q{i}"} for i in range(10)], working_dir=work
        )
        print(f"run {run}: ok, requests sent: {sum(calls.values()) - sent}")
    except ValueError as e:
        print(f"run {run}: {e} (requests sent: {sum(calls.values()) - sent})")