respkit

respkit is a small reusable Python SDK for structured LLM tasks over normalized text input.

Install

python3 -m pip install respkit

For local development:

git clone https://github.com/gracee3/respkit.git
cd respkit
python3 -m pip install -e .[dev]

Quick Start

from pathlib import Path

from respkit.inputs import NormalizedInput
from respkit.providers import OpenAICompatibleProvider
from respkit.runners import SingleInputRunner

from examples.demo_rename_proposal.task import build_tasks


proposal_task, _review_task = build_tasks()
runner = SingleInputRunner(
    task=proposal_task,
    provider=OpenAICompatibleProvider(endpoint="http://localhost:8000/v1/responses"),
    artifacts_root=Path(".respkit_demo"),
)

item = NormalizedInput(
    source_id="sample.txt",
    source_path=Path("sample.txt"),
    media_type="text/plain",
    decoded_text=Path("sample.txt").read_text(encoding="utf-8"),
)

result = runner.run(item)
print(result.status, result.validated_output)

result contains normalized status, structured output, validation report, and artifact directory path. Use ReviewRunner when you need a second-pass validator.

What the SDK contains

• respkit/ — reusable core:
  • providers (openai_compatible, base contract)
  • runners (single, batch, review)
  • tasks/contracts
  • validators and normalization
  • actions (markdown/json/manifest)
  • artifacts + manifest writers
• examples/ — safe synthetic examples only
• tests/ — SDK tests with synthetic fixtures

Supported shape

Every task uses:

• normalized input (source_id, source_path, decoded_text)
• prompt template + renderer
• schema validation
• deterministic action execution
• optional review pass

The core execution path remains generic and reusable:

• prompt rendering
• provider call
• response parsing + validation
• artifact capture
• manifest append
• optional review runner with optional concurrency

Synthetic developer example

The public example is intentionally synthetic:

• examples/demo_rename_proposal/
  • task.py — task wiring and prompt/context builders
  • schemas.py — proposal/review output models
  • prompts/ — synthetic prompts
  • __main__.py — CLI entrypoint

This example uses only synthetic names/entities and works without corpus-specific data.

For a generic SDK-level example of the new corpus adjudication ledger abstraction, use:

python examples/demo_ledger.py

Run the example

respkit-demo single /path/to/file.txt \
  --endpoint http://localhost:8000/v1/responses \
  --out .respkit_demo \
  --provider-timeout 30
  # or: python -m examples.demo_rename_proposal single /path/to/file.txt ...

respkit-demo batch /path/to/text-dir \
  --endpoint http://localhost:8000/v1/responses \
  --out .respkit_demo \
  --max-concurrency 4 \
  --review

You can run the review pass concurrently with:

respkit-demo batch /path/to/text-dir \
  --endpoint http://localhost:8000/v1/responses \
  --out .respkit_demo \
  --max-concurrency 8 \
  --review --review-max-concurrency 4 \
  --provider-timeout 30
  # or: python -m examples.demo_rename_proposal batch ...

Available flags:

• --max-concurrency: proposal batch parallelism
• --review-max-concurrency: review concurrency (default 1)
• --provider-timeout: request timeout seconds
• --review: enable optional review pass

Smoke scripts

make smoke-single   # single fixture
make smoke-batch    # batch fixtures
make smoke          # runs both

Env vars used by smoke targets:

• SMOKE_ENDPOINT (default http://localhost:8000/v1/responses)
• SMOKE_MAX_CONCURRENCY (default 1)
• SMOKE_REVIEW_MAX_CONCURRENCY (default 1)
• SMOKE_PROVIDER_TIMEOUT (default 30)
• SMOKE_REVIEW (set to any non-empty value to enable review)

scripts/smoke_single.sh and scripts/smoke_batch.sh call the synthetic example by default and can be used outside make.

Status vocabulary

Status values are consistent across runners and manifest rows:

• success
• preflight_model_not_found
• provider_error
• parse_error
• validation_failed
• action_failed
• review_failed

parse_error means the provider output could not be parsed into a JSON payload.

Artifact output

Each task run writes per-item artifacts under:

.respkit_demo/
  artifacts/
    <task_name>/<run_id>/
      prompt_template.md
      prompt.txt
      provider_request.json
      raw_response.json
      parsed_response.json
      validation_report.json
      validated_response.json
      action_results.json
      run_metadata.json
      manifest_row.json (if manifest action is configured)

The run metadata includes provider timing, status, and chosen model.

manifest.jsonl is append-only and one row is written per manifest action invocation.

Ledger API (SDK)

respkit ships a task-agnostic adjudication ledger for tasks that iterate over many items and need proposal/review/human/apply coordination.

• module: respkit.ledger
• canonical storage: SQLite (LedgerStore)
• machine/human state is split:
  • machine: not_run, proposed, reviewed, provider_error, apply_ready, applied, superseded
  • human: needs_review, approved, rejected
• task payloads are stored in JSON columns (proposal_payload, review_payload, apply_payload, human_decision_payload) plus extras
• per-stage provenance and run identifiers are captured
• optional apply hooks with optional clean-tree guard are supported

SQLite stores a current-state table and an event/history table so stage transitions are auditable and never overwrite history.

applied_in_commit is separate from apply_code_commit:

• apply_code_commit: code hash at the time apply callback is invoked (or before mutation for non-dry-run apply).
• applied_in_commit: commit that eventually captures mutation output if your workflow commits afterward (optional and often unavailable immediately).

Core API

• LedgerStore(ledger_path) — create/open a SQLite ledger
• create_or_update_row(...)
• record_proposal(...)
• record_review(...)
• record_human_decision(...)
• record_apply(...)
• mark_superseded(...)
• query_rows(LedgerQuery(...))
• run_apply(...)
• exports:
  • export_csv(path, query=...)
  • export_jsonl(path, query=...)
  • export_markdown(path, query=...)
• import_jsonl(source_jsonl) for one-time migration to SQLite

Query examples

• LedgerQuery(task_name=task_name, unresolved_only=True)
• LedgerQuery(task_name=task_name, provider_error_only=True)
• LedgerQuery(task_name=task_name, rejected_only=True)
• LedgerQuery(task_name=task_name, not_approved_only=True)
• LedgerQuery(task_name=task_name, unresolved_only=True, rerun_eligible_only=True)
• include/exclude controls:
  • LedgerQuery(include_approved=False)
  • LedgerQuery(include_superseded=True)

Generic usage

from pathlib import Path

from respkit.ledger import (
    ApplyPolicy,
    HumanDecision,
    LedgerQuery,
    LedgerStore,
)

ledger = LedgerStore(Path(".my_ledger.sqlite"))
task_name = "generic-corpus-task"

ledger.record_proposal(
    task_name=task_name,
    item_id="item-001",
    item_locator="docs/file-a.txt",
    proposal_payload={"op": "normalize_section_headers"},
    proposal_result={"status": "ok"},
)
ledger.record_review(
    task_name=task_name,
    item_id="item-001",
    review_payload={"risk": "low"},
    review_result={"accept": True},
)
ledger.record_human_decision(task_name=task_name, item_id="item-001", decision=HumanDecision.APPROVED)

ready = ledger.query_rows(LedgerQuery(task_name=task_name, unresolved_only=True, include_approved=False))
print([r.item_id for r in ready])

ledger.run_apply(
    query=LedgerQuery(task_name=task_name),
    callback=lambda _row, dry_run: (
        {"op": "apply"} if dry_run else {"op": "apply"},
        {"status": "ok"},
    ),
    dry_run=True,
)

ledger.run_apply(
    query=LedgerQuery(task_name=task_name),
    callback=lambda _row, dry_run: ({"op": "apply"}, {"status": "applied"}),
    dry_run=False,
    policy=ApplyPolicy(require_clean_working_tree=True, working_directory=Path(".")),
)

Programmatic resolver session

Use a session object directly from agents or scripts:

from pathlib import Path

from respkit.ledger import (
    DefaultResolverHooks,
    ResolverAction,
    ResolverSession,
    LedgerQuery,
    LedgerStore,
)


class MyHooks(DefaultResolverHooks):
    def risk_flags(self, row):
        if row.review_payload and isinstance(row.review_payload, dict) and row.review_payload.get("risk") == "high":
            return ["high risk"]
        return []

    def derive_approved_output(self, row, edits):
        return {"approved_output": edits or {"approved": True}}


    def validate_resolution(self, row, edits):
        return (True, None)

ledger = LedgerStore(Path(".my_ledger.sqlite"))
session = ResolverSession(store=ledger, hooks=MyHooks())

# 1) list pending rows
pending = session.list_pending(LedgerQuery(task_name="generic-corpus-task", unresolved_only=True))
for view in pending:
    print(view.item_id, view.machine_status, view.human_status, view.risk_flags)
    print("preview:", session.preview_row(view))

    # Recommendation mode (do not persist)
    recommendation = session.build_recommendation(
        view,
        action=ResolverAction.APPROVE_WITH_EDIT,
        edits={"approved": True},
        note="policy suggestion",
        decision_source="agent",
        decision_actor="policy-bot",
    )
    recommendation_result = session.apply_recommendation(recommendation, apply=False)
    print("recommendation:", recommendation_result.status)

    # Persist as an explicit decision.
    persisted = session.apply_recommendation(
        session.build_recommendation(
            view,
            action=ResolverAction.REJECT,
            note="blocked by policy",
            decision_source="agent",
            decision_actor="policy-bot",
        )
    )
    print("persisted:", persisted.status, persisted.action.value if persisted.action else None)

Interactive resolver example (hook extension)

The CLI flow remains available and is now a thin wrapper over ResolverSession.

respkit-ledger resolve --ledger .my_ledger.sqlite --task-name generic-corpus-task --unresolved-only

You can set decision provenance metadata for CLI-driven rows:

respkit-ledger resolve \
  --ledger .my_ledger.sqlite \
  --task-name generic-corpus-task \
  --unresolved-only \
  --decision-source cli \
  --decision-actor cli-user

For a runnable toy demo, run:

PYTHONPATH=. python3 examples/demo_ledger_session.py

Resolver and Export CLI

Installable script:

respkit-ledger resolve --ledger .my_ledger.sqlite --task-name generic-corpus-task --unresolved-only
respkit-ledger export --ledger .my_ledger.sqlite --task-name generic-corpus-task --format markdown --out review.md
respkit-ledger import-jsonl --ledger .my_ledger.sqlite --source /tmp/old_ledger.jsonl

Demo command

Run the generic ledger demos:

PYTHONPATH=. python3 examples/demo_ledger.py
PYTHONPATH=. python3 examples/demo_ledger_resolver.py
PYTHONPATH=. python3 examples/demo_ledger_session.py
PYTHONPATH=. python3 examples/demo_ledger_service.py

Target explicit paths:

PYTHONPATH=. python3 examples/demo_ledger.py --repo /tmp/corpus_repo --ledger /tmp/corpus_ledger.sqlite

Ledger service backend (JSON-RPC over stdio)

The SDK also ships a local machine-readable service surface for frontends (CLI, agents, or future Rust/desktop TUI) that drive resolver workflows without duplicating SDK logic.

Start the backend:

python -m respkit.service.backend --ledger .my_ledger.sqlite --stdio

Equivalent script entrypoint (after installation):

respkit-ledger-service --ledger .my_ledger.sqlite --stdio

Request envelope is JSON-RPC 2.0 with one JSON object per line:

{"jsonrpc":"2.0","id":"1","method":"rows.list","params":{"task_name":"generic-corpus-task","unresolved_only":true}}

Success and error examples:

{"jsonrpc":"2.0","id":"1","result":{"rows":[...]}}
{"jsonrpc":"2.0","id":"9","error":{"code":-32602,"message":"params required","data":"..."}}

Core methods:

• ledger.open — health/read metadata
• ledger.info — store/task metadata
• ledger.summary — dashboard counts and by-task buckets
• ledger.tasks — available tasks in the ledger
• rows.list — query rows with ledger filters
• rows.get — full row payload with rendered summary/context
• rows.history — event history
• rows.preview — task-specific preview payload
• rows.validate — validate edits for a row
• rows.derive — derive approved output from edits
• rows.decide — recommendation/apply decision API
• actions.list — list available actions for selected rows
• actions.invoke — run built-in or adapter actions
• export — csv/jsonl/markdown export (inline or to output path)
• system.shutdown — terminate backend process

Generic service usage examples:

import json

def req(method: str, params: dict) -> str:
    return json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params})

print(req("ledger.summary", {"task_name": "generic-corpus-task"}))
print(req("rows.list", {"task_name": "generic-corpus-task", "unresolved_only": True}))
print(req("rows.get", {"task_name": "generic-corpus-task", "item_id": "item-001"}))
print(req("rows.history", {"task_name": "generic-corpus-task", "item_id": "item-001"}))
print(req("rows.decide", {"task_name": "generic-corpus-task", "item_id": "item-001", "action": "approve_with_edit", "apply": False, "edits": {"approved": True}}))
print(req("rows.decide", {"task_name": "generic-corpus-task", "item_id": "item-001", "action": "approve", "apply": True, "decision_source": "agent", "decision_actor": "local-llm-agent", "decision_note": "approved by policy-bot"}))
print(req("actions.list", {"task_name": "generic-corpus-task"}))
print(req("actions.invoke", {"task_name": "generic-corpus-task", "action": "mark_checked", "item_ids": ["item-001"]}))

Decision provenance is preserved on persisted actions:

• decision_source (default: agent in service default config)
• decision_actor (e.g. local-llm-agent, cli-user, batch-policy)
• decision_note
• optional decision_code_commit

The recommendation mode returns proposed actions without mutating ledger state (apply: false).

Adapter-enabled actions are injected via TaskServiceAdapter implementations:

from respkit.service import DefaultTaskServiceAdapter, ActionDescriptor, ActionResult
from respkit.ledger import HumanDecision, LedgerRow


class ToyServiceAdapter(DefaultTaskServiceAdapter):
    def available_actions(self, row: LedgerRow) -> list[ActionDescriptor]:
        actions = super().available_actions(row)
        actions.append(ActionDescriptor(name="mark_checked", description="mark row checked"))
        return actions

    def execute_action(self, *, row: LedgerRow, action: str, params, store) -> ActionResult:
        if action != "mark_checked":
            return super().execute_action(row=row, action=action, params=params, store=store)
        store.record_human_decision(
            task_name=row.task_name,
            item_id=row.item_id,
            decision=HumanDecision.NEEDS_REVIEW,
            decision_payload={"mark_checked": True},
            decision_source="action",
            decision_actor="adapter",
            notes="marked by adapter action",
        )
        return ActionResult(success=True, message="marked")

Launch backend with this adapter:

respkit-ledger-service --ledger .my_ledger.sqlite --adapter module.path:ToyServiceAdapter --stdio

Use whichever concrete adapter module path fits your private task package. The backend contract remains generic and stable; task-specific behavior stays in the adapter.

For a runnable end-to-end toy example, see:

PYTHONPATH=. python3 examples/demo_ledger_service.py

Local test fixtures

tests/fixtures/rename_inputs/*.txt contains synthetic, non-sensitive material for local runs.

Notes

This repository intentionally does not bundle real corpus data or private task iterations. Those should live in a private task/corpus repo.

• ResolverSession(store, hooks) for programmatic row-by-row decision workflows
  • list_pending(...), get_row(...), get_next(...), peek_next(...)
  • preview_row(...), build_recommendation(...), apply_recommendation(...)
• ResolverAction (approve, approve_with_edit, reject, needs_review, skip)
• ResolverRecommendation, ResolverRowView, ResolverApplyResult, ValidationResult