📦 Pydantic Ai Tool System — 注册AI工具

技能文档

# PydanticAI Tool System ## 工具注册 根据是否需要上下文，有两种装饰器： ``python from pydantic_ai import Agent, RunContext agent = Agent('openai:gpt-4o') # @agent.tool - 第一个参数必须是 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 - 不允许出现 context 参数 @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. Docstrings：必须提供，以便 LLM 理解工具用途 4. Google 风格 docstrings：用于参数描述 ## Docstring 格式 Google 风格（默认）： `python @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 风格： `python @agent.tool_plain(docstring_format='sphinx') async def search(query: str) -> list[str]: """Search for items. :param query: The search query. """ ` ## 工具返回类型 工具可返回多种类型： `python # String（直接返回） @agent.tool_plain def get_info() -> str: return "Some information" # Pydantic 模型（序列化为 JSON） @agent.tool_plain def get_user() -> User: return User(name="John", age=30) # Dict（序列化为 JSON） @agent.tool_plain def get_data() -> dict[str, Any]: return {"key": "value"} # ToolReturn 用于自定义内容类型 from pydantic_ai import ToolReturn, ImageUrl @agent.tool_plain def get_image() -> ToolReturn: return ToolReturn(content=[ImageUrl(url="https://...")]) ` ## 访问上下文 RunContext 提供： `python @agent.tool async def my_tool(ctx: RunContext[MyDeps]) -> str: # 依赖项 db = ctx.deps.db api = ctx.deps.api_client # 模型信息 model_name = ctx.model.model_name # 用量追踪 tokens_used = ctx.usage.total_tokens # 重试信息 attempt = ctx.retry # 当前重试次数（从 0 开始） max_retries = ctx.max_retries # 消息历史 messages = ctx.messages return "result" ` ## 工具准备函数 根据请求动态修改工具： `python 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) ` ## 工具集 分组与组合工具： `python from pydantic_ai import FunctionToolset, CombinedToolset # 创建工具集 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.""" ... # 在 agent 中使用 agent = Agent('openai:gpt-4o', toolsets=[db_tools]) # 组合工具集 all_tools = CombinedToolset([db_tools, api_tools]) ` ## 常见错误 ### 错误：在 tool_plain 中使用上下文 `python @agent.tool_plain async def bad_tool(ctx: RunContext[MyDeps]) -> str: # ERROR! ... ` ### 错误：在 tool 中缺少上下文 `python @agent.tool def bad_tool(user_id: int) -> str: # ERROR! ... ` ### 错误：上下文不是第一个参数 `python @agent.tool def bad_tool(user_id: int, ctx: RunContext[MyDeps]) -> str: # ERROR! ... ` ## 异步 vs 同步 两者均可，但 I/O 操作推荐异步： `python # Async（I/O 操作首选） @agent.tool async def fetch_data(ctx: RunContext[Deps]) -> str: return await ctx.deps.client.get('/data') # Sync（CPU 密集型操作可接受） @agent.tool_plain def compute(x: int, y: int) -> int: return x * y ``