そのまま試せる
RAG×Jev統合の実装コード

# 1) 社内文書をスキルツリーへコンパイル（分岐率10、トップレベルは最大8）
python -m corpus2skill --input ./docs --output ./compiled --p 10 --max-top 8

# 2) Jev 用の判定表を作る（コンパイルし直したら毎回実行）
python3 build_maps.py ./compiled/.claude/skills ./jev_maps

# 3) キーは環境変数で渡す（コードやリポジトリに書かない）
export TYPESAFE_API_KEY="jv_live_..."      # Jev（再販エンドポイントの例）
export JEV_DECIDE_URL="https://jevtypesafeai.com/api/v1/decide"
export ANTHROPIC_API_KEY="sk-ant-..."      # 回答生成用
python3 answer.py "インボイスを再発行する手順を教えてください"

rag-jev/
├── compiled/                  # Corpus2Skill のコンパイル出力
│   ├── .claude/skills/        #   SKILL.md と INDEX.md の階層
│   └── documents.json         #   チャンク分割していない原文
├── jev_maps/                  # build_maps.py が作る判定表
│   ├── cluster_criteria.json
│   └── index_map.json
├── build_maps.py              # 前処理（コンパイルのたびに1回）
├── jev_client.py              # Jev /v1/decide の薄いクライアント
├── navigator.py               # 2ホップ探索（System 1）
├── answer.py                  # 回答生成（System 2）
├── calibrate.py               # 閾値校正とコスト集計
└── qa/benchmark.jsonl         # 自社QAセット

"""build_maps.py — Corpus2Skill のコンパイル出力から Jev 用の判定表を作る前処理

出力:
  cluster_criteria.json  {cluster_id: 判定基準の説明}   ← Hop1 の choice に渡す
  index_map.json         {cluster_id: {doc_id: タイトル/要約}} ← Hop2 の noul に渡す
INDEX.md の書式は Corpus2Skill のバージョンや設定で変わり得るため、
DOC_LINE の正規表現は自社の出力を開いて合わせてください。
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

DOC_LINE = re.compile(r"^\s*[-*]\s*`?(?P<doc_id>[\w\-.:]+)`?\s*[:：|-]\s*(?P<title>.+)$")


def first_paragraph(md: str, limit: int = 400) -> str:
    body = re.sub(r"^---.*?---\s*", "", md, flags=re.S)       # front matter を除く
    lines = [ln.strip() for ln in body.splitlines() if ln.strip() and not ln.startswith("#")]
    return " ".join(lines)[:limit]


def main(skills_root: str, out_dir: str) -> None:
    root, out = Path(skills_root), Path(out_dir)
    criteria: dict[str, str] = {}
    index_map: dict[str, dict[str, str]] = {}
    for top in sorted(p for p in root.iterdir() if p.is_dir()):   # トップレベルスキル = クラスタ
        skill_md = top / "SKILL.md"
        if not skill_md.exists():
            continue
        cid = top.name
        criteria[cid] = first_paragraph(skill_md.read_text(encoding="utf-8"))
        docs: dict[str, str] = {}
        for index_md in top.rglob("INDEX.md"):                     # 配下のリーフを平坦化
            for line in index_md.read_text(encoding="utf-8").splitlines():
                m = DOC_LINE.match(line)
                if m:
                    docs[m.group("doc_id")] = m.group("title").strip()[:200]
        index_map[cid] = docs
    if len(criteria) > 254:
        raise SystemExit("choice の選択肢は255個まで（other を含む）。--max-top を見直してください")
    criteria["other"] = "上記のどの分野にも当てはまらない質問、または社内文書では答えられない質問"
    out.mkdir(parents=True, exist_ok=True)
    (out / "cluster_criteria.json").write_text(json.dumps(criteria, ensure_ascii=False, indent=2), "utf-8")
    (out / "index_map.json").write_text(json.dumps(index_map, ensure_ascii=False, indent=2), "utf-8")
    print(f"clusters={len(criteria) - 1} docs={sum(map(len, index_map.values()))}")


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

{
  "model": "jev-1.13.0",
  "state": "インボイスを再発行する手順を教えてください",
  "questions": {
    "target_cluster": {
      "type": "choice",
      "instructions": "この質問に答えるために参照すべき知識クラスタを1つ選ぶ。",
      "criteria": {
        "billing": "請求・支払い・インボイス発行に関する規定と手順",
        "hr": "就業規則・休暇・給与・福利厚生",
        "other": "上記のどれにも当てはまらない質問"
      }
    },
    "is_urgent": { "type": "noul", "instructions": "質問者は至急の対応を求めている。" }
  }
}

{
  "model": "jev-1.13.0",
  "answers": {
    "target_cluster": {
      "choice": "billing",
      "confidence": 0.91,
      "probabilities": { "billing": 0.91, "hr": 0.02, "other": 0.07 }
    },
    "is_urgent": { "noul": 0.18 }
  },
  "usage": { "input_tokens": 412, "cost_usd": 0.00017, "credits_remaining_usd": 4.83 }
}

"""jev_client.py — Jev の /v1/decide を標準ライブラリだけで呼ぶ薄いクライアント"""
from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request

# 既定は再販 JevTypeSafeAI.com。公式API・互換ゲートウェイを使う場合は環境変数で差し替える
JEV_DECIDE_URL = os.environ.get("JEV_DECIDE_URL", "https://jevtypesafeai.com/api/v1/decide")
JEV_MODEL = os.environ.get("JEV_MODEL", "jev-1.13.0")  # 本番はバージョン固定（閾値がずれないように）
JEV_API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
RETRYABLE = {502, 503, 504}


class JevError(RuntimeError):
    def __init__(self, status: int, body: str):
        super().__init__(f"Jev HTTP {status}: {body[:200]}")
        self.status = status


def decide(state, questions: dict, *, timeout: float = 10.0, retries: int = 2) -> dict:
    """state と型付き質問を送り、{"answers": ..., "usage": ...} を返す。"""
    payload = json.dumps({"model": JEV_MODEL, "state": state, "questions": questions},
                         ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        JEV_DECIDE_URL, data=payload, method="POST",
        headers={"Authorization": f"Bearer {JEV_API_KEY}",
                 "Content-Type": "application/json",
                 "User-Agent": "rag-jev-navigator/1.0"})
    for attempt in range(retries + 1):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as res:
                return json.loads(res.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", "replace")
            # 401=キー不正 / 402=残高不足 / 403=停止 / 413=入力過大 は再試行しない
            if exc.code not in RETRYABLE or attempt == retries:
                raise JevError(exc.code, body) from exc
        except urllib.error.URLError:
            if attempt == retries:
                raise
        time.sleep(0.5 * (2 ** attempt))  # 502 などは指数バックオフ
    raise RuntimeError("unreachable")

"""navigator.py — Jev でスキルツリーを2ホップ探索し、原文全文を取り出す"""
from __future__ import annotations

import json
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from jev_client import decide

DATA = Path("./jev_maps")
CLUSTER_CRITERIA: dict[str, str] = json.loads((DATA / "cluster_criteria.json").read_text("utf-8"))
INDEX_MAP: dict[str, dict[str, str]] = json.loads((DATA / "index_map.json").read_text("utf-8"))
# Corpus2Skill の documents.json（チャンク分割されていない原文）。形式は {doc_id: 本文} を想定
DOCUMENTS: dict[str, str] = json.loads(Path("./compiled/documents.json").read_text("utf-8"))

DOC_BATCH = 40  # 入力上限（state＋全質問で64kトークン）を超えないよう候補を分割


def _noul_batch(query: str, items: list[tuple[str, str]]) -> dict[str, float]:
    questions = {
        doc_id: {"type": "noul",
                 "instructions": f"この質問に正しく答えるには、文書『{summary}』の本文を読む必要がある。"}
        for doc_id, summary in items
    }
    res = decide(query, questions)
    return {doc_id: float(a.get("noul", 0.0)) for doc_id, a in res.get("answers", {}).items()}


def navigate(query: str, *, high_conf: float = 0.75, min_conf: float = 0.35,
             branch_cutoff: float = 0.15, doc_threshold: float = 0.55, top_k: int = 3) -> dict:
    # ---- Hop 1: クラスタ選択（choice）と緊急度（noul）を1リクエストで同時に判定 ----
    hop1 = decide(query, {
        "target_cluster": {
            "type": "choice",
            "instructions": "この質問に答えるために参照すべき知識クラスタを1つ選ぶ。",
            "criteria": CLUSTER_CRITERIA,
        },
        "is_urgent": {
            "type": "noul",
            "instructions": "質問者は業務停止や期限切迫により、至急の対応を求めている。",
        },
    })
    ans = hop1["answers"]["target_cluster"]
    is_urgent = float(hop1["answers"]["is_urgent"].get("noul", 0.0))
    choice, conf = ans["choice"], float(ans.get("confidence", 0.0))

    # ゲート1: 該当なし・確信度が極端に低い → 生成LLMを呼ばずに打ち切る
    if choice == "other" or conf < min_conf:
        return {"status": "abstain", "documents": {}, "is_urgent": is_urgent, "confidence": conf}

    # ゲート2: 高確信なら単独、中確信なら上位最大3クラスタへ投機的ファンアウト
    if conf >= high_conf:
        clusters = [choice]
    else:
        ranked = sorted(ans.get("probabilities", {}).items(), key=lambda kv: kv[1], reverse=True)
        clusters = [cid for cid, p in ranked if cid != "other" and p >= branch_cutoff][:3] or [choice]

    # ---- Hop 2: 候補文書ごとの要否を noul で並列判定 ----
    candidates: dict[str, str] = {}
    for cid in clusters:
        candidates.update(INDEX_MAP.get(cid, {}))
    items = list(candidates.items())
    batches = [items[i:i + DOC_BATCH] for i in range(0, len(items), DOC_BATCH)]
    scores: dict[str, float] = {}
    with ThreadPoolExecutor(max_workers=4) as pool:
        for part in pool.map(lambda b: _noul_batch(query, b), batches):
            scores.update(part)

    picked = sorted((d for d in scores if scores[d] >= doc_threshold),
                    key=lambda d: scores[d], reverse=True)[:top_k]
    docs = {d: DOCUMENTS[d] for d in picked if d in DOCUMENTS}
    return {
        "status": "success" if docs else "no_matching_doc",
        "documents": docs,
        "clusters": clusters,
        "scores": {d: scores[d] for d in picked},
        "is_urgent": is_urgent,
        "confidence": conf,
    }

"""answer.py — 取り出した原文全文と質問だけを生成LLMに渡して回答を書かせる"""
from __future__ import annotations

import json
import os
import urllib.request

from navigator import navigate

ANTHROPIC_URL = "https://api.anthropic.com/v1/messages"
ANSWER_MODEL = os.environ.get("ANSWER_MODEL", "claude-sonnet-4-6")

SYSTEM = (
    "あなたは社内ナレッジの回答担当です。与えられた<document>の内容だけを根拠に日本語で答えてください。"
    "例外規定・ただし書きがあれば必ず反映し、各文の末尾に根拠の doc_id を [doc_id] 形式で付けます。"
    "文書から答えが分からない場合は、分からないと答えてください。"
)


def _call_llm(question: str, docs: dict[str, str]) -> str:
    context = "\n".join(f'<document id="{d}">\n{text}\n</document>' for d, text in docs.items())
    body = json.dumps({
        "model": ANSWER_MODEL,
        "max_tokens": 1200,
        "system": SYSTEM,
        "messages": [{"role": "user", "content": f"{context}\n\n質問: {question}"}],
    }, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(ANTHROPIC_URL, data=body, method="POST", headers={
        "x-api-key": os.environ["ANTHROPIC_API_KEY"],
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    })
    with urllib.request.urlopen(req, timeout=60) as res:
        data = json.loads(res.read().decode("utf-8"))
    return "".join(block.get("text", "") for block in data.get("content", []))


def answer(question: str) -> dict:
    nav = navigate(question)
    if nav["status"] != "success":
        text = "社内文書から該当する情報を見つけられませんでした。担当窓口へお問い合わせください。"
    else:
        text = _call_llm(question, nav["documents"])
    return {"answer": text, "sources": list(nav["documents"]), "is_urgent": nav["is_urgent"],
            "confidence": nav["confidence"]}


if __name__ == "__main__":
    import sys
    print(json.dumps(answer(sys.argv[1]), ensure_ascii=False, indent=2))

# Corpus2Skill 標準（LLMナビゲーション）での評価。品質指標とトークン・費用が記録される
python -m corpus2skill.eval --output-dir ./compiled --qa ./qa/benchmark.jsonl --output eval_results.json

"""calibrate.py — 自社QAセットで Jev の閾値を校正し、Jev 側コストを集計する"""
from __future__ import annotations

import json
import sys

import jev_client
import navigator

_orig_decide = jev_client.decide
COST = {"usd": 0.0, "calls": 0}


def _metered(state, questions, **kw):
    res = _orig_decide(state, questions, **kw)
    COST["usd"] += float((res.get("usage") or {}).get("cost_usd", 0.0))
    COST["calls"] += 1
    return res


navigator.decide = _metered  # navigator 内の呼び出しを計測付きに差し替え


def evaluate(qa_path: str, doc_threshold: float) -> dict:
    recall_sum = scored = total = abstain = 0
    with open(qa_path, encoding="utf-8") as f:
        for line in f:
            qa = json.loads(line)                      # {"question": ..., "doc_ids": [...]}
            nav = navigator.navigate(qa["question"], doc_threshold=doc_threshold)
            total += 1
            abstain += nav["status"] != "success"
            expected = set(qa.get("doc_ids", []))
            if expected:                               # 必要な文書のうち何割を取れたか
                recall_sum += len(set(nav["documents"]) & expected) / len(expected)
                scored += 1
    return {"threshold": doc_threshold, "context_recall": round(recall_sum / max(scored, 1), 3),
            "abstain_rate": round(abstain / max(total, 1), 3),
            "jev_usd_per_query": round(COST["usd"] / max(total, 1), 6)}


if __name__ == "__main__":
    for t in (0.45, 0.55, 0.65):
        COST.update(usd=0.0, calls=0)
        print(evaluate(sys.argv[1], t))