Skip to content

lpgcast.extract

One extraction call: an episode and a slice in, the model's flat reply out.

The reply is parsed into the wire models below and nothing more. Whether a type exists, a property belongs to its vertex, a value fits its field or a quote is in the text are checked afterwards; the wire models only make sure the reply has the flat shape.

Attributes

CHARS_PER_TOKEN = 4 module-attribute

INSTANCE_SCHEMA_NAME = 'lpg_instance' module-attribute

Scalar = str | int | float | bool module-attribute

Classes

EpisodeTooLongError

Bases: ValueError

An episode over max_episode_tokens; episodes are never chunked.

Source code in lpgcast/extract.py 
90
91
92
93
94
95
96
97
98
99
class EpisodeTooLongError(ValueError):
    """An episode over ``max_episode_tokens``; episodes are never chunked."""

    def __init__(self, source_id: str, estimated: int, ceiling: int) -> None:
        super().__init__(
            f"Episode {source_id!r} is ~{estimated} tokens, over the ceiling of {ceiling}."
        )
        self.source_id = source_id
        self.estimated = estimated
        self.ceiling = ceiling

Attributes

ceiling = ceiling instance-attribute
estimated = estimated instance-attribute
source_id = source_id instance-attribute

Methods:

__init__(source_id, estimated, ceiling)
Source code in lpgcast/extract.py 
93
94
95
96
97
98
99
def __init__(self, source_id: str, estimated: int, ceiling: int) -> None:
    super().__init__(
        f"Episode {source_id!r} is ~{estimated} tokens, over the ceiling of {ceiling}."
    )
    self.source_id = source_id
    self.estimated = estimated
    self.ceiling = ceiling

Extraction

Bases: BaseModel

One episode's reply, with how it was obtained.

Source code in lpgcast/extract.py 
82
83
84
85
86
87
class Extraction(BaseModel):
    """One episode's reply, with how it was obtained."""

    source_id: str
    wire: WireExtraction
    completion: JsonCompletion = Field(description="Raw reply, mode, cache and usage.")

Attributes

completion = Field(description='Raw reply, mode, cache and usage.') class-attribute instance-attribute
source_id instance-attribute
wire instance-attribute

WireEdge

Bases: BaseModel

An edge the model read, between two refs.

Source code in lpgcast/extract.py 
51
52
53
54
55
56
57
58
59
60
class WireEdge(BaseModel):
    """An edge the model read, between two refs."""

    model_config = ConfigDict(extra="ignore", populate_by_name=True)

    relation: str | None = None
    from_: str = Field(alias="from")
    to: str
    quote: str = ""
    props: list[WireProp] = Field(default_factory=list)

Attributes

from_ = Field(alias='from') class-attribute instance-attribute
model_config = ConfigDict(extra='ignore', populate_by_name=True) class-attribute instance-attribute
props = Field(default_factory=list) class-attribute instance-attribute
quote = '' class-attribute instance-attribute
relation = None class-attribute instance-attribute
to instance-attribute

WireExtraction

Bases: BaseModel

The model's whole reply. Missing lists read as empty; unknown keys are ignored.

Source code in lpgcast/extract.py 
72
73
74
75
76
77
78
79
class WireExtraction(BaseModel):
    """The model's whole reply. Missing lists read as empty; unknown keys are ignored."""

    model_config = ConfigDict(extra="ignore")

    vertices: list[WireVertex] = Field(default_factory=list)
    edges: list[WireEdge] = Field(default_factory=list)
    unmapped: list[WireUnmapped] = Field(default_factory=list)

Attributes

edges = Field(default_factory=list) class-attribute instance-attribute
model_config = ConfigDict(extra='ignore') class-attribute instance-attribute
unmapped = Field(default_factory=list) class-attribute instance-attribute
vertices = Field(default_factory=list) class-attribute instance-attribute

WireProp

Bases: BaseModel

A property the model read: name, value as written, and its quote.

Source code in lpgcast/extract.py 
29
30
31
32
33
34
35
36
class WireProp(BaseModel):
    """A property the model read: name, value as written, and its quote."""

    model_config = ConfigDict(extra="ignore")

    name: str
    value: Scalar | list[Scalar]
    quote: str = ""

Attributes

model_config = ConfigDict(extra='ignore') class-attribute instance-attribute
name instance-attribute
quote = '' class-attribute instance-attribute
value instance-attribute

WireShapeError

Bases: ValueError

A reply that is JSON but not the flat extraction shape.

Source code in lpgcast/extract.py 
102
103
104
105
106
107
108
109
110
111
class WireShapeError(ValueError):
    """A reply that is JSON but not the flat extraction shape."""

    def __init__(self, source_id: str, error: ValidationError, content: str) -> None:
        super().__init__(
            f"Reply for {source_id!r} does not have the extraction shape: {error}"
        )
        self.source_id = source_id
        self.error = error
        self.content = content

Attributes

content = content instance-attribute
error = error instance-attribute
source_id = source_id instance-attribute

Methods:

__init__(source_id, error, content)
Source code in lpgcast/extract.py 
105
106
107
108
109
110
111
def __init__(self, source_id: str, error: ValidationError, content: str) -> None:
    super().__init__(
        f"Reply for {source_id!r} does not have the extraction shape: {error}"
    )
    self.source_id = source_id
    self.error = error
    self.content = content

WireUnmapped

Bases: BaseModel

Something the model saw and could not type.

Source code in lpgcast/extract.py 
63
64
65
66
67
68
69
class WireUnmapped(BaseModel):
    """Something the model saw and could not type."""

    model_config = ConfigDict(extra="ignore")

    quote: str
    note: str = ""

Attributes

model_config = ConfigDict(extra='ignore') class-attribute instance-attribute
note = '' class-attribute instance-attribute
quote instance-attribute

WireVertex

Bases: BaseModel

A vertex the model read.

Source code in lpgcast/extract.py 
39
40
41
42
43
44
45
46
47
48
class WireVertex(BaseModel):
    """A vertex the model read."""

    model_config = ConfigDict(extra="ignore")

    ref: str
    type: str
    quote: str = ""
    mention: bool = False
    props: list[WireProp] = Field(default_factory=list)

Attributes

mention = False class-attribute instance-attribute
model_config = ConfigDict(extra='ignore') class-attribute instance-attribute
props = Field(default_factory=list) class-attribute instance-attribute
quote = '' class-attribute instance-attribute
ref instance-attribute
type instance-attribute

Functions:

estimate_episode_tokens(episode)

Estimated tokens of the text and its context.

Source code in lpgcast/extract.py 
114
115
116
117
def estimate_episode_tokens(episode: Episode) -> int:
    """Estimated tokens of the text and its context."""
    chars = len(episode.text) + sum(len(item.text) for item in episode.context)
    return math.ceil(chars / CHARS_PER_TOKEN)

extract(client, slice_, episode, *, user_instruction=None, max_episode_tokens=None) async

Make the one extraction call for episode against slice_.

Raises:

Type Description
EpisodeTooLongError

before any call, when the episode is over the ceiling.
WireShapeError

the reply is a JSON object of the wrong shape.
Any error of
Source code in lpgcast/extract.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def extract(
    client: LLMClient,
    slice_: Schema,
    episode: Episode,
    *,
    user_instruction: str | None = None,
    max_episode_tokens: int | None = None,
) -> Extraction:
    """Make the one extraction call for *episode* against *slice_*.

    Raises:
        EpisodeTooLongError: before any call, when the episode is over the ceiling.
        WireShapeError: the reply is a JSON object of the wrong shape.
        Any error of :meth:`LLMClient.acomplete_json`.
    """
    if max_episode_tokens is not None:
        estimated = estimate_episode_tokens(episode)
        if estimated > max_episode_tokens:
            raise EpisodeTooLongError(episode.source_id, estimated, max_episode_tokens)
    messages = build_messages(slice_, episode, user_instruction=user_instruction)
    completion = await client.acomplete_json(
        messages, instance_json_schema(slice_), schema_name=INSTANCE_SCHEMA_NAME
    )
    try:
        wire = WireExtraction.model_validate(completion.data)
    except ValidationError as error:
        raise WireShapeError(episode.source_id, error, completion.content) from error
    return Extraction(source_id=episode.source_id, wire=wire, completion=completion)