7. LangChainでLLMアプリを部品化してみる

連載： NMS開発者がLLMブートキャンプで学んだこと

← 前回 ｜ 次回 →

前回は、RAG を「検索して、見つけた文書を LLM に渡して、回答する仕組み」として整理しました。

ただし、実際に RAG を少し書いてみると、LLM を呼ぶ前後の処理が思ったより多いことに気づきます。

関連文書を検索し、context として整形し、プロンプトに差し込み、LLM の出力を文字列や JSON として扱いやすい形に戻す。さらに、回答の品質を評価したり、検索結果が弱いときの fallback を考えたりすると、単なる API 呼び出しでは済まなくなります。

最初は OpenAI API を直接呼ぶだけで十分でした。しかし、処理が増えてくると、コードのあちこちに「検索する処理」「プロンプトを組み立てる処理」「LLM を呼ぶ処理」「出力を整える処理」が散らばっていきます。

この状態で、NMS の障害分析のような業務機能を作ろうとすると、だんだん苦しくなります。

そこで出てくるのが LangChain です。

今回は、LangChain を「LLM アプリを作るための大きな魔法」としてではなく、LLM 処理を部品化して組み立てるための道具として見ていきます。

背景

OpenAI API を直接呼ぶコードは、最初の実験では分かりやすいです。

messages = [
    {"role": "system", "content": "あなたはネットワーク障害分析アシスタントです。"},
    {"role": "user", "content": alert_log},
]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
)

print(response.choices[0].message.content)

これくらいなら、まだ読みやすいです。

しかし、障害ログ分析を少し実務寄りにすると、処理はすぐ増えます。

障害ログを受け取る
  ↓
severity を分類する
  ↓
原因候補を出す
  ↓
運用者向けに要約する
  ↓
一次対応案を作る
  ↓
JSON として返す

さらに、前回の RAG を入れるなら、ここに検索処理も入ります。

障害ログを受け取る
  ↓
関連 Runbook を検索する
  ↓
検索結果を context にする
  ↓
LLM に回答させる
  ↓
出典付きで返す

このあたりから、単純な API 呼び出しだけでは、アプリケーションとして整理しづらくなります。

Java/Spring であれば、controller、service、repository のように責務を分けます。Node.js や FastAPI でも、router、service、client、schema を分けます。

LLM アプリでも同じことが必要になります。

プロンプト、モデル、出力変換、分岐、並列処理、memory、retriever、tool。

これらを一つの大きな関数に詰め込むと、あとで変更しづらくなります。

LangChain は、この LLM 周辺の処理を部品として扱いやすくするためのフレームワークだと理解しました。

全体像: LLM を一つの箱にしない

LLM アプリを作るとき、私は最初「LLM を呼ぶ関数」を一つ作れば十分だと思っていました。

しかし NMS の障害分析に近づけると、実際には分類、要約、Runbook 検索、一次対応案、チケット要約のように、役割の違う処理がいくつも出てきます。

このとき LangChain を使う意味は、LLM を賢くすることではなく、LLM の前後にある処理を部品として見える形にすることだと感じました。

この図のポイントは、LLM を一つの巨大な箱として置かないことです。

NMS 側では、syslog、SNMP Trap、監視イベント、チケット情報が入ってきます。そこから装置名や event type を抽出する前処理は、既存の backend 側でかなり安定して実装できます。

一方で、障害分類、要約、Runbook に基づく回答、一次対応案のように、言語理解が効く部分を chain として分けます。

こうしておくと、あとから prompt、retriever、parser、tool を差し替えやすくなります。

LangChain の最初の形

LangChain の基本として、まず次の import から見ていきます。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

最初に印象に残ったのは、この形です。

chain = prompt | llm | parser

最初はこの | が少し不思議でした。

Python で見ると、まるで Linux の pipe のようです。

prompt
  |
llm
  |
parser

入力を prompt に入れる。

prompt の結果を LLM に渡す。

LLM の出力を parser に渡す。

こうして一連の処理を chain として扱えます。

たとえば、障害ログを短く要約するだけなら、次のように書けます。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """
あなたは NMS の障害ログを読む運用支援アシスタントです。
回答は日本語で、短く、事実ベースで書いてください。
""".strip()),
    ("human", """
次の障害ログを1段落で要約してください。

障害ログ:
{alert_log}
""".strip()),
])

summary_chain = prompt | llm | StrOutputParser()

呼び出しはこうです。

alert_log = """
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
""".strip()

summary = summary_chain.invoke({"alert_log": alert_log})
print(summary)

API を直接呼ぶより少し遠回りに見えます。

しかし、処理が増えてくると、この「部品として持てる」ことが効いてきます。

障害ログ分析 Chain を作る

今回の例では、NMS の障害ログを入力して、次の3つを作る chain を考えます。

• 障害分類
• 運用者向け要約
• 一次対応案

まずは、分類用の chain です。

classify_prompt = ChatPromptTemplate.from_messages([
    ("system", """
あなたはネットワーク障害の一次分類を行うアシスタントです。
severity は critical, warning, info のいずれかにしてください。
category は routing, interface, monitoring, resource, unknown のいずれかにしてください。
""".strip()),
    ("human", """
障害ログ:
{alert_log}

出力:
severity:
category:
reason:
""".strip()),
])

classify_chain = classify_prompt | llm | StrOutputParser()

次に、要約用です。

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", """
あなたは NOC 担当者向けに障害状況を要約するアシスタントです。
装置名、発生しているイベント、影響の可能性を簡潔にまとめてください。
""".strip()),
    ("human", """
障害ログ:
{alert_log}

分類結果:
{classification}
""".strip()),
])

summary_chain = summary_prompt | llm | StrOutputParser()

最後に、一次対応案です。

action_prompt = ChatPromptTemplate.from_messages([
    ("system", """
あなたはネットワーク運用の一次対応案を作るアシスタントです。
危険な自動操作は提案せず、確認手順を優先してください。
不明な点は人間確認に回してください。
""".strip()),
    ("human", """
障害ログ:
{alert_log}

分類結果:
{classification}

要約:
{summary}

一次対応案を箇条書きで作ってください。
""".strip()),
])

action_chain = action_prompt | llm | StrOutputParser()

ここまでで、3つの部品ができました。

あとは、これを普通の Python 関数でつないでもよいです。

def analyze_alert(alert_log: str) -> dict:
    classification = classify_chain.invoke({"alert_log": alert_log})

    summary = summary_chain.invoke({
        "alert_log": alert_log,
        "classification": classification,
    })

    action = action_chain.invoke({
        "alert_log": alert_log,
        "classification": classification,
        "summary": summary,
    })

    return {
        "classification": classification,
        "summary": summary,
        "action": action,
    }

この形にすると、障害分析の流れが見えやすくなります。

classify_chain
  ↓
summary_chain
  ↓
action_chain

API 呼び出しをその場で何度も書くより、どの処理が何を担当しているかが分かりやすいです。

5分で試す NMS 障害ログ Chain

ここまでの話を、実際に手元で動かせる小さな Python 例にしてみます。

本来の LangChain では ChatPromptTemplate | llm | StrOutputParser のように chain を作ります。ただ、最初から OpenAI API Key や外部ライブラリを用意すると、記事を読んでいる途中で試すには少し重くなります。

そこで、ここでは LangChain 本体を使わず、Python だけで「部品を | でつなぐ感覚」を試します。

やることは単純です。

NMS 障害ログ
  ↓
preprocess
  ↓
classification_chain
  ↓
summary_chain
  ↓
action_chain
  ↓
ticket_formatter

コード全体は次の通りです。

# nms_alert_chain_minimal.py

from __future__ import annotations

import json
import re
import sys
from dataclasses import dataclass
from typing import Any, Callable

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")


@dataclass
class Step:
    name: str
    func: Callable[[Any], Any]

    def __or__(self, other: "Step") -> "Step":
        return Step(
            name=f"{self.name} | {other.name}",
            func=lambda value: other.func(self.func(value)),
        )

    def invoke(self, value: Any) -> Any:
        return self.func(value)


def extract_alert(raw_log: str) -> dict:
    device_match = re.search(r"\b(router|switch|fw)-[a-z0-9-]+\b", raw_log, re.I)
    peer_match = re.search(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", raw_log)

    text = raw_log.lower()
    events = []
    if "bgp" in text:
        events.append("bgp_down")
    if "snmp" in text:
        events.append("snmp_timeout")
    if "interface" in text or "gi0/" in text:
        events.append("interface_error")

    return {
        "raw_log": raw_log.strip(),
        "device": device_match.group(0) if device_match else "unknown",
        "peer_ip": peer_match.group(0) if peer_match else None,
        "events": events or ["unknown"],
    }


def classify_alert(alert: dict) -> dict:
    events = set(alert["events"])

    if "bgp_down" in events and "snmp_timeout" in events:
        severity = "critical"
        category = "routing_and_monitoring"
    elif "bgp_down" in events:
        severity = "critical"
        category = "routing"
    elif "snmp_timeout" in events:
        severity = "warning"
        category = "monitoring"
    elif "interface_error" in events:
        severity = "warning"
        category = "interface"
    else:
        severity = "info"
        category = "unknown"

    return {
        **alert,
        "severity": severity,
        "category": category,
    }


def summarize_for_noc(alert: dict) -> dict:
    peer = f" peer {alert['peer_ip']}" if alert["peer_ip"] else ""
    event_text = ", ".join(alert["events"])

    summary = (
        f"{alert['device']} で {event_text}{peer} が検知されました。"
        f"重要度は {alert['severity']}、分類は {alert['category']} です。"
    )

    return {
        **alert,
        "summary": summary,
    }


def recommend_first_actions(alert: dict) -> dict:
    actions = []

    if "bgp_down" in alert["events"]:
        actions.extend([
            "peer IP への疎通を確認する",
            "BGP state と直近の経路変更を確認する",
            "対象回線のメンテナンス予定を確認する",
        ])

    if "snmp_timeout" in alert["events"]:
        actions.extend([
            "監視サーバから対象装置への疎通を確認する",
            "SNMP community、ACL、装置 CPU を確認する",
        ])

    if "interface_error" in alert["events"]:
        actions.extend([
            "interface の error counter 推移を確認する",
            "光レベルと対向ポート状態を確認する",
        ])

    if not actions:
        actions.append("関連 Runbook がないため、NOC リーダーへ確認する")

    return {
        **alert,
        "first_actions": actions,
    }


def format_ticket(alert: dict) -> str:
    payload = {
        "device": alert["device"],
        "severity": alert["severity"],
        "category": alert["category"],
        "summary": alert["summary"],
        "first_actions": alert["first_actions"],
    }
    return json.dumps(payload, ensure_ascii=False, indent=2)


preprocess = Step("preprocess", extract_alert)
classify = Step("classification_chain", classify_alert)
summarize = Step("summary_chain", summarize_for_noc)
recommend = Step("action_chain", recommend_first_actions)
to_ticket = Step("ticket_formatter", format_ticket)

alert_chain = preprocess | classify | summarize | recommend | to_ticket


if __name__ == "__main__":
    sample_log = """
    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
    """

    print(alert_chain.invoke(sample_log))

実行方法は次の通りです。

python .nms_alert_chain_minimal.py

外部 API Key、LangChain、OpenAI SDK は不要です。Python 3.10 以上であれば、そのまま実行できます。

実行すると、次のような JSON が出ます。

{
  "device": "router-core-01",
  "severity": "critical",
  "category": "routing_and_monitoring",
  "summary": "router-core-01 で bgp_down, snmp_timeout, interface_error peer 10.10.0.2 が検知されました。重要度は critical、分類は routing_and_monitoring です。",
  "first_actions": [
    "peer IP への疎通を確認する",
    "BGP state と直近の経路変更を確認する",
    "対象回線のメンテナンス予定を確認する",
    "監視サーバから対象装置への疎通を確認する",
    "SNMP community、ACL、装置 CPU を確認する",
    "interface の error counter 推移を確認する",
    "光レベルと対向ポート状態を確認する"
  ]
}

この小さな例では、実際の LLM は呼んでいません。

しかし、NMS 開発者が LangChain を理解するうえで大事な感覚は確認できます。

• preprocess: syslog や SNMP Trap から装置名、peer、event を抽出する層
• classification_chain: LLM に置き換えると、障害カテゴリや重要度を分類する層
• summary_chain: NOC 担当者向けの短い説明を作る層
• action_chain: Runbook や RAG とつなげて一次対応案を作る層
• ticket_formatter: 既存のチケットシステムに渡しやすい JSON にする層

実務版では、classify_alert や summarize_for_noc の中身を LangChain の ChatPromptTemplate | llm | StrOutputParser に差し替えられます。

つまり、このミニプロジェクトは LangChain そのものの再実装ではありません。

LLM 処理を一つの巨大な関数にせず、NMS の障害対応フローに合わせて小さな部品に分ける練習です。

Output parser の役割

授業でよく出てきたのが StrOutputParser です。

chain = prompt | llm | StrOutputParser()

これは LLM のメッセージ出力から文字列を取り出してくれる parser です。

最初は、なくてもよいのではと思いました。

でも、chain を組み合わせると、出力の型が重要になります。

LLM の戻り値が AIMessage のままなのか。

文字列として次の prompt に渡すのか。

JSON として扱うのか。

この違いを曖昧にすると、後続処理で壊れます。

障害分類を本当にアプリケーションで使うなら、文字列より構造化された出力の方が扱いやすいです。

from pydantic import BaseModel, Field

class AlertClassification(BaseModel):
    severity: str = Field(description="critical, warning, info のいずれか")
    category: str = Field(description="routing, interface, monitoring, resource, unknown のいずれか")
    reason: str

LangChain では、モデルに structured output を指定することもできます。

structured_llm = llm.with_structured_output(AlertClassification)

structured_classify_chain = classify_prompt | structured_llm

これで、出力を Pydantic の model として扱えます。

result = structured_classify_chain.invoke({"alert_log": alert_log})

print(result.severity)
print(result.category)
print(result.reason)

NMS のような既存システムに組み込むなら、この差は大きいです。

自然文の回答は人間には読みやすいですが、システムには扱いづらいです。

人間向け:
router-core-01 で BGP peer down と SNMP timeout が同時に発生しています。

システム向け:
{
  "severity": "critical",
  "category": "routing",
  "device": "router-core-01"
}

LangChain の parser や structured output は、この「LLM の出力をアプリケーションのデータに戻す」部分を整理するために使えます。

RunnableParallel で並列に考える

LangChain には、RunnableParallel という仕組みもあります。

授業では、次のような形でした。

from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    summary=ChatPromptTemplate.from_template("{text}を1行で要約してください"),
    keywords=ChatPromptTemplate.from_template("{text}からキーワードを3つ抽出してください"),
)

これを障害ログ分析に置き換えると、同じ入力から複数の観点を同時に作るイメージになります。

たとえば、分類と要約は、必ずしも直列でなくてもよい場合があります。

from langchain_core.runnables import RunnableParallel

parallel_analysis = RunnableParallel(
    classification=classify_chain,
    short_summary=summary_chain,
)

ただし、この例では summary_chain が classification を必要としているので、そのままでは並列にできません。

そこで、分類に依存しない短い要約 chain を別に作ります。

quick_summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "障害ログを NOC 向けに1文で要約してください。"),
    ("human", "{alert_log}"),
])

quick_summary_chain = quick_summary_prompt | llm | StrOutputParser()

parallel_analysis = RunnableParallel(
    classification=classify_chain,
    quick_summary=quick_summary_chain,
)

result = parallel_analysis.invoke({"alert_log": alert_log})

結果は、次のような dict になります。

{
    "classification": "...",
    "quick_summary": "...",
}

業務アプリで考えると、これは少し便利です。

たとえば、障害チケット画面で次のような項目を同時に作れます。

左側:
障害分類

右側:
短い要約

下部:
次の確認候補

もちろん、並列にすれば常に速くなるとは限りません。API 呼び出し回数も増えます。

しかし、LLM 処理を「どの部品がどの入力から何を作るか」として見られるようになるのは大事です。

RunnableBranch で問い合わせを振り分ける

授業では、RunnableBranch を使った分岐も出てきました。

例として、質問内容に応じて技術サポート、料金、一般問い合わせに振り分ける chain がありました。

from langchain_core.runnables import RunnableBranch

tech_chain = ChatPromptTemplate.from_template(
    "技術支援チームです: {question}"
) | llm | StrOutputParser()

billing_chain = ChatPromptTemplate.from_template(
    "料金管理チームです: {question}"
) | llm | StrOutputParser()

general_chain = ChatPromptTemplate.from_template(
    "一般相談チームです: {question}"
) | llm | StrOutputParser()

def route_logic(x):
    text = x["question"]
    if "エラー" in text or "障害" in text:
        return "technical"
    elif "料金" in text or "請求" in text:
        return "billing"
    return "general"

branch = RunnableBranch(
    (lambda x: route_logic(x) == "technical", tech_chain),
    (lambda x: route_logic(x) == "billing", billing_chain),
    general_chain,
)

NMS の障害分析でも、この発想は使えます。

障害カテゴリに応じて chain を分けるのです。

routing:
BGP/OSPF/経路系の確認手順を優先

interface:
物理リンク、光レベル、対向ポートを優先

monitoring:
SNMP、監視経路、poller を優先

resource:
CPU、memory、process を優先

実装イメージはこうです。

routing_chain = ChatPromptTemplate.from_messages([
    ("system", "BGP/OSPF など routing 障害の一次対応案を作ってください。"),
    ("human", "{alert_log}"),
]) | llm | StrOutputParser()

interface_chain = ChatPromptTemplate.from_messages([
    ("system", "interface 障害の一次対応案を作ってください。"),
    ("human", "{alert_log}"),
]) | llm | StrOutputParser()

monitoring_chain = ChatPromptTemplate.from_messages([
    ("system", "SNMP や監視経路障害の一次対応案を作ってください。"),
    ("human", "{alert_log}"),
]) | llm | StrOutputParser()

def alert_route(x):
    text = x["alert_log"].lower()
    if "bgp" in text or "ospf" in text:
        return "routing"
    if "interface" in text or "gi0/" in text:
        return "interface"
    if "snmp" in text or "poller" in text:
        return "monitoring"
    return "general"

alert_branch = RunnableBranch(
    (lambda x: alert_route(x) == "routing", routing_chain),
    (lambda x: alert_route(x) == "interface", interface_chain),
    (lambda x: alert_route(x) == "monitoring", monitoring_chain),
    action_chain,
)

ここで大事なのは、分岐を全部 LLM に任せなくてもよいということです。

bgp や snmp のような明確なキーワードなら、アプリケーション側で分岐してもよいです。

LLM に任せるべきところと、コードで決めるべきところを分ける。

これは、これまでの Prompt Engineering や RAG でも繰り返し感じたことです。

Memory は便利だが、業務では注意が必要

LangChain では、会話履歴や memory も扱えます。

基本的な Gradio チャットの例では、history を使って過去の HumanMessage / AIMessage を組み立てるコードがありました。

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

def chat_with_llm(message, history):
    messages = [
        SystemMessage(content="あなたは親切な日本語アシスタントです。")
    ]

    for item in history:
        if item["role"] == "user":
            messages.append(HumanMessage(content=item["content"]))
        else:
            messages.append(AIMessage(content=item["content"]))

    messages.append(HumanMessage(content=message))

    return llm.invoke(messages).content

チャットボットとしては自然です。

しかし、NMS や障害対応で memory を使う場合は注意が必要です。

過去の会話を覚えていることは便利です。

運用者:
router-core-01 の BGP 障害を見ています。

LLM:
router-core-01 の BGP peer down として扱います。

運用者:
SNMP timeout も出ています。

このとき、LLM が前の文脈を覚えていれば、二つ目の発言だけでも router-core-01 の話だと分かります。

これは使いやすいです。

ただし、業務では memory が危険になる場面もあります。

前の障害チケットの情報が、次のチケットに混ざる。

別の装置名を引きずる。

古い判断を前提にしてしまう。

人間が訂正した内容を、どこまで保持するか曖昧になる。

障害対応では、文脈が混ざると危険です。

そのため、私は memory を使うなら、会話履歴を無制限に入れるのではなく、チケット単位、障害ID単位、セッション単位で明確に区切りたいです。

ticket_id = INC-20260310-001
device = router-core-01
session_scope = this_ticket_only

そして、LLM に渡す履歴も必要最小限にします。

直近のユーザー確認
現在の仮説
既に実施済みの確認
未確認の項目

単に過去会話を全部渡すのではなく、運用状態として整理してから渡す方が安全です。

この考え方は、後の LangGraph や Agent の状態管理にもつながります。

RAG と LangChain をつなげる

前回の RAG では、検索して context を作り、LLM に渡す関数を書きました。

LangChain を使うと、この流れも chain として組めます。

def format_docs(docs):
    return "\n\n".join(
        f"[source={doc.metadata.get('source')}, page={doc.metadata.get('page')}]\n"
        f"{doc.page_content}"
        for doc in docs
    )

retriever を用意します。

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

prompt を作ります。

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """
あなたは NMS 運用マニュアルに基づいて回答するアシスタントです。
必ず context に含まれる内容だけを使って回答してください。
context にない内容は「文書では確認できません」と答えてください。
回答の最後に参照した source を書いてください。
""".strip()),
    ("human", """
質問:
{question}

context:
{context}
""".strip()),
])

LCEL では、RunnablePassthrough を使って、入力をそのまま次に渡しながら context を作る形もあります。

from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

呼び出しはこうです。

answer = rag_chain.invoke(
    "router-core-01 で BGP peer down と SNMP timeout が同時に出ています。初動は？"
)

このコードを初めて見たときは、処理の流れを理解するのが少し難しく感じました。

でも、慣れると「入力がどう流れるか」を一つのパイプラインとして見られます。

question
  ├─ retriever -> format_docs -> context
  └─ passthrough -> question
       ↓
     prompt
       ↓
     llm
       ↓
     parser

RAG のように、検索、整形、プロンプト、生成がつながる処理では、この見える化が助かります。

Chain の中で何が起きているか追いづらい

LangChain を触っていて気になったのは、chain の中で何が実行されているか追いづらくなる場面があることです。

prompt | llm | parser だけなら分かります。

しかし、RAG のように dict、retriever、format 関数、RunnablePassthrough が入ると、最初は頭の中で流れを追うのが大変でした。

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

このコードを見て、すぐに「どこに何が渡るか」を理解できる人ばかりではないと思います。

そのため、私は学習中は次のように分解して確認する方がよいと感じました。

question = "BGP peer down の初動は？"

docs = retriever.invoke(question)
context = format_docs(docs)

prompt_value = rag_prompt.invoke({
    "question": question,
    "context": context,
})

llm_result = llm.invoke(prompt_value)
answer = StrOutputParser().invoke(llm_result)

これなら、各段階で print できます。

print(docs)
print(context)
print(prompt_value)
print(answer)

慣れてから chain にまとめればよいです。

LangChain は、処理を短く書ける一方で、デバッグ時には途中の値を見る意識が必要です。

特に業務システムでは、LLM の回答だけでなく、次の情報もログに残したいです。

• 入力
• 最終 prompt
• 使用モデル
• retrieved docs
• 出力
• parser の結果
• エラー
• token usage

これは前回の RAG の retrieval log と同じです。

抽象化でコードは短くなります。

でも、運用ログまで消してはいけません。

業務で使うなら

NMS や障害対応の業務アプリに LangChain を使うなら、私は次のように部品を分けたいです。

まず、入力の前処理です。

raw alert log
  ↓
device, interface, event_type, severity candidate を抽出
  ↓
LLM に渡しやすい alert_context にする

ここは、必ずしも LLM でなくてもよいです。

正規表現や既存 parser で取れるものは、コードで取った方が安定します。

次に、LLM chain です。

classification_chain
summary_chain
action_chain
rag_answer_chain
judge_chain

それぞれ責務を分けます。

classification_chain:
障害カテゴリと重要度を返す

summary_chain:
運用者向けに短くまとめる

action_chain:
一次確認項目を作る

rag_answer_chain:
Runbook に基づいて回答する

judge_chain:
回答が context に忠実か評価する

そして、アプリケーション側で orchestration します。

def handle_alert(alert_log: str) -> dict:
    classification = classification_chain.invoke({"alert_log": alert_log})

    related_docs = retriever.invoke(alert_log)
    context = format_docs(related_docs)

    summary = summary_chain.invoke({
        "alert_log": alert_log,
        "classification": classification,
    })

    action = rag_action_chain.invoke({
        "alert_log": alert_log,
        "classification": classification,
        "context": context,
    })

    return {
        "classification": classification,
        "summary": summary,
        "action": action,
        "sources": [doc.metadata for doc in related_docs],
    }

ここで、LangChain は LLM 処理の部品化に使います。

しかし、業務ルール、権限、保存、監査ログ、通知、承認フローは、既存のバックエンド側で管理したいです。

LLM に全部を任せるのではなく、既存システムの中に LLM 部品を差し込む。

この考え方が、NMS 開発者としては一番しっくりきました。

Tool Calling へのつながり

さらに LangChain では、@tool、StructuredTool、bind_tools を使って、LLM に外部処理を呼ばせる内容に進みました。

たとえば、天気や為替の例では、LLM が直接答えるのではなく、必要な tool call を作ります。

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """天気を照会する関数です"""
    weather_db = {
        "seoul": "晴れ, 22度, 湿度45%",
        "tokyo": "曇り, 19度, 湿度70%",
    }
    return weather_db.get(city, f"{city}: 情報なし")

llm_with_tools = llm.bind_tools([get_weather])
response = llm_with_tools.invoke("東京の天気は？")

この話は次回のテーマですが、LangChain の chain を先に理解しておくと、Tool Calling も見えやすくなります。

LLM アプリは、だんだん次のように広がります。

prompt | llm | parser
  ↓
retriever とつなぐ
  ↓
branch / parallel で流れを作る
  ↓
tool を呼ぶ
  ↓
agent や graph で状態を管理する

つまり、LangChain は単に「プロンプトを楽に書くライブラリ」ではありません。

LLM、retriever、parser、tool を同じような部品として扱い、組み合わせるための土台です。

NMS の例で言えば、次のような部品をつなげられます。

障害ログ
  ↓
classification_chain
  ↓
Runbook retriever
  ↓
rag_answer_chain
  ↓
NMS device info tool
  ↓
ticket summary chain

この先に、Mock NMS API や CMDB、チケットシステムがつながっていきます。

まとめ

LangChain は、最初は少し抽象化が多く感じました。

ChatPromptTemplate、StrOutputParser、RunnableParallel、RunnableBranch、RunnablePassthrough など、名前だけ見ると覚えることが多いです。

しかし、実際に障害ログ分析の例に当てはめると、役割はかなり実務的でした。

prompt、LLM、parser を chain にする。

分類、要約、一次対応案を別々の部品にする。

同じ入力から複数の分析を並列に作る。

障害カテゴリに応じて処理を分岐する。

RAG の retriever と回答生成をつなげる。

会話履歴を必要な範囲で扱う。

こうした部品化によって、LLM アプリは単発の API 呼び出しから、業務アプリケーションの一部に近づきます。

このとき大事なのは、LLM 処理も普通のアプリケーション部品として設計することです。

ただし、小さな実験では LangChain が過剰になることもあります。

単純な要約だけなら、直接 API を呼ぶ方が分かりやすいです。

LangChain を使う価値が出るのは、処理を組み合わせる必要が出てきたときです。

NMS の現場で考えるなら、障害分類、Runbook 検索、一次対応案、出典表示、評価、チケット要約のような処理を、独立した chain として持てるのは大きいです。

今回の結論は、こうです。

LangChain は LLM を賢くする道具ではありません。

LLM を業務アプリの中で扱いやすい部品にするための道具です。

次回は、この部品化した LLM に外部 API を使わせる Tool Calling を整理します。

LLM が NMS API や CMDB、チケットシステムを呼べるようになると、単なる回答生成から、業務アシスタントに少し近づきます。