pydantic-ai-tool-system

PydanticAI ツールシステム

ツールの登録

コンテキストが必要かどうかに応じて、2 種類のデコレーターを使います。

from pydantic_ai import Agent, RunContext

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

# @agent.tool - First param MUST be RunContext
@agent.tool
async def get_user_data(ctx: RunContext[MyDeps], user_id: int) -> str:
    """Get user data from database.

    Args:
        ctx: The run context with dependencies.
        user_id: The user's ID.
    """
    return await ctx.deps.db.get_user(user_id)

# @agent.tool_plain - NO context parameter allowed
@agent.tool_plain
def calculate_total(prices: list[float]) -> float:
    """Calculate total price.

    Args:
        prices: List of prices to sum.
    """
    return sum(prices)

重要なルール

1. @agent.tool：最初のパラメーターは必ず RunContext[DepsType] にします
2. @agent.tool_plain：RunContext パラメーターを指定してはいけません
3. Docstring：LLM がツールの目的を理解するために必要です
4. Google スタイルの docstring：パラメーターの説明に使用します

ゲート（記憶に頼らず、ファイル内で確認してください）

1. デコレーターとシグネチャの一致 — 最初のパラメーターが RunContext[...] の場合、デコレーターは @agent.tool（@agent.tool_plain ではありません）でなければなりません。合格条件： 同じ def 行のデコレーター列に @agent.tool が含まれ、最初のパラメーターの型が RunContext[...] であること。
2. プレーンツール — @agent.tool_plain を使う場合、パラメーターリストに RunContext を含めてはいけません。合格条件： シグネチャをざっと確認し、RunContext がないこと。
3. モデル向けの docstring — 空でない docstring が必要です。ツールにパラメーターがある場合は、その説明を記述します（docstring_format='sphinx' を使用する場合は、Google の Args: または Sphinx の :param を使用します）。合格条件： シグネチャ内の各パラメーターが docstring 本文で言及されていること。

Docstring の形式

Google スタイル（デフォルト）：

@agent.tool_plain
async def search(query: str, limit: int = 10) -> list[str]:
    """Search for items.

    Args:
        query: The search query.
        limit: Maximum results to return.
    """

Sphinx スタイル：

@agent.tool_plain(docstring_format='sphinx')
async def search(query: str) -> list[str]:
    """Search for items.

    :param query: The search query.
    """

ツールの戻り値の型

ツールはさまざまな型を返せます。

# String (direct)
@agent.tool_plain
def get_info() -> str:
    return "Some information"

# Pydantic model (serialized to JSON)
@agent.tool_plain
def get_user() -> User:
    return User(name="John", age=30)

# Dict (serialized to JSON)
@agent.tool_plain
def get_data() -> dict[str, Any]:
    return {"key": "value"}

# ToolReturn for custom content types
from pydantic_ai import ToolReturn, ImageUrl

@agent.tool_plain
def get_image() -> ToolReturn:
    return ToolReturn(content=[ImageUrl(url="https://...")])

コンテキストへのアクセス

RunContext が提供するもの：

@agent.tool
async def my_tool(ctx: RunContext[MyDeps]) -> str:
    # Dependencies
    db = ctx.deps.db
    api = ctx.deps.api_client

    # Model info
    model_name = ctx.model.model_name

    # Usage tracking
    tokens_used = ctx.usage.total_tokens

    # Retry info
    attempt = ctx.retry  # Current retry attempt (0-based)
    max_retries = ctx.max_retries

    # Message history
    messages = ctx.messages

    return "result"

ツールの Prepare 関数

リクエストごとにツールを動的に変更します。

from pydantic_ai.tools import ToolDefinition

async def prepare_tools(
    ctx: RunContext[MyDeps],
    tool_defs: list[ToolDefinition]
) -> list[ToolDefinition]:
    """Filter or modify tools based on context."""
    if ctx.deps.user_role != 'admin':
        # Hide admin tools from non-admins
        return [t for t in tool_defs if not t.name.startswith('admin_')]
    return tool_defs

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

ツールセット

ツールをグループ化して組み合わせます。

from pydantic_ai import FunctionToolset, CombinedToolset

# Create a toolset
db_tools = FunctionToolset()

@db_tools.tool
def query_users(name: str) -> list[dict]:
    """Query users by name."""
    ...

@db_tools.tool
def update_user(id: int, data: dict) -> bool:
    """Update user data."""
    ...

# Use in agent
agent = Agent('openai:gpt-4o', toolsets=[db_tools])

# Combine toolsets
all_tools = CombinedToolset([db_tools, api_tools])

よくある間違い

誤り：tool_plain でコンテキストを指定する

@agent.tool_plain
async def bad_tool(ctx: RunContext[MyDeps]) -> str:  # ERROR!
    ...

誤り：tool でコンテキストを指定しない

@agent.tool
def bad_tool(user_id: int) -> str:  # ERROR!
    ...

誤り：コンテキストが最初のパラメーターではない

@agent.tool
def bad_tool(user_id: int, ctx: RunContext[MyDeps]) -> str:  # ERROR!
    ...

非同期と同期

どちらも使えますが、I/O には非同期が推奨されます。

# Async (preferred for I/O operations)
@agent.tool
async def fetch_data(ctx: RunContext[Deps]) -> str:
    return await ctx.deps.client.get('/data')

# Sync (fine for CPU-bound operations)
@agent.tool_plain
def compute(x: int, y: int) -> int:
    return x * y