連載： NMS開発者がLLMブートキャンプで学んだこと
前回は、Embedding と Vector DB を中心に、RAG における検索の仕組みを整理しました。
今回は、検索結果を LLM に渡して回答を生成する流れと、その品質をどう管理するかを考えます。
RAG は、構造だけ見ればとてもシンプルです。
質問
↓
関連文書を検索
↓
検索結果を context として LLM に渡す
↓
回答を生成
最初にこの流れを見たとき、「思ったより単純だな」と感じました。
しかし、授業で PDF RAG、検索評価、rerank、LLM-as-Judge まで触ってみると、印象が変わりました。
RAG の実装はシンプルです。
でも、RAG の品質管理はかなり難しいです。
特に NMS や障害対応のような業務では、LLM がきれいな日本語で答えるだけでは足りません。
どの文書を根拠にしたのか。
検索結果は本当に質問に合っているのか。
文書にないことを推測していないか。
古い手順書を使っていないか。
回答が正しく見えても、根拠が間違っていれば危険です。
今回は、NMS 運用マニュアル RAG という例で、検索、出典、評価をどう扱うかを整理します。
背景
NMS の現場で、障害対応中によく起きることがあります。
アラートは出ている。
装置名も分かる。
イベント名も分かる。
でも、次に見るべき手順書や過去チケットにたどり着くまでに時間がかかる。
たとえば、次のようなアラートが出たとします。
2026-03-10 21:14:03 CRITICAL router-core-01 BGP neighbor 10.10.0.2 Down
2026-03-10 21:14:15 CRITICAL nms-poller SNMP timeout router-core-01
2026-03-10 21:14:22 WARNING router-core-01 Interface Gi0/1 input errors increased
運用者が知りたいのは、単なるログの意味ではありません。
この障害では、最初に何を確認すべきか。
どの手順書を見ればよいか。
顧客影響がある可能性はどこか。
過去に似た障害があったか。
これを LLM だけに聞くと、もっともらしい回答は返ってきます。
しかし、LLM は社内マニュアルの最新版を知りません。装置ごとの運用ルールも知りません。NOC の承認フローも知りません。
そこで RAG を使います。
LLM に自由に答えさせるのではなく、社内文書や運用マニュアルから関連する部分を検索し、その context に基づいて回答させる。
この考え方は、NMS 開発者の視点ではかなり自然です。
検索対象は、次のようなものです。
- 障害対応マニュアル
- Runbook
- 過去チケット
- ベンダー手順書
- 監視ルール説明
- FAQ
- 変更作業メモ
ただし、ここで落とし穴があります。
RAG は「LLM に文書を読ませる仕組み」ではありますが、実務では「検索システム」でもあります。
検索が外れれば、回答も外れます。
試したこと
授業では、まず素直な RAG を作りました。
文書を Document にする。FAISS に入れる。質問で top-k 検索する。検索結果をプロンプトに入れる。LLM に回答させる。
PDF RAG の実習では、PDF からテキストを取り出し、chunk に分け、Document の metadata に source、page_number、category、chunk_id を残し、FAISS に入れていました。
RAG の回答関数も、構造はかなり素直です。
def rag_answer(vs, query: str, k: int = 4) -> str:
docs = vs.similarity_search(query, k=k)
context = "\n\n".join(
f"[source={doc.metadata.get('source')}, page={doc.metadata.get('page_number')}]\n"
f"{doc.page_content}"
for doc in docs
)
messages = [
SystemMessage(content="""
あなたはネットワーク運用マニュアルに基づいて回答するアシスタントです。
必ず与えられたコンテキストだけを使って回答してください。
コンテキストにない内容は「文書では確認できません」と答えてください。
""".strip()),
HumanMessage(content=f"""
質問:
{query}
コンテキスト:
{context}
""".strip()),
]
return llm.invoke(messages).content
実装だけ見ると、拍子抜けするくらいです。
検索して、context を作って、プロンプトに入れる。
これが基本形です。
NMS の例に寄せるなら、まず小さな運用マニュアルを作ります。
from langchain_core.documents import Document
manual_docs = [
Document(
page_content="""
BGP neighbor down が発生した場合、まず peer IP への疎通を確認する。
次に BGP state、対象回線のアラート、直近の設定変更、対向装置側の障害情報を確認する。
顧客影響がある場合は、NOC リーダーへエスカレーションする。
""".strip(),
metadata={
"doc_id": "runbook_bgp_down",
"category": "routing",
"severity": "high",
"source": "nms_runbook",
"page": 12,
},
),
Document(
page_content="""
SNMP timeout が発生した場合、監視サーバから対象装置への疎通を確認する。
SNMP community、ACL、装置 CPU 高騰、監視経路の障害を確認する。
装置が ICMP に応答するが SNMP のみ失敗する場合は、SNMP 設定と ACL を優先する。
""".strip(),
metadata={
"doc_id": "runbook_snmp_timeout",
"category": "monitoring",
"severity": "middle",
"source": "nms_runbook",
"page": 18,
},
),
Document(
page_content="""
Interface down が発生した場合、物理リンク状態、光レベル、対向ポート、
ケーブル交換履歴、メンテナンス予定を確認する。
input errors の増加のみの場合は、即時断と判断せず、エラーカウンタ推移を見る。
""".strip(),
metadata={
"doc_id": "runbook_interface_down",
"category": "interface",
"severity": "middle",
"source": "nms_runbook",
"page": 25,
},
),
]
そして、FAISS に入れます。
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vs = FAISS.from_documents(manual_docs, embeddings)
質問します。
query = """
router-core-01 で BGP peer 10.10.0.2 が down し、
同時に SNMP timeout も出ています。最初に何を確認すべきですか？
""".strip()
answer = rag_answer(vs, query, k=3)
print(answer)
期待する回答は、BGP の peer 疎通、BGP state、回線アラート、直近設定変更、SNMP 到達性を分けて確認する、というようなものです。
ここまでで、RAG らしいものは動きます。
ただし、ここからが本題です。
RAG の回答には出典が必要
PDF RAG の実習では、回答だけでなく citation も返す関数を作りました。
これは実務ではかなり重要です。
LLM の回答だけを見ると、正しく見えるかもしれません。
しかし、障害対応では「どの文書のどの部分を根拠にしたのか」を確認できないと怖いです。
そこで、回答と一緒に source を返します。
def rag_answer_with_citations(vs, query: str, k: int = 4) -> dict:
docs = vs.similarity_search(query, k=k)
context = "\n\n".join(
f"[{i}] source={doc.metadata.get('source')} "
f"page={doc.metadata.get('page')} "
f"doc_id={doc.metadata.get('doc_id')}\n"
f"{doc.page_content}"
for i, doc in enumerate(docs, 1)
)
answer = llm.invoke([
SystemMessage(content="""
与えられたコンテキストだけで回答してください。
推測で手順を追加しないでください。
回答の最後に、参照した doc_id を列挙してください。
""".strip()),
HumanMessage(content=f"質問:\n{query}\n\nコンテキスト:\n{context}")
]).content
citations = [
{
"doc_id": doc.metadata.get("doc_id"),
"source": doc.metadata.get("source"),
"page": doc.metadata.get("page"),
"snippet": doc.page_content[:120],
}
for doc in docs
]
return {
"answer": answer,
"citations": citations,
"num_sources": len(citations),
}
この形にすると、NOC 担当者は回答と根拠を一緒に見られます。
回答:
まず BGP peer 10.10.0.2 への疎通、BGP state、対象回線アラート、
直近の設定変更を確認してください。同時に SNMP timeout が出ているため、
監視サーバから対象装置への疎通、SNMP community、ACL、装置 CPU も確認します。
参照:
- runbook_bgp_down
- runbook_snmp_timeout
RAG の業務利用では、この「参照」がかなり大事です。
LLM の答えを信じるのではなく、根拠文書に戻れるようにする。
これは、障害対応だけでなく、法務、金融、社内規定、セキュリティ運用でも同じだと思います。
ここまでの流れを図にすると、次のようになります。上段は「検索で外したまま回答する流れ」、下段は「検索、根拠、評価、人間確認を挟む流れ」です。
NMS 運用マニュアル RAG では、単に「検索して回答する」だけでは足りません。検索結果を評価し、出典を表示し、必要なら回答を抑制して人間確認に回す層が必要です。
検索品質を評価する
授業では、検索結果を precision_at_k、recall_at_k、mrr のような指標で見る練習をしました。
RAG では、この考え方がかなり重要です。
まず、小さくてもよいので評価データを作ります。
eval_dataset = [
{
"query": "BGP peer が down したときの初動は？",
"relevant": ["runbook_bgp_down"],
},
{
"query": "監視サーバからルータに SNMP が通らない",
"relevant": ["runbook_snmp_timeout"],
},
{
"query": "Gi0/1 の input errors が増えている",
"relevant": ["runbook_interface_down"],
},
{
"query": "ping は通るが SNMP だけ応答しない",
"relevant": ["runbook_snmp_timeout"],
},
]
検索を実行して、上位に正しい文書が入っているかを見ます。
def evaluate_retrieval(vs, eval_dataset, k=3):
rows = []
for item in eval_dataset:
docs = vs.similarity_search(item["query"], k=k)
retrieved = [doc.metadata["doc_id"] for doc in docs]
relevant = set(item["relevant"])
hit = any(doc_id in relevant for doc_id in retrieved)
first_rank = next(
(i + 1 for i, doc_id in enumerate(retrieved) if doc_id in relevant),
None
)
rows.append({
"query": item["query"],
"retrieved": retrieved,
"expected": item["relevant"],
"hit@k": hit,
"first_rank": first_rank,
})
return rows
これをやると、RAG の失敗が少し分解できます。
回答が悪いのか。
検索が悪いのか。
検索は合っているが、context のまとめ方が悪いのか。
文書自体に答えがないのか。
この切り分けができないと、RAG の改善はかなり苦しくなります。
NMS の障害対応で言えば、評価データは次のような観点で作りたいです。
- 典型的な障害名で検索する質問
- 自然文で症状を説明する質問
- 装置名や interface 名を含む質問
- Trap 名や OID を含む質問
- 複数障害が混ざった質問
- 文書に答えがない質問
最後の「文書に答えがない質問」も大事です。
RAG は、答えられないときに答えない能力も必要です。
BM25 と Vector search を分けて見る
前回も触れましたが、NMS の文書検索では BM25 と Vector search の両方が欲しくなります。
RAG にすると、この差がさらに重要になります。
bgpBackwardTransition、ifOperStatus、No Such Object、OID、interface 名、装置名のような文字列は、BM25 が強いです。
一方で、「経路が切れた」「監視から見えない」「リンクが不安定」「ping は通るが SNMP が返らない」のような自然文は、Vector search が効きやすいです。
授業では、BM25 retriever と vector retriever を組み合わせる EnsembleRetriever も扱いました。
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever
bm25 = BM25Retriever.from_documents(manual_docs)
bm25.k = 4
vector = vs.as_retriever(search_kwargs={"k": 4})
ensemble = EnsembleRetriever(
retrievers=[vector, bm25],
weights=[0.6, 0.4],
)
実務なら、クエリの種類によって重みを変えたくなります。
Trap 名、OID、装置名が含まれる:
BM25 を強める
自然文の症状説明:
Vector search を強める
両方が含まれる:
Hybrid search で上位候補を広めに取る
ただし、Hybrid search も万能ではありません。
検索結果が増えると、LLM に渡す context も増えます。
top_k を大きくすれば正しい文書が入る可能性は上がりますが、ノイズも増えます。
ここでも評価が必要です。
rerank は便利だが万能ではない
授業では、検索結果をそのまま使うのではなく、rerank で並べ替える考え方を扱いました。
RAG では、最初の retriever で候補を広めに取り、その後で reranker に質問との関連度を見てもらうことがあります。
イメージはこうです。
Vector search / BM25 で候補を 20件取る
↓
reranker で質問との関連度を再評価する
↓
上位 4件だけを LLM に渡す
NMS マニュアル RAG でも、これは効きそうです。
たとえば、質問が次のような場合です。
ping は通るが SNMP だけ応答しない。装置 CPU は高くない。
単純な検索では、SNMP timeout、装置到達性、CPU 高騰、ACL、interface down などが候補に出るかもしれません。
rerank で「ping は通る」「SNMP だけ」「CPU は高くない」という条件を見て、SNMP 設定や ACL の手順を上に持ってきたいです。
ただし、rerank を入れれば必ず良くなるわけではありません。
reranker もモデルです。
ドメイン知識が足りなければ、期待通りに並べ替えられないことがあります。候補文書がそもそも悪ければ、rerank しても正しい文書は出てきません。
また、コストと遅延も増えます。
RAG の品質を上げる方法は、rerank だけではありません。
- chunk を直す
- metadata を増やす
- query を書き換える
- BM25 と Vector を併用する
- top_k と fetch_k を調整する
- スコア閾値を入れる
- 評価データを増やす
rerank は、この中の一つです。
LLM-as-Judge で回答を評価する
授業でのマルチモーダル RAG の資料では、LLM-as-Judge で回答を評価するパターンが出てきました。
評価軸は、たとえば次の2つです。
faithfulness:
回答は与えられた context だけで正当化できるか
relevance:
回答は質問に直接答えているか
RAG の運用では、この考え方がかなり使えます。
from pydantic import BaseModel, Field
class JudgeVerdict(BaseModel):
faithfulness: int = Field(..., ge=1, le=5)
relevance: int = Field(..., ge=1, le=5)
reasoning: str
そして、別の LLM に評価させます。
def judge_answer(query: str, answer: str, sources: list[dict]) -> dict:
judge_llm = llm.with_structured_output(JudgeVerdict)
verdict = judge_llm.invoke([
SystemMessage(content="""
あなたは RAG 回答の評価者です。
faithfulness=5 は、回答が提示された sources だけで完全に正当化できる状態です。
relevance=5 は、質問に直接答えている状態です。
推測や sources にない手順が混ざる場合は faithfulness を下げてください。
""".strip()),
HumanMessage(content=f"""
質問:
{query}
回答:
{answer}
sources:
{sources}
""".strip()),
])
return verdict.model_dump()
もちろん、LLM-as-Judge も絶対ではありません。
評価者も LLM なので、見落としや甘い判定があります。
それでも、RAG の回答を全部人間が見続けるよりは、品質監視の入口になります。
NMS の運用で使うなら、次のようなルールにできます。
faithfulness < 4:
回答をそのまま表示せず、根拠不足として警告する
relevance < 4:
検索クエリの書き換え、再検索、または人間への確認に回す
sources が空:
「文書では確認できません」と返す
ここで大事なのは、LLM-as-Judge を「正解判定の神様」として使わないことです。
運用品質を監視するための signal として使う。
このくらいが現実的だと思いました。
Self RAG と Corrective RAG の考え方
授業での資料では、Adaptive / Self / Corrective RAG を LangGraph で統合する課題がありました。
ここで印象に残ったのは、RAG を一発で終わらせないことです。
通常の RAG は、検索して回答して終わりです。
しかし、Self RAG では、検索結果や回答を自分で評価します。
検索結果は質問に関連しているか
回答は十分な品質か
不十分なら再検索・再生成するか
Corrective RAG では、ローカル検索結果が弱いと判断したら、外部検索や別の情報源で補います。
授業ノートでは、関連度スコアが低い場合に grade="fail" とし、外部検索へフォールバックする流れがありました。
NMS の現場に置き換えると、次のようになります。
1. 社内 Runbook を検索
2. 関連度が低ければ、過去チケットも検索
3. それでも弱ければ、ベンダー資料やナレッジベースを検索
4. 回答の根拠が弱ければ、人間確認に回す
これはかなり現実的です。
障害対応では、社内マニュアルだけで答えられる場合もあります。
しかし、ファームウェア固有の問題、ベンダー障害、過去の特殊対応のようなケースでは、社内 FAQ だけでは足りません。
RAG を「一回の検索で答える仕組み」と見るより、「検索、評価、補正、再回答の流れ」と見る方が実務に近いです。
ただし、ループには上限が必要です。
授業の課題でも refinement_count を持ち、最大2回まで再生成するような設計がありました。
これは大事です。
Agent や RAG の自動改善ループは、止めどころを決めないと運用しづらいです。
NMS 運用マニュアル RAG の流れ
ここまでをまとめると、NMS の運用マニュアル RAG は、いきなりチャットボットとして作るより、段階的に作る方がよさそうです。
最初の形は、検索だけです。
入力:
router-core-01 で BGP peer down と SNMP timeout が同時に出ている
出力:
関連文書:
1. BGP neighbor down 初動
2. SNMP timeout 確認
3. Interface errors 確認
次の段階で、検索結果を LLM に渡して短い回答を作ります。
回答:
まず BGP peer 10.10.0.2 への疎通、BGP state、対象回線アラート、
直近の設定変更を確認してください。
同時に SNMP timeout が出ているため、監視サーバから router-core-01 への疎通、
SNMP community、ACL、装置 CPU の状態も確認してください。
根拠:
- runbook_bgp_down
- runbook_snmp_timeout
さらに、評価を入れます。
retrieval:
hit@3 = true
judge:
faithfulness = 5
relevance = 4
action:
回答を表示。根拠文書も表示。
もし文書が弱ければ、こうします。
retrieval:
hit@3 = false
judge:
faithfulness = 2
relevance = 3
action:
回答を抑制。
「関連する手順書が見つかりません。検索条件を変更するか、NOC リーダーへ確認してください」と表示。
この「回答しない判断」が、実務ではかなり重要です。
LLM は何か答えようとします。
しかし運用システムでは、分からないときに分からないと言える方が安全です。
5分で試す NMS Runbook RAG の品質管理
ここまでの話を、5〜15分で動かせる小さな Python 例にします。
本物の Embedding API や Vector DB は使いません。Python 標準ライブラリだけで、次の流れを再現します。
NMS 障害内容
↓
簡易 Hybrid search
↓
Runbook 検索結果
↓
RAG 風の回答
↓
出典表示
↓
検索結果と回答の簡易品質チェック
これは本番用の検索エンジンではありません。
ただ、RAG の品質管理で何を見るべきかを手元で確認するには十分です。
コード
#nms_rag_quality_check.py
import math
import re
import sys
from collections import Counter
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")
RUNBOOKS = [
{
"doc_id": "runbook_bgp_down",
"title": "BGP neighbor down 初動",
"category": "routing",
"page": 12,
"text": """
BGP neighbor down が発生した場合、まず peer IP への疎通を確認する。
次に BGP state、対象回線のアラート、直近の設定変更、対向装置側の障害情報を確認する。
顧客影響がある場合は、NOC リーダーへエスカレーションする。
""".strip(),
},
{
"doc_id": "runbook_snmp_timeout",
"title": "SNMP timeout 確認",
"category": "monitoring",
"page": 18,
"text": """
SNMP timeout が発生した場合、監視サーバから対象装置への疎通を確認する。
SNMP community、ACL、装置 CPU 高騰、監視経路の障害を確認する。
ping に応答するが SNMP のみ失敗する場合は、SNMP 設定と ACL を優先して確認する。
""".strip(),
},
{
"doc_id": "runbook_interface_errors",
"title": "Interface errors 確認",
"category": "interface",
"page": 25,
"text": """
Interface down や input errors が増加した場合、物理リンク状態、光レベル、対向ポート、
ケーブル交換履歴、メンテナンス予定を確認する。
input errors の増加のみの場合は、即時断と判断せず、エラーカウンタ推移を見る。
""".strip(),
},
{
"doc_id": "runbook_cpu_high",
"title": "CPU 高騰確認",
"category": "resource",
"page": 31,
"text": """
CPU 使用率が高い場合、直近のトラフィック増加、制御プレーンへの負荷、
ログ出力増加、攻撃兆候、SNMP polling の集中を確認する。
CPU 高騰がある場合、監視応答遅延や timeout が発生することがある。
""".strip(),
},
]
QUERY = """
router-core-01 は ping に応答しますが、SNMP だけ timeout しています。
ACL または SNMP community の問題でしょうか。最初に何を確認すべきですか？
""".strip()
EXPECTED_DOC_IDS = {"runbook_snmp_timeout"}
def tokenize(text):
return re.findall(r"[a-zA-Z0-9_.:-]+|[一-龥ぁ-んァ-ンー]+", text.lower())
def term_vector(text):
return Counter(tokenize(text))
def cosine_similarity(a, b):
keys = set(a) | set(b)
dot = sum(a[k] * b[k] for k in keys)
norm_a = math.sqrt(sum(v * v for v in a.values()))
norm_b = math.sqrt(sum(v * v for v in b.values()))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
def keyword_score(query_tokens, doc_tokens):
if not query_tokens:
return 0.0
hits = sum(1 for token in query_tokens if token in doc_tokens)
return hits / len(query_tokens)
def hybrid_search(query, docs, k=3, vector_weight=0.65, keyword_weight=0.35):
query_vec = term_vector(query)
query_tokens = set(query_vec)
results = []
for doc in docs:
doc_vec = term_vector(doc["text"] + " " + doc["title"] + " " + doc["category"])
doc_tokens = set(doc_vec)
vector = cosine_similarity(query_vec, doc_vec)
keyword = keyword_score(query_tokens, doc_tokens)
score = vector_weight * vector + keyword_weight * keyword
results.append({
**doc,
"vector_score": round(vector, 3),
"keyword_score": round(keyword, 3),
"hybrid_score": round(score, 3),
})
return sorted(results, key=lambda x: x["hybrid_score"], reverse=True)[:k]
def generate_answer(query, retrieved_docs):
top = retrieved_docs[0]
citations = [f"{doc['doc_id']} p.{doc['page']}" for doc in retrieved_docs]
if top["doc_id"] == "runbook_snmp_timeout":
answer = (
"ping に応答するが SNMP のみ timeout しているため、まず監視サーバから対象装置への "
"SNMP 到達性、SNMP community、ACL を確認します。装置 CPU と監視経路も併せて確認します。"
)
else:
answer = (
f"最上位に検索された {top['title']} に基づくと、まず関連する初動確認を行います。"
"ただし、質問との関連度が十分か確認してください。"
)
return {
"answer": answer,
"citations": citations,
}
def judge_retrieval(retrieved_docs, expected_doc_ids):
retrieved_ids = [doc["doc_id"] for doc in retrieved_docs]
hit = any(doc_id in expected_doc_ids for doc_id in retrieved_ids)
first_rank = next(
(idx + 1 for idx, doc_id in enumerate(retrieved_ids) if doc_id in expected_doc_ids),
None,
)
return {
"retrieved_ids": retrieved_ids,
"hit_at_k": hit,
"first_relevant_rank": first_rank,
}
def judge_answer(answer, retrieved_docs):
source_text = " ".join(doc["text"] for doc in retrieved_docs).lower()
answer_tokens = set(tokenize(answer))
source_tokens = set(tokenize(source_text))
supported = [token for token in answer_tokens if token in source_tokens]
support_ratio = len(supported) / len(answer_tokens) if answer_tokens else 0
return {
"faithfulness_like_score": round(support_ratio, 3),
"needs_human_review": support_ratio < 0.35,
}
def main():
print("=== Query ===")
print(QUERY)
print()
retrieved = hybrid_search(QUERY, RUNBOOKS, k=3)
print("=== Retrieved runbooks ===")
for rank, doc in enumerate(retrieved, 1):
print(
f"{rank}. {doc['doc_id']} | {doc['title']} | "
f"hybrid={doc['hybrid_score']} vector={doc['vector_score']} keyword={doc['keyword_score']}"
)
print()
retrieval_verdict = judge_retrieval(retrieved, EXPECTED_DOC_IDS)
print("=== Retrieval check ===")
print(retrieval_verdict)
print()
result = generate_answer(QUERY, retrieved)
print("=== RAG-style answer ===")
print(result["answer"])
print("citations:", ", ".join(result["citations"]))
print()
answer_verdict = judge_answer(result["answer"], retrieved)
print("=== Answer quality check ===")
print(answer_verdict)
if not retrieval_verdict["hit_at_k"] or answer_verdict["needs_human_review"]:
print("\nACTION: 根拠が弱いため、人間確認に回してください。")
else:
print("\nACTION: 根拠付き回答として NOC 担当者に提示できます。")
if __name__ == "__main__":
main()
実行方法
外部 API、Docker、DB、環境変数は不要です。
python nms_rag_quality_check.py
Windows PowerShell で日本語パスのフォルダから実行する場合は、Set-Location -LiteralPath を使うと安全です。
期待される結果
実行すると、だいたい次のような結果になります。
=== Query ===
router-core-01 は ping に応答しますが、SNMP だけ timeout しています。
ACL または SNMP community の問題でしょうか。最初に何を確認すべきですか？
=== Retrieved runbooks ===
1. runbook_snmp_timeout | SNMP timeout 確認 | hybrid=0.514 vector=0.583 keyword=0.385
2. runbook_cpu_high | CPU 高騰確認 | hybrid=0.155 vector=0.156 keyword=0.154
3. runbook_bgp_down | BGP neighbor down 初動 | hybrid=0.0 vector=0.0 keyword=0.0
=== Retrieval check ===
{'retrieved_ids': ['runbook_snmp_timeout', 'runbook_cpu_high', 'runbook_bgp_down'], 'hit_at_k': True, 'first_relevant_rank': 1}
=== RAG-style answer ===
ping に応答するが SNMP のみ timeout しているため、まず監視サーバから対象装置への SNMP 到達性、SNMP community、ACL を確認します。装置 CPU と監視経路も併せて確認します。
citations: runbook_snmp_timeout p.18, runbook_cpu_high p.31, runbook_bgp_down p.12
=== Answer quality check ===
{'faithfulness_like_score': 0.571, 'needs_human_review': False}
ACTION: 根拠付き回答として NOC 担当者に提示できます。
ここで見たいのは、スコアそのものではありません。
runbook_snmp_timeout が上位に取れているか。回答に citation が付いているか。検索結果が期待と違う場合に、人間確認へ回す判断ができるか。
この3点です。
NMS 開発との接続点
本番の NMS では、今回の RUNBOOKS は社内 Runbook、過去チケット、ベンダー手順書になります。
hybrid_search() は、FAISS、BM25、reranker に置き換えられます。
judge_answer() は、LLM-as-Judge に置き換えられます。
しかし、設計の考え方は同じです。
- 検索結果をそのまま信じない
- 出典を必ず表示する
- 関連文書が取れなければ回答を抑制する
- 根拠が弱い場合は NOC 担当者やリーダー確認に回す
- 障害対応では「答えること」より「危ない回答を出さないこと」を優先する
業務で使うなら
NMS や障害対応に RAG を入れるなら、私は次の順番で作りたいです。
まず、文書側の整備です。
- Runbook を障害種別ごとに分ける
- 見出し、手順番号、前提条件を残す
-
doc_id、category、device_type、severity、source、updated_atを metadata に入れる - 古い手順書と最新手順書を区別する
- 承認済み文書だけを本番検索対象にする
次に、検索側です。
- Vector search と BM25 の両方を試す
- Hybrid search の重みを評価する
- top_k と fetch_k を調整する
- スコア閾値を設定する
- rerank を入れるか検証する
- 文書なしの場合の fallback を決める
次に、回答生成です。
- context だけで回答させる
- 推測で手順を追加させない
- 出典を必ず表示する
- 回答を短くし、一次対応向けにする
-
temperatureは低めから始める
最後に、評価と運用監視です。
- eval dataset を作る
- hit@k、MRR、検索失敗率を見る
- LLM-as-Judge で faithfulness / relevance を見る
- ユーザーフィードバックを保存する
- どの文書がよく使われるかを見る
- 回答抑制や人間確認の条件を決める
ここまでやると、RAG は単なるデモではなく、業務機能に近づきます。
まとめ
今回学んで印象に残ったのは、RAG を LLM 単体の技術ではなく、検索と生成をつなぐ設計として見ることでした。
構造だけ見れば、「検索して、渡して、答える」だけです。
しかし、NMS の障害対応に入れるなら、価値は「LLM が賢く答えること」だけではありません。
正しい Runbook に早くたどり着くこと。根拠を表示すること。根拠が弱いときは回答を抑制すること。
障害対応の現場では、この地味な品質管理の方がずっと重要だと感じました。
次回は、RAG の周辺で増えてくる処理を、LangChain でどう部品化するかを整理します。
OpenAI API を直接呼ぶだけではつらくなってくるところから、chain、Runnable、output parser、memory の話につなげます。