Qwen3.5-2B-UD-japanese-imatrix developed by dahara1@webbigdata

この量子化シリーズの詳細はdahara1/Qwen3.5-4B-UD-japanese-imatrixを見てください
please see dahara1/Qwen3.5-4B-UD-japanese-imatrix for detail.

サンプルサーバーコマンド（マルチモーダル/画像認識を含む) / sample sever upload command(Multimodal demo including image recognition)

./llama-server -m Qwen3.5-2B-UD-Q6_K_XL.gguf --top_p 0.8 --top_k 20 --min_p 0.0 -c 8000 --mmproj mmproj-f32.gguf --presence_penalty 1.5 --chat-template-kwargs '{"enable_thinking":true}' --jinja --chat-template-file chat_template.jinja --port 8080 --host 0.0.0.0

サンプルクライアントスクリプト（マルチモーダル/画像認識を含む) / sample client demo script.(Multimodal demo including image recognition)

![Qwen3.5 2B ローカル AI デモ動画]

"""
ATMガーディアン — 振り込め詐欺防止AIデモ v3
LLM Function Call ツールチェーンデモ

判定を個別のシンプルなプロンプトに分離：
  Phase 1: 前回と同一人物か？（前回画像がある場合のみ）
  Phase 2: スマートフォンを持っているか？
  Phase 3: 高齢者に見えるか？（Phase 2がYESの場合のみ）

該当するものがあれば、LLMにツール選択させる：
  - notify_emergency_support: 緊急お客様サポート（振り込め詐欺の疑い）
  - notify_support: お客様サポート（操作に苦戦）

使い方:
  python atm_guardian.py
  python atm_guardian.py --debug
"""

import json
import os
import shutil
import base64
import time
import glob
import argparse
from datetime import datetime
from openai import OpenAI

# ============================================================
# 設定
# ============================================================
client = OpenAI(base_url="http://127.0.0.1:8080/v1", api_key="dummy", timeout=120)
MODEL = "qwen3.5"

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
IMAGES_DIR = os.path.join(SCRIPT_DIR, "images")
PREVIOUS_DIR = os.path.join(IMAGES_DIR, "previous")
NORMAL_DIR = os.path.join(IMAGES_DIR, "normal")

# ============================================================
# デモシナリオ（5分間隔の撮影を模擬）
# ============================================================

# all these image are AI generated images.
SCENARIOS = [
    {"time": "10:00", "image": "1.jpg",  "desc": "スマホを持ちながらATM操作しているお婆さん"},
    {"time": "10:05", "image": "2.jpg",  "desc": "スマホを持ちながらATM操作しているお爺さん"},
# need more prompt engineering
#    {"time": "10:10", "image": "3.jpg",  "desc": "スマホを持ちながらATM操作しているお婆さん(1と同じ人) + 銀行員が対応中"},
    {"time": "10:15", "image": "4.jpg",  "desc": "ATM操作している若い女性"},
# need more prompt engineering
#    {"time": "10:20", "image": "5.jpg",  "desc": "ATM操作しているお婆さんを横から覗いているマスクの男"},
    {"time": "10:25", "image": "6.jpg",  "desc": "スマホを見ながらATM操作をしているビジネスマン"},
    {"time": "10:30", "image": "8.jpg",  "desc": "ATM操作しているお爺さん"},
    {"time": "10:35", "image": "9.jpg",  "desc": "ATM操作しているお爺さん(8と同じ人)"},
    {"time": "10:40", "image": "10.jpg", "desc": "ATM操作している若い男性"},
]

# ============================================================
# 色定義
# ============================================================
class C:
    BOLD       = "\033[1m"
    DIM        = "\033[2m"
    RESET      = "\033[0m"
    CYAN       = "\033[96m"
    YELLOW     = "\033[93m"
    GREEN      = "\033[92m"
    BLUE       = "\033[94m"
    MAGENTA    = "\033[95m"
    RED        = "\033[91m"
    WHITE      = "\033[97m"
    BG_RED     = "\033[41m"
    BG_GREEN   = "\033[42m"
    BG_BLUE    = "\033[44m"
    BG_MAGENTA = "\033[45m"
    BG_CYAN    = "\033[46m"
    BG_YELLOW  = "\033[43m"
    GRAY       = "\033[90m"

# ============================================================
# 画像ユーティリティ
# ============================================================
def encode_image(path):
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def get_mime(path):
    ext = os.path.splitext(path)[1].lower()
    return {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}.get(ext, "image/jpeg")

def make_image_content(path):
    """画像ファイルをOpenAI互換のimage_url要素に変換"""
    b64 = encode_image(path)
    mime = get_mime(path)
    return {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}

# ============================================================
# ディレクトリ管理
# ============================================================
def init_dirs():
    for d in [PREVIOUS_DIR, NORMAL_DIR]:
        if os.path.exists(d):
            shutil.rmtree(d)
        os.makedirs(d, exist_ok=True)

def get_previous_image():
    files = glob.glob(os.path.join(PREVIOUS_DIR, "*"))
    return files[0] if files else None

def set_previous_image(src_path):
    for f in glob.glob(os.path.join(PREVIOUS_DIR, "*")):
        os.remove(f)
    dst = os.path.join(PREVIOUS_DIR, os.path.basename(src_path))
    shutil.copy2(src_path, dst)


def move_to_normal(src_path):
    dst = os.path.join(NORMAL_DIR, os.path.basename(src_path))
    shutil.copy2(src_path, dst)

# ============================================================
# ツール実装
# ============================================================
def notify_support(reason):
    return json.dumps({
        "status": "notified", "to": "お客様サポート",
        "reason": reason,
        "timestamp": datetime.now().isoformat(),
    }, ensure_ascii=False)

def notify_emergency_support(reason):
    return json.dumps({
        "status": "notified", "to": "緊急お客様サポート",
        "reason": reason, "priority": "CRITICAL",
        "timestamp": datetime.now().isoformat(),
    }, ensure_ascii=False)

def execute_tool(func_name, args):
    if func_name == "notify_support":
        return notify_support(args.get("reason", ""))
    elif func_name == "notify_emergency_support":
        return notify_emergency_support(args.get("reason", ""))
    else:
        return json.dumps({"error": f"Unknown tool: {func_name}"}, ensure_ascii=False)

# ============================================================
# ツール定義
# ============================================================
tools = [
    {
        "type": "function",
        "function": {
            "name": "notify_support",
            "description": "お客様サポート係に連絡します。ATM操作に苦戦している人（長時間滞在など）を発見した場合に使用します。",
            "parameters": {
                "type": "object",
                "properties": {
                    "reason": {"type": "string", "description": "連絡理由の詳細"},
                },
                "required": ["reason"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "notify_emergency_support",
            "description": "緊急お客様サポートに通報します。振り込め詐欺の疑いが濃厚な場合に使用します。",
            "parameters": {
                "type": "object",
                "properties": {
                    "reason": {"type": "string", "description": "通報理由の詳細"},
                },
                "required": ["reason"],
            },
        },
    },
]


# ============================================================
# ストリーミングLLM呼び出し（判定用 — ツールなし）
# ============================================================
def ask_judge(messages, label, debug=False):
    """YES/NO判定をLLMに聞く。ストリーミングで思考過程も表示。"""
    print(f"\n  {C.BOLD}{C.CYAN}🔍 判定: {label}{C.RESET}")

    if debug:
        print(f"    {C.DIM}メッセージ数: {len(messages)}{C.RESET}")
        for i, msg in enumerate(messages):
            content = msg.get("content", "")
            if isinstance(content, list):
                img_count = sum(1 for c in content if isinstance(c, dict) and c.get("type") == "image_url")
                texts = [c.get("text", "")[:40] for c in content if isinstance(c, dict) and c.get("type") == "text"]
                print(f"    {C.DIM}  [{i}] {msg['role']}: {texts} + {img_count}画像{C.RESET}")
            elif isinstance(content, str):
                preview = content[:60] + "..." if len(content) > 60 else content
                print(f"    {C.DIM}  [{i}] {msg['role']}: \"{preview}\"{C.RESET}")

    try:
        stream = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            temperature=0.7,
            stream=True,
            max_tokens=4800,
        )
    except Exception as e:
        print(f"    {C.RED}❌ API ERROR: {e}{C.RESET}")
        return None

    full_content = ""
    in_reasoning = False
    in_content = False

    for chunk in stream:
        delta = chunk.choices[0].delta if chunk.choices else None
        if not delta:
            continue

        reasoning_text = getattr(delta, "reasoning_content", None)
        if reasoning_text:
            if not in_reasoning:
                in_reasoning = True
                print(f"    {C.MAGENTA}{C.DIM}💭 <think>{C.RESET}")
                print(f"    {C.MAGENTA}{C.DIM}", end="", flush=True)
            print(f"{C.MAGENTA}{C.DIM}{reasoning_text}{C.RESET}", end="", flush=True)

        if delta.content:
            text = delta.content
            full_content += text
            if in_reasoning:
                in_reasoning = False
                print(f"{C.RESET}")
                print(f"    {C.MAGENTA}{C.DIM}💭 </think>{C.RESET}")
            if not in_content:
                in_content = True
                print(f"    {C.WHITE}→ ", end="", flush=True)
            print(f"{C.WHITE}{text}{C.RESET}", end="", flush=True)

    if in_reasoning:
        print(f"{C.RESET}")
        print(f"    {C.MAGENTA}{C.DIM}💭 </think>{C.RESET}")
    if in_content:
        print(f"{C.RESET}")

    # YES/NO判定を抽出
    answer = full_content.strip().upper()
    is_yes = answer.startswith("YES")

    if not is_yes and not answer.startswith("NO"):
        # YESでもNOでもない場合、内容にYESが含まれるかで判定
        is_yes = "YES" in answer

    if is_yes:
        print(f"    {C.YELLOW}⚡ 判定結果: YES{C.RESET}")
    else:
        print(f"    {C.GREEN}✅ 判定結果: NO{C.RESET}")

    return is_yes


# ============================================================
# ストリーミングLLM呼び出し（ツール選択用）
# ============================================================
def call_with_tools(messages, debug=False):
    """LLMにツール選択させる。ストリーミング表示。"""
    print(f"\n  {C.BOLD}{C.YELLOW}🤖 ツール選択中...{C.RESET}")

    if debug:
        print(f"    {C.DIM}メッセージ数: {len(messages)}{C.RESET}")

    try:
        stream = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=tools,
            tool_choice="required",
            temperature=0.7,
            stream=True,
        )
    except Exception as e:
        print(f"    {C.RED}❌ API ERROR: {e}{C.RESET}")
        return None, None, None

    full_content = ""
    full_reasoning = ""
    tool_calls_map = {}
    in_reasoning = False
    in_content = False

    for chunk in stream:
        delta = chunk.choices[0].delta if chunk.choices else None
        if not delta:
            continue

        reasoning_text = getattr(delta, "reasoning_content", None)
        if reasoning_text:
            full_reasoning += reasoning_text
            if not in_reasoning:
                in_reasoning = True
                print(f"    {C.MAGENTA}{C.DIM}💭 <think>{C.RESET}")
                print(f"    {C.MAGENTA}{C.DIM}", end="", flush=True)
            print(f"{C.MAGENTA}{C.DIM}{reasoning_text}{C.RESET}", end="", flush=True)

        if delta.content:
            text = delta.content
            full_content += text
            if in_reasoning:
                in_reasoning = False
                print(f"{C.RESET}")
                print(f"    {C.MAGENTA}{C.DIM}💭 </think>{C.RESET}")
            if not in_content:
                in_content = True
                print(f"    {C.WHITE}💬 ", end="", flush=True)
            print(f"{C.WHITE}{text}{C.RESET}", end="", flush=True)

        if delta.tool_calls:
            for tc_delta in delta.tool_calls:
                idx = tc_delta.index
                if idx not in tool_calls_map:
                    tool_calls_map[idx] = {"id": tc_delta.id or "", "name": "", "arguments": ""}
                if tc_delta.id:
                    tool_calls_map[idx]["id"] = tc_delta.id
                if tc_delta.function:
                    if tc_delta.function.name:
                        tool_calls_map[idx]["name"] = tc_delta.function.name
                    if tc_delta.function.arguments:
                        tool_calls_map[idx]["arguments"] += tc_delta.function.arguments

    if in_reasoning:
        print(f"{C.RESET}")
        print(f"    {C.MAGENTA}{C.DIM}💭 </think>{C.RESET}")
    if in_content:
        print(f"{C.RESET}")

    tool_calls_list = [tool_calls_map[idx] for idx in sorted(tool_calls_map.keys())]
    return full_content, full_reasoning, tool_calls_list


# ============================================================
# 判定プロンプト構築
# ============================================================
JUDGE_SYSTEM = "あなたはカメラ画像を分析する専門家です。イロハ銀行渋谷店のATMで撮影された画像を解析してください。与えられた質問に対してYESまたはNOのみで回答してください。 最初の単語は必ずYESかNOにしてください。その後に短い理由を添えてください。"

def build_previous_check(current_image_path, previous_image_path):
    """前回同一人物チェック: 5分前と同じ人がまだいるか？"""
    return [
        {"role": "system", "content": JUDGE_SYSTEM},
        {"role": "user", "content": [
            {"type": "text", "text": "2枚の画像の人物が同一人物かを判定してください。1枚目が今回のATM利用者、2枚目が5分前のATM利用者です。\n判定基準：顔立ち、性別、年齢層、髪型、髪の色、持ち物で比較してください。\n注意：服装や髪型が似ているだけでは同一人物と判定しないでください。「何をしているか？」や「何処か？」は判定に使わないでください。\n同一人物の場合はYESと答えます"},
            make_image_content(current_image_path),
            make_image_content(previous_image_path),
        ]},
    ]

def build_phone_check(current_image_path):
    """スマホチェック: ATM利用者がスマートフォンを持っているか？"""
    return [
        {"role": "system", "content": JUDGE_SYSTEM},
        {"role": "user", "content": [
            {"type": "text", "text": "この画像のATM利用者はスマートフォン(電話)等の通話装置を手に持っていますか？持っている場合はYESと回答してください。"},
            make_image_content(current_image_path),
        ]},
    ]

def build_elderly_check(current_image_path):
    """高齢者チェック: ATM利用者が高齢者に見えるか？"""
    return [
        {"role": "system", "content": JUDGE_SYSTEM},
        {"role": "user", "content": [
            {"type": "text", "text": "この画像のATM利用者は明らかにカジュアルな格好をした高齢者(70歳以上)に見えますか？白髪、しわ、姿勢、服装など外見的特徴から判断してください。高齢者の場合はYESと回答してください。"},
            make_image_content(current_image_path),
        ]},
    ]


# ============================================================
# ツール選択プロンプト構築
# ============================================================
def build_tool_call_message(detection_type, detail):
    """検知内容に基づいてLLMにツール選択させるメッセージ"""
    if detection_type == "same_person_long_stay":
        instruction = (
            "ATMで同一人物が長時間（5分以上）滞在しています。操作に苦戦している可能性があります。\n"
            f"状況: {detail}\n"
            "適切なツールを使って対応してください。"
        )
    elif detection_type == "elderly_with_phone":
        instruction = (
            "高齢者がスマートフォンを持ちながらATM操作をしています。振り込め詐欺の疑いがあります。\n"
            f"状況: {detail}\n"
            "適切なツールを使って対応してください。"
        )
    else:
        instruction = f"状況: {detail}\n適切なツールを使って対応してください。"

    return [
        {
            "role": "system",
            "content": (
                "あなたはATM監視AIです。報告された状況に対して適切なツールを呼び出してください。\n"
            ),
        },
        {"role": "user", "content": instruction},
    ]


# ============================================================
# シーン実行
# ============================================================
def run_scene(scene_index, scenario, debug=False):
    scene_time = scenario["time"]
    image_name = scenario["image"]
    image_path = os.path.join(IMAGES_DIR, image_name)
    desc = scenario["desc"]

    # ヘッダー表示
    print(f"\n{'='*62}")
    print(f"{C.BOLD}{C.BG_CYAN}{C.WHITE} 📷 シーン {scene_index + 1} — {scene_time} {C.RESET}")
    print(f"  {C.CYAN}📁 画像: {image_name}{C.RESET}")
    print(f"  {C.DIM}{desc}{C.RESET}")
    print(f"{'='*62}")

    if not os.path.exists(image_path):
        print(f"  {C.RED}❌ 画像ファイルが見つかりません: {image_path}{C.RESET}")
        return None

    # 状態表示
    prev_image = get_previous_image()

    if prev_image:
        print(f"  {C.DIM}📂 前回画像: {os.path.basename(prev_image)}{C.RESET}")
    else:
        print(f"  {C.DIM}📂 前回画像: なし（初回）{C.RESET}")

    detection_type = None
    detection_detail = ""

    # ─── Phase 1: 前回同一人物チェック ───
    if prev_image:
        msgs = build_previous_check(image_path, prev_image)
        is_same = ask_judge(msgs, "前回と同一人物か？（長時間滞在チェック）", debug)
        if is_same:
            detection_type = "same_person_long_stay"
            detection_detail = f"同一人物が5分以上ATMに滞在中。画像: {image_name}"

    # ─── Phase 2: スマホを持っているか？ ───
    has_phone = False
    if detection_type is None:
        msgs = build_phone_check(image_path)
        has_phone = ask_judge(msgs, "スマートフォンを持っているか？", debug)

    # ─── Phase 3: 高齢者に見えるか？（スマホ持ちの場合のみ） ───
    if detection_type is None and has_phone:
        msgs = build_elderly_check(image_path)
        is_elderly = ask_judge(msgs, "高齢者に見えるか？", debug)
        if is_elderly:
            detection_type = "elderly_with_phone"
            detection_detail = f"高齢者がスマートフォンを持ちながらATM操作中。画像: {image_name}"

    # ─── 結果処理 ───
    if detection_type is None:
        print(f"\n  {C.GREEN}✅ 全判定クリア — 異常なし{C.RESET}")
        move_to_normal(image_path)
        print(f"  {C.GREEN}📂 → 問題なしDIR: {image_name}{C.RESET}")
        set_previous_image(image_path)
        return False

    # ─── ツール選択フェーズ ───
    print(f"\n  {C.BG_RED}{C.WHITE} ⚠️  検知: {detection_type} {C.RESET}")
    print(f"  {C.RED}{detection_detail}{C.RESET}")

    tool_msgs = build_tool_call_message(detection_type, detection_detail)
    full_content, full_reasoning, tool_calls_list = call_with_tools(tool_msgs, debug)

    if tool_calls_list:
        print(f"\n  {C.YELLOW}{'─'*50}{C.RESET}")
        print(f"  {C.YELLOW}⚡ ツール実行: {len(tool_calls_list)}件{C.RESET}")

        for tc in tool_calls_list:
            func_name = tc["name"]
            try:
                tc_args = json.loads(tc["arguments"])
            except json.JSONDecodeError:
                tc_args = {}

            if func_name == "notify_emergency_support":
                print(f"\n    {C.BG_RED}{C.WHITE} 🚨 緊急通報 {C.RESET}")
                print(f"    {C.RED}📞 notify_emergency_support{C.RESET}")
            elif func_name == "notify_support":
                print(f"\n    {C.BG_YELLOW}{C.WHITE} 📞 サポート連絡 {C.RESET}")
                print(f"    {C.YELLOW}📞 notify_support{C.RESET}")

            for k, v in tc_args.items():
                val_str = str(v)[:60]
                print(f"    {C.CYAN}   {k}: {val_str}{C.RESET}")

            result = execute_tool(func_name, tc_args)
            print(f"    {C.GREEN}   ✅ 完了{C.RESET}")

        print(f"  {C.YELLOW}{'─'*50}{C.RESET}")

    # 問題ありDIRに移動（記録用）
    move_to_normal(image_path)
    set_previous_image(image_path)
    return True


# ============================================================
# メイン
# ============================================================
def main():
    parser = argparse.ArgumentParser(description="ATMガーディアン デモ")
    parser.add_argument("--debug", action="store_true", help="デバッグ情報を表示")
    args = parser.parse_args()
    debug = args.debug

    print(f"\n{C.BOLD}{C.CYAN}{'='*62}{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}  🛡️  ATMガーディアン — 振り込め詐欺防止AI{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}  🔗 LLM Function Call ツールチェーンデモ v3{C.RESET}")
    if debug:
        print(f"{C.BOLD}{C.YELLOW}  🔍 デバッグモード ON{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}{'='*62}{C.RESET}")
    print(f"\n  {C.DIM}📹 5分間隔のATMカメラ撮影をシミュレーション{C.RESET}")
    print(f"  {C.DIM}📁 画像ディレクトリ: {IMAGES_DIR}{C.RESET}")
    print(f"  {C.DIM}🤖 モデル: {MODEL} @ {client.base_url}{C.RESET}")
    print(f"  {C.DIM}📷 シーン数: {len(SCENARIOS)}{C.RESET}")

    init_dirs()

    log_file = os.path.join(SCRIPT_DIR, "incident_log.json")
    if os.path.exists(log_file):
        os.remove(log_file)

    results = []
    for i, scenario in enumerate(SCENARIOS):
        result = run_scene(i, scenario, debug)
        results.append({"scene": i + 1, "time": scenario["time"], "image": scenario["image"], "action_taken": result})
        if i < len(SCENARIOS) - 1:
            time.sleep(1)

    # サマリー表示
    print(f"\n\n{C.BOLD}{C.CYAN}{'='*62}{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}  📊 デモ結果サマリー{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}{'='*62}{C.RESET}")

    for r in results:
        if r["action_taken"] is True:
            icon = f"{C.RED}🚨 通報あり{C.RESET}"
        elif r["action_taken"] is False:
            icon = f"{C.GREEN}✅ 異常なし{C.RESET}"
        else:
            icon = f"{C.YELLOW}⚠️  エラー{C.RESET}"
        print(f"  シーン{r['scene']:2d} ({r['time']}) {r['image']:8s} → {icon}")

    if os.path.exists(log_file):
        with open(log_file, "r") as f:
            logs = json.load(f)
        print(f"\n  {C.CYAN}📋 インシデントログ: {len(logs)}件{C.RESET}")
        for log in logs:
            desc = log['description'][:50] + "..." if len(log['description']) > 50 else log['description']
            print(f"  {C.DIM}  [{log['id']}] {log['incident_type']} - {desc}{C.RESET}")

    normal_count = len(glob.glob(os.path.join(NORMAL_DIR, "*.jpg")))
    print(f"\n  {C.DIM}📂 処理済みDIR: {normal_count}件{C.RESET}")
    print(f"{C.BOLD}{C.CYAN}{'='*62}{C.RESET}\n")


if __name__ == "__main__":
    main()

ベンチマーク結果/benchmark result

shisa-ai/M-IFEval を使って計測した日本語における指示追従性能は以下です。
Ability to follow Japanese instructions measured using shisa-ai/M-IFEval is as follows.

Unslothは量子化モデルで世界的に有名であるため、今回、彼らのモデルに挑戦しました。
英語をメインに使用する場合はUnslothのモデルの方が性能が高いと思われるので留意してください。

Since Unsloth are world-renowned experts in quantization models, I decided to try their models this time.
Please note that their models are likely to perform better if you primarily use English.

update

• 2026/04/09 fix prompt template for cache reuse issue

謝辞 / Acknowledgments

• Qwen
• Unsloth
• llama.cpp
• Thank you to all AI researchers and practitioners.

作成者 / Developer

開発：dahara1@Webbigdata / Developed by dahara1@Webbigdata