Antigravity × DSPy で『プロンプト職人芸』から抜け出す — 自己最適化するLLMプログラムを実務に乗せる

# Antigravity のターミナルから実行
python -m venv .venv
source .venv/bin/activate
 
# DSPy 本体と評価用ユーティリティ（パッケージ名は dspy。古い記事にある dspy-ai は旧名です）
pip install "dspy>=3.0" "litellm>=1.50.0" python-dotenv
 
# 構造化出力の検証用
pip install pydantic

# scripts/dspy_smoke.py
import os
from dotenv import load_dotenv
import dspy
 
load_dotenv()
 
# Gemini 3.8 Flash（GA）。2.5 Flash は既存利用者向けに提供が絞られているため、
# 新規プロジェクトでは 3.8 Flash か 3.5 Flash-Lite を選びます
lm = dspy.LM(
    "gemini/gemini-3.8-flash",
    api_key=os.environ["GEMINI_API_KEY"],
    temperature=0.0,
    max_tokens=1024,
)
dspy.configure(lm=lm)
 
# 最小の Signature: 質問 → 回答
class SimpleQA(dspy.Signature):
    """質問に1〜2文で簡潔に回答する。"""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="1〜2文の日本語回答")
 
predict = dspy.Predict(SimpleQA)
result = predict(question="DSPy は何を解決するために作られた？")
print(result.answer)

# pipelines/invoice_extraction.py
from typing import List
from pydantic import BaseModel, Field
import dspy
 
class LineItem(BaseModel):
    description: str = Field(description="商品名・サービス名")
    quantity: float = Field(description="数量。不明なら 1.0")
    unit_price: float = Field(description="単価（税抜・JPY）")
 
class Invoice(BaseModel):
    vendor: str = Field(description="取引先名")
    invoice_date: str = Field(description="発行日 YYYY-MM-DD")
    total_amount: float = Field(description="合計金額（税込・JPY）")
    line_items: List[LineItem] = Field(default_factory=list)
 
class ExtractInvoice(dspy.Signature):
    """OCR テキストから請求書のフィールドを抽出する。
    数値は半角・整数または小数で返し、日付は YYYY-MM-DD に正規化する。
    自信がない場合でも null は返さず、最も確からしい値を埋める。"""
    ocr_text: str = dspy.InputField(desc="請求書 PDF を OCR したプレーンテキスト")
    invoice: Invoice = dspy.OutputField(desc="抽出済みの構造化データ")
 
class InvoiceExtractor(dspy.Module):
    def __init__(self):
        super().__init__()
        # ChainOfThought で内部的に推論を展開させ、抽出精度を底上げする
        self.extract = dspy.ChainOfThought(ExtractInvoice)
 
    def forward(self, ocr_text: str) -> Invoice:
        return self.extract(ocr_text=ocr_text).invoice

# agents/exchange_rate_agent.py
import requests
import dspy
 
def get_exchange_rate(from_ccy: str, to_ccy: str) -> str:
    """為替レートを取得する。失敗時は理由を文字列で返す。"""
    try:
        r = requests.get(
            "https://api.exchangerate.host/convert",
            params={"from": from_ccy, "to": to_ccy},
            timeout=5.0,
        )
        r.raise_for_status()
        rate = r.json().get("result")
        return f"1 {from_ccy} = {rate} {to_ccy}" if rate else "rate not available"
    except requests.RequestException as e:
        return f"error: {e}"
 
class FxAnswer(dspy.Signature):
    """ユーザーの為替に関する質問に、必要なら為替レートを取得してから答える。"""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="日本語の自然な文章で答える")
 
agent = dspy.ReAct(FxAnswer, tools=[get_exchange_rate], max_iters=4)
 
if __name__ == "__main__":
    print(agent(question="100 ドルって今日いくら？").answer)

# eval/invoice_metric.py
from typing import Any
from pipelines.invoice_extraction import Invoice
 
def invoice_match(example, pred, trace=None) -> float:
    """ground-truth と pred を突き合わせ、0.0〜1.0 のスコアを返す。
    重要なフィールドを重みづけして評価する。"""
    gt: Invoice = example.invoice
    pr: Invoice = getattr(pred, "invoice", None)
    if pr is None:
        return 0.0
 
    score = 0.0
    if gt.vendor.strip() == pr.vendor.strip():
        score += 0.3
    if gt.invoice_date == pr.invoice_date:
        score += 0.2
    # 金額は ±0.5% 以内を正解扱い（OCR の桁ズレと数学的丸め誤差を吸収）
    if pr.total_amount and abs(gt.total_amount - pr.total_amount) / max(gt.total_amount, 1) < 0.005:
        score += 0.3
    # ライン明細は件数一致＋金額合計が一致すれば加点
    if len(gt.line_items) == len(pr.line_items):
        score += 0.1
    gt_sum = sum(li.quantity * li.unit_price for li in gt.line_items)
    pr_sum = sum(li.quantity * li.unit_price for li in pr.line_items)
    if abs(gt_sum - pr_sum) / max(gt_sum, 1) < 0.01:
        score += 0.1
    return score

# train/optimize_invoice.py
import json
import dspy
from dspy.teleprompt import BootstrapFewShotWithRandomSearch, MIPROv2
from pipelines.invoice_extraction import InvoiceExtractor, Invoice
from eval.invoice_metric import invoice_match
 
# 教師データの読み込み（最低 30 件、できれば 100 件以上を推奨）
def load_examples(path: str):
    out = []
    with open(path) as f:
        for line in f:
            row = json.loads(line)
            out.append(
                dspy.Example(
                    ocr_text=row["ocr_text"],
                    invoice=Invoice(**row["invoice"]),
                ).with_inputs("ocr_text")
            )
    return out
 
trainset = load_examples("data/invoice_train.jsonl")
valset = load_examples("data/invoice_val.jsonl")
 
# Step 1: BootstrapFewShotWithRandomSearch で初期最適化
bs_optimizer = BootstrapFewShotWithRandomSearch(
    metric=invoice_match,
    max_bootstrapped_demos=4,
    max_labeled_demos=2,
    num_candidate_programs=8,
    num_threads=4,
)
bs_program = bs_optimizer.compile(InvoiceExtractor(), trainset=trainset, valset=valset)
 
# Step 2: MIPROv2 で指示文と例を協調最適化
mipro = MIPROv2(
    metric=invoice_match,
    auto="medium",   # light / medium / heavy
    num_threads=4,
)
optimized = mipro.compile(
    bs_program,
    trainset=trainset,
    valset=valset,
)
# 以前ここに書いていた requires_permission_to_run 引数は非推奨になり、
# True を渡すと例外になります。付けないのが現行の正解です
 
# compile() の戻り値に score 属性は付かないので、Evaluate で改めて測る
evaluate = dspy.Evaluate(devset=valset, metric=invoice_match, num_threads=4, display_progress=True)
print("val score:", evaluate(optimized).score)
 
# 成果物（指示文・Few-shot・LM 設定）を JSON で保存し、本番ではこれをロードする
optimized.save("artifacts/invoice_v3.json")

# eval/invoice_metric_gepa.py
import dspy
from eval.invoice_metric import invoice_match
 
def invoice_match_with_feedback(gold, pred, trace=None, pred_name=None, pred_trace=None):
    """GEPA 用: 点数に加えて、どのフィールドが外れたかを文章で返す。"""
    score = invoice_match(gold, pred, trace)
    gt = gold.invoice
    pr = getattr(pred, "invoice", None)
    if pr is None:
        return dspy.Prediction(score=0.0, feedback="invoice が返らなかった。出力の型を守ること。")
 
    notes = []
    if gt.vendor.strip() != pr.vendor.strip():
        notes.append(f"vendor が不一致（正: {gt.vendor} / 予測: {pr.vendor}）")
    if gt.invoice_date != pr.invoice_date:
        notes.append(f"invoice_date が不一致（正: {gt.invoice_date} / 予測: {pr.invoice_date}）")
    if gt.total_amount:
        drift = abs(gt.total_amount - (pr.total_amount or 0.0)) / gt.total_amount
        if drift >= 0.005:
            notes.append(f"total_amount が {drift:.1%} ずれている")
    if len(gt.line_items) != len(pr.line_items):
        notes.append(f"line_items が {len(gt.line_items)} 件のところ {len(pr.line_items)} 件")
 
    feedback = "；".join(notes) if notes else "全フィールド一致"
    return dspy.Prediction(score=score, feedback=feedback)

# train/optimize_invoice_gepa.py（抜粋・trainset / valset は前掲と同じ）
import dspy
from pipelines.invoice_extraction import InvoiceExtractor
from eval.invoice_metric_gepa import invoice_match_with_feedback
 
# 反省用モデルは温度を上げ、長い出力を許す
reflection_lm = dspy.LM("gemini/gemini-3.8-flash", temperature=1.0, max_tokens=8000)
 
gepa = dspy.GEPA(
    metric=invoice_match_with_feedback,
    auto="light",              # auto / max_full_evals / max_metric_calls のうち 1 つだけを指定
    reflection_lm=reflection_lm,
    num_threads=4,
)
optimized = gepa.compile(InvoiceExtractor(), trainset=trainset, valset=valset)
optimized.save("artifacts/invoice_gepa_v1.json")

# scripts/diff_optimized.py
import dspy
from pipelines.invoice_extraction import InvoiceExtractor
 
baseline = InvoiceExtractor()
# 状態だけを保存した JSON は、同じクラスを組み立ててから .load() で読み込む
# （dspy.load() は save_program=True で丸ごと保存したディレクトリ用）
optimized = InvoiceExtractor()
optimized.load("artifacts/invoice_v3.json")
 
samples = [...]  # 代表的な OCR テキストを 5〜10 件
for s in samples:
    print("==== sample ====")
    print("baseline:", baseline(ocr_text=s).model_dump())
    print("optimized:", optimized(ocr_text=s).model_dump())
 
dspy.inspect_history(n=5)

# .github/workflows/dspy-eval.yml
name: dspy-eval
 
on:
  pull_request:
    paths:
      - "pipelines/**"
      - "agents/**"
      - "eval/**"
      - "data/eval/**"
 
jobs:
  evaluate:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - name: Run evaluation
        env:
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
        run: |
          python eval/run_holdout.py --module pipelines.invoice_extraction:InvoiceExtractor \
            --artifact artifacts/invoice_latest.json \
            --data data/eval/invoice_holdout.jsonl \
            --report eval-report.md
      - name: Comment evaluation report on PR
        uses: marocchino/sticky-pull-request-comment@v2
        with:
          path: eval-report.md
      - name: Enforce non-regression
        run: python eval/check_regression.py eval-report.md --tolerance 0.01

# eval/check_regression.py
import argparse
import json
import re
import sys
from pathlib import Path
 
BASELINE = Path("eval/baseline.json")   # {"score": 0.89, "artifact": "invoice_v3.json"}
SCORE_RE = re.compile(r"^\|\s*mean_score\s*\|\s*([0-9.]+)\s*\|", re.M)
 
def read_report_score(report: Path) -> float:
    m = SCORE_RE.search(report.read_text(encoding="utf-8"))
    if not m:
        sys.exit("report に mean_score の行が見つかりません（run_holdout.py の出力形式を確認）")
    return float(m.group(1))
 
def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("report", type=Path)
    ap.add_argument("--tolerance", type=float, default=0.01)
    ap.add_argument("--promote", action="store_true",
                    help="本番デプロイ成功後に手で叩く。baseline を今回の値で上書きする")
    args = ap.parse_args()
 
    current = read_report_score(args.report)
    baseline = json.loads(BASELINE.read_text())["score"] if BASELINE.exists() else 0.0
    floor = baseline - args.tolerance
    print(f"baseline={baseline:.4f} current={current:.4f} floor={floor:.4f}")
 
    if args.promote:
        BASELINE.write_text(json.dumps({"score": current}, indent=2) + "\n")
        print("baseline を更新しました")
        return
    if current < floor:
        sys.exit(f"REGRESSION: {current:.4f} < {floor:.4f}")
    print("OK: 回帰なし")
 
if __name__ == "__main__":
    main()