Skip to content

Token Usage Tracking

ExtractionResult.usage contains raw provider usage objects grouped by schema-generation and extraction calls, plus computed operation totals. See the Token Tracking Guide for examples.

ExtractorUsage

Raw provider usage objects grouped by extraction step.

Source code in structx/utils/usage.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class ExtractorUsage(BaseModel):
    """Raw provider usage objects grouped by extraction step."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    steps: Dict[ExtractionStep, List[Any]] = Field(default_factory=dict)
    _lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)

    @field_serializer("steps")
    def serialize_steps(
        self, steps: Dict[ExtractionStep, List[Any]]
    ) -> Dict[str, List[Any]]:
        with self._lock:
            return {step.value: list(usages) for step, usages in steps.items()}

    def add_step_usage(self, step: ExtractionStep, usage: Any) -> None:
        """Associate a provider usage object with a pipeline step."""
        if usage is not None:
            with self._lock:
                self.steps.setdefault(step, []).append(usage)

    def merge(self, other: "ExtractorUsage") -> None:
        """Append another tracker's raw usages while preserving step order."""
        for step in ExtractionStep:
            for usage in other.get_step(step):
                self.add_step_usage(step, usage)

    def get_step(self, step: Union[ExtractionStep, str]) -> List[Any]:
        """Return the raw usage objects recorded for a pipeline step."""
        step = ExtractionStep(step) if isinstance(step, str) else step
        with self._lock:
            return list(self.steps.get(step, []))

    def _all_usages(self) -> Iterable[Any]:
        with self._lock:
            return [usage for usages in self.steps.values() for usage in usages]

    @staticmethod
    def _prompt_tokens(usage: Any) -> int:
        return _token_count(usage, "prompt_tokens", "input_tokens")

    @staticmethod
    def _completion_tokens(usage: Any) -> int:
        return _token_count(usage, "completion_tokens", "output_tokens")

    @classmethod
    def _total_tokens(cls, usage: Any) -> int:
        total = _optional_token_count(usage, "total_tokens")
        if total is not None:
            return total
        return cls._prompt_tokens(usage) + cls._completion_tokens(usage)

    @staticmethod
    def _reasoning_tokens(usage: Any) -> Optional[int]:
        details = _read_value(
            usage, "completion_tokens_details", "output_tokens_details"
        )
        nested = _optional_token_count(details, "reasoning_tokens", "thinking_tokens")
        return (
            nested
            if nested is not None
            else _optional_token_count(usage, "reasoning_tokens", "thinking_tokens")
        )

    @staticmethod
    def _cached_tokens(usage: Any) -> Optional[int]:
        details = _read_value(usage, "prompt_tokens_details", "input_tokens_details")
        nested = _optional_token_count(
            details,
            "cached_tokens",
            "cache_read_tokens",
            "cache_read_input_tokens",
        )
        return (
            nested
            if nested is not None
            else _optional_token_count(
                usage,
                "cached_tokens",
                "cache_read_tokens",
                "cache_read_input_tokens",
            )
        )

    @computed_field
    @property
    def total_tokens(self) -> int:
        return sum(self._total_tokens(usage) for usage in self._all_usages())

    @computed_field
    @property
    def prompt_tokens(self) -> int:
        return sum(self._prompt_tokens(usage) for usage in self._all_usages())

    @computed_field
    @property
    def completion_tokens(self) -> int:
        return sum(self._completion_tokens(usage) for usage in self._all_usages())

    @computed_field
    @property
    def thinking_tokens(self) -> Optional[int]:
        values = [
            value
            for usage in self._all_usages()
            if (value := self._reasoning_tokens(usage)) is not None
        ]
        return sum(values) if values else None

    @computed_field
    @property
    def cached_tokens(self) -> Optional[int]:
        values = [
            value
            for usage in self._all_usages()
            if (value := self._cached_tokens(usage)) is not None
        ]
        return sum(values) if values else None

total_tokens property

prompt_tokens property

completion_tokens property

thinking_tokens property

cached_tokens property

get_step(step)

Return the raw usage objects recorded for a pipeline step.

Source code in structx/utils/usage.py 
72
73
74
75
76
def get_step(self, step: Union[ExtractionStep, str]) -> List[Any]:
    """Return the raw usage objects recorded for a pipeline step."""
    step = ExtractionStep(step) if isinstance(step, str) else step
    with self._lock:
        return list(self.steps.get(step, []))

ExtractionStep

A model-backed step in the extraction pipeline.

Source code in structx/utils/usage.py 
15
16
17
18
19
class ExtractionStep(Enum):
    """A model-backed step in the extraction pipeline."""

    SCHEMA_GENERATION = "schema_generation"
    EXTRACTION = "extraction"

Only schema_generation and extraction are tracked. With a custom model, schema generation is absent. Each RowResult also exposes the extraction usage for its own input row.