Skip to content

lpgcast.sink

The same shape as graflo's DocErrorSink: a small async protocol and a JSON Lines implementation. Three record kinds share one file — finding (an item left out after repairs), episode (an episode that produced nothing, e.g. no schema matched or the reply never parsed) and unmapped (what the model saw and no type covers: proposals for the schema, not failures).

Classes

JsonlQuarantineSink

Appends records to a JSON Lines file.

Source code in lpgcast/sink.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class JsonlQuarantineSink:
    """Appends records to a JSON Lines file."""

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self._lock = asyncio.Lock()

    async def write_records(self, records: list[QuarantineRecord]) -> None:
        if not records:
            return

        def _append() -> None:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.path, "a", encoding="utf-8") as handle:
                handle.writelines(
                    record.model_dump_json(exclude_none=True) + "\n"
                    for record in records
                )

        async with self._lock:
            await asyncio.to_thread(_append)

Attributes

path = Path(path) instance-attribute

Methods:

__init__(path)
Source code in lpgcast/sink.py 
54
55
56
def __init__(self, path: str | Path) -> None:
    self.path = Path(path)
    self._lock = asyncio.Lock()
write_records(records) async
Source code in lpgcast/sink.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def write_records(self, records: list[QuarantineRecord]) -> None:
    if not records:
        return

    def _append() -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.path, "a", encoding="utf-8") as handle:
            handle.writelines(
                record.model_dump_json(exclude_none=True) + "\n"
                for record in records
            )

    async with self._lock:
        await asyncio.to_thread(_append)

MemoryQuarantineSink

Keeps records in memory; the default when no file is given.

Source code in lpgcast/sink.py 
74
75
76
77
78
79
80
81
class MemoryQuarantineSink:
    """Keeps records in memory; the default when no file is given."""

    def __init__(self) -> None:
        self.records: list[QuarantineRecord] = []

    async def write_records(self, records: list[QuarantineRecord]) -> None:
        self.records.extend(records)

Attributes

records = [] instance-attribute

Methods:

__init__()
Source code in lpgcast/sink.py 
77
78
def __init__(self) -> None:
    self.records: list[QuarantineRecord] = []
write_records(records) async
Source code in lpgcast/sink.py 
80
81
async def write_records(self, records: list[QuarantineRecord]) -> None:
    self.records.extend(records)

QuarantineRecord

Bases: BaseModel

One quarantined item.

Source code in lpgcast/sink.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class QuarantineRecord(BaseModel):
    """One quarantined item."""

    record: Literal["finding", "episode", "unmapped"]
    source_id: str
    kind: str = Field(
        description="Finding kind, episode failure kind, or ``unmapped``."
    )
    detail: str | None = None
    path: str | None = Field(default=None, description="Location in the reply.")
    message: str = ""
    quote: str | None = None
    char_start: int | None = None
    char_end: int | None = None
    fragment: dict[str, Any] | None = Field(
        default=None, description="The item as the model wrote it."
    )
    attempts: int = Field(default=0, description="Model requests made for the episode.")
    raw_output: str | None = Field(
        default=None, description="The reply the item came from."
    )
    schema_ref: str | None = None

Attributes

attempts = Field(default=0, description='Model requests made for the episode.') class-attribute instance-attribute
char_end = None class-attribute instance-attribute
char_start = None class-attribute instance-attribute
detail = None class-attribute instance-attribute
fragment = Field(default=None, description='The item as the model wrote it.') class-attribute instance-attribute
kind = Field(description='Finding kind, episode failure kind, or ``unmapped``.') class-attribute instance-attribute
message = '' class-attribute instance-attribute
path = Field(default=None, description='Location in the reply.') class-attribute instance-attribute
quote = None class-attribute instance-attribute
raw_output = Field(default=None, description='The reply the item came from.') class-attribute instance-attribute
record instance-attribute
schema_ref = None class-attribute instance-attribute
source_id instance-attribute

QuarantineSink

Bases: Protocol

Where quarantined records go.

Source code in lpgcast/sink.py 
43
44
45
46
47
48
@runtime_checkable
class QuarantineSink(Protocol):
    """Where quarantined records go."""

    async def write_records(self, records: list[QuarantineRecord]) -> None:
        """Persist *records*; safe to call from concurrent tasks."""

Methods:

write_records(records) async

Persist records; safe to call from concurrent tasks.

Source code in lpgcast/sink.py 
47
48
async def write_records(self, records: list[QuarantineRecord]) -> None:
    """Persist *records*; safe to call from concurrent tasks."""