Pydanticaiで生産グレードのLLM駆動型アプリケーションを構築します

ソフトウェア開発の急速に進化する状況では、人工知能、データ検証、およびデータベース管理の交差点が前例のない可能性を開きました。このブログ投稿では、最新のPydanticaiフレームワークとGoogleのGemini-1.5モデルを使用して、SQLコード生成とSQLコードの説明に対する革新的なアプローチを調査し、最先端のAIテクノロジーがデータベースクエリ開発を合理化および強化する方法を示します。

開発者、データ科学者、およびデータアナリストのために、この探索は、自然言語処理からのインテリジェントコード生成の将来を垣間見ることができます。ここでは、複雑なデータベースクエリを簡単かつ正確に作成できます。

学習目標

• PydanticとPydanticaiの基礎を理解してください。
• AI搭載のSQLコード生成システムを実装する方法を学びます。
• SQL翻訳のために、自然言語でGemini-1.5-Flashの機能を調べてください。
• データベースインタラクションのためのインテリジェントAIエージェントの構築に関する洞察を得る。

この記事は、データサイエンスブログソンの一部として公開されました。

目次

• Pydanticaiとは何ですか？
• 活動中のピダンティン派の例
• AIエージェントとは何ですか？
• エージェントワークフローとは何ですか？
• AIエージェントとエージェントワークフローの最新の使用
• Pydanticaiフレームワークとは何ですか？
• あなたのプロジェクトを始めましょう
• プロジェクトを実装するための段階的なガイド
• 結論
• よくある質問

Pydanticaiとは何ですか？

Pydanticaiは、データの検証とタイプチェックに革命をもたらす強力なPythonライブラリです。データモデルを定義するための宣言的なアプローチを提供し、複雑なデータ構造を簡単に作成および検証できます。

Pydanticの重要な機能は次のとおりです。

カスタマイズ

• プリミティブタイプや複雑なネストされた構造を含む幅広いデータ型を検証します。
• 検証とシリアル化のために、ほぼすべてのPythonオブジェクトをサポートします

柔軟性

データ検証の厳格性を制御できます。

• 予想されるタイプにデータを強制します
• 必要に応じて厳格なタイプチェックを強制します

シリアル化

• Pydanticオブジェクト、辞書、およびJSON間のシームレスな変換をサポートします。
• 自己文書化APIとJSONスキーマを使用するツールとの互換性を有効にします。

パフォーマンス

• コア検証ロジックは、並外れた速度と効率のために錆で記述されています。
• Scalable Rest APIなどのハイスループットアプリケーションに最適です。

生態系

• Fastapi、Langchain、Llamaindexなどの人気のあるPythonライブラリで広く使用されています。
• 最新のエージェントLLMは、Pydanticなしでは実装できません。

活動中のピダンティン派の例

Pydanticaiは、Pythonでデータの検証とタイプチェックを簡素化し、堅牢なデータモデルを作成するための強力なツールになります。その機能を紹介するいくつかの実用的な例を探りましょう。

基本的なデータ検証

Pydantic Import Basemodelから


クラスユーザー（BASEMODEL）：
    名前：str
    年齢：int


＃有効なデータ
user = user（name = "alice"、age = 30）
印刷（ユーザー）

print（ "================================"）
＃無効なデータ（年齢は文字列です）
試す：
    user = user（name = "Alice"、age = "30"）
eとしての例外を除く：
    印刷（e）

上記のコードは、Pydanticのベースメーモデルを使用してユーザーモデルを定義し、名前を文字列として施行し、年齢を整数として施行します。正しいデータを検証しますが、無効なデータ（年齢の文字列）が提供されると、検証エラーが発生します。

出力：

自動タイプの強制

Pydantic Import Basemodelから
クラス製品（BaseModel）：
    価格：フロート
    数量：int

＃不一致のタイプのデータ
Product = product（Price = "19.99"、数量= "5"）
印刷（製品）

印刷（type（product.price））
print（type（product.quantity））

ここでは、価格が浮かんでいる製品モデル、整数としての量の製品モデル。 Pydanticは、文字列入力（「19.99」と「5」）を正しいタイプ（float and int）に自動的に強制し、その型変換機能を実証します。

出力：

ネストされたモデル

Pydantic Import Basemodelから
クラスアドレス（BaseModel）：
    ストリート：str
    都市：str
クラスユーザー（BASEMODEL）：
    名前：str
    アドレス：アドレス

＃有効なデータ
user = user（name = "bob"、address = {"street"： "123 Main St"、 "City"： "Wonderland"}）
印刷（ユーザー）

＃ネストされた属性にアクセスします
print（user.address.city）

ここでは、アドレスモデルを含むネストされたユーザーモデルを定義します。 Pydanticは、ネストされた検証を許可し、辞書をモデルに自動変換します。有効なデータはユーザーオブジェクトを初期化し、「 user.address.city」などのネストされた属性に直接アクセスできます。

出力：

カスタムルールを使用した検証

Pydantic Import Basemodel、Field、field_validatorから

クラスユーザー（BASEMODEL）：
    名前：str
    年齢：int = field（...、gt = 0、description = "年齢はゼロより大きくなければなりません"）

    @field_validator（ "name"）
    def name_must_be_non_empty（cls、value）：
        value.strip（）ではない場合：
            Raise ValueRerr（「名前は空にすることはできません」）
        返品値

＃有効なデータ
user = user（name = "Charlie"、age = 25）
印刷（ユーザー）

＃無効なデータ
試す：
    user = user（name = ""、age = -5）
eとしての例外を除く：
    印刷（e）

ここでは、検証ルールを使用してユーザーモデルを定義します。年齢は0より大きく、名前を空にすることはできません（ name_must_be_non_emptyメソッドを介して検証されます）。有効なデータはユーザーインスタンスを作成しますが、無効なデータ（空の名前または負の年齢）は詳細な検証エラーを引き起こし、Pydanticの検証機能を実証します。

出力：

これらは、Pydanticの中核例のいくつかです。データ検証の基本原則を理解するのに役立つことを願っています。

AIエージェントとは何ですか？

AIエージェントは、特定の目的を達成するために、タスクを自律的に実行し、意思決定を行い、環境と対話するように設計されたインテリジェントシステムです。これらのエージェントは新しいものではありませんが、生成AIの最近の急速な発展とそれをエージェントと組み合わせることで、新しい時代のエージェントソフトウェア開発が行われます。これで、エージェントは入力を処理し、アクションを実行し、動的に適応できます。彼らの行動は、人間のような問題解決を模倣し、最小限の人間の介入でさまざまなドメインで機能することを可能にします。

エージェントワークフローとは何ですか？

エージェントワークフローとは、1つまたは複数のAIエージェントによって管理および実行されたタスクの目標駆動型シーケンス、構造、目標主導のシーケンスを指します。非ラインの剛性のある従来のワークフロー、エージェントワークフローは、適応性、自律性、およびコンテキスト認識を示します。これらのワークフロー内のAIエージェントは、独立して意思決定を行い、サブタスクを委任し、フィードバックから学習し、効率的で最適化された結果につながる可能性があります。

AIエージェントとエージェントワークフローの最新の使用

AIエージェントとエージェントワークフローの統合により、複雑なタスクを自動化し、意思決定を強化し、推進効率を高めることにより、産業に革命をもたらしました。これらのインテリジェントシステムは動的に適応し、多様なドメイン全体でよりスマートなソリューションを可能にします。

ビジネスオートメーション

AIエージェントは、チャットボット、電子メール管理、販売パイプラインの最適化を通じて、カスタマーサポートなどの繰り返しタスクを自動化します。それらは、より価値のあるタスクから人的資源を解放することにより、生産性を向上させます。

ソフトウェア開発

AIを搭載したエージェントは、コードを生成、テスト、デバッグすることにより、ソフトウェアのライフサイクルを加速し、開発時間とヒューマンエラーを短縮します。

健康管理

AIエージェントは、医療診断、患者の監視、および治療のパーソナライズを支援し、医療提供と運用効率の改善を支援します。

ファイナンス

金融システムのエージェントワークフローは、詐欺検出、リスク評価、投資分析を自動化し、より速く、より信頼性の高い意思決定を可能にします。

eコマース

インテリジェンスエージェンシーは、ショッピングエクスペリエンスのパーソナライズを強化し、製品の推奨事項と顧客サービスを最適化します。

AIエージェントとエージェントワークフローの台頭は、複雑なプロセスを管理できる非常に自律的なシステムへのシフトを意味します。それらの適応性と学習能力により、現代の産業にとって不可欠なものになり、革新、スケーラビリティ、およびドメイン全体の効率を促進します。 AIが進化し続けると、AIエージェントは毎日のワークフローにさらに統合され、タスクの管理方法を変換します。

Pydanticaiフレームワークとは何ですか？

Pydanticaiは、Pydantic、Fastapiの作成者によって開発されたPythonエージェントフレームワークであり、生成AIを利用した生産グレードアプリケーションの構築を合理化し、タイプの安全性、モデルに依存しない設計、および大規模な言語モデル（LLM）とのシームレスな統合を強調しています。

Pydanticaiが含まれる重要な機能：

• モデルと存在するサポート： Pydanticaiは、Openai、Antropic、Gemini、Groq、Mistral、Ollamaなどのさまざまなモデルと互換性があり、追加のモデルを組み込むための簡単なインターフェイスを備えています。
• タイプセーフティ： PythonのタイプシステムとPydanticの検証を活用して、Pydanticaiは堅牢でスケーラブルなエージェント開発を保証します。
• 依存関係注射システム：新規のTy-Safe依存関係注入メカニズムを導入し、テストと評価駆動型開発を強化します。
• 構造化された応答の検証： Pydanticの検証機能を利用すると、正確で信頼できる構造応答が保証されます。
• logfire統合： LLM駆動のアプリケーションのデバッグと監視を強化するために、Pydantic Logfireとの統合を提供します。

Pydanticaiの最小限の例を次に示します。

 OSをインポートします
pydantic_aiインポートエージェントから
pydantic_ai.models.geminiインポートGeminimodelから
dotenvインポートload_dotenvから

load_dotenv（）

gemini_api_key = os.getenv（ "<google_api_key>"）

モデル= geminimodel（
    「Gemini-1.5-flash」、
    api_key = gemini_api_key、
））

agent = agent（
    モデル=モデル、
    System_prompt = "簡潔に、1つの文で返信してください。"、
））

result = agent.run_sync（ '「hello world」はどこから来るの？」）
印刷（result.data）</google_api_key>

出力：

さあ、いくつかの本当のことをする時が来ました。 Pydanticaiエージェントフレームワークを使用して、 Postgres SQLクエリ生成を構築します。

あなたのプロジェクトを始めましょう

不可欠なツールと環境をセットアップするための段階的なガイドを使用して、プロジェクトの基盤を築きます。

環境の設定

プロジェクトのコンドマ環境を作成します。

 #create env
$ conda create  - 名前sql_gen python = 3.12

＃ENVをアクティブにします
$ condaはsql_genをアクティブにします

次に、プロジェクトフォルダーを作成します

＃フォルダーを作成します
$ mkdir sql_code_gen
＃フォルダーに変更します
$ cd sql_code_gen

Postgresとロードデータベースをインストールします

Postgres、PSQL-Command-Tools、およびPGADMIN-4をインストールするには、システム用にインストーラーをEDBDownLoadLoadに移動し、すべてのツールを一度にインストールしてください。

ここで、dvdrentalデータベースをherとhersからダウンロードして、これらの手順に従ってください。

ステップ1：ターミナルを開きます

psql -u postgres

＃パスワードを要求します

ステップ2：データベースを作成します

＃postgres =＃

データベースdvdrentalを作成します。

ステップ3：ターミナルのコマンド

次に、PSQLコマンドを終了してから、端末を入力します

PG_RESTORE -U POSTGRES -D DVDRENTAL D：/SAMPLEDB/POSTGRES/DVDRENTAL.TAR

ステップ4：PSQLに接続します

次に、PSQLに接続し、データベースがロードされているかどうかを確認します。

 psql -u postgres

＃dvdrentalと接続します
\ c dvdrental

＃テーブルを見てみましょう

\ dt

出力：

上記のテーブルが表示されている場合は、大丈夫です。私たちは皆、メインプロジェクトを開始する予定です。

必要なPythonライブラリをSQL_Gen Conda envにインストールします。

 condaはsql_genをアクティブにします

＃ライブラリをインストールします
ピップインストールpydantic asyncpg asyncio pydantic-ai 

PIPSインストールpython-dotenv fastapi google-generativeai

PIPインストールdevtools注釈付きタイプタイプタイプの拡張

プロジェクト構造

私たちのプロジェクトには、メイン、モデル、サービス、スキーマ、つまり4つのファイルがあります。

 sql_query_gen/
|
|  -  main.py
|  -  models.py
|  -  schema.py
|  -  service.py
|  - 。env
| -__ init__.py
|  - 。gitignore

プロジェクトを実装するための段階的なガイド

この包括的な実装ガイドを使用して、プロジェクトをコンセプトから現実に導くための詳細な手順と実用的なテクニックに飛び込みます。

Pydanticモデル

Models.pyファイルでデータモデルを作成することから始めます

DataClassesからImport Dataclassから
インポート注釈付きタイプから
asyncpgをインポートします
Annotated_typesからImport Minlenから
Pydantic Import Basemodel、Fieldから

@dataClass
クラスDEPS：
    conn：asyncpg.connection

クラスの成功（BaseModel）：
    sql_query：注釈[str、minlen（1）]
    説明：str = field（ ""、description = "SQLクエリの説明、マークダウンとして"）

クラスInvalidRequest（BaseModel）：
    error_message：str

上記のコードでは、

• DEPSクラスは、データベース接続の依存関係を管理します。 @DataClassは、__init__や__Repr__などの特別な方法を自動的に生成します。 connは「asyncpg.connection」と入力され、アクティブなpostgreSQL接続を表します。この設計は、依存関係噴射パターンに従い、コードをよりテストしやすく保守可能にします。
• サクセスクラスは、SQL-Queryの成功を表します。SQL_Queryは空の文字列（ Minlen （1））でなければならず、注釈付きで検証制約を追加する必要があります。説明は、デフォルトの空の文字列を備えたオプションのフィールドです。
• InvalidRequestクラスはエラー応答モデルであり、失敗したSQLクエリ生成の試みを表します。

このコードは、データベース接続管理、入力検証、構造化された応答処理、エラー処理の基礎を確立しました。

サービスモジュール

ここで、SQL GenerationのPydanticaiサービスをサービスモジュールに実装します。

ライブラリと構成をインポートします

OSをインポートします
インポートユニオンのタイピングから
dotenvインポートload_dotenvから
asyncpgをインポートします
typing_extensionsからImport typealiasから
Pydantic_aiインポートエージェント、ModelRetry、RunContextから
pydantic_ai.models.geminiインポートGeminimodelから
スキーマインポートDB_Schemaから
モデルから、DEPS、成功、InvalidRequestをインポートします

構成するには、プロジェクトルートに.ENVファイルを作成し、Gemini APIキーをそこに配置します

＃.env

gemini_api_key = "asgfhkdhjy457gthjhajbsd"

次に、service.pyファイルで：

 load_dotenv（）

gemini_api_key = os.getenv（ "google_api_key"）

`からGoogle APIキーをロードします。 env `ファイル。

モデルとエージェントの作成

応答：typealias = union [success、invalidRequest]

モデル= geminimodel（
    「Gemini-1.5-flash」、
    api_key = gemini_api_key、
））

agent = agent（
    モデル、
    result_type = response、＃タイプ：無視
    deps_type = deps、
））

• 最初に成功またはinvalidRequestのいずれかである可能性のある応答タイプを定義します
• APIキーを使用してGemini 1.5 Flashモデルを初期化します
• 指定された応答と依存関係のタイプを使用してPydanticaiエージェントを作成する

システムプロンプト定義

次に、SQLクエリ生成のシステムプロンプトを定義します。

 @agent.system_prompt
async def system_prompt（） - > str：
    f "" "\を返します

以下のPostgreSQLの記録表を考えると、あなたの仕事は
ユーザーのリクエストに合ったSQLクエリを書きます。

データベーススキーマ：
{db_schema}

例
    リクエスト：4.00ドルを超えるレンタル料金と「PG」の格付けのあるすべての映画を見つける
    応答：[タイトル]、RENTAL_RATEを選択します
    映画から
    rental_rate> 4.00およびrating = 'pg';
例
    リクエスト：最長の長さのフィルムを見つけます
    応答：タイトル、長さを選択します
    映画から
    ここで、長さ=（フィルムから最大（長さ）を選択）;
例
    リクエスト：各カテゴリの映画の平均レンタル期間を見つける
    応答：c.name、avg（f.rental_duration）をaverage_rental_durationとして選択します
    カテゴリからc
    c.category_id = fc.category_idにfilm_category fcに参加してください
    FC.FILM_ID = F.FILM_IDでフィルムFに参加してください
    C.Nameによるグループ
    Average_rental_duration descによる注文;
"" "

ここでは、AIモデルのベースコンテキストを定義し、モデルの応答をガイドするためのクエリの例を提供します。また、モデルがスキーマを分析し、より良い応答を生成できるように、モデルにデータベーススキーマ情報も含めます。

応答検証

AIモデルからのエラーのない応答をプロジェクトの要件までにするために、応答を検証するだけです。

 @agent.result_validator
async def validate_result（ctx：runco​​ntext [deps]、result：response） - > response：
    ISInstance（result、invalidRequest）の場合：
        返品結果

    ＃Geminiは、しばしばSQLにExtraneosバックラッシュを追加します
    result.sql_query = result.sql_query.replace（ "\\"、 ""）
    result.sql_query.upper（）。startswith（ "select"）：
        ModelRetryを上げる（「選択クエリを作成してください」）

    試す：
        await ctx.deps.conn.execute（f "説明{result.sql_query}"）
    asyncpg.exceptions.postgreserrorをe：
        eからモデルレトリ（f "無効なsql：{e}"）を上げます
    それ以外：
        返品結果

ここでは、生成されたSQLクエリを検証および処理します

重要な検証手順：

• 結果が無効な場合はすぐに戻ります。余分なバックスラッシュをクリーンアップします
• クエリが選択されたステートメントであることを確認してください
• PostgreSQL説明を使用してSQL構文を検証します
• 無効なクエリのモデルレトリを上げます

データベーススキーマ

データベーススキーマを取得するには、Postgresセットアップ中にインストールしたPGADMIN4を開き、「 DVDRENTAL」データベースに移動し、右クリックして、 [データベース]の[ERD]をクリックします。
以下のERD図を取得し、ERDからSQLを生成します（画像の丸い黒いマーキングを参照）。

スキーマをschema.pyモジュールにコピーします。

 ＃schema.py
db_schema = "" "
始める;


存在しない場合はテーブルを作成します
（
    actor_id serial not null、
    first_name文字変化（45）はpg_catalogを照合します。 "default" not null、
    last_name文字変化（45）はpg_catalogを照合します。 "default" not null、
    last_updateタイムスタンプタイムゾーンなしでnull default now（）、
    Constraint Actor_pkeyプライマリキー（Actor_id）
）;
。
。
。
  
。
。
。
"" "

上記のコードブロックはひどく切り捨てられています。完全なコードを取得するには、プロジェクトリポジトリにアクセスしてください。

これで、必要なすべてのモジュールが完了しているため、メインメソッドとテストを実装する時間です。

メインの実装

メイン関数定義とプロンプト処理を行います。

 Asyncioをインポートします
OSをインポートします
sysをインポートします
インポートユニオンのタイピングから
dotenvインポートload_dotenvから
asyncpgをインポートします
DevToolsからインポートデバッグから
typing_extensionsからImport typealiasから
pydantic_aiインポートエージェントから
pydantic_ai.models.geminiインポートGeminimodelから
モデルから、DEPS、成功、InvalidRequestをインポートします

load_dotenv（）

gemini_api_key = os.getenv（ "google_api_key"）


応答：typealias = union [success、invalidRequest]

モデル= geminimodel（
    「Gemini-1.5-flash」、
    api_key = gemini_api_key、
））

agent = agent（
    モデル、
    result_type = response、＃タイプ：無視
    deps_type = deps、
））


async def main（）：
    Len（Sys.Argv）== 1の場合：
        PROMP = "選択クエリを作成してください"
    それ以外：
        PROMPT = sys.Argv [1]

    ＃データベースへの接続
    conn = asyncpg.connectを待ちます（
        user = "postgres"、
        パスワード= "avizyt"、
        host = "localhost"、
        port = 5432、
        データベース= "dvdrental"、
    ））
    試す：
        deps = deps（conn）
        result = waing agent.run（prompt、deps = deps）
        result = debug（result.data）
        print（ "========あなたのクエリ=========="）
        print（debug（result.sql_query）））
        print（ "=========説明========"）
        print（debug（result.explanation））

    ついに：
        conn.close（）を待っています


__name__ == "__main__"の場合：
    asyncio.run（main（））

ここで、最初に、非同期メイン関数を定義し、クライアントクエリのコマンドライン引数を確認します。 argsが提供されていない場合は、デフォルトのプロンプトを使用します。

次に、Postgres接続パラメーターを設定して、DVDrentalデータベースサービスに接続します。

Tryブロックで、データベース接続を備えたDEPSインスタンスを作成し、プロンプトを使用してAIエージェントを実行し、デバッグ関数（ PIPインストールDevTools ）を使用して結果を処理します。次に、生成されたSQLクエリとクエリの説明を含むフォーマットされた出力を印刷します。その後、最終的にデータベース接続を閉じました。

次のように、以下のようにメインモジュールを実行します。

 ＃ターミナルで
python main.py「各顧客のレンタルの総数を取得」

出力：

PGADMIN4のSQLクエリをテストした後：

おお！それは私たちが望むように機能しています。このようなクエリをもっとテストし、学習をお楽しみください。

結論

このプロジェクトは、データベースの相互作用をより直感的でアクセスしやすくする際の重要な前進を表しています。 AIの力を堅牢なソフトウェアエンジニアリングの原則と組み合わせることにより、SQLクエリを生成するだけでなく、実世界の使用に安全で教育的で実用的な方法でそれを行うツールを作成しました。

この実装の成功は、AIが従来のデータベース操作を置き換えるのではなく、強化する可能性を示しており、学習と生産性の両方に貴重なツールを提供します。

Project Repo - このプロジェクトで使用されるすべてのコードはこちらから入手できます。

キーテイクアウト

• Pydanticaiは、インテリジェントなコンテキストを意識したコード生成を可能にします。
• Gemini-1.5-Flashは、技術的なタスクに高度な自然言語理解を提供します。
• AIエージェントは、データベースとの対話方法を変換し、コードを生成できます。
• AIで生成されたコードシステムでは、堅牢な検証が重要です。

よくある質問

Q 1。SQL世代のPydanticaiの利点は何ですか？

A. Pydanticaiは、組み込みのエラーチェックとコンテキスト理解を備えたタイプセーフ、検証済みのコード生成を提供します。

Q2。Gemini-1.5-Flashはプロジェクトにどのように貢献しますか？

A.ジェミニモデルは、高度な自然言語処理を提供し、複雑な人間のクエリを正確なSQLステートメントに変換します。

Q 3。このプロジェクトは他のAIアプリケーションに拡張できますか？

A.絶対に！アーキテクチャは、さまざまなドメインでコード生成、データ変換、およびインテリジェントオートメーションに適合させることができます。

この記事に示されているメディアは、Analytics Vidhyaが所有しておらず、著者の裁量で使用されています。

以上がPydanticaiで生産グレードのLLM駆動型アプリケーションを構築しますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。