Backfill historical logs

Use a backfill when generations already exist in another database or observability system. Each historical generation becomes a Klu log associated with an existing Action.

For interactive filtering, feedback, Datasets, and exports, see Manage logs and datasets.

Before you begin

You need:

  • a workspace API key
  • an App and Action in Klu
  • the Action GUID
  • historical input and output values
  • a stable identifier from your source system, stored in metadata, so you can audit or deduplicate the import.

There is no separate SDK backfill endpoint. Create each historical item with the Data client and identify the import through metadata. Creating the same record twice creates two Klu logs, so checkpoint successful source IDs before retrying a batch.

Import a batch

The examples limit concurrency so a large import does not create an unbounded number of requests. Replace the synthetic records list with rows from your source system.

Backfill generations

import asyncio
from typing import Dict

from klu import Klu


records = [
    {
        "id": "history-001",
        "input": "Summarize order 1042",
        "output": "Order 1042 shipped on Monday.",
        "rating": "Positive",
    },
    {
        "id": "history-002",
        "input": "Summarize order 1043",
        "output": "Order 1043 is awaiting payment.",
        "rating": "Negative",
    },
]


async def main() -> None:
    klu = Klu("YOUR_API_KEY")
    limit = asyncio.Semaphore(10)

    async def import_record(record: Dict[str, str]) -> str:
        async with limit:
            data = await klu.data.create(
                input=record["input"],
                output=record["output"],
                action_guid="YOUR_ACTION_GUID",
                meta_data={
                    "source": "backfill",
                    "source_record_id": record["id"],
                },
            )
            await klu.feedback.create(
                type="rating",
                value=record["rating"],
                data_guid=data.guid,
                created_by="YOUR_USER_ID",
                source="backfill",
            )
            return data.guid

    imported_guids = await asyncio.gather(
        *(import_record(record) for record in records)
    )
    print(imported_guids)


asyncio.run(main())

The generation must be created before its feedback because feedback requires the returned data GUID. If feedback creation fails after data creation succeeds, retain that GUID and retry only the feedback request.

Verify the import

  1. Open the App and select Logs.
  2. Filter by the backfill source or search for a known input.
  3. Open a row and confirm the Action, input, output, metadata, and feedback.
  4. Compare the imported count with your checkpointed successful source IDs.
  5. Select the imported rows and use Save Dataset if they should become an evaluation or training set.

Missing or invalid Action GUIDs cause data creation to fail. Invalid API keys fail every request. Partial network failures can leave a batch partly imported, which is why the source identifier and per-record checkpoint are required for safe retries.