論文 PDF を構造化 JSON にするパイプラインを構成してみる (その1)

はじめに

最近興味本位で AWS 認定の AIP-C01 を取得してみたのですが、それなりに知見は溜まったものの、業務でその辺りに触れる機会はなかなか少なく、せっかくなので雑にありそうなものを作りながらよりリアルな雰囲気を掴んでみたいと思いました。

ちなみに資格取得は本当に興味のある人以外は別にやらなくていいと思うのですが、やってみて思うのは、興味のある人はもちろん、自分が属する組織やチームの運用改善などを考えている人にはお薦めしたいかもしれません。これまでなんとなく雰囲気で使っていたり、理解していたものの解像度が少し上がるので、より実用的なアイディアが生まれる可能性があるかもしれません。

今回取り上げた題材としては、社内に溜まっている運用ドキュメント、設計書、障害対応記録などを構造化して RAG などのソースに使える形にすることを目指し、論文 PDF を構造化 JSON に変換するパイプラインを構成してみたので、その過程を記録できたらと思っています。

※論文データは見出し・表・参考文献・二段組レイアウトのように、構造化を検討する際に扱いたい要素が一通り揃っており今回の検証に適していたため

---

PDF を構造化する理由について

ここでまず PDF を構造化する理由について少し整理しておきたいと思います。

少し乱暴な表現になりますが、PDF は「紙に印刷したときの見え方」を保存する形と認識すると個人的には腑に落ちるものがあり、それを前提に考えると、どこが見出しで、どの数字がどの列のものか、という情報をコンピュータ側で判別するのは難しかったりします。

よって、これらの情報を構造化してあげることで、その文章を章や節といった意味のまとまり (文字数ではなく) として表現でき、後工程で効率良く検索ができたりするなどのメリットがあったりします。

また、構造化して整理しておくことで、文章だけでなく、その文章の出所や、信頼度といったメタデータ的な情報も付加できます。※個人的にはこれが利点だと思っていたり

RAG などのソースデータとして大量の文書を扱う場合でも、意味の単位で切られたデータに出所と信頼度が付いている場合、探しやすく、答えの根拠も示せるような仕組みにすることもできます。この下処理があることで、質問に関係する節や表だけを LLM に渡せるため、文書を丸ごと渡す場合と比べて入力トークンを抑えられ、回答の質だけでなくコスト効率も考慮した設計ができたりします。

---

全体構成

以下が構成例です。抽出には Textract と Bedrock を利用しています。※Bedrock は現時点では比較的新しいサービスであるため、今回はこれを試したい動機もありました

• S3 に PDF を置くと EventBridge が Step Functions を起動し、validator → preprocessor → 抽出 (2 経路) → finalizer の順に Lambda が処理するような形にしています。
• 制御 (順序・分岐・リトライ) は Step Functions、処理は Lambda、抽出は Textract と Bedrock、生成したデータや状態管理は S3 と DynamoDB が担います。

---

2 つの抽出経路

前提として、Textract は日本語に対応していないため、日本語 PDF は後述の経路 B だけを通ります。英語 PDF は両方の経路を持てる構成にしています。※英語 PDF は Textract 経由と Bedrock 直読で出力にどんな違いが出るかも見たかったため

抽出は Step Functions の Parallel で 2 つの経路に分かれています。どちらも preprocessor が作ったページ画像とテキストレイヤーを入力にし、最終的に同じ形の JSON を出力します。

経路 A: Textract で読み取り、Bedrock で整える

Textract の非同期 API (StartDocumentAnalysis) に PDF を渡し、LAYOUT (見出し・本文・図表の配置) と TABLES (表のセル) を読み取ります。解析は数十秒〜数分かかるため、Lambda の中では待たず、完了を SNS で受け取って同じ Lambda が再起動される形にしています。

読み取った結果 (Block) は Go 側で読み順や表の行列に組み立て直し、座標と確信度を保持します。その上で Bedrock (Claude) には「どれが見出しで、本文がどこからどこまでか、どれが参考文献か」という構造の判断だけを任せ、本文そのものはモデルに書き写させない形にしています。

※座標: その要素がページのどこにあるかを、ページの幅と高さを 1 としたときの左上と右下の位置 (例: [0.54, 0.26, 0.84, 0.46]) で表したもの。後から原本の該当箇所に戻るための手がかりになる

※確信度: Textract が「この文字列をこう読んだ」という結果に対して自分でどれくらい自信があるかを 0〜1 で返す値。低い箇所だけ人が確認する、といった使い方ができる

経路 B: ページ画像を Bedrock に直接読ませる

ページごとの PNG を Step Functions の Map で 1 ページ 1 起動、並列 5 で bedrock-parser に渡します。

各 Lambda は画像と英語の指示文を Bedrock (Claude) に送り、そのページに印字されている書誌・章・図・表・参考文献を tool use で決めた形で返させます。※本来は外部ツールを呼ばせる仕組みですが、今回は出力結果を固定する目的で利用しています

ページ単位の結果は S3 に保存し、最後に finalizer が「見出しの有無」と「前ページの続きかのフラグ」だけでページをまたぐ章や参考文献を機械的に結合します。この段階ではモデルを呼び直しません。

各経路の設定とその理由は、その2 (経路 A)、その3 (経路 B) で詳しく書きたいと思います。

---

S3 のキー設計と jobId

S3 は 1 バケットで、プレフィックスで役割を分けています。

{bucket}/
├── uploads/{jobId}/
│   └── original.pdf            受領した PDF。イベント発火点
├── work/{jobId}/
│   ├── pages/
│   │   └── page-NNNN.png       ページ画像 (A4 1240 x 1754 px)
│   ├── text/
│   │   └── layer.txt           テキストレイヤー
│   ├── textract/
│   │   ├── raw.json            Textract の生出力
│   │   ├── callback.json       完了通知への応答に要る情報 (タスクトークンなど)
│   │   └── document.json       経路 A の正規化前の結果
│   └── bedrock/
│       └── page-NNNN.json      経路 B のページ単位の結果
└── outputs/{jobId}/
    ├── result-textract.json    経路 A の抽出結果
    ├── result-bedrock.json     経路 B の抽出結果
    └── comparison.json         両経路の結果を 1 つにまとめたもの

uploads/ は受け取り、work/ は処理途中の中間物、outputs/ は最終的な成果物です。

Step Functions を起動するイベント通知は uploads/ だけに絞っています。※これを怠ると work/ や outputs/ への書き込みで自分自身が再起動し、無限ループになります

jobId は PDF の SHA-256 から導出しています。同じ PDF を再度アップロードしても同じ jobId に落ちるため、DynamoDB の条件付き書き込み 1 回で二重処理を防げます。

DynamoDB の条件付き書き込みについて

このパイプラインは S3 のイベント通知で起動し、各処理を Lambda で行っているため、同じ PDF に対して処理が 2 回走る可能性があります。※同じファイルを再度アップロードした場合、イベント通知が重複して届いた場合、Lambda が一時的なエラーで再試行された場合など。Lambda は「同じ入力で何度呼ばれても結果が変わらない」ように作るのが前提になります

そのため、処理状況を保持する DynamoDB に「このジョブはもう動いている、または終わっている」かを記録し、2 回目以降は先に進まないようにする必要があります。

DynamoDB には条件付き書き込み (書き込む前に条件を評価し、満たさなければ書き込みを失敗させる) という機能があるので、これを使って「未処理か失敗したジョブだけ登録を通す」ようにしています。

参考: DynamoDB condition expression CLI example - Amazon DynamoDB

以下がジョブを登録する関数です。validator は PDF の検証を行う前にこの関数を呼びます。※2 回目の起動で PDF のダウンロードや検証まで到達させないため

// RegisterJob は条件付き書き込みでジョブを登録する
func (c *Client) RegisterJob(ctx context.Context, jobID, filename string) (Job, error) {
	now := c.now().UTC()
	job := Job{
		JobID:     jobID, // PDF の SHA-256。同じ PDF は同じ jobId になる
		Status:    StatusProcessing,
		Filename:  filename,
		CreatedAt: now,
		UpdatedAt: now,
	}
	item, err := job.item()
	if err != nil {
		return Job{}, err
	}

	_, err = c.api.PutItem(ctx, &awsdynamodb.PutItemInput{
		TableName: new(c.tableName),
		Item:      item,
		// レコードが無い、または status が FAILED のときだけ書き込みを通す
		// 成功済み (COMPLETED / REVIEW_PENDING) と処理中 (PROCESSING) はここで弾かれる
		ConditionExpression: new("attribute_not_exists(#jobId) OR #status = :failed"),
		ExpressionAttributeNames: map[string]string{
			"#jobId":  attrJobID,
			"#status": attrStatus,
		},
		ExpressionAttributeValues: map[string]awsdynamodbtypes.AttributeValue{
			":failed": &awsdynamodbtypes.AttributeValueMemberS{Value: string(StatusFailed)},
		},
		// 弾かれたときに旧レコードをエラーに載せて返してもらう (status を見るための GetItem が要らない)
		ReturnValuesOnConditionCheckFailure: awsdynamodbtypes.ReturnValuesOnConditionCheckFailureAllOld,
	})
	if err != nil {
		// 条件に弾かれた場合は失敗ではなく「既に存在する」として呼び出し側に判断を委ねる
		if condErr, ok := errors.AsType[*awsdynamodbtypes.ConditionalCheckFailedException](err); ok {
			existing, parseErr := jobFromItem(condErr.Item) // ALL_OLD で返った旧レコードを復元する
			return Job{}, &JobExistsError{JobID: jobID, Existing: existing, ExistingErr: parseErr}
		}
		return Job{}, fmt.Errorf("dynamo: put job %s: %w", jobID, err)
	}
	return job, nil
}

ポイントは以下の 2 点です。

1. ConditionExpression は「レコードが無い、または status が FAILED」。この 1 行で「未処理か失敗したものだけ通し、成功済み・処理中は弾く」を成立させています。jobId が PDF の SHA-256 なので、同じ PDF は必ず同じレコードに当たります。
2. ReturnValuesOnConditionCheckFailure: ALL_OLD を付けると、弾かれたときに既存のレコードがエラーに乗って返ってきます。呼び出し側はそれを見て「処理中なら IN_PROGRESS、完了済みなら ALREADY_PROCESSED として SKIPPED」と判定でき、追加の GetItem が要りません。

弾かれた場合も Lambda のエラーにはせず、Step Functions には decision: SKIPPED として返し、Choice で Skipped (Succeed) に流しています。

---

出力される JSON

経路ごとに outputs/{jobId}/result-textract.json / result-bedrock.json を書きます。

JSON はこちらで定義した共通の構造体で決めており、finalizer (Lambda) がどちらの経路の結果もその形に組み立てます。この 2 つが構造化の最終成果物で、RAG などの後工程にはここから節やメタデータを取り出して渡せるようにしています。

骨格だけ抜き出すと以下のようになっています。※配列は 1 要素ずつに省略し、本文は伏せています

{
  "jobId": "3f2a9c...e41b",
  "schemaVersion": "1.0",
  "source": {
    "filename": "original.pdf",
    "sha256": "3f2a9c...e41b",
    "language": "en",
    "pageCount": 12,
    "hasTextLayer": true
  },
  "metadata": {
    "title": "論文の題名",
    "authors": [{ "name": "著者名" }],
    "abstract": "要旨の本文...",
    "year": 2026
  },
  "sections": [
    {
      "level": 1,
      "heading": "1 Introduction",
      "text": "節の本文...",
      "pages": [1, 2]
    }
  ],
  "figures": [
    {
      "id": "figure-1",
      "caption": "Figure 1: ...",
      "page": 1,
      "bbox": [0.54, 0.26, 0.84, 0.46]
    }
  ],
  "tables": [
    {
      "id": "table-1",
      "caption": "Table 1: ...",
      "page": 3,
      "header": [["Threshold", "Model A", "Model B"]],
      "rows": [["0.85", "0.0%", "8"]]
    }
  ],
  "references": [
    {
      "raw": "Author A and Author B. 2020. Title of the paper. In Proceedings of ...",
      "title": "Title of the paper",
      "authors": ["Author A", "Author B"],
      "year": 2020,
      "doi": null
    }
  ],
  "provenance": {
    "route": "textract",
    "confidence": {
      "title": 0.94,
      "sections": 0.85,
      "tables": 0.87,
      "references": 0.99
    },
    "cost": {
      "textractPages": 12,
      "bedrockInputTokens": 14201,
      "bedrockOutputTokens": 3240
    },
    "warnings": ["figure-2: キャプションを特定できなかった"]
  }
}

各キーの意味は以下のとおりです。

書き終えると DynamoDB の状態を COMPLETED (完了) か REVIEW_PENDING (人の確認が要る) にします。信頼度が閾値を下回った項目がある、原本に見つからない値がある、ページが欠けた、のいずれかで REVIEW_PENDING になります。

comparison.json は両経路の結果を 1 つにまとめたもので、後工程には渡さず、2 経路を見比べるための補助ファイルです。骨格は以下のとおりです。※値は伏せています

{
  "jobId": "3f2a9c...e41b",
  "finalizedAt": "2026-08-22T13:00:00Z",
  "status": "REVIEW_PENDING",
  "needsReview": true,
  "routes": {
    "textract": {
      "status": "succeeded",
      "resultKey": "outputs/3f2a9c...e41b/result-textract.json",
      "needsReview": false,
      "durationMs": 50235,
      "cost": { "textractPages": 12, "bedrockInputTokens": 14201, "bedrockOutputTokens": 3240 },
      "warnings": ["..."],
      "report": { "fields": "...", "hallucinations": [], "crossref": "..." }
    },
    "bedrock": {
      "status": "succeeded",
      "resultKey": "outputs/3f2a9c...e41b/result-bedrock.json",
      "error": "PageDecodeError",
      "cause": "...",
      "missingPages": 1,
      "needsReview": true,
      "durationMs": 63216,
      "cost": { "bedrockInputTokens": 33726, "bedrockOutputTokens": 11711 },
      "warnings": ["..."],
      "report": { "fields": "...", "hallucinations": [], "crossref": "..." }
    }
  },
  "diff": {
    "title": { "textract": "...", "bedrock": "...", "equal": true },
    "authors": { "textract": ["..."], "bedrock": ["..."], "equal": true },
    "abstract": { "textract": "...", "bedrock": "...", "equal": false },
    "counts": {
      "sections": { "textract": 0, "bedrock": 0, "equal": false },
      "figures": { "textract": 0, "bedrock": 0, "equal": true },
      "tables": { "textract": 0, "bedrock": 0, "equal": true },
      "references": { "textract": 0, "bedrock": 0, "equal": true }
    },
    "headings": { "textract": ["..."], "bedrock": ["..."], "equal": false }
  }
}

---

diff に関しては、現時点ではどちらが正しいかを判定せず、並べて equal を付けるだけです。どちらの経路がどの項目で正確かの評価は、別の機会に扱えたらと思っています。

---

おわりに

今回は PDF を構造化する目的と、パイプラインのざっくりな全体像を整理しました。

次回は経路 A (Textract → Bedrock) の詳細な内容を書いていきたいと思います。