Claude API で契約書レビュー自動化システムを構築する — リスク条項検出・差分追跡・修正提案を本番運用する実装ガイド

# contract_loader.py
# 解決すること: テキスト埋め込みPDFとスキャン画像PDFを自動判別し、
# それぞれに最適な抽出経路に振り分ける。
import base64
import pdfplumber
from pathlib import Path
from anthropic import Anthropic
 
client = Anthropic()
TEXT_THRESHOLD = 100  # 1ページあたり100文字未満ならスキャン型と判定
 
def extract_text_pages(pdf_path: Path) -> list[str]:
    """テキスト埋め込み型から抽出。失敗時は空リストを返す。"""
    pages: list[str] = []
    try:
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                text = page.extract_text() or ""
                pages.append(text)
    except Exception as e:
        # 破損PDFは空で返し、上位層でスキャン型として扱う
        print(f"[loader] pdfplumber failed: {e}")
        return []
    return pages
 
def is_scanned(pages: list[str]) -> bool:
    if not pages:
        return True
    avg = sum(len(p) for p in pages) / len(pages)
    return avg < TEXT_THRESHOLD
 
def extract_via_vision(pdf_path: Path) -> list[str]:
    """スキャンPDFをClaude Visionでページ単位にOCR。"""
    import fitz  # PyMuPDF
    pages: list[str] = []
    doc = fitz.open(pdf_path)
    for i, page in enumerate(doc):
        pix = page.get_pixmap(dpi=200)
        b64 = base64.b64encode(pix.tobytes("png")).decode()
        try:
            resp = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4000,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "image", "source": {
                            "type": "base64", "media_type": "image/png", "data": b64,
                        }},
                        {"type": "text", "text": "このページを忠実にテキスト化してください。レイアウトは保持し、表は『|』区切りで再現してください。本文以外の挿入は禁止です。"},
                    ],
                }],
            )
            pages.append(resp.content[0].text)
        except Exception as e:
            print(f"[loader] vision failed on page {i}: {e}")
            pages.append("")
    return pages
 
def load_contract(pdf_path: Path) -> list[str]:
    """戻り値: ページごとのテキスト配列。"""
    pages = extract_text_pages(pdf_path)
    if is_scanned(pages):
        print("[loader] scanned PDF detected, falling back to Vision")
        pages = extract_via_vision(pdf_path)
    return pages
 
if __name__ == "__main__":
    p = Path("samples/nda_v3.pdf")
    pages = load_contract(p)
    print(f"loaded {len(pages)} pages, total {sum(len(x) for x in pages)} chars")
# 期待出力例: loaded 12 pages, total 28433 chars

# clause_splitter.py
# 解決すること: ページ配列を条項単位に分割し、ID付きの構造化データに変換する。
from pydantic import BaseModel, Field, ValidationError
from anthropic import Anthropic
import json
 
client = Anthropic()
 
class Clause(BaseModel):
    clause_id: str = Field(..., description="例: '第3条第2項'")
    title: str
    body: str
    page_range: list[int]
 
class ClauseList(BaseModel):
    clauses: list[Clause]
 
SPLIT_PROMPT = """あなたは法務文書を構造化するアシスタントです。
入力された契約書テキストを条項単位に分割し、以下のJSONスキーマで出力してください。
 
スキーマ:
{
  "clauses": [
    {"clause_id": "第N条第M項", "title": "条項タイトル", "body": "本文(原文ママ)", "page_range": [開始ページ, 終了ページ]}
  ]
}
 
ルール:
- 本文は原文をそのまま転記してください。要約や言い換えは禁止です
- 条項番号がない場合は文書順に '段落-N' を割り当ててください
- 表紙・目次・署名欄は出力に含めないでください
- 出力はJSONのみ。前後の説明文は禁止です"""
 
def split_into_clauses(pages: list[str], max_retry: int = 2) -> ClauseList:
    joined = "\n".join(f"[Page {i+1}]\n{p}" for i, p in enumerate(pages))
    last_err = ""
    for attempt in range(max_retry + 1):
        prompt = SPLIT_PROMPT
        if last_err:
            prompt += f"\n\n前回の出力は次の理由で却下されました: {last_err}\n再度出力してください。"
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=8000,
            messages=[
                {"role": "user", "content": prompt + "\n\n[契約書本文]\n" + joined},
            ],
        )
        raw = resp.content[0].text.strip()
        # コードブロック装飾を除去
        if raw.startswith("```"):
            raw = raw.split("```")[1].lstrip("json\n").rstrip()
        try:
            parsed = ClauseList.model_validate_json(raw)
            return parsed
        except (ValidationError, json.JSONDecodeError) as e:
            last_err = str(e)[:300]
            print(f"[splitter] attempt {attempt+1} failed: {last_err}")
    raise RuntimeError(f"条項分割に{max_retry+1}回失敗しました: {last_err}")
 
if __name__ == "__main__":
    pages = ["本契約は…", "第1条 (定義) 本契約において…", "第2条 (秘密保持) 受領者は…"]
    result = split_into_clauses(pages)
    for c in result.clauses:
        print(f"{c.clause_id}: {c.title} (pages {c.page_range})")
# 期待出力例:
# 第1条: 定義 (pages [1, 1])
# 第2条: 秘密保持 (pages [1, 2])

# risk_analyzer.py
# 解決すること: 各条項を「リスクカテゴリ × レベル」で分類し、根拠と引用を必須化する。
from pydantic import BaseModel, Field
from typing import Literal
from anthropic import Anthropic
 
client = Anthropic()
 
RiskCategory = Literal[
    "liability_cap",      # 責任制限
    "indemnification",    # 補償義務
    "ip_assignment",      # 知的財産
    "termination",        # 解除条件
    "confidentiality",    # 秘密保持範囲
    "governing_law",      # 準拠法・管轄
    "data_protection",    # データ保護
    "auto_renewal",       # 自動更新
]
RiskLevel = Literal["low", "medium", "high"]
 
class RiskFinding(BaseModel):
    clause_id: str
    category: RiskCategory
    level: RiskLevel
    rationale: str = Field(..., description="日本語で200文字以内")
    quote: str = Field(..., description="判定根拠となる原文の一部抜粋")
 
class ClauseAnalysis(BaseModel):
    clause_id: str
    findings: list[RiskFinding]
 
ANALYZE_PROMPT_TEMPLATE = """あなたは契約書のリスク監査担当です。次の条項を分析し、JSONで出力してください。
 
【許可されたリスクカテゴリ】
- liability_cap: 損害賠償の上限が買主側に著しく不利
- indemnification: 過大または非対称な補償義務
- ip_assignment: 知的財産の包括的譲渡や永続的使用許諾
- termination: 一方的・即時解除や違約金条項
- confidentiality: 秘密保持の範囲・期間が不当
- governing_law: 不利な準拠法・管轄
- data_protection: GDPR/個人情報保護法に抵触する規定
- auto_renewal: 通知期間が短い/料金変動を伴う自動更新
 
【出力スキーマ】
{
  "clause_id": "<条項ID>",
  "findings": [
    {
      "clause_id": "<同上>",
      "category": "<上記8カテゴリのいずれか>",
      "level": "low|medium|high",
      "rationale": "200文字以内で根拠を述べる",
      "quote": "判定根拠となる原文の一部"
    }
  ]
}
 
【ルール】
- リスクが見当たらない場合は findings を [] にしてください
- カテゴリは上記8つに限定。当てはまらない懸念は出力しないでください
- quote は原文からの直接抜粋のみ（要約・言い換え禁止）
- 出力はJSONのみ"""
 
def analyze_clause(clause_id: str, body: str) -> ClauseAnalysis:
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT_TEMPLATE + f"\n\n【条項】\nclause_id: {clause_id}\nbody:\n{body}",
        }],
    )
    raw = resp.content[0].text.strip()
    if raw.startswith("```"):
        raw = raw.split("```")[1].lstrip("json\n").rstrip()
    return ClauseAnalysis.model_validate_json(raw)

# diff_engine.py
# 解決すること: 同一契約の旧版・新版を条項単位で比較し、変更種別を分類する。
from dataclasses import dataclass
from difflib import SequenceMatcher
 
@dataclass
class ClauseDiff:
    clause_id: str
    change: str  # "added" | "removed" | "modified" | "unchanged"
    similarity: float
    old_body: str | None = None
    new_body: str | None = None
 
def diff_clauses(old: list[dict], new: list[dict], modify_threshold: float = 0.85) -> list[ClauseDiff]:
    old_map = {c["clause_id"]: c["body"] for c in old}
    new_map = {c["clause_id"]: c["body"] for c in new}
    results: list[ClauseDiff] = []
    for cid, body in new_map.items():
        if cid not in old_map:
            results.append(ClauseDiff(cid, "added", 0.0, None, body))
            continue
        sim = SequenceMatcher(None, old_map[cid], body).ratio()
        if sim >= modify_threshold:
            results.append(ClauseDiff(cid, "unchanged" if sim > 0.99 else "modified", sim, old_map[cid], body))
        else:
            # 類似度が極端に低い場合は新規挿入扱い
            results.append(ClauseDiff(cid, "modified", sim, old_map[cid], body))
    for cid, body in old_map.items():
        if cid not in new_map:
            results.append(ClauseDiff(cid, "removed", 0.0, body, None))
    return results

# remediation.py
# 解決すること: 検出されたリスクから、書き換え案・理由・根拠の3点セットを生成する。
from anthropic import Anthropic
from pydantic import BaseModel
 
client = Anthropic()
 
class Remediation(BaseModel):
    proposed_clause: str
    why_safer: str
    references: list[str]
 
PROMPT = """あなたは社内法務担当の補助AIです。次の条項について、検出されたリスクを解消する修正案を作成してください。
 
【入力】
- 元の条項: {body}
- 検出リスク: {category} ({level})
- 根拠引用: {quote}
- 社内ポリシー要約: 当社は責任上限を契約金額の1倍に制限。GDPRに準拠した個人情報処理を要求。
 
【出力スキーマ】
{{
  "proposed_clause": "書き換え後の条文(日本語、必要十分な長さで)",
  "why_safer": "なぜこの書き換えがリスクを下げるか(200文字以内)",
  "references": ["民法第415条", "社内ポリシーv2.3 §4"]
}}
JSONのみを出力してください。"""
 
def propose_remediation(body: str, category: str, level: str, quote: str) -> Remediation:
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": PROMPT.format(body=body, category=category, level=level, quote=quote),
        }],
    )
    raw = resp.content[0].text.strip()
    if raw.startswith("```"):
        raw = raw.split("```")[1].lstrip("json\n").rstrip()
    return Remediation.model_validate_json(raw)