連載： NMS開発者がLLMブートキャンプで学んだこと
前回は、Tool Calling を「LLM に外部 API を使わせる仕組み」として整理しました。
NMS API で装置状態を取る。
CMDB API で拠点や担当を調べる。
過去チケットを検索する。
Runbook を参照する。
これらを tool として定義すると、LLM はユーザーの質問に応じて、どの tool をどの引数で呼ぶべきかを判断できるようになります。
ただ、Tool Calling を学んだあとにすぐ感じたのは、次のことでした。
「tool を呼べる」だけでは、障害対応の業務フローにはまだ足りない。
障害対応は、一回の API 呼び出しで終わる処理ではありません。
障害を受け付ける。
装置情報を確認する。
アラートの種類を分類する。
関連する Runbook を探す。
原因候補を整理する。
一次対応案を作る。
危険な操作なら人間に確認する。
チケットに記録する。
このように、複数のステップがあります。
しかも、毎回同じ順番で進むとは限りません。
critical な障害なら通知を急ぐ必要があります。メンテナンス時間中なら、自動復旧を止めたいです。装置情報が取れなければ、別の確認ルートに切り替える必要があります。同じ障害チケットがすでにあるなら、新規作成ではなく既存チケットに追記したいです。
ここで必要になるのは、単なる tool の一覧ではなく、状態と分岐を持った流れです。
今回扱う LangGraph は、そのための道具として理解するとかなり腑に落ちました。
背景
授業では、LangGraph を使った法律エージェントを作りました。
テーマは法律文書ベースの検索エージェントでした。
法律相談の質問を受け取り、質問タイプを分類し、必要なら法律文書を検索し、回答を作る。さらに、ReAct パターンで tool を呼び、MemorySaver で会話履歴を保持し、Human-in-the-Loop で tool 実行前に止めるところまで扱いました。
授業資料の学習目標は、だいたい次のような内容でした。
StateGraph と TypedDict で Agent の状態を定義する
classify -> search -> analyze -> answer のように node を分ける
条件付き edge で質問タイプごとに経路を変える
ReAct パターンで agent と tools を往復させる
MemorySaver で thread_id ごとの履歴を保持する
interrupt で人間確認を挟む
扱っているドメインは法律でしたが、構造としては NMS の障害対応にかなり近いと感じました。
法律エージェントでは、質問を分類してから検索します。
NMS の障害対応でも、アラートを分類してから確認手順を変えます。
法律エージェントでは、法律文書検索 tool を呼びます。
NMS の障害対応では、装置情報、interface 状態、過去チケット、Runbook 検索 tool を呼びます。
法律エージェントでは、最終回答に免責文を付けます。
NMS の障害対応では、自動判断の限界や人間承認の有無を明示する必要があります。
この対応関係が見えたことで、LangGraph は「LLM 用の特別な何か」というより、業務フローを graph として書くための仕組みだと理解できました。
Tool Callingだけでは足りないところ
前回の Tool Calling では、Mock NMS API を tool として定義する例を考えました。
@tool
def get_device_status(device_id: str) -> dict:
"""NMS から装置状態を取得する。"""
return {
"device_id": device_id,
"site": "Tokyo DC",
"status": "critical",
"last_alarm": "BGP_SESSION_DOWN",
}
LLM は、障害ログを読んでこの tool を呼び出せます。
しかし、障害対応ではこのあとが問題です。
装置状態を取ったあと、何をするのか
critical ならどの経路に進むのか
Runbook を検索する条件は何か
tool が失敗したらどうするのか
同じ tool を何回まで呼んでよいのか
人間承認はどこで入れるのか
途中状態をどこに保存するのか
これを普通の Python の if と while だけで書くこともできます。
実際、最初はそれで十分に見えます。
alert = receive_alert()
device = get_device_status(alert.device_id)
runbook = search_runbook(alert.message)
draft = generate_response(alert, device, runbook)
でも、少し条件が増えると読みにくくなります。
if critical:
notify_oncall()
if maintenance_window:
stop_auto_action()
else:
...
elif warning:
...
else:
...
さらに LLM と tool が入ると、流れはもっと複雑になります。
LLM が tool を呼ぶかどうかを判断する。
tool の結果を LLM に戻す。
LLM が追加 tool を呼ぶか、最終回答を出すかを判断する。
必要なら人間確認で止める。
このような処理は、単純な関数呼び出しというより、状態を持った workflow に近いです。
LangGraph は、この workflow を node と edge で表現します。
ここまでの流れを図にすると、次のようになります。
この図で大事なのは、Tools が終点ではないことです。
NMS API や Runbook 検索 tool を呼んだあと、その結果は StateGraph の状態に戻ります。Agent はその状態を見て、追加確認が必要か、原因候補を作れるか、人間承認に進むべきかを判断します。
つまり LangGraph は、LLM に tool を使わせるだけでなく、障害対応の途中状態と次の分岐を管理するための枠組みとして見ると分かりやすいです。
StateGraphを障害対応に置き換える
授業では、まず TypedDict で Agent の状態を定義しました。
法律エージェントでは、messages、query、question_type、search_results、answer のような項目を持っていました。
NMS の障害対応に置き換えるなら、私は次のような状態を考えます。
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class IncidentAgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
incident_id: str
raw_alert: str
device_id: str
severity: Literal["critical", "major", "minor", "unknown"]
device_status: dict
related_tickets: list[dict]
runbook_results: list[dict]
hypothesis: list[str]
action_plan: list[str]
approval_required: bool
final_note: str
ここで大事なのは、LLM に渡す文章だけを状態にするのではなく、業務フロー上で必要な情報を状態として持つことです。
たとえば、messages は LLM との会話履歴です。
一方で、device_status や related_tickets は tool の結果です。
severity は分類結果です。
approval_required は、次に人間確認へ進むかどうかの判断に使えます。
NMS のアプリケーションとして見ると、これはかなり普通の設計です。
障害チケットの内部状態を持ち、それに応じて画面や処理を切り替えるのと似ています。
LangGraph では、その状態を node 間で渡しながら更新していきます。
nodeを業務ステップとして見る
授業の基本グラフでは、次のような流れを作っていました。
START -> classify -> search -> analyze -> END
これを障害対応に置き換えると、たとえば次のようになります。
START
-> classify_alert
-> fetch_device
-> search_knowledge
-> analyze_cause
-> draft_action
-> END
コードの形は、かなり素直です。
from langgraph.graph import StateGraph, START, END
def build_basic_incident_graph():
builder = StateGraph(IncidentAgentState)
builder.add_node("classify_alert", classify_alert_node)
builder.add_node("fetch_device", fetch_device_node)
builder.add_node("search_knowledge", search_knowledge_node)
builder.add_node("analyze_cause", analyze_cause_node)
builder.add_node("draft_action", draft_action_node)
builder.add_edge(START, "classify_alert")
builder.add_edge("classify_alert", "fetch_device")
builder.add_edge("fetch_device", "search_knowledge")
builder.add_edge("search_knowledge", "analyze_cause")
builder.add_edge("analyze_cause", "draft_action")
builder.add_edge("draft_action", END)
return builder.compile()
この段階では、まだ一直線の workflow です。
それでも、処理の責務が node として分かれるだけで見通しがよくなります。
classify_alert_node はアラート分類だけを見る。
fetch_device_node は装置情報取得だけを見る。
search_knowledge_node は Runbook や過去チケット検索だけを見る。
analyze_cause_node は原因候補を作る。
draft_action_node は一次対応案を作る。
Java/Spring のサービス設計に近い感覚があります。
全部を一つの巨大な handleIncident() に詰め込むのではなく、処理単位を分ける。
LangGraph では、その分けた処理を graph としてつなげます。
条件分岐を入れる
障害対応では、すべてのアラートを同じ流れで処理するわけではありません。
たとえば、次のような分岐が考えられます。
BGP Down -> peer 状態と回線アラートを確認
Interface Down -> interface counters と直近作業履歴を確認
SNMP Timeout -> 監視経路、ACL、community を確認
Disk Full -> サーバ運用 Runbook を確認
unknown -> 一般的な一次切り分けへ
授業では、add_conditional_edges を使って、質問タイプごとに search、term、general へ分岐していました。
障害対応なら、同じ考え方で route 関数を書けます。
def route_by_alert_type(state: IncidentAgentState):
alert = state["raw_alert"].lower()
if "bgp" in alert or "peer" in alert:
return "bgp_flow"
if "interface" in alert or "link down" in alert:
return "interface_flow"
if "snmp" in alert or "timeout" in alert:
return "monitoring_flow"
return "general_flow"
graph 側では、この route 関数の戻り値に応じて次の node を決めます。
builder.add_conditional_edges(
"classify_alert",
route_by_alert_type,
{
"bgp_flow": "check_bgp_peer",
"interface_flow": "check_interface",
"monitoring_flow": "check_monitoring_path",
"general_flow": "search_knowledge",
},
)
ここで面白いのは、分岐の判断をすべて LLM に任せる必要はないことです。
文字列や severity のようにルールで判断できるものは、普通のコードで十分です。
LLM は、曖昧なログを分類したり、検索結果を読んで原因候補を整理したりするところで使えばよいです。
この分担は、NMS の実務ではかなり重要です。
ルールで確実にできることまで LLM に任せると、結果が揺れます。
逆に、人間が毎回読まないと判断しづらいところだけ LLM に任せると、価値が出やすいです。
ReActパターンを障害対応に使う
授業では、ReAct パターンも扱いました。
基本の形は、次のようなループです。
agent -> tools -> agent -> tools -> agent -> END
LLM が tool を呼ぶ必要があると判断したら tools node に進みます。
tool 実行結果が ToolMessage として戻ると、もう一度 agent に戻ります。
最後に tool 呼び出しがなくなれば終了します。
授業資料の構造は、だいたい次のような形でした。
def agent_node(state):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state):
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return END
builder.add_edge(START, "agent")
builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", END: END},
)
builder.add_edge("tools", "agent")
NMS で考えると、この ReAct ループはかなり自然です。
障害ログを読んだ LLM が、まず装置状態を調べる。
装置状態を見たあと、次に過去チケットを検索する。
過去チケットを見たあと、Runbook を検索する。
必要な情報が揃ったら、原因候補と一次対応案を出す。
この流れです。
たとえば、ユーザーがこう入力したとします。
router-tokyo-01 で BGP_SESSION_DOWN が発生。
影響範囲と一次対応案を整理して。
Agent は最初に get_device_status を呼ぶかもしれません。
次に search_related_tickets を呼ぶかもしれません。
さらに search_runbook を呼ぶかもしれません。
最後に、それらの結果をまとめて回答します。
これは単なる RAG ではありません。
文書検索だけでなく、現在状態、過去履歴、手順書を組み合わせるからです。
Tool Calling だけでも実現できますが、LangGraph にすると「今どの状態で、次にどの node に進むのか」が明示されます。
Agentが不要なtoolを呼ぶ問題
「tool を呼びすぎる」「危険な tool を渡しすぎる」という論点自体は、前回の Tool Calling の回で一度整理しました。
tool schema を分けること、API 結果を構造化して返すこと、更新系 tool は read-only tool と切り離すこと。これらは Tool Calling の設計原則として、すでに書いた通りです。
今回 LangGraph を触っていて気づいたのは、その原則を「守ろうとする」のと「graph の構造で守らざるを得なくする」のは別物だということでした。
前回の ReAct ループは、こういう形でした。
for turn in range(max_turns):
response = llm_with_tools.invoke(messages)
messages.append(response)
if not response.tool_calls:
return response.content
for tc in response.tool_calls:
result = tool_map[tc["name"]].invoke(tc["args"])
messages.append(
ToolMessage(content=str(result), tool_call_id=tc["id"])
)
この形だと、「tool を何回まで呼んでいいか」「危険な tool の前で止めるか」は、コメントか運用ルールでしか表現できません。for ループの中に if を足していけば書けますが、それは結局、前回書いた「増えると読みにくくなる if/while」に戻ってしまいます。
LangGraph だと、この制約を state と edge に持たせられます。
tool 呼び出し回数の上限は、state の中にカウンタとして持ち、should_continue の条件に組み込めます。
危険な tool は、そもそも別の node（別の graph）に分けて、そこに到達する edge の手前に interrupt_before を置けます。
つまり、「tool を呼びすぎない」「危険な tool の前で止める」という前回の結論を、コードレビューで気をつける話から、graph の構造として強制する話に変えられる。これが、Tool Calling から LangGraph に進む一番の実務的な理由だと感じました。
MemorySaverで障害ごとの履歴を持つ
授業では、MemorySaver を checkpointer として使い、thread_id ごとに会話履歴を保持しました。
法律エージェントでは、user-A と user-B の履歴を分ける例がありました。
NMS では、この thread_id は障害チケット ID や incident ID に近いです。
config = {
"configurable": {
"thread_id": "INC-20260616-001"
}
}
同じ障害について、最初のアラート、追加ログ、担当者の確認結果、復旧確認を順番に扱うなら、履歴が必要になります。
たとえば、最初にこう聞きます。
router-tokyo-01 の BGP_SESSION_DOWN を分析して。
次に、同じ incident でこう追加します。
対向 peer への ping は通っています。直近で回線メンテナンス予定はありません。
このとき、Agent が前の文脈を覚えていないと、また最初から装置情報を取り直したり、同じ説明を繰り返したりします。
MemorySaver によって thread ごとの state を持てば、少なくとも検証段階ではこの流れを再現できます。
graph = build_memory_graph(tools)
config = {"configurable": {"thread_id": "INC-20260616-001"}}
graph.invoke(
{"messages": [HumanMessage(content="BGP Down を分析して")]},
config=config,
)
graph.invoke(
{"messages": [HumanMessage(content="peer ping は通っています")]},
config=config,
)
state = graph.get_state(config)
print(len(state.values["messages"]))
もちろん、実務では MemorySaver だけでは足りません。
プロセスを再起動したら消えるメモリではなく、DB や Redis のような永続ストアが必要になるはずです。
それでも、授業で thread_id ごとに履歴が分かれる様子を見たことで、障害チケット単位で Agent state を持つイメージがかなり具体的になりました。
Human-in-the-Loopを入れる
障害対応 Agent で一番大事なのは、人間確認をどこに入れるかです。
授業では、interrupt_before=["tools"] を使って、tool 実行前に graph を止める例がありました。
return builder.compile(
checkpointer=MemorySaver(),
interrupt_before=["tools"],
)
この設定をすると、Agent が tool を呼ぼうとした時点で一度停止できます。
状態を確認し、問題なければ再開します。
state = graph.get_state(config)
print(state.next)
# ユーザー承認後に再開
graph.invoke(None, config=config)
NMS の文脈では、これはかなり重要です。
たとえば read-only tool なら、毎回止める必要はないかもしれません。
装置状態の取得や Runbook 検索は、自動実行してもよいでしょう。
しかし、次のような操作は必ず止めたいです。
チケットを作成する
チケットを close する
オンコール通知を送る
顧客向け通知文を送信する
装置設定を変更する
復旧コマンドを実行する
ここで人間承認を挟むと、Agent は「完全自動化」ではなく「半自動化」の道具になります。
私は、NMS への LLM 導入はこの半自動化から始めるのが現実的だと思いました。
いきなり自動復旧まで進めるのではなく、まずは次のような形です。
障害内容を整理する
関連情報を集める
原因候補を出す
一次対応案を draft する
人間が確認する
承認されたらチケットに追記する
このくらいなら、既存の運用フローを大きく壊さずに試せます。
そして、どこで人間が止めたか、どの提案が採用されたかをログに残せば、次の改善にも使えます。
障害対応Agentの小さな設計例
ここからは、授業内容を NMS 向けに置き換えた小さな設計例を考えてみます。
目的は、次の流れを作ることです。
障害受付
-> アラート分類
-> 装置情報取得
-> 過去チケット検索
-> Runbook 検索
-> 原因候補作成
-> 一次対応案作成
-> 人間承認
-> チケット追記 draft
まず、read-only tool を用意します。
from langchain_core.tools import tool
@tool
def get_device_status(device_id: str) -> dict:
"""NMS から装置の現在状態を取得する。"""
return {
"device_id": device_id,
"site": "Tokyo DC",
"role": "edge-router",
"status": "critical",
"last_seen": "2026-06-16T01:10:00+09:00",
}
@tool
def search_related_tickets(device_id: str, keyword: str) -> list[dict]:
"""過去の関連障害チケットを検索する。"""
return [
{
"ticket_id": "INC-20260512-014",
"summary": "BGP peer down after carrier maintenance",
"resolution": "carrier 側作業完了後に peer 復旧を確認",
}
]
@tool
def search_runbook(keyword: str) -> list[dict]:
"""障害対応 Runbook を検索する。"""
return [
{
"title": "BGP Session Down 初動対応",
"steps": [
"peer IP への疎通確認",
"interface error counter 確認",
"直近メンテナンス予定確認",
"対向キャリア障害情報確認",
],
}
]
次に、Agent と tools の ReAct graph を作ります。
def build_incident_react_graph(tools):
builder = StateGraph(IncidentAgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", END: END},
)
builder.add_edge("tools", "agent")
return builder.compile(checkpointer=MemorySaver())
ここまでは授業の法律エージェントとほぼ同じです。
違うのは、tool の中身と prompt の役割です。
法律エージェントでは、法律文書検索と回答生成が中心でした。
障害対応 Agent では、次のような system message を置きたくなります。
あなたはネットワーク運用センターの一次対応支援 Agent です。
障害原因を断定せず、確認済み情報と仮説を分けて整理してください。
危険な復旧操作や外部通知は提案までに留め、人間承認が必要であることを明示してください。
回答には、参照した装置情報、過去チケット、Runbook を分けて含めてください。
これにより、Agent の役割を「自動復旧する人」ではなく、「一次対応を整理する補助者」に寄せます。
この立ち位置が大事だと思います。
障害対応では、LLM が自信満々に間違えるのが一番危険です。
だから、Agent には「確定」と「仮説」を分けさせる必要があります。
5分で試す最小プロジェクト
ここまで LangGraph の考え方を説明してきましたが、最初から LangGraph、LangChain、LLM API を全部入れると、どこが workflow の本質なのか分かりづらくなります。
そこで、まずは外部ライブラリなしの Python だけで、障害対応 Agent の状態遷移を小さく再現してみます。
この実習で確認したいのは、モデルの回答品質ではありません。
次の流れです。
Alert
-> classify_alert
-> fetch_device
-> search_tickets
-> search_runbook
-> analyze_cause
-> draft_action
-> approval_gate
-> make_ticket_draft
#incident_agent_demo.py
import argparse
import json
import re
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Callable
MAX_TOOL_CALLS = 5
@dataclass
class IncidentState:
incident_id: str
raw_alert: str
device_id: str = ""
alert_type: str = "unknown"
severity: str = "unknown"
device_status: dict = field(default_factory=dict)
related_tickets: list = field(default_factory=list)
runbook: list = field(default_factory=list)
hypotheses: list = field(default_factory=list)
action_plan: list = field(default_factory=list)
approval_required: bool = False
approved: bool = False
ticket_draft: str = ""
trace: list = field(default_factory=list)
tool_calls: list = field(default_factory=list) # 監査ログ
warnings: list = field(default_factory=list)
# --- Mock external systems -------------------------------------------------
# 実際の実装では、それぞれ NMS API / ticket system API / RAG 検索に置き換える。
# 戻り値は ok / error_code を持つ構造にし、「tool の結果をそのまま信じない」前提にする。
MOCK_NMS = {
"router-tokyo-01": {
"site": "Tokyo DC",
"role": "edge-router",
"status": "critical",
"last_seen": "2026-08-04T01:10:00+09:00",
},
"switch-osaka-02": {
"site": "Osaka Branch",
"role": "access-switch",
"status": "major",
"last_seen": "2026-08-04T01:08:00+09:00",
},
}
MOCK_TICKETS = [
{
"ticket_id": "INC-20260721-014",
"device_id": "router-tokyo-01",
"alert_type": "bgp",
"summary": "BGP peer down after carrier maintenance",
"resolution": "Carrier maintenance completed. Peer recovered after route refresh.",
}
]
MOCK_RUNBOOK = {
"bgp": [
"Check peer IP reachability.",
"Check interface error counters.",
"Check recent maintenance window.",
"Confirm carrier-side incident notice.",
],
"interface": [
"Check physical link status.",
"Check CRC/input/output error counters.",
"Confirm recent cabling or switch work.",
],
"monitoring": [
"Check SNMP community or credential changes.",
"Check ACL/firewall path from monitoring server.",
"Check device CPU and polling timeout trend.",
],
}
def get_device_status(device_id: str) -> dict:
device = MOCK_NMS.get(device_id)
if not device:
return {"ok": False, "error_code": "device_not_found", "retryable": False}
return {"ok": True, **device}
def search_related_tickets(device_id: str, alert_type: str) -> dict:
tickets = [
t for t in MOCK_TICKETS
if t["device_id"] == device_id and t["alert_type"] == alert_type
]
return {"ok": True, "tickets": tickets}
def search_runbook(alert_type: str) -> dict:
steps = MOCK_RUNBOOK.get(alert_type)
if not steps:
return {"ok": True, "steps": ["Collect latest alarms.", "Check device reachability."]}
return {"ok": True, "steps": steps}
# --- Tool call wrapper -------------------------------------------------
# tool 呼び出し回数の上限と、監査ログ（何を・どんな引数で・成功したか・何 ms かかったか）を
# ここで一括管理する。node 側は個別に上限チェックを書かなくてよい。
def call_tool(state: IncidentState, tool_name: str, func: Callable, **kwargs) -> dict:
if len(state.tool_calls) >= MAX_TOOL_CALLS:
state.warnings.append(f"tool_call_limit_reached before {tool_name}")
return {"ok": False, "error_code": "tool_call_limit_reached", "retryable": False}
started = time.monotonic()
result = func(**kwargs)
elapsed_ms = round((time.monotonic() - started) * 1000, 2)
state.tool_calls.append({
"tool": tool_name,
"args": kwargs,
"ok": result.get("ok", False),
"elapsed_ms": elapsed_ms,
})
if not result.get("ok", False):
state.warnings.append(f"{tool_name} failed: {result.get('error_code')}")
return result
# --- Nodes: 業務ステップ単位で処理を分ける -------------------------------
def classify_alert(state: IncidentState) -> str:
state.trace.append("classify_alert")
match = re.search(r"(router|switch)-[a-z]+-\d+", state.raw_alert)
state.device_id = match.group(0) if match else "unknown-device"
alert_lower = state.raw_alert.lower()
if "bgp" in alert_lower or "peer" in alert_lower:
state.alert_type, state.severity = "bgp", "critical"
elif "interface" in alert_lower or "link down" in alert_lower:
state.alert_type, state.severity = "interface", "major"
elif "snmp" in alert_lower or "timeout" in alert_lower:
state.alert_type, state.severity = "monitoring", "major"
else:
state.alert_type, state.severity = "general", "unknown"
return "fetch_device"
def fetch_device(state: IncidentState) -> str:
state.trace.append("fetch_device")
result = call_tool(state, "get_device_status", get_device_status, device_id=state.device_id)
if result.get("ok"):
state.device_status = result
else:
# tool が失敗しても graph は止めない。分かっている範囲で先に進める。
state.device_status = {"site": "unknown", "role": "unknown", "status": "unknown"}
return "search_tickets"
def search_tickets(state: IncidentState) -> str:
state.trace.append("search_tickets")
result = call_tool(
state, "search_related_tickets", search_related_tickets,
device_id=state.device_id, alert_type=state.alert_type,
)
state.related_tickets = result.get("tickets", []) if result.get("ok") else []
return "search_runbook"
def search_runbook_node(state: IncidentState) -> str:
state.trace.append("search_runbook")
result = call_tool(state, "search_runbook", search_runbook, alert_type=state.alert_type)
state.runbook = result.get("steps", []) if result.get("ok") else []
return "analyze_cause"
def analyze_cause(state: IncidentState) -> str:
state.trace.append("analyze_cause")
hypotheses_by_type = {
"bgp": [
"Carrier-side maintenance or transient peer failure",
"Physical or logical issue on the uplink interface",
"Recent routing policy or configuration change",
],
"interface": [
"Physical link or cabling issue",
"Interface error increase caused by optics or media problem",
"Planned work not reflected in the maintenance calendar",
],
"monitoring": [
"Monitoring path or ACL problem",
"SNMP credential mismatch",
"Device overload causing polling timeout",
],
}
state.hypotheses = hypotheses_by_type.get(
state.alert_type, ["Cause is not clear yet. Manual triage is required."]
)
if state.device_status.get("status") == "unknown":
state.hypotheses.append("Device status could not be confirmed via NMS API.")
return "draft_action"
def draft_action(state: IncidentState) -> str:
state.trace.append("draft_action")
state.action_plan = list(state.runbook)
state.action_plan.append("Record confirmed facts and unknowns in the ticket.")
state.approval_required = state.severity in {"critical", "major"}
return "approval_gate"
def approval_gate(state: IncidentState) -> str:
state.trace.append("approval_gate")
if state.approval_required and not state.approved:
state.ticket_draft = (
"Approval required before ticket update or external notification.\n"
"Run again with --approve to simulate operator approval."
)
return "END"
return "make_ticket_draft"
def make_ticket_draft(state: IncidentState) -> str:
state.trace.append("make_ticket_draft")
related = ", ".join(t["ticket_id"] for t in state.related_tickets) or "none"
hypotheses = "\n".join(f"- {h}" for h in state.hypotheses)
actions = "\n".join(f"{i + 1}. {a}" for i, a in enumerate(state.action_plan))
state.ticket_draft = f"""## Incident Draft
- incident_id: {state.incident_id}
- device: {state.device_id}
- alert_type: {state.alert_type}
- severity: {state.severity}
- site: {state.device_status.get("site")}
- role: {state.device_status.get("role")}
- related_tickets: {related}
## Hypotheses
{hypotheses}
## First Actions
{actions}
## Note
This is an operator-support draft. Do not execute recovery actions without approval.
"""
return "END"
# --- Graph: node と edge を分離して定義する ---------------------------
# LangGraph の StateGraph / add_edge に相当する部分を、ここでは dict で表現する。
# 「どの node が存在し、それぞれの node が何を実行するか」だけをここに集約し、
# 「次にどこへ進むか」は各 node の戻り値（edge の役割）に任せる。
GRAPH: dict[str, Callable[["IncidentState"], str]] = {
"classify_alert": classify_alert,
"fetch_device": fetch_device,
"search_tickets": search_tickets,
"search_runbook": search_runbook_node,
"analyze_cause": analyze_cause,
"draft_action": draft_action,
"approval_gate": approval_gate,
"make_ticket_draft": make_ticket_draft,
}
def run_graph(state: IncidentState, entry: str = "classify_alert") -> IncidentState:
current = entry
while current != "END":
node = GRAPH[current]
current = node(state)
return state
def to_output(state: IncidentState, fmt: str) -> str:
if fmt == "json":
return json.dumps(asdict(state), ensure_ascii=False, indent=2)
return state.ticket_draft
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--alert",
default="router-tokyo-01 BGP_SESSION_DOWN peer 10.0.0.2",
help="Raw NMS alert text",
)
parser.add_argument(
"--approve",
action="store_true",
help="Simulate operator approval before ticket draft creation",
)
parser.add_argument(
"--format",
choices=["markdown", "json"],
default="markdown",
help="Ticket draft output format",
)
args = parser.parse_args()
state = IncidentState(
incident_id="INC-" + datetime.now().strftime("%Y%m%d-%H%M%S"),
raw_alert=args.alert,
approved=args.approve,
)
run_graph(state)
print("=== Trace ===")
print(" -> ".join(state.trace))
print("\n=== Tool Calls (audit log) ===")
print(json.dumps(state.tool_calls, ensure_ascii=False, indent=2))
if state.warnings:
print("\n=== Warnings ===")
for w in state.warnings:
print(f"- {w}")
print("\n=== State Summary ===")
print(f"device_id: {state.device_id}")
print(f"alert_type: {state.alert_type}")
print(f"severity: {state.severity}")
print(f"approval_required: {state.approval_required}")
print(f"approved: {state.approved}")
print(f"\n=== Ticket Draft ({args.format}) ===")
print(to_output(state, args.format))
if __name__ == "__main__":
main()
実行方法はシンプルです。
python incident_agent_demo.py
最初の実行では、critical 障害なので人間承認前で止まります。あわせて、get_device_status を実際に呼んだ記録が監査ログとして残ります。
=== Trace ===
classify_alert -> fetch_device -> search_tickets -> search_runbook -> analyze_cause -> draft_action -> approval_gate
=== Tool Calls (audit log) ===
[
{
"tool": "get_device_status",
"args": {"device_id": "router-tokyo-01"},
"ok": true,
"elapsed_ms": 0.01
},
{
"tool": "search_related_tickets",
"args": {"device_id": "router-tokyo-01", "alert_type": "bgp"},
"ok": true,
"elapsed_ms": 0.01
},
{
"tool": "search_runbook",
"args": {"alert_type": "bgp"},
"ok": true,
"elapsed_ms": 0.0
}
]
=== State Summary ===
device_id: router-tokyo-01
alert_type: bgp
severity: critical
approval_required: True
approved: False
=== Ticket Draft (markdown) ===
Approval required before ticket update or external notification.
Run again with --approve to simulate operator approval.
承認済みとしてチケット draft まで進める場合は、次のように実行します。
python incident_agent_demo.py --approve
この場合は、approval_gate のあとに make_ticket_draft まで進みます。
=== Ticket Draft (markdown) ===
## Incident Draft
- incident_id: INC-...
- device: router-tokyo-01
- alert_type: bgp
- severity: critical
- site: Tokyo DC
- role: edge-router
- related_tickets: INC-20260721-014
## Hypotheses
- Carrier-side maintenance or transient peer failure
- Physical or logical issue on the uplink interface
- Recent routing policy or configuration change
## First Actions
1. Check peer IP reachability.
2. Check interface error counters.
3. Check recent maintenance window.
4. Confirm carrier-side incident notice.
5. Record confirmed facts and unknowns in the ticket.
NMS API から装置が見つからないケースも試せます。前回の記事で「tool の結果をそのまま信じない」「エラーは構造化して返す」と書いた部分が、実際にどう state へ流れるかを確認できます。
python incident_agent_demo.py --alert "switch-tokyo-99 unknown alert" --approve
=== Warnings ===
- get_device_status failed: device_not_found
=== Ticket Draft (markdown) ===
...
- site: unknown
- role: unknown
...
## Hypotheses
- Cause is not clear yet. Manual triage is required.
- Device status could not be confirmed via NMS API.
装置情報が取れなくても graph は止まらず、「不明」を明示した状態でチケット draft まで進みます。何が確認できて、何が確認できなかったかを分けて出す、という前回からの原則をコードで再現した形です。
出力を JSON で欲しい場合は --format json を付けます。チケット本文としてではなく、画面表示や DB 保存に使う想定です。
python incident_agent_demo.py --approve --format json
別のアラートも試せます。
python incident_agent_demo.py --alert "switch-osaka-02 interface Gi0/1 link down" --approve
この小さな実習では、LLM も LangGraph も使っていません。
あえて外部依存なしにした理由は、まず「障害対応 Agent に必要な状態遷移」を手で確認するためです。
実際の NMS 開発に置き換えると、各関数は次のようになります。
get_device_status -> NMS API / inventory API
search_related_tickets -> ticket system API
search_runbook -> RAG / vector search / internal wiki search
call_tool -> LangGraph の tool node 実行部分に相当（呼び出し上限・ログもここに集約）
approval_gate -> operator approval UI / interrupt_before
make_ticket_draft -> ticket comment draft / incident report draft
LangGraph に移行する場合は、GRAPH dict の各エントリがそのまま add_node に、node の戻り値（次に進む node 名）がそのまま add_edge / add_conditional_edges に対応します。call_tool の中身は、LangGraph の tool node と should_continue の条件判定に分かれる想定です。
つまり、この Python 実習は LangGraph の代替ではなく、LangGraph を導入する前に「どの状態を持ち、どこで分岐し、どこで人間承認を入れるか」を設計するための下書きです。
出力はチケットに貼れる形にする
NMS で使うなら、最終回答は会話として自然なだけでは足りません。
チケットや運用ログに貼れる形が必要です。
たとえば、次のような形式です。
## 障害概要
- device: router-tokyo-01
- alert: BGP_SESSION_DOWN
- severity: critical
## 確認済み情報
- NMS 上の装置状態: critical
- 拠点: Tokyo DC
- role: edge-router
- 過去類似チケット: INC-20260512-014
## 原因候補
1. 対向 peer または carrier 側の一時障害
2. 対象 interface の物理/link 問題
3. 直近メンテナンスまたは設定変更の影響
## 一次対応案
1. peer IP への疎通確認
2. interface error counter 確認
3. メンテナンス予定と carrier 障害情報確認
4. 必要に応じて担当者へ escalation
## 注意
- 現時点では原因未確定
- 復旧操作は人間承認後に実施
この形式にしておくと、NOC 担当者が読みやすいです。
また、チケットシステムに入れるときにも扱いやすくなります。
JSON と Markdown のどちらにするかは用途次第です。
画面表示や DB 保存なら JSON が便利です。
人間が読むチケット本文なら Markdown が便利です。
LLM に最終出力を任せる場合でも、どの形式で返すかはアプリケーション側で強く指定した方がよいです。
業務で使うなら
NMS の現場で LangGraph を使うなら、私はいきなり自動復旧 Agent を作るより、次のような小さな機能から始めたいです。
障害チケットの初期整理 Agent
入力は、NMS から上がってきたアラートと装置 ID です。
Agent は read-only tool だけを使います。
装置状態取得
interface 状態取得
過去チケット検索
Runbook 検索
メンテナンス予定確認
出力は、チケットに貼れる初期メモです。
障害概要
確認済み情報
未確認情報
原因候補
一次対応チェックリスト
人間承認が必要な操作
この段階では、装置に対して変更操作はしません。
通知も自動送信しません。
チケット close もしません。
まずは、人間が情報収集に使う時間を減らすところを狙います。
この使い方なら、LLM の価値とリスクのバランスが取りやすいです。
次の段階として、承認付きの write tool を追加できます。
チケットに初期メモを追記する
担当チームに通知 draft を作る
オンコール通知の候補を表示する
さらに成熟したら、限定された復旧操作を候補に入れることも考えられます。
ただし、そこまで進めるには、権限制御、監査ログ、dry-run、rollback、承認フロー、テスト環境での検証が必要です。
LangGraph は、その流れを作る助けになります。
しかし、運用責任まで LangGraph が引き受けてくれるわけではありません。
そこは、普通の業務システムと同じように設計する必要があります。
まとめ
LangGraph は、LLM と tool を含む複数ステップの処理を、状態付きの graph として組み立てるための仕組みです。
授業では、法律文書検索エージェントを題材に、StateGraph、TypedDict、node、edge、条件分岐、ReAct パターン、MemorySaver、Human-in-the-Loop を学びました。
NMS の障害対応に置き換えると、次のような流れを作れます。
障害受付
-> アラート分類
-> 装置情報取得
-> 過去チケット検索
-> Runbook 検索
-> 原因候補整理
-> 一次対応案作成
-> 人間承認
-> チケット追記 draft
ただし、実務で使うなら、最初から完全自動化を目指すべきではないと思います。
まずは read-only tool に限定し、障害チケットの初期整理や一次対応チェックリスト作成から始めるのが現実的です。
危険な操作には interrupt のような人間確認ポイントを入れる。
障害単位で thread_id を分け、履歴を残す。
tool 呼び出し回数や権限を制御する。
原因候補は断定せず、根拠と未確認事項を分けて出す。
このあたりを守れば、LangGraph は NMS の障害対応を少しずつ半自動化するための土台になりそうです。
今回の結論は、こうです。
Agent は「LLM に自由に任せる仕組み」ではありません。
むしろ、LLM を業務フローの中で安全に使うために、状態、分岐、tool、人間承認を設計する仕組みです。
NMS 開発者としては、この見方の方がずっと現実的でした。
次回は、ここで作った Agent の流れから少し離れて、PDF、Word、PowerPoint、HTML のような社内文書を LLM に読ませる前処理を見ていきます。
障害対応 Agent が本当に役に立つには、Runbook や報告書や過去資料を読める必要があります。
その入口になるのが、unstructured を使った文書処理です。