Skip to content

ontocast.agent.convert_document

Document conversion agent for OntoCast.

This module provides functionality for converting various document formats into structured data that can be processed by the OntoCast system.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

convert_document(state, tools)

Convert a single raw input payload on the state into a DoclingDocument.

Strict 1-in / 1-out agent node: state.raw_input must contain exactly one {filename: bytes} entry. Batch fan-out (e.g. JSONL) is the caller's responsibility and must happen before this node is invoked.

Parameters:

Name Type Description Default
state AgentState

The current agent state with a single raw input payload.

 required
tools ToolBox

The toolbox instance providing utility functions.

 required

Returns:

Name Type Description
AgentState AgentState

Updated state with docling_doc populated, or with
AgentState

status == Status.FAILED if conversion could not be performed.
Source code in ontocast/agent/convert_document.py 
 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
def convert_document(state: AgentState, tools: ToolBox) -> AgentState:
    """Convert a single raw input payload on the state into a DoclingDocument.

    Strict 1-in / 1-out agent node: ``state.raw_input`` must contain exactly
    one ``{filename: bytes}`` entry. Batch fan-out (e.g. JSONL) is the caller's
    responsibility and must happen before this node is invoked.

    Args:
        state: The current agent state with a single raw input payload.
        tools: The toolbox instance providing utility functions.

    Returns:
        AgentState: Updated state with ``docling_doc`` populated, or with
        ``status == Status.FAILED`` if conversion could not be performed.
    """
    logger.debug("Converting document")

    state.status = Status.SUCCESS
    raw_input = state.raw_input

    if len(raw_input) != 1:
        logger.error(
            "convert_document expects exactly one raw input entry, received %d",
            len(raw_input),
        )
        state.status = Status.FAILED
        return state

    filename, file_content = next(iter(raw_input.items()))
    file_extension = pathlib.Path(filename).suffix.lower()
    logger.debug("Converting %s with extension %s", filename, file_extension)

    if file_extension in tools.converter.supported_extensions:
        doc = tools.converter(file_content, filename=filename)
        state.set_docling_doc(doc)
        blocked = _fail_when_fixed_catalog_ontology_missing(state)
        return blocked if blocked is not None else state

    if file_extension == ".json":
        result_json = json.loads(file_content.decode("utf-8"))
        json_text = _extract_json_payload_text(result_json, state, filename)
        if json_text is None:
            state.status = Status.FAILED
            return state
        state.set_docling_doc(plain_text_to_docling_doc(json_text, filename))
        blocked = _fail_when_fixed_catalog_ontology_missing(state)
        return blocked if blocked is not None else state

    if file_extension == ".txt":
        text = file_content.decode("utf-8")
        state.set_docling_doc(plain_text_to_docling_doc(text, filename))
        blocked = _fail_when_fixed_catalog_ontology_missing(state)
        return blocked if blocked is not None else state

    logger.error("Unsupported file extension %s for %s", file_extension, filename)
    state.status = Status.FAILED
    return state