Jevを会計・監査で使う｜摘要の分類と仕訳テストのトリアージ

import pandas as pd

df = pd.read_csv("journal.csv")
print(df.head())

        date voucher_no account  amount        description dept created_by
0 2026-03-31     J-1001     交際費  132000  ㈱大和田商事 接待 銀座  営業部         山田
1 2026-03-31     J-1002     交際費   98000  ゴルフコンペ参加費 取引先同伴  営業部         山田
2 2026-03-28     J-1003     会議費    4800  部内打合せ 弁当代 6名  経理部         佐藤
3 2026-03-28     J-1004   旅費交通費   38600  大阪出張 新幹線往復  営業部         鈴木
4 2026-03-29     J-1005   支払手数料  500000  コンサルティング料 3月分 経営企画         田中

import pandas as pd
from typesafe_sdk import Choice, Noul, Score, TypeSafeClient

QUESTIONS = {
    "expense_type": Choice(
        instructions="この支出の費用区分はどれか。摘要と勘定科目だけから判断する",
        criteria={
            "meeting": "社内または取引先との打合せにかかる飲食・会場費",
            "entertainment": "取引先の接待・贈答・供応。ゴルフや会食を含む",
            "travel": "出張・移動にかかる交通費と宿泊費",
            "supplies": "文具・備品などの消耗品",
            "outsourcing": "外部への業務委託、コンサルティング",
            "other": "上記のいずれにも当てはまらない、または判断できない",
        },
    ),
    "description_detail": Score(
        instructions="摘要が、第三者が取引内容を特定できる程度に具体的か",
        criteria=[
            "「一式」「その他」など、内容がまったく特定できない",
            "費目は分かるが、相手先や目的が書かれていない",
            "相手先・目的・数量のいずれかが具体的に書かれている",
        ],
    ),
    "private_use_doubt": Noul(
        instructions="摘要の文言だけから見て、個人的な支出が混ざっている疑いを説明する必要がある",
    ),
}

def ask(client: TypeSafeClient, row: pd.Series) -> dict:
    state = {
        "勘定科目": row["account"],
        "摘要": row["description"],
        "計上部門": row["dept"],
    }
    res = client.system_one(state=state, questions=QUESTIONS)
    a = res.answers
    return {
        "expense_type": a["expense_type"].choice,
        "type_conf": a["expense_type"].confidence,
        "detail_score": a["description_detail"].score,
        "private_doubt": a["private_use_doubt"].noul,
        "input_tokens": res.usage.input_tokens,
    }


df = pd.read_csv("journal.csv")
with TypeSafeClient() as client:
    judged = pd.DataFrame([ask(client, row) for _, row in df.iterrows()])

out = pd.concat([df, judged], axis=1)

voucher_no account  amount     description  expense_type  type_conf  detail_score  private_doubt
    J-1001     交際費  132000    ㈱大和田商事 接待 銀座 entertainment      0.952          1.81           0.04
    J-1002     交際費   98000 ゴルフコンペ参加費 取引先同伴 entertainment      0.952          1.81           0.87
    J-1003     会議費    4800    部内打合せ 弁当代 6名       meeting      0.952          1.81           0.04
    J-1005   支払手数料  500000   コンサルティング料 3月分   outsourcing      0.952          1.81           0.04
    J-1007     交際費  180000          懇親会 一式 entertainment      0.952          0.28           0.87
    J-1008      雑費   50000             お車代         other      0.016          1.81           0.87
    J-1010   業務委託費  990000     業務委託 3月分 一式   outsourcing      0.952          0.28           0.87

def route(row: pd.Series) -> str:
    if row["type_conf"] < 0.5:
        return "人が判断"
    if row["type_conf"] < 0.9:
        return "採用（要確認フラグ）"
    return "自動採用"


out["routing"] = out.apply(route, axis=1)
print(out["routing"].value_counts())

def triage(out: pd.DataFrame) -> pd.DataFrame:
    out = out.copy()

    # 数値と日付の判定はコード側で持つ
    out["date"] = pd.to_datetime(out["date"])
    out["is_month_end"] = out["date"].dt.is_month_end
    out["over_limit"] = out["amount"] >= 100_000

    # Jev の判断はしきい値でフラグに落とす
    out["needs_review"] = out["private_doubt"] >= 0.7
    out["vague_description"] = out["detail_score"] < 1.0
    out["type_mismatch"] = out["type_conf"] < 0.5

    reasons = []
    for _, r in out.iterrows():
        rs = []
        if r["needs_review"]:
            rs.append("私的流用の疑いあり")
        if r["vague_description"]:
            rs.append("摘要が不十分")
        if r["type_mismatch"]:
            rs.append("費用区分を特定できず")
        if r["over_limit"] and r["is_month_end"]:
            rs.append("期末日・高額")
        reasons.append(" / ".join(rs))

    out["reason"] = reasons
    return out.loc[out["reason"] != "", ["voucher_no", "account", "amount", "description", "reason"]]


print(triage(out).to_string(index=False))

voucher_no account  amount     description                      reason
    J-1001     交際費  132000    ㈱大和田商事 接待 銀座                      期末日・高額
    J-1002     交際費   98000 ゴルフコンペ参加費 取引先同伴                   私的流用の疑いあり
    J-1007     交際費  180000          懇親会 一式 私的流用の疑いあり / 摘要が不十分 / 期末日・高額
    J-1008      雑費   50000             お車代      私的流用の疑いあり / 費用区分を特定できず
    J-1010   業務委託費  990000     業務委託 3月分 一式 私的流用の疑いあり / 摘要が不十分 / 期末日・高額

import asyncio
import pandas as pd
from typesafe_sdk import AsyncTypeSafeClient, RetryPolicy, TypeSafeAPIError

CONCURRENCY = 16
RETRY = RetryPolicy(max_retries=4, backoff_initial=0.5, backoff_max=8.0)


async def ask_one(client, sem, row):
    state = {"勘定科目": row["account"], "摘要": row["description"], "計上部門": row["dept"]}
    async with sem:
        try:
            res = await client.system_one(state=state, questions=QUESTIONS, retry=RETRY)
        except TypeSafeAPIError as e:
            return {"voucher_no": row["voucher_no"], "error": type(e).__name__}
    a = res.answers
    return {
        "voucher_no": row["voucher_no"],
        "expense_type": a["expense_type"].choice,
        "type_conf": a["expense_type"].confidence,
        "detail_score": a["description_detail"].score,
        "private_doubt": a["private_use_doubt"].noul,
        "input_tokens": res.usage.input_tokens,
        "error": None,
    }


async def judge_all(df: pd.DataFrame) -> pd.DataFrame:
    sem = asyncio.Semaphore(CONCURRENCY)
    async with AsyncTypeSafeClient() as client:
        results = await asyncio.gather(*(ask_one(client, sem, r) for _, r in df.iterrows()))
    return pd.DataFrame(results)


res = asyncio.run(judge_all(df))
print("エラー:", res["error"].notna().sum())

uniq = (
    df[["account", "description", "dept"]]
    .drop_duplicates()
    .reset_index(drop=True)
)
uniq["voucher_no"] = [f"U-{i:06d}" for i in range(len(uniq))]  # 判定用の一時キー

judged = asyncio.run(judge_all(uniq))
uniq_judged = uniq.merge(judged, on="voucher_no")

merged = df.merge(
    uniq_judged.drop(columns=["voucher_no"]),
    on=["account", "description", "dept"],
    how="left",
)

out["model"] = res.model
out["judged_at"] = pd.Timestamp.now()
out.to_csv("judged_2026-03.csv", index=False)