Skip to content

ontocast.tool.chunk.section_llm

LLM section-label backfill for chunk preparation.

ChunkSectionClassification

Bases: BasePydanticModel

LLM output mapping one excerpt to a canonical section label.

Source code in ontocast/tool/chunk/section_llm.py 
33
34
35
36
37
38
39
class ChunkSectionClassification(BasePydanticModel):
    """LLM output mapping one excerpt to a canonical section label."""

    label: str | None = Field(
        default=None,
        description="Canonical section label or null if not classifiable",
    )

classify_section_with_llm(text, tools, schema, *, document_type_hint=None) async

Classify a text fragment with the section-label LLM prompt.

Source code in ontocast/tool/chunk/section_llm.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def classify_section_with_llm(
    text: str,
    tools: "ToolBox",
    schema: SectionLabelSchema,
    *,
    document_type_hint: str | None = None,
) -> str | None:
    """Classify a text fragment with the section-label LLM prompt."""
    fragment = fragment_for_text(text)
    if not fragment:
        return None
    parser = PydanticOutputParser(pydantic_object=ChunkSectionClassification)
    allowed = ", ".join(canonical_labels(schema))
    prompt = CHUNK_SECTION_CLASSIFICATION_PROMPT.format_prompt(
        allowed_labels=allowed,
        format_instructions=parser.get_format_instructions(),
        document_context=document_type_context(document_type_hint),
        fragment=fragment,
    )
    response = await tools.llm(prompt)
    parsed = parser.parse(response.content or "")
    return normalise_llm_label(parsed.label, schema)

fragment_for_text(text)

Return a short excerpt suitable for LLM section classification.

Source code in ontocast/tool/chunk/section_llm.py 
42
43
44
45
46
47
48
49
def fragment_for_text(text: str) -> str:
    """Return a short excerpt suitable for LLM section classification."""
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            return stripped[:_FRAGMENT_MAX_CHARS]
    snippet = text.strip()
    return snippet[:_FRAGMENT_MAX_CHARS]

llm_backfill_section_labels(segments, tools, *, section_schema_id, document_type_hint=None, section_tag_min_chars=80) async

Set section_label on segments that are still unlabeled (mutates in place).

Source code in ontocast/tool/chunk/section_llm.py 
 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
async def llm_backfill_section_labels(
    segments: list,
    tools: "ToolBox",
    *,
    section_schema_id: str | None,
    document_type_hint: str | None = None,
    section_tag_min_chars: int = 80,
) -> None:
    """Set ``section_label`` on segments that are still unlabeled (mutates in place)."""
    schema_id = resolve_section_schema_id(
        section_schema_id=section_schema_id,
        document_type_hint=document_type_hint,
    )
    schema = load_section_label_schema(schema_id)
    min_chars = max(0, section_tag_min_chars)

    def _needs_llm_backfill(index: int) -> bool:
        segment = segments[index]
        if segment.section_label is not None:
            return False
        text = segment.text.strip()
        fragment = fragment_for_text(segment.text)
        if not fragment:
            return False
        if len(text) >= min_chars:
            return True
        if fragment.lstrip().startswith("#"):
            return True
        return bool(segment.headings)

    unlabeled_indices = [
        index for index in range(len(segments)) if _needs_llm_backfill(index)
    ]
    if not unlabeled_indices:
        return

    worker_limit = max(1, tools.config.server.parallel_workers)
    semaphore = asyncio.Semaphore(worker_limit)

    async def classify_index(index: int) -> tuple[int, str | None]:
        async with semaphore:
            segment = segments[index]
            try:
                label = await classify_section_with_llm(
                    segment.text,
                    tools,
                    schema,
                    document_type_hint=document_type_hint,
                )
                return index, label
            except Exception as exc:
                logger.warning(
                    "LLM section classification failed for segment %s: %s",
                    index,
                    exc,
                )
                return index, None

    results = await asyncio.gather(
        *[classify_index(index) for index in unlabeled_indices]
    )
    for index, label in results:
        if label is not None:
            segments[index].section_label = label