Backfill historical logs

Use a backfill when generations already exist in another 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.

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 process a limited number of requests at once. 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())

Create each generation before its feedback because the feedback request uses the returned data GUID. If feedback creation fails, 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.

Invalid credentials or identifiers cause requests to fail. Network interruptions can leave a batch partly imported, so use the source identifier and a per-record checkpoint for safe retries.