ローカルLLMで心理学論文のPDFを翻訳するPythonスクリプトを作った

まあChatGPTとGeminiに書いてもらったんですが。

大阪出張のときに参加したOSC2026 Osakaのいくやさんのセミナーに刺激されて、がんばって作ってみました。ChatGPTが書くコードそのまんまだとあまりうまくいかなかったので、Geminiにいろいろ直してもらいました。僕は1行もコードを書いていません。すげえ時代だ……。

こんな感じでセットアップします(Ubuntu 24.04.4 LTS; CUDAセットアップ済み):

(1) llama.cppをビルド。
(2) grapevine-AI/plamo-2-translate-ggufを~/AI/models/plamo/にダウンロード。当環境ではQ8_0版がいい感じ。
(3) 仮想環境をつくる:

$ mkdir -p ~/AI/pdf_translate
$ uv venv
$ source .venv/bin/activate

(4) MinerUとrequestsをインストール:

$ uv pip install -U "mineru[all]" requests

(5) vim pdf2jp.py して次のように書く:

#!/usr/bin/env python3
import argparse
import subprocess
import sys
import time
import requests
import hashlib
import sqlite3
import shutil
import re
from pathlib import Path

# =====================
# 設定
# =====================
LLAMA_PORT = "8080"
LLAMA_URL = f"http://127.0.0.1:{LLAMA_PORT}/completion"
LLAMA_BIN = str(Path.home() / "llama.cpp/build/bin/llama-server")
LLAMA_MODEL = str(Path.home() / "AI/models/plamo/Plamo-2-Translate-Q8_0.gguf")

LLAMA_ARGS = [
    LLAMA_BIN,
    "-m", LLAMA_MODEL,
    "-ngl", "99",
    "--threads", "8",
    "--parallel", "1",
    "--ctx-size", "8192",
    "--port", LLAMA_PORT,
    "--flash-attn", "on",
    "--repeat-penalty", "1.0", # 繰り返しペナルティなし（専門用語用）
    "--no-mmap"  # メモリ読み込み安定化
]

CACHE_DB = "pdf2jp_cache.db"
MAX_WORDS = 100   # 1回の翻訳単位（単語数）
N_PREDICT = 1024  # 生成トークン数（512だと稀に切れるので1024に統一）
STARTUP_TIMEOUT = 120
REQUEST_TIMEOUT = 600

# =====================
# キャッシュ
# =====================
class Cache:
    def __init__(self):
        self.conn = sqlite3.connect(CACHE_DB)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS cache(
                hash TEXT PRIMARY KEY,
                src TEXT,
                dst TEXT
            )
        """)

    def key(self, text):
        return hashlib.sha256(text.encode()).hexdigest()

    def get(self, text):
        k = self.key(text)
        cur = self.conn.execute("SELECT dst FROM cache WHERE hash=?", (k,))
        r = cur.fetchone()
        return r[0] if r else None

    def set(self, text, out):
        k = self.key(text)
        self.conn.execute(
            "INSERT OR REPLACE INTO cache VALUES(?,?,?)",
            (k, text, out)
        )
        self.conn.commit()

cache = Cache()

# =====================
# ユーティリティ
# =====================
def is_llama_alive():
    try:
        r = requests.get(f"http://127.0.0.1:{LLAMA_PORT}/health", timeout=2)
        return r.status_code == 200
    except:
        return False

def clean_text(text):
    text = text.replace("®", "fi")
    text = text.replace("fi", "fi")
    text = text.replace("fl", "fl")
    text = re.sub(r"[^\x09\x0A\x0D\x20-\x7E]", "", text)
    return text.strip()

# =====================
# llama 起動 / 停止
# =====================
def start_llama():
    if is_llama_alive():
        print("✅ llama-server already running")
        return None

    print("🦙 Starting llama-server...")
    proc = subprocess.Popen(
        LLAMA_ARGS,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )

    start = time.time()
    while time.time() - start < STARTUP_TIMEOUT:
        if is_llama_alive():
            print("✅ llama-server ready")
            return proc
        time.sleep(2)

    print("❌ llama-server startup timeout")
    proc.terminate()
    sys.exit(1)

def stop_llama(proc):
    if not proc:
        return
    print("🛑 Stopping llama-server...")
    proc.terminate()
    try:
        proc.wait(timeout=10)
    except:
        proc.kill()

# =====================
# 翻訳
# =====================
def translate(text):
    text = clean_text(text)
    cached = cache.get(text)
    if cached:
        return cached

    text = text.strip()
    if not text:
        return ""

    # <|plamo:op|>dataset を削除し、直接指示から入る
    prompt = (
        "<|plamo:bos|><|plamo:op|>translation\n" # ← ここを直接 translation に
        "学術論文の翻訳タスクです。以下の英語を自然な日本語に訳してください。解説は一切不要です。\n"
        "<|plamo:op|>input lang=English\n"
        f"{text}\n"
        "<|plamo:op|>output lang=Japanese"
    )

    payload = {
        "prompt": prompt,
        "n_predict": N_PREDICT, # ここをグローバル変数に統一
        "temperature": 0.1,
        "cache_prompt": True,   # 高速化の鍵
        "stop": ["<|plamo:op|>", "<|plamo:bos|>", "<|plamo:eos|>"]
    }

    try:
        r = requests.post(LLAMA_URL, json=payload, timeout=REQUEST_TIMEOUT)
        r.raise_for_status()
        out = r.json().get("content", "").strip()

        # ===== バリデーション =====
        if not out:
            print(f"⚠️ Empty output for: {text[:20]}...")
            return text 

        # テンプレート漏れの除去
        out = re.sub(r"<\|.*?\|>", "", out).strip()

        # 短すぎる出力のチェック（ただし入力も短ければ許可）
        if len(out) < 3 and len(text) > 10:
             # 原文が長いのに訳が極端に短い場合はエラー扱いせず、原文を返す（安全策）
             print(f"⚠️ Output too short for: {text[:20]}...")
             return text

        cache.set(text, out)
        return out

    except Exception as e:
        print(f"⚠️ Translation failed: {e}")
        # 失敗時は原文を返す（「翻訳失敗」と書かれるより読みやすい）
        return text

# =====================
# 分割（文単位）
# =====================
def split_text_by_sentences(text, max_words=MAX_WORDS):
    # 改行をスペースに置換してから分割
    clean_text = text.replace('\n', ' ')
    # ピリオド等の後ろのスペースで分割（肯定先読み）
    sentences = re.split(r'(?<=[.!?]) +', clean_text)

    chunks = []
    buf = []
    current_count = 0

    for s in sentences:
        words_in_sent = len(s.split())
        if current_count + words_in_sent > max_words and buf:
            chunks.append(" ".join(buf))
            buf = [s]
            current_count = words_in_sent
        else:
            buf.append(s)
            current_count += words_in_sent

    if buf:
        chunks.append(" ".join(buf))
    return chunks

# =====================
# Markdown 翻訳
# =====================
def translate_md(in_md, out_md, with_en):
    with open(in_md, encoding="utf-8") as f:
        content = f.read()

    # 段落（空行区切り）ごとに処理
    blocks = [b.strip() for b in content.split('\n\n') if b.strip()]

    total = len(blocks)
    start_time = time.time()
    out_md.parent.mkdir(parents=True, exist_ok=True)

    with open(out_md, "w", encoding="utf-8") as out:
        for i, block in enumerate(blocks, 1):
            if block.startswith("#"):
                out.write(block + "\n\n")
                continue

            # 文単位で分割して翻訳
            parts = split_text_by_sentences(block)
            jp_parts = []
            for p in parts:
                jp_parts.append(translate(p))

            jp = " ".join(jp_parts)

            if with_en:
                out.write("**Original**\n\n")
                out.write(block + "\n\n")
                out.write("**Japanese**\n\n")
                out.write(jp + "\n\n---\n\n")
            else:
                out.write(jp + "\n\n")

            elapsed = time.time() - start_time
            eta = (elapsed / i) * (total - i)
            eta_s = time.strftime("%H:%M:%S", time.gmtime(eta))
            print(f"📑 {i}/{total} 完了 (ETA {eta_s})")

# =====================
# MinerU 関連 / main
# =====================
def run_mineru(pdf, outdir):
    print("📄 Running MinerU...")
    cmd = ["mineru", "-p", str(pdf), "-o", str(outdir)]
    subprocess.run(cmd)

def find_md_file(pdf_name, base_dir="tmp_mineru"):
    for p in [Path(base_dir) / pdf_name / "auto", Path(base_dir) / pdf_name / "hybrid_auto"]:
        if p.exists():
            md_files = list(p.glob("*.md"))
            if md_files: return md_files[0]
    return None

def copy_images(pdf_name, base_dir, output_md):
    search_path = Path(base_dir) / pdf_name
    images_src = next(search_path.rglob("images"), None)
    if images_src:
        images_dst = Path(output_md).parent / "images"
        if images_dst.exists(): shutil.rmtree(images_dst)
        shutil.copytree(images_src, images_dst)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("pdf")
    ap.add_argument("out_md")
    ap.add_argument("--with-en", action="store_true")
    ap.add_argument("--workdir", default="tmp_mineru")
    args = ap.parse_args()

    pdf = Path(args.pdf)
    md = find_md_file(pdf.stem, args.workdir)
    if not md:
        run_mineru(pdf, Path(args.workdir))
        md = find_md_file(pdf.stem, args.workdir)

    if not md:
        print("❌ No Markdown produced"); sys.exit(1)

    llama_proc = start_llama()
    try:
        translate_md(md, Path(args.out_md), args.with_en)
    finally:
        stop_llama(llama_proc)

    copy_images(pdf.stem, args.workdir, args.out_md)
    print("✅ Done:", args.out_md)

if __name__ == "__main__":
    main()

(6) 実行権限をつける。

$ chmod +x pdf2jp.py

(7) 以下のように実行する:

$ ./pdf2jp.py input.pdf output.md --with-en

これでPDFがMarkdownに変換されたあとそれを対訳モードで翻訳したMarkdownが生成されます。ouput.mdと同じディレクトリにimagesディレクトリがコピーされるので、ouput.mdをpandocなりVSCodiumなりでPDFに変換すると英日対訳で図入りのきれいなPDFを作ることができます。

MinerUがとっても強力でした。はじめはYomiTokuを使ってたんですが、英語のPDFならMinerUがよいですね。AIパワーであっという間にPDFを解析してMarkdownとかに変換してしまいます。

今回はChatGPTとGeminiを使ったけど、せっかくだからllama.cppでgpt-oss-20bを使えばよかったと今更ながら後悔。まあ今回はこれでいいや。

試せる環境がある方はお試しあれ。

​