連載： NMS開発者がLLMブートキャンプで学んだこと
← 前回 ｜ [次回 →]
前回までは、LLM を外部 API や Agent の流れに接続する話をしてきました。
Tool Calling で NMS API を呼ぶ。
LangGraph で障害対応の流れを作る。
装置情報、過去チケット、Runbook、担当者確認をひとつのワークフローにする。
ここまで来ると、次に必ず出てくるのが「そもそも LLM に読ませたい社内文書をどう準備するのか」という問題です。
NMS やネットワーク運用の現場には、構造化された API データだけがあるわけではありません。
PDF の運用マニュアルがあります。
Word の障害報告書があります。
PowerPoint の設計資料があります。
HTML の製品ページや社内 Wiki もあります。
しかも、それらは人間が読む前提で作られています。LLM がそのまま理解しやすい形ではありません。
今回のテーマは、unstructured を使って PDF・Word・PowerPoint などの非構造文書を LLM/RAG に渡せる形へ変換することです。
背景
授業では、HuggingFace、Fine-tuning、LoRA と並んで、文書処理の実習がありました。
その中で扱った重要なライブラリが unstructured です。
最初は、単に「PDF からテキストを抜くライブラリ」くらいに思っていました。
しかし実際に触ってみると、もう少し違う位置づけだと分かりました。
unstructured は、文書をただの長い文字列として読むのではなく、文書の中にある要素を分解してくれます。
Title
NarrativeText
ListItem
Table
Header
Footer
PageBreak
このような単位に分けてくれるので、RAG の前処理として使いやすくなります。
授業では、まず PDF を対象にしました。
from unstructured.partition.auto import partition
pdf_path = "business_report.pdf"
elements = partition(filename=pdf_path)
さらに PDF 専用の parser も使いました。
from unstructured.partition.pdf import partition_pdf
elements = partition_pdf(filename=pdf_path, strategy="fast")
この elements は、単なる文字列のリストではありません。
それぞれの要素に category や metadata.page_number が付いています。
つまり、LLM に渡す前の段階で「これはタイトル」「これは本文」「これは表」「これは何ページ目の情報」という文脈を持てるようになります。
NMS の運用マニュアルを RAG 化するとき、この差はかなり大きいです。
単純に PDF 全体をテキスト化して分割すると、章タイトル、注意書き、手順、表、フッターが混ざります。
その結果、検索で引っかかった chunk が「ページ番号だけ」「会社名だけ」「表の一部だけ」になることがあります。
しかし、文書要素として扱えば、少なくとも次のような判断ができます。
Header / Footer は除外する
Title を chunk の境界として使う
Table は本文と別の形式で保存する
page_number を metadata に残す
このあたりから、文書処理は RAG の「前処理」ではなく、RAG の品質を決める本体に近いと感じるようになりました。
ここまでの流れを図にすると、次のようになります。
この図で見せたいのは、PDF、Word、PowerPoint をそのまま LLM に投げるのではなく、Elements、Chunks、Metadata という段階を通して、NMS の Runbook 検索や障害対応 Agent が使える知識に変換する流れです。
特に重要なのは、最後の Metadata です。
LLM が「BGP_SESSION_DOWN の対応では interface error を確認してください」と答えたとしても、それがどの文書の何ページに基づくのかが分からなければ、現場では使いにくいです。
source、page_number、category を残しておくことで、回答と元資料をつなげられます。
PDFをただのテキストとして読まない
最初に試したのは、PDF の中身を partition_pdf で Element に分解することでした。
from collections import Counter
from unstructured.partition.pdf import partition_pdf
elements = partition_pdf(filename=pdf_path, strategy="fast")
stats = Counter(el.category for el in elements)
print(len(elements))
print(stats)
授業では、business_report.pdf のようなサンプル PDF を使い、Element の種類と数を確認しました。
自分がここで学んだのは、PDF には「見た目」と「テキスト構造」のズレがあるということです。
人間には、見出し、本文、表、注釈が自然に見えます。
しかしプログラムから見ると、PDF はかなり扱いにくい形式です。
文字の順番が見た目通りとは限りません。
複数カラムの文書では順序が崩れることがあります。
表は表として取れず、ただの行テキストになることがあります。
画像化された PDF では、そもそも文字が取れないこともあります。
この問題を知らずに RAG を作ると、検索や回答の品質が不安定になります。
例えば NMS の運用マニュアルで、次のようなページがあるとします。
BGP_SESSION_DOWN 対応手順
1. peer 状態を確認する
2. interface error を確認する
3. 直近の config change を確認する
注意:
メンテナンス時間中は自動復旧を実行しない
このページが正しく chunk 化されていれば、LLM は「BGP_SESSION_DOWN の対応」と「メンテナンス中の注意」を一緒に参照できます。
しかし chunk が雑に分かれると、対応手順だけが検索され、注意書きが抜けます。
障害対応の文脈では、これは小さな問題ではありません。
LLM の回答がそれっぽくても、必要な注意事項が抜ければ、運用上は危険です。
だから PDF を扱うときは、最初に「どれくらい正しく Element 化できているか」を見るべきだと思いました。
授業では、次のような分析関数も作りました。
from collections import Counter, defaultdict
from unstructured.partition.pdf import partition_pdf
def analyze_document(pdf_path: str, strategy: str = "fast") -> dict:
els = partition_pdf(filename=pdf_path, strategy=strategy)
type_stats = dict(Counter(el.category for el in els))
page_dist = defaultdict(lambda: Counter())
for el in els:
page = int(el.metadata.page_number or 0)
page_dist[page][el.category] += 1
narratives = [el for el in els if el.category == "NarrativeText"]
narratives.sort(key=lambda e: len(str(e)), reverse=True)
top3 = [
{
"length": len(str(el)),
"page": int(el.metadata.page_number or 0),
"text_preview": str(el)[:80],
}
for el in narratives[:3]
]
return {
"total_elements": len(els),
"type_statistics": type_stats,
"page_distribution": {p: dict(c) for p, c in page_dist.items()},
"top3_longest_narratives": top3,
}
これは地味ですが、とても実務的です。
RAG を作る前に、文書がどのように読めているかを可視化できます。
Element 数が異常に少ないなら、PDF が画像化されている可能性があります。
Header や Footer が多すぎるなら、除外ルールが必要かもしれません。
Table が取れていないなら、表抽出の別手段を考える必要があります。
NMS の現場で言えば、いきなり「AI にマニュアルを読ませる」のではなく、まず「マニュアルが機械にどう見えているか」を確認する工程です。
この確認を飛ばすと、あとで回答品質の悪さを LLM のせいにしてしまいます。
しかし実際には、モデルではなく入力文書のパースが壊れているだけ、ということが十分にあり得ます。
fastとhi_resの違い
partition_pdf には strategy があります。
授業で主に扱ったのは fast と hi_res です。
elements_fast = partition_pdf(filename=pdf_path, strategy="fast")
elements_hi = partition_pdf(
filename=pdf_path,
strategy="hi_res",
infer_table_structure=True,
)
fast は名前の通り速いです。
テキストベースの PDF であれば、まずはこちらで十分なことが多いです。
一方、hi_res はレイアウト解析寄りです。
表や複雑なレイアウトを扱いたい場合に検討します。
ただし、hi_res は依存関係が重くなりがちです。
実行時間も長くなります。
ローカル環境やサーバー環境によっては、追加のライブラリやモデルが必要になります。
ここで、自分の中でひとつ整理ができました。
まず fast で全体を読む
表やレイアウトが重要な文書だけ hi_res を試す
さらに表が重要なら PyMuPDF や Camelot も比較する
すべての PDF に対して最初から重い処理をかける必要はありません。
NMS の社内マニュアルでも、文章中心の手順書なら fast で十分かもしれません。
しかし、設計資料や障害報告書では表が重要になることがあります。
例えば次のような情報です。
Interface | Status | Error Count | Action
Gi0/1 | down | 1203 | cable check
Gi0/2 | up | 0 | none
この表がただの文字列として崩れると、検索も回答も不安定になります。
その場合は、表抽出を別ルートで扱ったほうがよいです。
PyMuPDF の find_tables() や Camelot も試しました。
import fitz
doc = fitz.open(PDF_BUSINESS)
all_tables = []
for page_idx, page in enumerate(doc):
tables = page.find_tables()
for table in tables.tables:
df = table.to_pandas()
all_tables.append(
{
"page": page_idx + 1,
"shape": df.shape,
"df": df,
}
)
Camelot では、線がはっきりした表に対して lattice を使う例もありました。
import camelot
tables = camelot.read_pdf(PDF_BUSINESS, pages="all", flavor="lattice")
このあたりは、ひとつのツールで全部を解決するというより、文書の性質に合わせて組み合わせる感覚です。
自分なら、運用マニュアル RAG の取り込みパイプラインを次のように分けます。
文章中心の PDF:
unstructured fast
表が重要な PDF:
unstructured hi_res
PyMuPDF find_tables
Camelot
画像化された PDF:
OCR または Vision model
PowerPoint:
partition_pptx
Word:
partition_docx
この分岐を作っておくと、あとで文書が増えても対応しやすくなります。
Word・PowerPoint・HTMLも同じ入口で扱う
PDF だけではなく、授業では HTML、DOCX、PPTX も扱いました。
from unstructured.partition.html import partition_html
from unstructured.partition.docx import partition_docx
from unstructured.partition.pptx import partition_pptx
html_path = "product_page.html"
docx_path = "quarterly_report.docx"
pptx_path = "investor_deck.pptx"
html_els = partition_html(filename=html_path)
docx_els = partition_docx(filename=docx_path)
pptx_els = partition_pptx(filename=pptx_path)
この部分で面白かったのは、形式が違っても最終的には Element として扱えることです。
PDF、Word、PowerPoint、HTML はファイル形式としてはまったく違います。
しかし RAG の観点では、最終的に必要なのは次のような情報です。
本文テキスト
タイトル
リスト
表
ページ番号またはスライド番号
元ファイル名
unstructured を使うと、この入口をかなり揃えられます。
授業では、形式ごとの Element 数や category を DataFrame で比較しました。
import pandas as pd
from collections import Counter
rows = []
for fmt, els in [
("html", html_els),
("docx", docx_els),
("pptx", pptx_els),
]:
cats = Counter(el.category for el in els)
rows.append({"format": fmt, "total": len(els), **dict(cats)})
df = pd.DataFrame(rows).fillna(0)
この比較は、社内文書を取り込むときにも使えます。
例えば、Word の障害報告書は Title と NarrativeText が多いかもしれません。
PowerPoint の設計資料は、Title と ListItem が多いかもしれません。
HTML の製品ページは、見出し構造が比較的きれいに取れるかもしれません。
形式ごとの特徴を見てから chunking を変えると、RAG の検索品質が上がります。
NMS の業務に置き換えると、文書の種類はだいたい次のように分かれます。
運用マニュアル:
PDF / Word
手順と注意事項が重要
設計資料:
PowerPoint / PDF
構成図、箇条書き、表が重要
障害報告書:
Word / PDF
時系列、原因、対策が重要
製品仕様:
HTML / PDF
見出し、仕様表、バージョン情報が重要
全部を同じ chunk サイズで処理するのではなく、文書タイプごとに処理を変える必要があります。
例えば、障害報告書では「原因」と「再発防止策」を別々の chunk にしすぎると、回答が浅くなります。
設計資料では、スライド単位で metadata を残したほうが、あとで「どの資料の何枚目か」を追いやすくなります。
運用マニュアルでは、見出し単位の chunking がかなり効きます。
このように考えると、unstructured は単にファイルを読む道具ではなく、文書タイプごとの RAG 設計を始めるための入口になります。
chunk_by_titleで意味の境界を残す
文書を Element に分解したあと、次に必要なのが chunking です。
授業では、chunk_by_title と chunk_elements を比較しました。
from unstructured.chunking.title import chunk_by_title
from unstructured.chunking.basic import chunk_elements
chunks_title = chunk_by_title(
elements,
max_characters=1000,
new_after_n_chars=800,
combine_text_under_n_chars=200,
)
chunks_basic = chunk_elements(
elements,
max_characters=500,
overlap=50,
)
chunk_elements は、文字数ベースの chunking に近いです。
一方、chunk_by_title はタイトル境界を意識して chunk を作ります。
RAG では chunk サイズが重要だと何度も学びましたが、ここでさらに分かったのは、サイズだけでは足りないということです。
意味の境界を残す必要があります。
運用マニュアルでは、章タイトルが重要です。
3.2 BGP障害対応
3.2.1 peer down の確認
3.2.2 interface error の確認
3.2.3 経路再広告の注意点
このタイトルを無視して 500 文字ごとに切ると、LLM は「どの手順の話か」を失いやすくなります。
逆に、タイトル境界を意識して chunk を作れば、検索結果に章の文脈が残ります。
授業では、文書の平均長に応じて chunk サイズを変える簡単な関数も作りました。
from unstructured.chunking.title import chunk_by_title
def adaptive_chunk(elements: list) -> list:
if not elements:
return []
avg_len = sum(len(str(e)) for e in elements) / len(elements)
if avg_len >= 200:
max_chars = 1000
elif avg_len >= 50:
max_chars = 500
else:
max_chars = 300
return chunk_by_title(
elements,
max_characters=max_chars,
combine_text_under_n_chars=200,
)
実務で使うなら、これをもう少し拡張したいです。
文書タイプごとに chunking を変えます。
def chunk_for_nms_document(elements: list, doc_type: str) -> list:
if doc_type == "runbook":
return chunk_by_title(
elements,
max_characters=900,
new_after_n_chars=700,
combine_text_under_n_chars=150,
)
if doc_type == "incident_report":
return chunk_by_title(
elements,
max_characters=1200,
new_after_n_chars=900,
combine_text_under_n_chars=200,
)
if doc_type == "slide":
return chunk_by_title(
elements,
max_characters=700,
new_after_n_chars=500,
combine_text_under_n_chars=100,
)
return chunk_by_title(elements, max_characters=800)
障害対応 Runbook は、短く検索しやすい chunk が向いています。
障害報告書は、時系列や原因分析が分断されすぎないほうがよいです。
スライド資料は、1 スライドごとの情報密度が低い場合があるので、短い要素をある程度まとめたほうがよいです。
このように、chunking は「文字数の設定」ではなく「業務文書の読み方の設計」だと考えるほうが自然でした。
cleanerで余計なノイズを落とす
文書から取り出したテキストには、細かいノイズが入ります。
余分な空白。
全角・半角の揺れ。
引用符の揺れ。
ヘッダーやフッター。
ページ番号。
授業では、unstructured.cleaners.core も使いました。
from unstructured.cleaners.core import (
clean,
clean_extra_whitespace,
replace_unicode_quotes,
)
cleaned = clean_extra_whitespace(
replace_unicode_quotes(text)
)
RAG では、ノイズがそのまま embedding に入ります。
人間が読めば無視できるノイズでも、検索品質には影響します。
特に社内文書では、次のような文字列が混ざりがちです。
Confidential
Page 12
Copyright 2026
株式会社...
更新日: ...
これらがすべての chunk に入ると、検索時に似たような chunk が大量に出ることがあります。
NMS の障害対応 RAG で欲しいのは、著作権表記ではなく手順です。
そのため、自分なら前処理で category ベースの除外も入れます。
def filter_elements(elements: list) -> list:
skip_categories = {"Header", "Footer", "PageBreak"}
filtered = []
for el in elements:
if el.category in skip_categories:
continue
text = str(el).strip()
if not text:
continue
if text.lower() in {"confidential", "internal use only"}:
continue
filtered.append(el)
return filtered
ただし、ノイズ除去にも注意が必要です。
消しすぎると、必要な情報まで消えます。
例えば「注意」「禁止」「メンテナンス時間中のみ実施」のような短い文は、文字数だけで見るとノイズに見えることがあります。
しかし障害対応では非常に重要です。
だから cleaner は、強くかけすぎないほうがよいと感じました。
自分なら、最初は次の程度に留めます。
空白を整える
Header / Footer / PageBreak を除く
ページ番号だけの要素を除く
極端に短いが重要語を含む要素は残す
重要語の例は、運用文書なら次のようなものです。
注意
禁止
必須
停止
復旧
再起動
承認
メンテナンス
rollback
do not
must
このような語を含む短文は残す、というルールを入れるだけでも、運用上の安全性が上がります。
LangChain Documentに変換する
Element と chunk ができたら、RAG に渡すために Document に変換します。
授業では、UnstructuredFileLoader も扱いました。
from langchain_community.document_loaders import UnstructuredFileLoader
loader = UnstructuredFileLoader(
pdf_path,
mode="elements",
strategy="fast",
)
documents = loader.load()
mode には single、elements、paged のような違いがあります。
for mode in ["single", "elements", "paged"]:
loader = UnstructuredFileLoader(pdf_path, mode=mode, strategy="fast")
docs = loader.load()
avg = sum(len(d.page_content) for d in docs) / max(len(docs), 1)
print(mode, len(docs), avg)
ただ、実務では自分で Document に変換するほうが制御しやすいと感じました。
特に metadata をきちんと設計したいからです。
from langchain_core.documents import Document
def chunks_to_documents(chunks: list, source: str, doc_type: str) -> list[Document]:
docs = []
for i, chunk in enumerate(chunks):
text = str(chunk).strip()
if not text:
continue
metadata = {
"source": source,
"doc_type": doc_type,
"page": getattr(chunk.metadata, "page_number", None),
"category": getattr(chunk.metadata, "category", None),
"chunk_id": i,
}
docs.append(
Document(
page_content=text,
metadata=metadata,
)
)
return docs
RAG では metadata がかなり重要です。
回答時に出典を返すためです。
回答:
BGP_SESSION_DOWN の一次対応では、peer 状態、interface error、直近の設定変更を確認します。
メンテナンス時間中は自動復旧を実行しないでください。
参照:
- runbook_bgp.pdf p.12
- nms_operation_manual.docx section 3.2
このような回答を作るには、元ファイル名やページ番号が必要です。
「どこに書いてあったか」が出せない RAG は、現場では使いにくいです。
特に障害対応では、担当者が最終判断をします。
LLM の回答だけではなく、元資料に戻れることが大事です。
だから Document の metadata には、最低でも次を入れたいです。
source
doc_type
page または slide_number
category
chunk_id
ingested_at
version
社内マニュアルでは、版数も重要です。
古い手順書と新しい手順書が混ざると危険です。
可能であれば、文書の更新日やバージョンも metadata に入れます。
PDF RAGの最小パイプライン
授業では、PDF を読み、chunking し、embedding し、FAISS で検索し、LLM で回答する関数も作りました。
流れは次の通りです。
PDF
-> partition_pdf
-> Header/Footer除外
-> chunk_by_title
-> LangChain Document
-> Embedding
-> FAISS
-> similarity search
-> LLM answer
コードにすると、だいたい次の形です。
import os
from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.messages import SystemMessage, HumanMessage
def build_docs_from_pdf(pdf_path: str) -> list[Document]:
elements = partition_pdf(filename=pdf_path, strategy="fast")
cleaned = [
el
for el in elements
if el.category not in ["Header", "Footer", "PageBreak"]
]
chunks = chunk_by_title(
cleaned,
max_characters=600,
combine_text_under_n_chars=150,
)
docs = []
for i, chunk in enumerate(chunks):
text = str(chunk).strip()
if not text:
continue
docs.append(
Document(
page_content=text,
metadata={
"source": os.path.basename(pdf_path),
"page": getattr(chunk.metadata, "page_number", None),
"category": getattr(chunk.metadata, "category", None),
"chunk_id": i,
},
)
)
return docs
def answer_from_pdf(pdf_path: str, question: str, k: int = 3) -> dict:
docs = build_docs_from_pdf(pdf_path)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)
hits = vectorstore.similarity_search(question, k=k)
context = "\n\n".join(
f"[p{h.metadata.get('page', '?')}] {h.page_content}"
for h in hits
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
msg = llm.invoke(
[
SystemMessage(
content="与えられたコンテキストに基づいて回答してください。根拠がない場合は不明と言ってください。"
),
HumanMessage(
content=f"コンテキスト:\n{context}\n\n質問: {question}"
),
]
)
return {
"answer": msg.content,
"sources": [
{
"page": h.metadata.get("page", "?"),
"preview": h.page_content[:80],
}
for h in hits
],
}
これは最小構成です。
本番で使うなら、毎回 FAISS を作り直すのではなく、取り込み処理と検索処理を分けます。
また、複数文書を扱うなら、metadata に文書タイプやバージョンを持たせ、検索時に filter できるようにしたいです。
例えば、障害対応 Agent から使うなら、次のように呼びたいです。
result = search_runbook(
query="BGP_SESSION_DOWN peer down interface error",
device_vendor="cisco",
doc_type="runbook",
)
この tool の裏側で、unstructured で処理された文書 chunk を検索します。
前回の LangGraph とつなげると、次のような流れになります。
alertを受信
-> severity分類
-> device情報取得
-> unstructuredで取り込んだRunbookを検索
-> 類似チケットを検索
-> 原因候補を整理
-> 一次対応案を作成
-> 危険操作なら人間承認
ここで unstructured は表に出る主役ではありません。
しかし、Agent が参照する知識ベースの品質を支える重要な部品です。
表はそのまま捨てない
PDF や Word には表がよく出てきます。
ネットワーク運用文書でも表は多いです。
アラーム名 | 重大度 | 確認項目 | 一次対応
装置種別 | OID | MIB名 | 説明
拠点 | 回線ID | ベンダー | 保守窓口
表をただの文章として embedding すると、列の関係が壊れることがあります。
そのため、抽出した DataFrame を Markdown に変換する例もありました。
def table_to_markdown(df, max_rows: int = 10) -> str:
df = df.fillna("")
header = "| " + " | ".join(str(c) for c in df.columns) + " |"
sep = "| " + " | ".join(["---"] * len(df.columns)) + " |"
rows = df.head(max_rows).astype(str).apply(
lambda r: "| " + " | ".join(r) + " |",
axis=1,
).tolist()
out = "\n".join([header, sep] + rows)
if len(df) > max_rows:
out += f"\n... ({len(df) - max_rows} more rows)"
return out
これは RAG でかなり使いやすい形です。
LLM は Markdown table を比較的読みやすいからです。
表を Document にするときは、metadata に content_type="table" を入れるとよいです。
Document(
page_content=table_to_markdown(df),
metadata={
"source": "operation_manual.pdf",
"page": 18,
"content_type": "table",
"table_id": "p18_t1",
},
)
本文 chunk と表 chunk を同じ vector store に入れるか、別 index にするかは用途次第です。
ただし、障害対応では表が回答の根拠になることが多いので、表を捨てるのは危険です。
例えばアラーム一覧表がある場合、LLM が「このアラームは major です」と答えるには、表の行を正しく参照する必要があります。
表が崩れていると、重大度や対応手順を取り違える可能性があります。
この意味で、表処理は見た目以上に重要です。
画像化された情報は別扱いする
PDF や PowerPoint には、画像として埋め込まれた情報もあります。
構成図。
ネットワークトポロジ。
グラフ。
画面キャプチャ。
スキャンされた手順書。
unstructured だけで十分に読めない場合、画像抽出や Vision model を組み合わせる必要があります。
PyMuPDF で画像を抽出し、Vision model に読ませる流れも扱いました。
import base64
from langchain_core.messages import HumanMessage
def analyze_image(image_path, llm_vision):
with open(image_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
msg = HumanMessage(
content=[
{
"type": "text",
"text": "この画像に見えるテキストを抽出し、内容を要約してください。",
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{b64}",
},
},
]
)
return llm_vision.invoke([msg]).content
NMS の資料では、構成図が非常に重要です。
例えば、障害が発生した装置がどの上位ルータにつながっているのか、どの拠点に影響するのかは、文章より図に書かれていることがあります。
この情報を RAG に入れないと、LLM は文書の半分しか読んでいない状態になります。
ただし、図の理解はテキスト抽出より難しいです。
最初から完全自動化を狙うより、次のように段階を分けるほうが現実的だと思います。
1. PDF/DOCX/PPTX のテキスト要素を取り込む
2. 表を Markdown 化して取り込む
3. 画像を抽出してファイルとして保存する
4. 重要な画像だけ Vision model で説明文を作る
5. 画像説明文を Document として index する
6. metadata に元ページ・画像パスを残す
この形なら、テキスト RAG と multimodal 処理を無理なくつなげられます。
5分で試す小さなプロジェクト
ここまでの話は、実際に手元で動かしてみると理解しやすくなります。
そこで、最小構成の実習として「unstructured で NMS Runbook 検索の前処理を作る」スクリプトを作ります。
やることはシンプルです。
PDF / DOCX / PPTX / HTML を読む
-> unstructured で Element に分解する
-> Header / Footer / PageBreak を除外する
-> source / page / category を metadata として残す
-> NMS っぽい query で簡易検索する
Embedding や Vector DB は使いません。
5分から15分で試せるように、まずは lexical search だけにしています。
#nms_unstructured_mini_project.py
from __future__ import annotations
import argparse
import re
from collections import Counter
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
WORKSPACE_DIR = SCRIPT_DIR.parent
DEFAULT_DOCS = [
WORKSPACE_DIR / "10W" / "business_report.pdf",
WORKSPACE_DIR / "10W" / "quarterly_report.docx",
WORKSPACE_DIR / "10W" / "investor_deck.pptx",
]
DEMO_RUNBOOK_HTML = """<!doctype html>
<html>
<head><meta charset="utf-8"><title>NMS Runbook</title></head>
<body>
<h1>BGP_SESSION_DOWN Runbook</h1>
<h2>Primary checks</h2>
<ol>
<li>Check BGP peer state on the affected edge router.</li>
<li>Check interface error counters and optical level.</li>
<li>Check recent configuration changes in the NMS audit log.</li>
</ol>
<h2>Safety rule</h2>
<p>Do not run automatic recovery during a maintenance window. Human approval is required.</p>
<h2>NMS metadata</h2>
<table>
<tr><th>Alarm</th><th>Severity</th><th>Owner</th></tr>
<tr><td>BGP_SESSION_DOWN</td><td>critical</td><td>NOC L2</td></tr>
</table>
</body>
</html>
"""
def ensure_demo_runbook() -> Path:
path = SCRIPT_DIR / "demo_nms_runbook.html"
if not path.exists():
path.write_text(DEMO_RUNBOOK_HTML, encoding="utf-8")
return path
def partition_file(path: Path):
ext = path.suffix.lower()
if ext == ".pdf":
from unstructured.partition.pdf import partition_pdf
return partition_pdf(filename=str(path), strategy="fast")
if ext == ".docx":
from unstructured.partition.docx import partition_docx
return partition_docx(filename=str(path))
if ext == ".pptx":
from unstructured.partition.pptx import partition_pptx
return partition_pptx(filename=str(path))
if ext in {".html", ".htm"}:
from unstructured.partition.html import partition_html
return partition_html(filename=str(path))
from unstructured.partition.auto import partition
return partition(filename=str(path))
def tokenize(text: str) -> set[str]:
return set(re.findall(r"[a-zA-Z0-9_]+", text.lower()))
def parse_documents(paths: list[Path]) -> list[dict]:
records: list[dict] = []
skip_categories = {"Header", "Footer", "PageBreak"}
for path in paths:
if not path.exists():
print(f"[skip] missing: {path}")
continue
print(f"[parse] {path.name}")
elements = partition_file(path)
categories = Counter(getattr(el, "category", el.__class__.__name__) for el in elements)
print(f" elements={len(elements)} categories={dict(categories)}")
for i, element in enumerate(elements):
metadata = getattr(element, "metadata", None)
category = getattr(element, "category", element.__class__.__name__)
text = str(element).strip()
if category in skip_categories or not text:
continue
records.append(
{
"text": text,
"source": path.name,
"category": category,
"page": getattr(metadata, "page_number", None),
"index": i,
}
)
return records
def search_records(records: list[dict], query: str, k: int = 5) -> list[dict]:
query_terms = tokenize(query)
scored = []
for record in records:
score = len(query_terms & tokenize(record["text"]))
if score:
scored.append((score, record))
scored.sort(key=lambda item: item[0], reverse=True)
return [record | {"score": score} for score, record in scored[:k]]
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("paths", nargs="*")
parser.add_argument(
"--query",
default="BGP_SESSION_DOWN critical interface error maintenance approval",
)
parser.add_argument("--no-demo", action="store_true")
args = parser.parse_args()
paths = [Path(p).resolve() for p in args.paths] if args.paths else [p for p in DEFAULT_DOCS if p.exists()]
if not args.no_demo:
paths.append(ensure_demo_runbook())
if not paths:
raise SystemExit("No input documents found.")
try:
records = parse_documents(paths)
except ImportError as exc:
print('Missing dependency. Run: pip install "unstructured[pdf,docx,pptx]"')
raise SystemExit(1) from exc
print(f"\n[index] searchable records={len(records)}")
print(f"[query] {args.query}\n")
for n, hit in enumerate(search_records(records, args.query), start=1):
page = hit["page"] if hit["page"] is not None else "-"
preview = hit["text"].replace("\n", " ")[:220]
print(f"{n}. score={hit['score']} source={hit['source']} page={page} category={hit['category']}")
print(f" {preview}")
if __name__ == "__main__":
main()
実行方法は次の通りです。
pip install "unstructured[pdf,docx,pptx]"
python nms_unstructured_mini_project.py
自分の Runbook や障害報告書を指定する場合は、ファイルパスを渡します。
python nms_unstructured_mini_project.py ./runbook_bgp.pdf ./incident_report.docx
検索語を変える場合は --query を使います。
python nms_unstructured_mini_project.py --query "interface error maintenance approval"
期待される出力は、だいたい次のような形です。
[parse] business_report.pdf
elements=... categories={'Title': ..., 'NarrativeText': ..., 'Table': ...}
[parse] quarterly_report.docx
elements=... categories={'Title': ..., 'NarrativeText': ..., 'ListItem': ...}
[parse] investor_deck.pptx
elements=... categories={'Title': ..., 'ListItem': ...}
[parse] demo_nms_runbook.html
elements=... categories={'Title': ..., 'ListItem': ..., 'NarrativeText': ..., 'Table': ...}
[index] searchable records=...
[query] BGP_SESSION_DOWN critical interface error maintenance approval
1. score=... source=demo_nms_runbook.html page=- category=ListItem
Check interface error counters and optical level.
2. score=... source=demo_nms_runbook.html page=- category=NarrativeText
Do not run automatic recovery during a maintenance window. Human approval is required.
この小さなプロジェクトで確認したいのは、検索アルゴリズムの精度ではありません。
NMS の Runbook や障害報告書を、LLM が扱いやすい単位に分解し、あとで根拠を追える metadata を残すことです。
実務では、この search_records の部分を embedding + vector store に置き換えれば、前回までに作った Tool Calling や LangGraph Agent に接続できます。
NMS業務にどうつなげるか
今回の unstructured 学習を、NMS/SNMP の業務に置き換えると、最初に作りたいのは「社内文書取り込みパイプライン」です。
イメージは次のようなものです。
docs/
runbook/
bgp_operation_manual.pdf
ospf_troubleshooting.docx
design/
tokyo_dc_network_design.pptx
incident/
2026_q1_major_incidents.docx
ingest pipeline:
file scan
format detection
partition
clean
chunk
metadata付与
embedding
vector store保存
文書タイプごとに処理を変えます。
from pathlib import Path
def detect_doc_type(path: Path) -> str:
name = path.name.lower()
if "runbook" in name or "manual" in name:
return "runbook"
if "design" in name or "構成" in name:
return "design"
if "incident" in name or "障害" in name:
return "incident_report"
return "general"
ファイル拡張子で parser を切り替えます。
from unstructured.partition.pdf import partition_pdf
from unstructured.partition.docx import partition_docx
from unstructured.partition.pptx import partition_pptx
from unstructured.partition.html import partition_html
def partition_file(path: Path) -> list:
ext = path.suffix.lower()
if ext == ".pdf":
return partition_pdf(filename=str(path), strategy="fast")
if ext == ".docx":
return partition_docx(filename=str(path))
if ext == ".pptx":
return partition_pptx(filename=str(path))
if ext in {".html", ".htm"}:
return partition_html(filename=str(path))
raise ValueError(f"unsupported file type: {ext}")
そして、metadata を厚めに付けて Document 化します。
def ingest_file(path: Path) -> list[Document]:
doc_type = detect_doc_type(path)
elements = partition_file(path)
elements = filter_elements(elements)
chunks = chunk_for_nms_document(elements, doc_type=doc_type)
return chunks_to_documents(
chunks,
source=path.name,
doc_type=doc_type,
)
このようにして作った vector store は、前回までの Agent から tool として使えます。
@tool
def search_nms_documents(query: str, doc_type: str = "runbook") -> list[dict]:
"""NMS関連文書から、質問に近い手順や説明を検索する。"""
hits = vectorstore.similarity_search(
query,
k=5,
filter={"doc_type": doc_type},
)
return [
{
"text": h.page_content,
"source": h.metadata.get("source"),
"page": h.metadata.get("page"),
"chunk_id": h.metadata.get("chunk_id"),
}
for h in hits
]
これで、障害対応 Agent は社内文書を参照できます。
User:
BGP_SESSION_DOWN が Tokyo DC の edge-router-01 で発生しました。
一次対応を整理してください。
Agent:
1. 装置状態を確認
2. BGP peer 状態を確認
3. interface error を確認
4. 直近の config change を確認
5. Runbook の注意事項を確認
6. メンテナンス時間中なら自動復旧を止める
参照:
- bgp_operation_manual.pdf p.12
- tokyo_dc_network_design.pptx slide 5
ここまで来ると、LLM は単なるチャットではなく、社内文書を読んだ運用支援ツールに近づきます。
まとめ
今回学んだ unstructured は、PDF や Word を LLM に読ませるための地味な道具に見えます。
しかし、RAG や Agent を実務に使うなら、かなり重要な層です。
LLM の回答品質は、モデルだけで決まりません。
どの文書を読むか。
どの形式から読むか。
どの単位で chunk にするか。
metadata をどれだけ残すか。
表や画像をどう扱うか。
この設計で大きく変わります。
特に NMS や障害対応のような業務では、社内文書の多くが PDF、Word、PowerPoint です。
API だけでは取れない知識が、そこにあります。
Runbook。
設計資料。
障害報告書。
ベンダー資料。
保守手順。
これらを LLM に読ませるには、ただテキストを抜くだけでは足りません。
文書の構造をなるべく残し、必要な metadata を付けて、検索しやすい単位に整える必要があります。
今回の学びを一言でまとめるなら、次のようになります。
RAG の品質は、LLM に質問する前の「文書の読ませ方」でかなり決まる。
前回までに作ってきた Tool Calling や LangGraph の Agent は、外部 API を使えるようになりました。
今回の unstructured によって、そこに社内文書という知識源をつなげられます。
次に必要になるのは、テキスト以外の情報です。
構成図。
画面キャプチャ。
音声議事録。
動画。
次回は、Whisper や Multimodal を使って、音声・画像・動画を LLM の入力として扱うところに進みます。