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"""Observation models for the remediation pipeline.
Observations are the atomic unit of the remediation pipeline. They represent
discrepancies between visual presentation and semantic markup, generated by:
- Analysis agent (initial document scan)
- Specialized agents (figures, tables, structure, typography)
- Human reviewers (manual observations)
Lifecycle (simplified to 2 fields):
- status: "open" or "closed"
- resolution: "fixed", "kept_original", or "skipped" (only when closed)
The ProcessingTrace captures full details of who/when/how.
"""
from datetime import UTC, datetime
from typing import Literal
from uuid import uuid4
from pydantic import BaseModel, ConfigDict, Field
class ObservationLocation(BaseModel):
"""Location reference for an observation.
Identifies where in the document an observation applies. Supports three
location types for different use cases:
- element: CSS-like selector for specific element
- range: Line range in markdown (e.g., "10-15")
- region: Human-readable description of region
Attributes:
location_type: Type of location reference
value: The actual location reference (selector, range, or description)
page_num: 1-indexed page number where observation occurs
Example:
>>> loc = ObservationLocation(
... location_type="element",
... value="img[alt='']",
... page_num=3
... )
"""
location_type: Literal["element", "range", "region"] = Field(
default="region",
description="Type of location reference"
)
value: str = Field(
...,
min_length=1,
description="Selector, line range, or region description"
)
page_num: int = Field(..., ge=1, description="1-indexed page number")
model_config = ConfigDict(
json_schema_extra={
"example": {
"location_type": "element",
"value": "img[alt='']",
"page_num": 3
}
}
)
class Observation(BaseModel):
"""A discrepancy between visual presentation and semantic markup.
Observations are the atomic unit of the remediation pipeline.
They can be generated by:
- Analysis agent (initial document scan)
- Specialized agents (figures, tables, structure, typography)
- Human reviewers (manual observations)
Observations do NOT map directly to WCAG criteria - that mapping
can be derived later for compliance reporting.
SIMPLIFIED LIFECYCLE (2 fields only):
- status: "open" or "closed"
- resolution: What was the outcome? (only set when closed)
- "fixed": Correction was applied (auto or human)
- "kept_original": Human chose to keep as-is
- "skipped": Force-applied with unreviewed items
The ProcessingTrace captures full details of who/when/how.
Attributes:
id: UUID for this observation
job_id: Associated job ID
agent: Agent that created this observation
source: Whether created by agent or human
visual_description: What is visually presented in the PDF
markup_description: How it's currently represented in markup
location: Where in the document this applies
affected_pages: For issues spanning multiple pages
confidence: Agent's confidence in this observation (0.0-1.0)
severity: Impact severity (critical, major, minor)
category: Category for grouping (single source of truth)
status: "open" or "closed"
resolution: Outcome when closed
created_at: When observation was created
human_comment: Comment from human when submitting observation
Example:
>>> obs = Observation(
... job_id="job-123",
... agent="figures",
... visual_description="Image shows flowchart with 5 steps",
... markup_description="Image has empty alt text",
... location=ObservationLocation(
... location_type="element",
... value="img[src='figure-1.png']",
... page_num=3
... ),
... category="alt_text",
... confidence=0.9,
... severity="major"
... )
"""
# Identification
id: str = Field(
default_factory=lambda: str(uuid4()),
description="UUID for this observation"
)
job_id: str = Field(..., description="Associated job ID")
# Source
agent: str = Field(
...,
description="Agent that created: figures, tables, structure, typography"
)
source: Literal["agent", "human"] = Field(
default="agent",
description="Whether created by AI agent or human reviewer"
)
# What the AI/human perceived
visual_description: str = Field(
...,
min_length=1,
description="What is visually presented in the PDF"
)
markup_description: str = Field(
...,
min_length=1,
description="How it's currently represented in markup"
)
# Location
location: ObservationLocation
affected_pages: list[int] = Field(
default_factory=list,
description="For issues spanning multiple pages"
)
# Assessment
confidence: float = Field(
default=0.8,
ge=0.0,
le=1.0,
description="Agent's confidence in this observation"
)
severity: Literal["critical", "major", "minor"] = Field(
default="major",
description="Impact severity of this observation"
)
# Category for grouping (single source of truth - ReviewItem derives from this)
category: str = Field(
default="general",
description="Category: reading_order, heading, ocr, formatting, alt_text, table, etc."
)
# SIMPLIFIED LIFECYCLE (2 fields only)
# ProcessingTrace already captures who/when/how
status: Literal["open", "closed"] = Field(
default="open",
description="Is this observation resolved?"
)
resolution: Literal["fixed", "kept_original", "skipped"] | None = Field(
default=None,
description="What was the outcome? Only set when status='closed'"
)
# Metadata
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
human_comment: str | None = Field(
default=None,
description="Comment from human when submitting observation"
)
def close(self, resolution: Literal["fixed", "kept_original", "skipped"]) -> None:
"""Close this observation with the given resolution.
Args:
resolution: The outcome
- "fixed": Correction was applied (auto or human)
- "kept_original": Human chose to keep as-is
- "skipped": Force-applied with unreviewed items
Raises:
ValueError: If observation is already closed
"""
if self.status == "closed":
raise ValueError("Observation already closed")
self.status = "closed"
self.resolution = resolution
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "550e8400-e29b-41d4-a716-446655440000",
"job_id": "job-123-uuid",
"agent": "figures",
"source": "agent",
"visual_description": "Image shows a flowchart with 5 connected boxes showing registration process",
"markup_description": "Image has empty alt text: ",
"location": {
"location_type": "element",
"value": "img[src='figure-1.png']",
"page_num": 3
},
"affected_pages": [3],
"confidence": 0.92,
"severity": "major",
"category": "alt_text",
"status": "open",
"resolution": None,
"created_at": "2024-12-10T10:30:00Z",
"human_comment": None
}
}
)