カスタムAIソリューションの開発：企業の課題に合わせたAIシステムの作り方

2026年10月8日に、NISTの現行AI RMFページを再読しました。AI RMFは任意利用で、AI製品・サービス・システムの設計、開発、利用、評価へ信頼性の観点を組み込む枠組みです。AI RMF 1.0は改訂中で、2026年4月7日には重要インフラ向けプロファイルのコンセプトノートが公開されています。この記事の7段階、保留フロー、架空20件の分岐は実務教材であり、NISTやLangChainが個別の実装・正確性・安全性を保証したことを示しません。

企業向けのカスタムAIは、モデルを呼べば完成する製品ではありません。誰の情報を扱い、どこまで自動化し、判断できないときに誰へ返すかを、業務の手順として設計する仕事です。既存サービスや通常の検索・ルール処理で足りる場合もあります。

この記事では、社内資料から回答候補を作る小さな業務を例に、要件、構成、実装、評価、運用をつなぎます。数値とコード中のデータは著者が作った架空教材です。実企業の導入成果や、外部モデルの性能測定ではありません。

30秒で要点

• カスタム開発の判断は「AIを使いたい」ではなく、既存手段との差、対象外、費用と確認作業で行います。
• 認可・入力上限・回答の検査をモデルの外側に置き、根拠がないものは保留します。
• 時間短縮と誤りを別々に測り、変更後も同じ評価例で比較します。改善率を導入前から約束しません。

既存サービスと自作を、条件をそろえて比べる

表が収まらない場合は、横にスクロールできます

自作なら必ず安い、品質が高い、競争優位になるとは限りません。初期費用だけでなく、確認、訂正、問い合わせ対応まで比較します。LLMの基礎はChatGPTの基本、実装の前提はプログラミング入門で確認できます。

要件から改善までの七段階

次の工程は一方向の通過儀礼ではありません。テストで根拠不足が分かれば、要件やデータの選び方へ戻ります。NIST AI RMFの「Govern・Map・Measure・Manage」も、固定順のチェックリストとして示されてはいません。[S1][S2]

1 要件: 利用者、対象業務、対象外、失敗時の戻し先
2 設計: 権限、データ、モデル、検査、保存期間の境界
3 試作: 小さな固定データで流れを確認
4 テスト: 正常・拒否・根拠不足・期限超過を比較
5 実装: 認証、監視、利用上限、取り消し手段を接続
6 運用: 誤り、保留、再作業、費用を観測
7 改善: 変更理由を記録し、同じ評価例で再確認

要件は「できること」と「しないこと」を対にする

1 概要: 社内手順書から回答候補と参照IDを提示する
2 課題・目標: 探す手間を減らす。最終判断は担当者が行う
3 機能: 許可済み資料の検索、根拠表示、根拠なしの保留
4 非機能: 入力上限、期限、監視、記録の最小化を決める
5 制約: 個人情報を試作データに入れない。外部送信は別審査
6 リスク: 誤回答、資料の古さ、権限漏れ、過剰利用への対応

応答2秒未満やエラー1%未満といった数字は、全業務に共通の標準ではありません。対象端末、データ量、測定区間、許容する失敗を定義してから目標を決めます。失敗を防ぐ計画も合わせて確認してください。

構成はモデルの前後に判断を置く

ブラウザ: 問い合わせを送る。プロバイダーの秘密キーは持たない
API境界: 認証 → 入力の型・長さ → 利用上限
業務処理: 利用者の権限で資料を絞る → 検索 → 回答候補
検査: 参照ID・出力形式・期限を確認 → 採用候補 または 保留
記録: 処理種別・結果・時間。質問と回答の全文は既定で保存しない

検索資料に「以前の指示を無視せよ」と書かれていても、業務上の権限が増える設計にしません。検索拡張や入力の整形だけでプロンプト注入を防げるわけではありません。権限と実行可能な操作をコード側で制限します。[S9]

表が収まらない場合は、横にスクロールできます

保留は2＋5＋3＝10件、担当者へ渡す10件と合わせて20件です。通過率50%は、この架空データの経路の割合であり、精度や削減率ではありません。

小さな実装で、成功以外の結果を返す

以下のPython教材は標準ライブラリだけを使います。各Python枠を掲載順に一つのファイルへ置いて実行できます。外部API、LangChain、埋め込みモデルは呼びません。実際のSDK接続はAPI実装ガイドと、使用版の公式仕様で別に確認します。

1. 回答候補を作る部分を交換可能にする

class AIService:
    def __init__(self, generate):
        self.generate = generate  # ローカル教材では固定応答関数

    def process(self, question, allowed_ids):
        if not isinstance(question, str) or not 1 <= len(question.strip()) <= 200:
            return {"status": "reject", "reason": "input"}
        if not allowed_ids:
            return {"status": "hold", "reason": "no_evidence"}
        try:
            value = self.generate(question.strip(), tuple(allowed_ids))
        except TimeoutError:
            return {"status": "hold", "reason": "timeout"}
        if not isinstance(value, dict):
            return {"status": "hold", "reason": "shape"}
        answer, refs = value.get("answer"), value.get("refs")
        if (not isinstance(answer, str) or not 1 <= len(answer) <= 500
                or not isinstance(refs, list) or not refs
                or any(not isinstance(x, str) or x not in allowed_ids for x in refs)):
            return {"status": "hold", "reason": "output"}
        return {"status": "candidate", "answer": answer, "refs": refs}

service = AIService(lambda q, ids: {"answer": "担当窓口へ確認してください", "refs": [ids[0]]})
assert service.process("申請先は？", ["D1"])["status"] == "candidate"
assert service.process("", ["D1"])["status"] == "reject"
assert service.process("申請先は？", [])["status"] == "hold"

参照IDが許可済みでも、文章が資料に忠実かはこの検査だけでは分かりません。candidateは回答の正しさを保証する状態名ではなく、次の確認へ渡せるという意味です。

2. HTTPの境界を業務処理と分ける

def handle_request(method, payload, session_user, remaining, allowed_ids):
    # session_user / remaining / allowed_ids はサーバーで確定した値
    if method != "POST":
        return 405, {"error": "method"}
    if session_user is None:
        return 401, {"error": "authentication"}
    if remaining <= 0:
        return 429, {"error": "quota"}
    if not isinstance(payload, dict) or set(payload) != {"question"}:
        return 400, {"error": "payload"}
    result = service.process(payload["question"], allowed_ids)
    return (400 if result["status"] == "reject" else 200), result

assert handle_request("GET", {}, "u1", 1, ["D1"])[0] == 405
assert handle_request("POST", {}, None, 1, ["D1"])[0] == 401
assert handle_request("POST", {"question": "申請先は？"}, "u1", 1, ["D1"])[0] == 200

これはFlask等へ接続する前の判断関数です。実認証、同時利用時の原子的な上限消費、リクエスト全体のサイズ制限、秘密管理、通信期限は実サーバー側の責務として残ります。予期しない例外の文字列をそのまま利用者へ返しません。

3. 観測は本文の収集と分ける

from time import perf_counter

def observed(call, events):
    started = perf_counter()
    try:
        result = call()
    except Exception:
        events.append({"event": "candidate_error", "seconds": perf_counter() - started})
        raise
    events.append({"event": "candidate_end", "status": result["status"],
                   "seconds": perf_counter() - started})
    return result

events = []
observed(lambda: service.process("申請先は？", ["D1"]), events)
assert set(events[0]) == {"event", "status", "seconds"}
assert events[0]["seconds"] >= 0

ログの目的は失敗の発見です。質問や回答、アクセストークンを既定で保存する必要はありません。別途詳しい調査が必要なら、閲覧権限・保管期限・マスキングを決めます。この例は入力全文を記録しないことを示す教材で、漏えいが起きないことの監査ではありません。

改善を、評価データと結ぶ

結果を集める → 失敗理由を分ける → 優先順位を決める
→ 変更を一つ試す → 同じ評価例で比べる → 採用・保留・取り消し

class ImprovementCycle:
    def __init__(self):
        self.metrics = []

    def add(self, seconds, held):
        if not isinstance(seconds, (int, float)) or seconds < 0:
            raise ValueError("seconds")
        self.metrics.append((seconds, bool(held)))

    def analyze(self):
        if not self.metrics:
            return {"count": 0, "mean_seconds": None, "held": 0}
        return {"count": len(self.metrics),
                "mean_seconds": sum(x[0] for x in self.metrics) / len(self.metrics),
                "held": sum(x[1] for x in self.metrics)}

cycle = ImprovementCycle()
assert cycle.analyze()["mean_seconds"] is None
for seconds, held in [(1, False), (2, True), (3, False)]:
    cycle.add(seconds, held)
assert cycle.analyze() == {"count": 3, "mean_seconds": 2.0, "held": 1}

平均2秒だけでは、最も遅い利用者の体験も、候補の正しさも分かりません。実運用では分位点、期限超過、根拠不足、誤った採用、確認時間などを別の指標として定義します。評価条件の記録と継続的な検査はNISTの測定機能でも重視されています。[S2]

業務知識ベースは、検索前に権限で絞る

class KnowledgeBaseSearch:
    def __init__(self, documents):
        self.documents = documents

    def search(self, query, allowed_ids):
        if not isinstance(query, str) or not query.strip():
            return []
        return [d for d in self.documents
                if d["id"] in allowed_ids and query in d["text"]]

kb = KnowledgeBaseSearch([
    {"id": "D1", "text": "申請は担当窓口へ確認"},
    {"id": "D2", "text": "申請に関する非公開の架空メモ"},
])
assert [d["id"] for d in kb.search("申請", {"D1"})] == ["D1"]
assert kb.search("申請", set()) == []

ここは単純な文字列検索です。意味検索やRAGの性能を再現してはいません。資料を外部埋め込みへ送る実装に進むなら、送信対象と提供条件を別に確認します。企業向けRAGとベクトルDBで検索の設計を、LangChain入門で部品の接続を整理できます。

三つの用途で、成果と保留を測り分ける

表が収まらない場合は、横にスクロールできます

例えば架空の検索作業が従来120分、新手順の検索40分＋確認30分＋再作業20分なら、比較対象は90分です。削減率は（120−90）÷120＝25%。検索だけの40分と比べる66.7%とは測定範囲が違います。実績ではなく、比較範囲を学ぶための計算です。

次へ進む条件は「導入した」ではなく、対象業務と許容する誤りについて合意し、保留時に戻せることです。AIエージェント開発、プロジェクト事例の読み方、First byteのAIアプローチへ進む際も、方法の説明と効果の実証を分けてください。

参考資料と確認範囲

• S1 NIST AI RMF：任意のリスク管理枠組み。導入成果の保証ではありません。
• S2 AI RMF Core：文脈、測定、リスク対応の必要範囲。
• S3 LangChain概要・S4 v1移行：フレームワークの現行入口と旧版の区別。この記事のPythonはローカル教材です。
• S9 OWASP Prompt Injection：入力と取得資料の指示を信用しすぎないための境界。

確認日2026年10月8日。NISTの現行案内を追加で本文再読し、必要な本文範囲の読解とURLの取得確認は別に行いました。ブラウザ操作、外部モデル、実認証、顧客業務での効果はここで検証していません。