Posted inAI

Regex（正規表現）との格闘はやめよう：LLMとPydanticでインテリジェントなETLを構築する

AI tutorial - IT technology blog
AI tutorial - IT technology blog

非構造化データの厄介な現実

従来のETL（Extract, Transform, Load）パイプラインは、ソースがクリーンなCSVや構造化されたAPIであれば完璧に機能します。しかし、クライアントから不規則なメールやPDFの医療報告書、Slackのログが詰まったフォルダが送られてきた瞬間、旧来の手法は限界に突き当たります。かつて私は、顧客のフィードバックを解析するために、200行もの正規表現（Regex）スクリプトと格闘して週末を潰したことがあります。LlamaParseのようなツールを使用した高度なデータ抽出手法を知る前の話です。それは正確に2時間だけ機能しました。ユーザーが別の日付形式を使うまでの間だけ。

コードは決定論的で厳格ですが、自然言語は流動的で予測不可能です。人間のコミュニケーションとSQLデータベースの厳格な要件の間の架け橋が必要です。大規模言語モデル（LLM）とPydanticを組み合わせることで、処理するデータを実際に「理解」するパイプラインを構築できます。

モダンなETLスタック：LLM + Pydantic

このスタックを3つのパートによるハーモニーと考えてください。

  • 推論エンジン (LLM): GPT-4oやClaude 3.5 Sonnetのようなモデルは、単に文字列をマッチングさせるだけではありません。文脈を理解します。「4番の部屋の患者」と「ヘンダーソン氏」が同じ人物を指していることを理解できるのです。
  • バリデーター (Pydantic): LLMは「ハルシネーション（幻覚）」を起こしたり、不要な会話のつなぎ言葉を加えたりする傾向があります。Pydanticは厳格なゲートキーパーとして機能し、LLM出力バリデーションを強制し、定義したスキーマに正確に一致するデータを返すようモデルに求めます。AIが名前のフィールドに年齢を入れようとすれば、Pydanticがそれを拒否します。
  • 永続レコード (SQLAlchemy): データがクリーンアップされ検証されたら、長期保存やBI分析のためにPostgreSQLやSQLiteなどの構造化環境に送ります。

私はこのパターンを導入して数千ものドキュメントを処理してきました. あるケースでは、1日5時間を要していた手動のデータ入力作業を、約3分間の自動処理に短縮することができました。

パイプラインのセットアップ

今回はPythonを使用して構築します。LLMとPydanticのギャップを埋めるには、instructorライブラリが最適です。これは標準のOpenAIクライアントにパッチを当て、生の文字列ではなく実際のPydanticオブジェクトを返せるようにします。

開始するには、依存関係をインストールします。

pip install openai instructor pydantic sqlalchemy

データ設計図の定義

まずはスキーマです。抽出ロジックを書く前に、「成功」の定義を行う必要があります。乱雑な紹介状を解析する場合を想定してみましょう。患者の名前、年齢、症状のリスト、および優先度が必要です。

from pydantic import BaseModel, Field
from typing import List

class PatientExtraction(BaseModel):
    name: str = Field(..., description="患者のフルネーム")
    age: int = Field(..., description="患者の年齢（歳）")
    symptoms: List[str] = Field(..., description="具体的な医療症状のリスト")
    priority: str = Field(..., description="緊急度: Low (低), Medium (中), または High (高)")

Fieldの記述は単なるドキュメント用ではありません。instructorライブラリは、これらの文字列を指示（インストラクション）として直接LLMに渡し、ノイズの中から正しいデータポイントを特定するのを助けます。

抽出エンジン

抽出ロジックは驚くほどシンプルです。「パッチ」を当てたクライアントを使用することで、モデルはPydanticスキーマに従わざるを得なくなります。モデルが有効なJSONを生成できなかった場合、instructorは自動的にリクエストを再試行し、AIに対してバリデーションエラーの箇所を正確に提示します。

import instructor
from openai import OpenAI

# パッチを当てたクライアントを初期化
client = instructor.from_openai(OpenAI(api_key="YOUR_OPENAI_API_KEY"))

messy_text = """
ジョン・ドウ（45歳）から電話を受けました。
深刻な腰痛と左足のしびれを訴えています。
緊急性が高いと思われるため、本日中に対応してください。
"""

def extract_patient_data(text: str) -> PatientExtraction:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=PatientExtraction,
        max_retries=3,
        messages=[{"role": "user", "content": text}],
    )

extracted = extract_patient_data(messy_text)
print(f"抽出結果: {extracted.name} | 優先度: {extracted.priority}")

このセルフヒーリング（自己修復）ループこそが「秘伝のソース」です。これにより、不正な形式のテキストを返す可能性がある標準的なAPI呼び出しよりも、システムの堅牢性が大幅に向上します。

検証済みデータをSQLに保存する

クリーンなPythonオブジェクトが得られたので、それをデータベースに保存できます。SQLAlchemyを使用して、PydanticデータをSQLiteテーブルにマッピングします。これにより、AIの「思考プロセス」からディスクへの保存まで、データ型の一貫性が保たれます。

from sqlalchemy import create_engine, Column, Integer, String, JSON
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()
engine = create_engine("sqlite:///patients.db")
Session = sessionmaker(bind=engine)

class PatientRecord(Base):
    __tablename__ = 'patients'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    symptoms = Column(JSON)
    priority = Column(String)

Base.metadata.create_all(engine)

def save_to_db(data: PatientExtraction):
    with Session() as session:
        new_record = PatientRecord(
            name=data.name,
            age=data.age,
            symptoms=data.symptoms,
            priority=data.priority
        )
        session.add(new_record)
        session.commit()

save_to_db(extracted)

本番環境へのスケーリング

ローカルスクリプトから本番パイプラインに移行するには、いくつかの調整が必要です。まず、コストを考慮してください。GPT-4oで1,000件のドキュメントを処理すると高額になる可能性がありますが、OpenAIの利用料金を50%削減できるBatch APIや、GPT-4o-miniを活用することで、構造化抽出の結果を維持しつつコストを劇的に抑えることができます。

次に、Pydanticのfield_validatorを使用してビジネスロジックを適用します。例えば、LLMが許可リストにない優先度を抽出した場合、デフォルト値に強制したり、データベースに保存される前にエラーを発生させたりすることができます。

from pydantic import field_validator

# PatientExtractionクラス内
@field_validator('priority')
@classmethod
def validate_priority(cls, v: str) -> str:
    allowed = ['Low', 'Medium', 'High']
    if v not in allowed:
        return 'Medium' # 安全なフォールバック
    return v

まとめ

LLMの推論能力とPydanticの厳格なバリデーションを組み合わせることで、ETLの常識が変わります。もう、乱雑で非構造化なテキストを恐れる必要はありません。明確なスキーマを定義し、解釈をAIに任せることで、柔軟かつ型安全なパイプラインを構築できます。

まずは小さく始めましょう。手動のデータ入力タスクを1つ選び、Pydanticモデルを定義して、いくつかのテストを実行してみてください。このパターンの信頼性を目の当たりにすれば、もう二度と複雑な正規表現を書きたいとは思わなくなるはずです。

Share:
Tags: