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

# Antigravity のターミナルから実行
python -m venv .venv
source .venv/bin/activate
 
# DSPy 本体と評価用ユーティリティ
pip install "dspy-ai>=2.6.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 2.5 Flash を使う例（コスト最適化用）
lm = dspy.LM(
    "gemini/gemini-2.5-flash",
    api_key=os.environ["GEMINI_API_KEY"],
    temperature=0.0,
    max_tokens=1024,
)
dspy.settings.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=False,
)
 
# 成果物を JSON で保存し、本番ではこれをロードする
optimized.save("artifacts/invoice_v3.json")
print("compiled:", optimized.score)

# scripts/diff_optimized.py
import dspy
from pipelines.invoice_extraction import InvoiceExtractor
 
baseline = InvoiceExtractor()
optimized = dspy.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)