業務自動化×LLM の安全な統合パターン — 監査ログ・PII マスキング・ガードレール

前回の続きから

前回、n8n・Zapier・Make を比較した末に、SMB の業務自動化基盤として n8n を選んだ話を書いた。ところが記事を出した翌週、最初の PoC を回し始めたクライアントの情報システム担当から、こんな指摘が飛んできた。

「これ、顧客の氏名と電話番号、そのまま Anthropic に飛んでませんか?」

反論できなかった。ノートに描いた設計図では、監査ログや PII マスキングの層がスカスカだったのだ。正直、動くところまで作って安心してしまっていた。「動くこと」と「業務で使えること」の間には、思ったより広い川がある。

本稿はその川を越えるためのメモだ。n8n + Claude API を前提に、PII マスキング middleware、監査ログを Postgres / Supabase に流すパターン、そして LLM 出力の guardrail (JSON schema 検証 + retry) を、実行できるコードで残しておく。図解と最小の sample repo は連載完結時にまとめて公開する予定。

設計の全体像 — 3 つの層

整理すると、やることは 3 つしかない。

• 入力側の防御: PII を検出・マスキングしてから LLM に渡す

• 実行の記録: 誰が・いつ・どの workflow で・どのモデルを・何 token 使ったかを Postgres に残す

• 出力側の防御: LLM の返答を JSON schema で検証し、失敗したら retry する

この 3 つを n8n のノード配置に落とすと、Claude ノード (もしくは HTTP Request ノード) の前後に薄い middleware を挟む形になる。前段が masking、後段が validation、両方が audit log を叩く。図に描くと退屈なくらい単純だ。単純なことを、単純に配置する。それが本題。

1. PII マスキング middleware

Claude を叩くノードの直前に、Python の小さな masking サービスを挟む。実装は Microsoft Presidio が現時点でいちばんストレスが少ない。n8n 公式にも「Postgres に原文を置き、マスク後だけを Claude に流す」テンプレートがあるが、あの構成の masking 部分を Presidio に置き換えたイメージだと思えばいい。

ただし Presidio は英語前提の recognizer が強く、日本語の氏名検出は正直まだ弱い。そこで筆者は、電話番号・郵便番号・メールアドレスといった機械的に拾えるものは正規表現で足し、氏名だけは既存の Ja NER (GiNZA) を後段に噛ませている。完璧を目指すと沼だが、この 4 種で実案件に流れる PII の 8 割は掴める。「8 割」と書くと少なく感じるが、100 件の顧客データが 20 件まで減れば、目視レビューが現実的な作業量に変わる。

from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern
from presidio_anonymizer import AnonymizerEngine
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
日本の電話番号 (090/080/070-XXXX-XXXX, 03-XXXX-XXXX 等)
jp_phone = PatternRecognizer(
supported_entity="JP_PHONE",
patterns=[Pattern(name="jp_phone", regex=r"0\d{1,4}-\d{1,4}-\d{4}", score=0.85)],
supported_language="ja",
)
郵便番号
jp_zip = PatternRecognizer(
supported_entity="JP_ZIP",
patterns=[Pattern(name="jp_zip", regex=r"\d{3}-\d{4}", score=0.7)],
supported_language="ja",
)
analyzer.registry.add_recognizer(jp_phone)
analyzer.registry.add_recognizer(jp_zip)
def mask(text: str) -> tuple[str, list[dict]]:
results = analyzer.analyze(text=text, language="ja")
masked = anonymizer.anonymize(text=text, analyzer_results=results).text
entities = [
{"type": r.entity_type, "start": r.start, "end": r.end, "score": r.score}
for r in results
]
return masked, entities

これを FastAPI で薄く包み、n8n の HTTP Request ノードから叩く。n8n の Function (Code) ノードで JavaScript から直接処理する手もあるが、モデルや辞書を差し替える運用を考えると、独立プロセスの方が後で楽になる。ここで 1 度ハマった。n8n の Docker ネットワーク内に置いたつもりが host 側にバインドしていて、丸半日 502 と戦った。docker-compose.yml を最初に統一しておくと事故が減る。地味だが本当に効く。

2. 監査ログを Postgres / Supabase に流す

PGAudit は基盤としては優秀だが、LLM 呼び出しの粒度で欲しいのは「どの workflow の、どの user の、どの prompt が、どういう出力になったか」であり、これはアプリケーション層で書く方が素直だ。liteLLM の Supabase callback を使う手もあるが、n8n から呼ぶ薄いサービス側に自前で書いた方が、後述の retry ロジックと 1 つのトランザクション境界で扱えるので都合がいい。

schema はミニマルに始める。あとから列を足すのは簡単だが、最初に肥大化させると更新が回らなくなる。

create table llm_audit_log (
id            bigserial primary key,
created_at    timestamptz not null default now(),
workflow_id   text not null,
user_ref      text,               -- 顧客の内部 ID など (原文氏名は入れない)
model         text not null,
input_masked  text,
output        jsonb,
pii_types     text[],             -- ['JP_PHONE','EMAIL_ADDRESS'] 等
status        text not null,      -- 'ok' | 'schema_error' | 'timeout' | ...
latency_ms    int,
input_tokens  int,
output_tokens int
);
create index on llm_audit_log (workflow_id, created_at desc);
create index on llm_audit_log (status) where status <> 'ok';

input_masked にはマスク後の文字列を入れる。原文は絶対に入れない。もし原文の突き合わせが必要なら、Presidio の結果と原文のペアを別スキーマ (RLS で強く保護) に置く。同じテーブルに混ぜると、監査ログの閲覧権限を営業に出した瞬間に PII が漏れる。分離が正解。

ログ書き込みは Python 側の decorator にしておくと呼び出し漏れが減る。

import time, os
from functools import wraps
from supabase import create_client
sb = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_SERVICE_KEY"])
class SchemaError(RuntimeError): ...
def audit(workflow_id: str):
def deco(fn):
@wraps(fn)
def wrapper(payload: dict):
t0 = time.perf_counter()
status, output, usage = "ok", None, {}
try:
output, usage = fn(payload)
return output
except SchemaError:
status = "schema_error"
raise
except Exception:
status = "error"
raise
finally:
sb.table("llm_audit_log").insert({
"workflow_id": workflow_id,
"user_ref": payload.get("user_ref"),
"model": payload.get("model"),
"input_masked": payload.get("input_masked"),
"output": output,
"pii_types": payload.get("pii_types", []),
"status": status,
"latency_ms": int((time.perf_counter() - t0) * 1000),
"input_tokens": usage.get("input_tokens"),
"output_tokens": usage.get("output_tokens"),
}).execute()
return wrapper
return deco

細かい話だが、Supabase は 2026-07-09 以降に作られた新規プロジェクトでは log_connections がデフォルト off になっている。既存プロジェクトから引き継いだ運用マニュアルどおりに設定していると、「接続ログが出ない」となるのは大体これだ。筆者はこれで小 1 時間溶かした。

3. Guardrail — JSON schema 検証と retry

LLM の返答を後段のシステムに流し込むなら、9 割方 JSON で受けたい。Anthropic の SDK は tool_use や structured output で clean JSON を促せるが、それでも稀に schema 違反は起きる。retry 戦略を持たずに本番に出すと、月に数回のインシデントで運用側が疲弊する。

基本形はシンプル。Pydantic モデルで validate し、失敗したらエラー文面を次の user message に添えて再送する。

from pydantic import BaseModel, ValidationError
from anthropic import Anthropic
client = Anthropic()
class InvoiceSummary(BaseModel):
total: int
currency: str
line_items: list[str]
SYSTEM = "あなたは請求書要約器。必ず JSON のみを返す。前置き・後書き禁止。"
def summarize(masked_text: str, max_retries: int = 2) -> InvoiceSummary:
messages = [{"role": "user", "content": masked_text}]
last_err = None
for attempt in range(max_retries + 1):
resp = client.messages.create(
model="claude-sonnet-4-6",
system=SYSTEM,
max_tokens=1024,
messages=messages,
)
text = resp.content[0].text
try:
return InvoiceSummary.model_validate_json(text)
except ValidationError as e:
last_err = e
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": (
"直前の出力が schema と一致しなかった。\n"
f"エラー:\n{e}\n\n"
"JSON のみを返す。コードブロック等の装飾も禁止。"
),
})
raise SchemaError(str(last_err))

max_retries は 2 で止める。3 以上に伸ばしても、経験上、成功率は目に見えて上がらない。上がらないのに料金だけ増える。retry で治らない失敗は prompt を直すサインだと思っていい。

もう一つ、retry を打つ前に rescue parsing を挟むと安上がりだ。LLM が json ... で囲んで返してきた、末尾に日本語で余計な説明を付けてきた——そういう軽症の malformed は、正規表現で JSON 部分だけ切り出せば救える。追加の 1 token も使わずに成功に転じる。

import re, json
_JSON_RE = re.compile(r"{.*}", re.DOTALL)
def rescue(text: str):
m = _JSON_RE.search(text)
if not m:
return None
try:
return json.loads(m.group(0))
except json.JSONDecodeError:
return None

コスト感覚もメモしておく。Sonnet 4.6 で 5K token の入力を 3 回 retry すると概算で $0.06 前後。1 件なら気にならないが、月 10 万件走らせる workflow で 3 % が retry を踏むと $180 / 月。rescue parsing で半分削れば $90。「30 % の retry を rescue でどうにかする」と言うと地味だが、規模が出ると効いてくる。

n8n workflow への組み込み

ここまで書いたものを 1 本の n8n workflow に並べると、流れはこうなる。

• Trigger (Webhook or Cron)

• HTTP Request → PII masking service (/mask)

• HTTP Request → LLM caller service (/summarize) ← Pydantic 検証 + retry を内包

• Function → 後段の Google Sheets / Notion / Slack へ配布

• Error Trigger → status <> 'ok' を検索する Slack 通知

Claude ノードを直接使わない構成に見えるが、これは意図的だ。中間に自前サービスを挟むことで、モデルの差し替え・masking 辞書の追加・prompt caching の有効化を、n8n の workflow を触らずに済ませられる。n8n は「業務の流れ」を書く場所として使い、LLM 周りの技術的な進化はサービス側で吸収する。この分離が半年後に効いてくる。前回の ツール選定の記事で n8n を「変更に強いから選んだ」と書いたが、その「変更に強さ」を実際に発揮させるのは、こういう境界の引き方だ。

まとめと次回

ツールを選び、middleware で守り、ログで記録し、schema で受け止める。派手な話は何もない。ただ、これを最初の PoC の段階で組み込んでおくかどうかで、6 ヶ月後の運用の消耗度がまったく違ってくる。

次回 (第 3 回) は、こうして安全に回り始めた自動化が、ビジネスとしてどれだけ効いているのかを測る話を書く。メトリクス設計と Grafana 連携で、ROI を「なんとなく効いてる」から「数字で示せる」に持っていく回になる予定だ。

筆者は 5years+ で韓国・日本の企業向けに AI・業務自動化の実装を担当している。上のコード片は実案件から抽出したもので、GitHub sample は連載完結時にまとめて公開予定。