はじめに
PydanticAIを用いると、構造化した状態でデータ出力できます。ただし、他システムのデータ構造が変動する場合、その状態に応じて定義を動的に作成したい場合がありました。
システム連携の都合上、二つのシステムのテーブル定義をそれぞれで定義するような形だと運用上の負担が大きいという課題がありました。
そのため型情報を取得するようなAPIなどがあれば、それを元にして動的構造化出力できます。
例えば、salesforceはオブジェクトの定義情報を取得できるため、それを元にして動的構造化出力できます。
pyproject.toml
とりあえず以下のような構成で環境を作成しました。
[project]
name = "dynamic-model-pydantic"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"pydantic>=2.8.2",
"pydantic-ai>=0.8.1",
"pydantic-settings>=2.10.1",
]
[dependency-groups]
dev = [
"ruff>=0.12.11",
]
コード例
以下が動的にモデルを出力するコード例を示します。
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Dict, List, Tuple
import copy
from pydantic import BaseModel, Field, create_model as pydantic_create_model
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.providers.google import GoogleProvider
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
GEMINI_API_KEY: str = Field(default="")
settings = Settings()
# 型名文字列から Python 型への簡易マッピング
TYPE_MAP: Dict[str, Any] = {
"str": str,
"int": int,
"float": float,
"bool": bool,
"date": date,
"datetime": datetime,
"decimal": Decimal,
"dict": dict,
"list": list,
}
def _parse_type(type_spec: Any) -> Any:
"""与えられた型指定を Python の型へ変換する。
サポート:
- 直接の型: int, str など
- 文字列: "int", "str", "datetime"
- Optional 指定: 末尾に '?' を付ける (例: "str?", "list[int]?")
- list[T] 記法: 例 "list[str]", "list[int]?"
"""
if isinstance(type_spec, type):
return type_spec
if not isinstance(type_spec, str):
raise TypeError(f"Unsupported type spec: {type_spec!r}")
# Optional 判定
is_optional = type_spec.endswith("?")
core = type_spec[:-1] if is_optional else type_spec
core = core.strip().lower()
# list[T] 記法
if core.startswith("list[") and core.endswith("]"):
inner_str = core[5:-1].strip()
inner_type = _parse_type(inner_str)
from typing import List as TypingList # 局所 import で循環を避ける
resolved = TypingList[inner_type]
else:
resolved = TYPE_MAP.get(core)
if resolved is None:
raise ValueError(f"Unknown type name: {core}")
if is_optional:
from typing import Optional as TypingOptional
return TypingOptional[resolved]
return resolved
def build_model_from_spec(
model_name: str,
fields_spec: List[Dict[str, Any]],
) -> type[BaseModel]:
"""物理名・論理名・型を含む仕様から Pydantic モデルを動的生成する。
fields_spec の各要素は以下のキーを持つ辞書:
- physical_name: フィールド物理名 (コード上の属性名)
- logical_name: フィールド論理名 (スキーマ title に反映)
- type: 型指定。例 "str", "int", "datetime", "list[str]", Optional は末尾に '?'。
"""
field_definitions: Dict[str, Tuple[Any, Any]] = {}
for spec in fields_spec:
physical_name = spec["physical_name"]
logical_name = spec.get("logical_name", physical_name)
type_spec = spec["type"]
py_type = _parse_type(type_spec)
is_optional = isinstance(type_spec, str) and type_spec.endswith("?")
# デフォルトの扱い
has_default = "default" in spec
has_factory = "default_factory" in spec
if has_factory:
default_factory = spec["default_factory"]
field_definitions[physical_name] = (
py_type,
Field(default_factory=default_factory, title=logical_name),
)
continue
if has_default:
default_value = spec["default"]
# default が None で非 Optional の場合は Optional に拡張
if default_value is None and not is_optional:
from typing import Optional as TypingOptional
py_type = TypingOptional[py_type]
if isinstance(default_value, (list, dict)):
# ミュータブルは deepcopy を返すファクトリに変換
def _factory(v=default_value):
return copy.deepcopy(v)
field_definitions[physical_name] = (
py_type,
Field(default_factory=_factory, title=logical_name),
)
else:
field_definitions[physical_name] = (
py_type,
Field(default_value, title=logical_name),
)
continue
# デフォルト指定なし
default_required_or_none = None if is_optional else ...
field_definitions[physical_name] = (
py_type,
Field(default_required_or_none, title=logical_name),
)
# Pydantic v2 対応の create_model を使用
return pydantic_create_model(model_name, **field_definitions)
def main():
# サンプル ここは別から取得し、整形・pydanticモデルを作成する
spec = [
{"physical_name": "user_id", "logical_name": "ユーザーID", "type": "int"},
{
"physical_name": "name",
"logical_name": "氏名",
"type": "str",
"default": "不明",
},
{
"physical_name": "tags",
"logical_name": "タグ",
"type": "list[str]?",
"default_factory": list,
},
{
"physical_name": "joined_at",
"logical_name": "入会日時",
"type": "datetime",
"default_factory": datetime.now,
},
{
"physical_name": "active",
"logical_name": "有効",
"type": "bool?",
"default": True,
},
]
UserModel = build_model_from_spec("UserModel", spec)
agent = Agent(
model=GoogleModel(
model_name="gemini-2.0-flash",
provider=GoogleProvider(api_key=settings.GEMINI_API_KEY),
),
system_prompt="You are a helpful assistant that can create Pydantic models from JSON data.",
output_type=UserModel,
)
result = agent.run_sync("""
以下からデータを抽出してください。
user_id: test
name: 山田太郎
tags: 社員、チームメンバー
joined_at: 2025-01-01
active: 有効
""")
print(result)
if __name__ == "__main__":
main()
おわりに
簡単にですが、スニペット共に記載しました。何かの参考になれば幸いです。
Discussion