Zenn
Penetrator テックブログPublicationへの投稿
🗒️

PydanticAI入門：型安全なLLMアプリケーション開発とデータベース連携の実践

今川
2025/01/25に公開
Python
Pydantic
LLM
生成 AI
tech

はじめに

みなさん、こんにちは。株式会社 Penetrator の今川です。今回は Pydantic Team によって開発された PydanticAI というフレームワークを使ってみました。PydanticAI は Python で生成 AI アプリケーションをより簡単に構築できるように設計されたフレームワークです。また、Pydantic の型安全な特徴を引き継いでおり、LLM のレスポンスを構造化して扱うことができます。

今回は、PydanticAI の初歩的な導入例と、Web アプリケーションの開発に欠かせないデータベースとの連携を例に、PydanticAI の使い方を紹介します。

環境構築例

  1. Python 3.12 のプロジェクトを作成

    uv init -p 3.12

  2. 依存パッケージをインストール

    add pydantic pydantic-ai sqlalchemy

  3. .env ファイルに OPENAI_API_KEY を設定

    OPENAI_API_KEY=sk-xxxxxx

クイックスタート

hello.py
from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    system_prompt="簡潔に、一文で回答してください。",
)

result = agent.run_sync("「Hello World」の由来を教えてください。")
print(result.data)

実行：

uv run --env-file .env -- python hello.py

実行結果:

「Hello World」は、1970年代にコンピュータ科学者のブライアン・カーニハンが書いたプログラミング入門書『The C Programming Language』で初めて紹介され、一番単純なプログラム例として広まったフレーズです。

データベースとの連携例

今回は、SQLAlchemy を使用した Todo リスト管理アプリケーションを例に、PydanticAI とデータベースの連携方法を紹介します。簡単化のため、SQLite を使用し、プロジェクトのルートディレクトリに todo.db というデータベースファイルを作成します。ユーザーとタスクを管理するシンプルなデータモデルを定義し、AI エージェントがユーザーの要求に応じて Todo リストを取得する機能を実装します。

db_example.py
from dataclasses import dataclass
from datetime import datetime
from pprint import pprint

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from sqlalchemy import TEXT, Column, ForeignKey, Integer, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Session

# SQLAlchemy の設定
db_url = "sqlite+pysqlite:///todo.db"
engine = create_engine(db_url)


# SQLAlchemy モデルの定義
class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(TEXT, nullable=False)


class Todo(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=False)
    priority = Column(Integer, nullable=False, default=1)  # 1-3


def init_db():
    # 既存のテーブルを削除して新しいテーブルを作成
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    # テストデータの挿入
    with Session(engine) as session:
        session.add_all(
            [
                User(id=1, name="アリス"),
                User(id=2, name="ボブ"),
                Todo(
                    id=1,
                    user_id=1,
                    title="本を読む",
                    priority=1,
                ),
                Todo(
                    id=2,
                    user_id=1,
                    title="買い物に行く",
                    priority=2,
                ),
                Todo(
                    id=3,
                    user_id=2,
                    title="宿題をする",
                    priority=3,
                ),
            ]
        )
        session.commit()


@dataclass
class Dependencies:
    session: Session
    user_id: int


class SupportResult(BaseModel):
    message: str = Field(..., description="ユーザーに送信するメッセージ")
    current_time: datetime = Field(..., description="現在時刻")


support_agent = Agent(
    "openai:gpt-4",
    deps_type=Dependencies,
    result_type=SupportResult,
    system_prompt=(
        "あなたは Todo リスト管理の AI エージェントです。"
        "ユーザーの名前を使って返信してください。"
    ),
)


@support_agent.system_prompt
def add_user_name(ctx: RunContext[Dependencies]) -> str:
    user_name = (
        ctx.deps.session.query(User.name).filter_by(id=ctx.deps.user_id).scalar()
    )
    return f"ユーザーの名前は {user_name!r} です"


@support_agent.tool
def get_todos(ctx: RunContext[Dependencies], priority: int | None = None) -> str:
    """ユーザーの Todo リストを取得します"""
    q = ctx.deps.session.query(Todo).filter(Todo.user_id == ctx.deps.user_id)
    if priority is not None:
        q = q.filter(Todo.priority == priority)
    todos = q.all()
    if not todos:
        return "Todo はありません"
    return "\n".join(f"- {todo.title}" for todo in todos)


@support_agent.tool
def get_current_time(ctx: RunContext[Dependencies]) -> datetime:
    """現在時刻を取得します"""
    return datetime.now()


if __name__ == "__main__":
    init_db()
    with Session(engine) as session:
        result = support_agent.run_sync(
            "全ての Todo を全て教えて",
            deps=Dependencies(session=session, user_id=1),
        )
        pprint(result.data.model_dump())

        result = support_agent.run_sync(
            "優先度が 2 の Todo を教えて",
            deps=Dependencies(session=session, user_id=1),
        )
        pprint(result.data.model_dump())

        result = support_agent.run_sync(
            "優先度が 3 の Todo を教えて",
            deps=Dependencies(session=session, user_id=1),
        )
        pprint(result.data.model_dump())

実行：

uv run --env-file .env -- python db_example.py

実行結果：

{'current_time': datetime.datetime(2025, 1, 25, 16, 5, 2, 294409),
 'message': 'アリスさんのTodoリストです:\n1. 本を読む\n2. 買い物に行く'}
{'current_time': datetime.datetime(2025, 1, 25, 16, 5, 9, 324103),
 'message': 'アリスさん、優先度2のTodoは「買い物に行く」です。'}
{'current_time': datetime.datetime(2025, 1, 25, 16, 5, 14, 858650),
 'message': 'アリスさん、優先度3のTodoは現在ありません。'}

さいごに

私たちは現在、「宇宙から地球の不動産市場を変える」挑戦に共感し、一緒に「おっ！」をつくる情熱的なエンジニアを募集しています。新しい技術に挑戦し、人々を驚かせるプロダクトを生み出したい方は、ぜひ 採用ページ をご覧ください。

これから定期的に記事を更新してまいりますので、ぜひブックマークや SNS でのフォローをお願いいたします。皆さまからのフィードバックやご意見もお待ちしております。

今後とも、どうぞよろしくお願いいたします。

Penetrator テックブログ
Penetrator テックブログPublication

Discussion