PydanticAI: 本番環境に対応した AI アプリケーションを構築するための包括的なガイド

PydanticAI は、Generative AI を使用して運用グレードのアプリケーションの開発を合理化するように設計された強力な Python フレームワーク です。これは、広く使用されているデータ検証ライブラリである Pydantic と同じチームによって構築されており、FastAPI の革新的で人間工学に基づいた設計を AI アプリケーション開発の分野に導入することを目的としています。 PydanticAI は、タイプ セーフティ、モジュール性、他の Python ツールとのシームレスな統合に重点を置いています。

中心となる概念

PydanticAI は、いくつかの重要な概念を中心に展開します。

エージェント

エージェントは、大規模言語モデル (LLM) と対話するための 主要なインターフェースです。エージェントは、次のようなさまざまなコンポーネントのコンテナとして機能します。

• システム プロンプト: LLM の指示。静的文字列または動的関数として定義されます。
• 関数ツール: LLM が追加情報を取得したり、アクションを実行したりするために呼び出すことができる関数。
• 構造化結果タイプ: LLM が実行の最後に返さなければならないデータ型。
• 依存関係の種類: システム プロンプト関数、ツール、結果バリデーターが使用する可能性のあるデータまたはサービス。
• LLM モデル: エージェントが使用する LLM。エージェントの作成時または実行時に設定できます。

エージェントは再利用できるように設計されており、通常は一度インスタンス化され、アプリケーション全体で再利用されます。

システムプロンプト

システム プロンプトは、開発者によって LLM に提供される指示です。それらは次のとおりです:

• 静的システム プロンプト: エージェントの作成時に、エージェント コンストラクターの system_prompt パラメーターを使用して定義されます。
• 動的システム プロンプト: @agent.system_prompt で修飾された関数によって定義されます。これらは、RunContext オブジェクトを介して、依存関係などの実行時情報にアクセスできます。

単一のエージェントは静的システム プロンプトと動的システム プロンプトの両方を使用でき、これらは実行時に定義された順序で追加されます。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

機能ツール

関数ツールを使用すると、LLM は外部情報にアクセスしたり、システム プロンプト自体では利用できないアクションを実行したりできます。ツールはいくつかの方法で登録できます:

• @agent.tool デコレータ: RunContext 経由でエージェントのコンテキストにアクセスする必要があるツール用。
• @agent.tool_plain デコレータ: エージェントのコンテキストにアクセスする必要のないツール用。
エージェント コンストラクターの
• tools キーワード引数: Tool クラスのプレーン関数またはインスタンスを受け取ることができ、ツール定義をより詳細に制御できます。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

ツール パラメーターは関数シグネチャから抽出され、ツールの JSON スキーマの構築に使用されます。関数の docstring は、ツールの説明とスキーマ内のパラメーターの説明を生成するために使用されます。

依存関係

依存関係は、依存関係注入システムを介して、エージェントのシステム プロンプト、ツール、結果バリデーターにデータとサービスを提供します。依存関係には RunContext オブジェクトを通じてアクセスします。任意の Python タイプを使用できますが、データクラスは複数の依存関係を管理する便利な方法です。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

結果

結果は、エージェントの実行から返される最終値です。これらは RunResult (同期および非同期実行の場合) または StreamedRunResult (ストリーミング実行の場合) にラップされ、使用状況データとメッセージ履歴へのアクセスを提供します。結果はプレーンテキストまたは構造化データであり、Pydantic を使用して検証されます。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

@agent.result_validator デコレータを介して追加された結果バリデータは、特に検証に IO が必要で非同期である場合に、さらに検証ロジックを追加する方法を提供します。

主な特長

PydanticAI は、AI アプリケーション開発にとって魅力的な選択肢となるいくつかの重要な機能を備えています。

• モデルに依存しない: PydanticAI は、OpenAI、Anthropic、Gemini、Ollama、Groq、Mistral などのさまざまな LLM をサポートしています。また、他のモデルのサポートを実装するためのシンプルなインターフェイスも提供します。
• 型安全性: mypy や pyright などの静的型チェッカーとシームレスに動作するように設計されています。これにより、依存関係と結果の型の型チェックが可能になります。
• Python 中心の設計: 使い慣れた Python 制御フローとエージェント構成を活用して AI プロジェクトを構築し、標準的な Python プラクティスを簡単に適用できます。
• 構造化された応答: Pydantic を使用してモデル出力を検証および構造化し、一貫した応答を保証します。
• 依存関係注入システム: エージェントのコンポーネントにデータとサービスを提供する依存関係注入システムを提供し、テスト容易性と反復開発を強化します。
• ストリーミング応答: 即時検証によるストリーミング LLM 出力をサポートし、迅速かつ正確な結果を可能にします。

エージェントとの連携

エージェントの実行

エージェントはいくつかの方法で実行できます:

• run_sync(): 同期実行用。
• run(): 非同期実行の場合。
• run_stream(): ストリーミング応答用。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

会話

エージェントの実行は会話全体を表す場合がありますが、特に対話間の状態を維持する場合には、会話が複数の実行で構成されることもあります。 message_history 引数を使用して以前の実行からのメッセージを渡し、会話を続行できます。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

使用制限

PydanticAI は、トークンとリクエストの数を制限するための settings.UsageLimits 構造を提供します。これらの設定は、usage_limits 引数を介して実行関数に適用できます。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

モデル設定

settings.ModelSettings 構造体を使用すると、温度、max_tokens、タイムアウトなどのパラメーターを通じてモデルの動作を微調整できます。これらは、実行関数の model_settings 引数を介して適用できます。

from pydantic import BaseModel
from pydantic_ai import Agent

class CityLocation(BaseModel):
    city: str
    country: str

agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'

機能ツールの詳細

ツールの登録

ツールは、@agent.tool デコレータ (コンテキストが必要なツールの場合)、@agent.tool_plain デコレータ (コンテキストのないツールの場合)、または Agent コンストラクタの tools 引数を使用して登録できます。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

# Synchronous run
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome

# Asynchronous run
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    async with agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

ツールスキーマ

パラメータの説明は docstring から抽出され、ツールの JSON スキーマに追加されます。ツールに JSON スキーマのオブジェクトとして表現できるパラメーターが 1 つある場合、スキーマはそのオブジェクトのみに簡略化されます。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')
result1 = agent.run_sync('Tell me a joke.')
print(result1.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync('Explain?', message_history=result1.new_messages())
print(result2.data)
#> This is an excellent joke invent by Samuel Colvin, it needs no explanation.

動的ツール

ツールは準備関数を使用してカスタマイズできます。この関数は、ツール定義を変更するか、そのステップからツールを省略するために各ステップで呼び出されます。

from pydantic_ai import Agent
from pydantic_ai.settings import UsageLimits
from pydantic_ai.exceptions import UsageLimitExceeded

agent = Agent('claude-3-5-sonnet-latest')
try:
    result_sync = agent.run_sync(
        'What is the capital of Italy? Answer with a paragraph.',
        usage_limits=UsageLimits(response_tokens_limit=10),
    )
except UsageLimitExceeded as e:
    print(e)
    #> Exceeded the response_tokens_limit of 10 (response_tokens=32)

メッセージとチャット履歴

メッセージへのアクセス

エージェントの実行中に交換されるメッセージには、RunResult オブジェクトと StreamedRunResult オブジェクトの all_messages() メソッドと new_messages() メソッドを介してアクセスできます。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')
result_sync = agent.run_sync(
    'What is the capital of Italy?',
    model_settings={'temperature': 0.0},
)
print(result_sync.data)
#> Rome

メッセージの再利用

メッセージを message_history パラメータに渡して、複数のエージェント実行にわたって会話を続けることができます。 message_history が設定されていて空でない場合、新しいシステム プロンプトは生成されません。

メッセージフォーマット

メッセージ形式はモデルに依存しないので、異なるエージェントでメッセージを使用したり、異なるモデルを使用する同じエージェントでメッセージを使用したりできます。

デバッグとモニタリング

ピダンティック・ログファイア

PydanticAI は、アプリケーション全体の監視とデバッグを可能にする可観測性プラットフォームである Pydantic Logfire と統合します。 Logfire は次の目的で使用できます:

• リアルタイム デバッグ: アプリケーションで何が起こっているかをリアルタイムで確認します。
• アプリケーションのパフォーマンスの監視: SQL クエリとダッシュボードを使用します。

Logfire で PydanticAI を使用するには、logfire オプション グループを使用してインストールします: pip install 'pydantic-ai[logfire]'。次に、Logfire プロジェクトを構成し、環境を認証する必要があります。

インストールとセットアップ

インストール

PydanticAI は pip を使用してインストールできます:

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

スリム インストールは、次のような特定のモデルを使用することもできます。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

Logfire の統合

Logfire で PydanticAI を使用するには、logfire オプション グループを使用してインストールします。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

例

サンプルは別のパッケージとして入手できます:

from pydantic import BaseModel
from pydantic_ai import Agent

class CityLocation(BaseModel):
    city: str
    country: str

agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'

テストと評価

単体テスト

単体テストでは、アプリケーション コードが期待どおりに動作するかどうかを検証します。 PydanticAI の場合は、次の戦略に従ってください:

• pytest をテスト ハーネスとして使用します。
• 実際のモデルの代わりに TestModel または FunctionModel を使用します。
• Agent.override を使用して、アプリケーション ロジック内のモデルを置き換えます。
• テスト以外のモデルへの誤った呼び出しを防ぐために、ALLOW_MODEL_REQUESTS=False をグローバルに設定します。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

# Synchronous run
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome

# Asynchronous run
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    async with agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

エヴァルス

評価は LLM のパフォーマンスを測定するために使用され、単体テストというよりはベンチマークに似ています。 Evals は、特定のアプリケーションに対して LLM がどのように実行されるかを測定することに重点を置いています。これは、エンドツーエンドのテスト、合成自己完結型テスト、LLM を使用した LLM の評価、または運用環境でのエージェントのパフォーマンスの測定によって実行できます。

使用例の例

PydanticAI はさまざまなユースケースで使用できます:

• ルーレット ホイール: 整数の依存関係とブール値の結果を持つエージェントを使用してルーレット ホイールをシミュレートします。
• チャット アプリケーション: 複数の実行でチャット アプリケーションを作成し、message_history を使用して以前のメッセージを渡します。
• 銀行サポート エージェント: ツール、依存関係の注入、構造化された応答を使用して銀行のサポート エージェントを構築します。
• 天気予報: 関数ツールと依存関係を使用して、場所と日付に基づいて天気予報を返すアプリケーションを作成します。
• SQL 生成: 結果バリデータを使用した検証を行い、ユーザー プロンプトから SQL クエリを生成します。

結論

PydanticAI は、型安全性とモジュール性を重視した AI アプリケーション開発のための堅牢かつ柔軟なフレームワークを提供します。データの検証と構造化に Pydantic を使用し、依存関係注入システムと組み合わせることで、信頼性が高く保守可能な AI アプリケーションを構築するための理想的なツールになります。 PydanticAI は、広範な LLM サポートと Pydantic Logfire などのツールとのシームレスな統合により、開発者が強力で本番環境に対応した AI 主導のプロジェクトを効率的に構築できるようにします。

以上がPydanticAI: 本番環境に対応した AI アプリケーションを構築するための包括的なガイドの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。