1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416"""Review checklist models for human review interface.
These models define the human review workflow with multiple choice options
and free input. ReviewItems are generated by specialized agents when they
need human judgment (lower confidence or semantic decisions).
Key design decisions:
- Category is NOT stored on ReviewItem - derived from linked Observation
- Options include implicit "Other" for custom input
- search_text field enables replacement after human decision
"""
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Literal
from uuid import uuid4
from pydantic import BaseModel, ConfigDict, Field
if TYPE_CHECKING:
from .observation import Observation
class ReviewOption(BaseModel):
"""One option in a review item (multiple choice).
Each ReviewItem has 2-4 predefined options. The UI implicitly adds
an "Other" option that allows custom text input.
Attributes:
id: UUID for this option
label: Display text for this option
action: What happens if selected (replace, keep, skip, other)
replacement_text: For 'replace' actions, the text to use
is_recommended: Agent's top choice (shown first in UI)
Example:
>>> option = ReviewOption(
... label="Yes, replace with 'Enzo'",
... action="replace",
... replacement_text="Enzo",
... is_recommended=True
... )
"""
id: str = Field(
default_factory=lambda: str(uuid4()),
description="UUID for this option"
)
label: str = Field(
...,
description="Display text for this option"
)
action: Literal["replace", "keep", "skip", "other"] = Field(
...,
description="What happens if selected"
)
replacement_text: str | None = Field(
default=None,
description="For 'replace' actions, the text to use"
)
is_recommended: bool = Field(
default=False,
description="Agent's top choice (shown first in UI)"
)
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "opt-550e8400-e29b-41d4-a716-446655440000",
"label": "Yes, replace with 'Enzo'",
"action": "replace",
"replacement_text": "Enzo",
"is_recommended": True
}
}
)
class ReviewItem(BaseModel):
"""Item needing human decision - multiple choice with free input.
ReviewItems are generated by specialized agents when they need human
judgment. Each item presents a question with predefined options.
IMPORTANT: category is NOT stored on ReviewItem. It's derived from
the linked Observation at checklist construction time via
ReviewChecklist.from_items_and_observations().
Attributes:
id: UUID for this review item
observation_id: Links to the observation
agent: Which agent generated this item
question: Human-readable question
options: Predefined choices (2-4 options, "Other" implicit)
search_text: Exact text to find in markdown for replacement
context: Surrounding text ~500 chars
page_num: Page number where issue occurs
visual_context_url: S3 URL to page image snippet
agent_recommendation: What agent thinks should happen (glass box)
agent_confidence: Agent's confidence in recommendation
selected_option_id: Human's selected option (filled after review)
custom_input: If 'Other' selected, the custom text
reviewed_by: Who reviewed this item
reviewed_at: When review occurred
Example:
>>> item = ReviewItem(
... observation_id="obs-123",
... agent="typography",
... question="Is 'Exxon' a typo for 'Enzo'?",
... options=[
... ReviewOption(label="Yes, replace", action="replace", replacement_text="Enzo"),
... ReviewOption(label="No, keep 'Exxon'", action="keep")
... ],
... search_text="Exxon",
... context="Both Exxon and yt are developed...",
... page_num=3,
... agent_recommendation="Likely OCR error - document discusses Enzo project",
... agent_confidence=0.85
... )
"""
id: str = Field(
default_factory=lambda: str(uuid4()),
description="UUID for this review item"
)
observation_id: str = Field(
...,
description="Links to the observation"
)
agent: str = Field(
...,
description="Which agent generated this item"
)
# NOTE: category is NOT on ReviewItem - derived from linked Observation
# at checklist construction time via from_items_and_observations()
# The question
question: str = Field(
...,
description="Human-readable question"
)
# Multiple choice options (2-4 options, "Other" implicit in UI)
options: list[ReviewOption] = Field(
...,
min_length=2,
max_length=4,
description="Predefined choices"
)
# For applying replacements
search_text: str = Field(
...,
description="Exact text to find in markdown for replacement"
)
# Context for decision
context: str = Field(
...,
description="Surrounding text ~500 chars"
)
page_num: int = Field(
...,
ge=1,
description="Page number where issue occurs"
)
visual_context_url: str | None = Field(
default=None,
description="S3 URL to page image snippet"
)
# Agent's recommendation (glass box)
agent_recommendation: str = Field(
...,
description="What agent thinks should happen"
)
agent_confidence: float = Field(
...,
ge=0.0,
le=1.0,
description="Agent's confidence in recommendation"
)
# Human's decision (filled after review)
selected_option_id: str | None = Field(
default=None,
description="ID of selected option"
)
custom_input: str | None = Field(
default=None,
description="If 'Other' selected, the custom text"
)
reviewed_by: str | None = Field(
default=None,
description="Who reviewed this item"
)
reviewed_at: datetime | None = Field(
default=None,
description="When review occurred"
)
def submit_review(
self,
option_id: str | None,
custom_input: str | None,
reviewed_by: str,
) -> None:
"""Submit a human review decision.
Args:
option_id: ID of selected option (None if custom input only)
custom_input: Custom text if "Other" was selected
reviewed_by: Identifier of the reviewer
Raises:
ValueError: If neither option_id nor custom_input provided
"""
if option_id is None and custom_input is None:
raise ValueError("Must provide either option_id or custom_input")
self.selected_option_id = option_id
self.custom_input = custom_input
self.reviewed_by = reviewed_by
self.reviewed_at = datetime.now(UTC)
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "ri-550e8400-e29b-41d4-a716-446655440000",
"observation_id": "obs-550e8400-e29b-41d4-a716-446655440001",
"agent": "typography",
"question": "Is 'Exxon' a typo for 'Enzo'?",
"options": [
{
"id": "opt-1",
"label": "Yes, replace with 'Enzo'",
"action": "replace",
"replacement_text": "Enzo",
"is_recommended": True
},
{
"id": "opt-2",
"label": "No, 'Exxon' is correct",
"action": "keep",
"is_recommended": False
}
],
"search_text": "Exxon",
"context": "...Both Exxon and yt are developed using...",
"page_num": 3,
"visual_context_url": None,
"agent_recommendation": (
"This appears to be an OCR error. The document discusses the "
"'Enzo' project, not the oil company 'Exxon'."
),
"agent_confidence": 0.85,
"selected_option_id": None,
"custom_input": None,
"reviewed_by": None,
"reviewed_at": None
}
}
)
class ReviewChecklist(BaseModel):
"""Human review interface - exposed via API.
Contains all items needing human review, grouped by category, agent,
and page for easy navigation in the UI.
IMPORTANT: Use from_items_and_observations() to construct this model.
It derives categories from linked Observations and excludes items
whose observations are already closed.
Attributes:
items: All items needing review
summary: Quick summary (e.g., "5 items need review: 2 figures, 2 OCR")
by_category: Items grouped by category (derived from Observations)
by_agent: Items grouped by agent
by_page: Items grouped by page number
total_items: Total number of items
critical_items: Items with confidence < 0.7
completed_items: Items that have been reviewed
Example:
>>> checklist = ReviewChecklist.from_items_and_observations(
... items=[item1, item2],
... observations=[obs1, obs2]
... )
>>> print(checklist.summary)
"2 items need review: 1 figures, 1 typography"
"""
# All items needing review
items: list[ReviewItem] = Field(
default_factory=list,
description="All items needing review"
)
# Quick summary
summary: str = Field(
default="",
description="e.g., '5 items need review: 2 figures, 2 OCR, 1 table'"
)
# Grouped for UI navigation
# NOTE: by_category is derived from linked Observations at construction time
by_category: dict[str, list[ReviewItem]] = Field(
default_factory=dict,
description="Items grouped by category (from linked Observations)"
)
by_agent: dict[str, list[ReviewItem]] = Field(
default_factory=dict,
description="Items grouped by agent"
)
by_page: dict[int, list[ReviewItem]] = Field(
default_factory=dict,
description="Items grouped by page number"
)
# Stats
total_items: int = Field(
default=0,
description="Total number of items"
)
critical_items: int = Field(
default=0,
description="Items with confidence < 0.7"
)
completed_items: int = Field(
default=0,
description="Items that have been reviewed"
)
@classmethod
def from_items_and_observations(
cls,
items: list[ReviewItem],
observations: list["Observation"],
) -> "ReviewChecklist":
"""Build checklist from items with category derived from linked observations.
Category is NOT stored on ReviewItem - it's derived at checklist construction
by looking up the linked Observation. This ensures single source of truth.
NOTE: Items whose linked Observation is already closed (status="closed") are
excluded from the checklist. Closed observations don't need review - they've
already been resolved (e.g., via auto-correction).
Args:
items: List of ReviewItems to include
observations: List of Observations for category lookup
Returns:
ReviewChecklist with items grouped by category, agent, and page
"""
# Build observation lookup
obs_by_id = {obs.id: obs for obs in observations}
by_category: dict[str, list[ReviewItem]] = {}
by_agent: dict[str, list[ReviewItem]] = {}
by_page: dict[int, list[ReviewItem]] = {}
included_items: list[ReviewItem] = []
for item in items:
# Look up linked observation
obs = obs_by_id.get(item.observation_id)
# Skip items whose observation is already closed (no review needed)
if obs and obs.status == "closed":
continue
# Derive category from linked observation
category = obs.category if obs else "unknown"
by_category.setdefault(category, []).append(item)
by_agent.setdefault(item.agent, []).append(item)
by_page.setdefault(item.page_num, []).append(item)
included_items.append(item)
# Build summary
parts = [f"{len(v)} {k}" for k, v in by_agent.items()]
summary = f"{len(included_items)} items need review: {', '.join(parts)}" if parts else "No items need review"
return cls(
items=included_items,
summary=summary,
by_category=by_category,
by_agent=by_agent,
by_page=by_page,
total_items=len(included_items),
critical_items=sum(1 for i in included_items if i.agent_confidence < 0.7),
completed_items=sum(1 for i in included_items if i.reviewed_at is not None),
)
model_config = ConfigDict(
json_schema_extra={
"example": {
"items": [],
"summary": "3 items need review: 1 figures, 2 typography",
"by_category": {"alt_text": [], "ocr": []},
"by_agent": {"figures": [], "typography": []},
"by_page": {1: [], 3: []}, # type: ignore[dict-item] # int keys valid at runtime
"total_items": 3,
"critical_items": 0,
"completed_items": 0
}
}
)
__all__ = ["ReviewOption", "ReviewItem", "ReviewChecklist"]