PydanticAI : un guide complet pour créer des applications d'IA prêtes pour la production-Tutoriel Python-php.cn

PydanticAI: A Comprehensive Guide to Building Production-Ready AI Applications

PydanticAI est un puissant framework Python conçu pour rationaliser le développement d'applications de production à l'aide de l'IA générative. Il est construit par la même équipe derrière Pydantic, une bibliothèque de validation de données largement utilisée, et vise à apporter la conception innovante et ergonomique de FastAPI au domaine du développement d'applications d'IA. PydanticAI se concentre sur la sécurité des types, la modularité et l'intégration transparente avec d'autres outils Python.

Concepts de base

PydanticAI s'articule autour de plusieurs concepts clés :

Agents

Les agents sont l'interface principale pour interagir avec les grands modèles linguistiques (LLM). Un agent agit comme un conteneur pour divers composants, notamment :

Invites système : instructions pour le LLM, définies sous forme de chaînes statiques ou de fonctions dynamiques.
Outils de fonctions : Fonctions que le LLM peut appeler pour obtenir des informations supplémentaires ou effectuer des actions.
Types de résultats structurés : types de données que le LLM doit renvoyer à la fin d'une exécution.
Types de dépendances : données ou services que les fonctions d'invite du système, les outils et les validateurs de résultats peuvent utiliser.
Modèles LLM : Le LLM que l'agent utilisera, qui peut être défini lors de la création de l'agent ou au moment de l'exécution.

Les agents sont conçus pour être réutilisables et sont généralement instanciés une seule fois et réutilisés tout au long d'une application.

Invites système

Les invites système sont des instructions fournies au LLM par le développeur. Ils peuvent être :

Invites système statiques : définies lors de la création de l'agent, à l'aide du paramètre system_prompt du constructeur de l'agent.
Invites système dynamiques : définies par des fonctions décorées avec @agent.system_prompt. Ceux-ci peuvent accéder aux informations d'exécution, telles que les dépendances, via l'objet RunContext.

Un seul agent peut utiliser des invites système statiques et dynamiques, qui sont ajoutées dans l'ordre dans lequel elles sont définies au moment de l'exécution.

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.

Copier après la connexion

Outils de fonction

Les outils fonctionnels permettent aux LLM d'accéder à des informations externes ou d'effectuer des actions non disponibles dans l'invite système elle-même. Les outils peuvent être enregistrés de plusieurs manières :

@agent.tool decorator : Pour les outils qui nécessitent un accès au contexte de l'agent via RunContext.
@agent.tool_plain decorator : Pour les outils qui n'ont pas besoin d'accéder au contexte de l'agent.
Argument de mot-clé tools dans le constructeur d'agent : peut prendre des fonctions simples ou des instances de la classe Tool, donnant plus de contrôle sur les définitions d'outils.

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.

Copier après la connexion

Les paramètres de l'outil sont extraits de la signature de la fonction et sont utilisés pour construire le schéma JSON de l'outil. Les docstrings de fonctions sont utilisées pour générer les descriptions de l'outil et les descriptions des paramètres dans le schéma.

Dépendances

Les dépendances fournissent des données et des services aux invites système, aux outils et aux validateurs de résultats de l'agent via un système d'injection de dépendances. Les dépendances sont accessibles via l'objet RunContext. Ils peuvent être de n'importe quel type Python, mais les classes de données constituent un moyen pratique de gérer plusieurs dépendances.

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!

Copier après la connexion

Résultats

Les résultats sont les valeurs finales renvoyées par une exécution d'agent. Ils sont encapsulés dans RunResult (pour les exécutions synchrones et asynchrones) ou StreamedRunResult (pour les exécutions en streaming), donnant accès aux données d'utilisation et à l'historique des messages. Les résultats peuvent être du texte brut ou des données structurées et sont validés à l'aide de 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.

Copier après la connexion

Les validateurs de résultats, ajoutés via le décorateur @agent.result_validator, fournissent un moyen d'ajouter une logique de validation supplémentaire, en particulier lorsque la validation nécessite des E/S et est asynchrone.

Principales fonctionnalités

PydanticAI possède plusieurs fonctionnalités clés qui en font un choix incontournable pour le développement d'applications d'IA :

Model Agnostic : PydanticAI prend en charge une variété de LLM, notamment OpenAI, Anthropic, Gemini, Ollama, Groq et Mistral. Il fournit également une interface simple pour implémenter la prise en charge d'autres modèles.
Type Safety : conçu pour fonctionner de manière transparente avec les vérificateurs de type statiques comme mypy et pyright. Il permet la vérification de type des dépendances et des types de résultats.
Conception centrée sur Python : exploite le flux de contrôle Python et la composition des agents familiers pour créer des projets d'IA, facilitant ainsi l'application des pratiques Python standard.
Réponses structurées : utilise Pydantic pour valider et structurer les sorties du modèle, garantissant des réponses cohérentes.
Système d'injection de dépendances : offre un système d'injection de dépendances pour fournir des données et des services aux composants d'un agent, améliorant ainsi la testabilité et le développement itératif.
Réponses diffusées : prend en charge le streaming des sorties LLM avec validation immédiate, permettant des résultats rapides et précis.

Travailler avec des agents

Agents en cours d'exécution

Les agents peuvent être exécutés de plusieurs manières :

run_sync() : Pour une exécution synchrone.
run() : Pour une exécution asynchrone.
run_stream() : pour diffuser les réponses.

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.

Copier après la connexion

Conversations

Une exécution d'agent peut représenter une conversation entière, mais les conversations peuvent également être composées de plusieurs exécutions, en particulier lors du maintien de l'état entre les interactions. Vous pouvez transmettre des messages d'exécutions précédentes en utilisant l'argument message_history pour poursuivre une conversation.

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!

Copier après la connexion

Limites d'utilisation

PydanticAI fournit une structure settings.UsageLimits pour limiter le nombre de jetons et de requêtes. Vous pouvez appliquer ces paramètres via l'argument usage_limits aux fonctions d'exécution.

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.

Copier après la connexion

Paramètres du modèle

La structure settings.ModelSettings vous permet d'affiner le comportement du modèle grâce à des paramètres tels que la température, max_tokens et le délai d'attente. Vous pouvez les appliquer via l'argument model_settings dans les fonctions d'exécution.

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'

Copier après la connexion

Outils de fonction en détail

Enregistrement des outils

Les outils peuvent être enregistrés à l'aide du décorateur @agent.tool (pour les outils nécessitant un contexte), du décorateur @agent.tool_plain (pour les outils sans contexte) ou via l'argument tools dans le constructeur de l'agent.

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

Copier après la connexion

Schéma de l'outil

Les descriptions des paramètres sont extraites des docstrings et ajoutées au schéma JSON de l'outil. Si un outil a un seul paramètre qui peut être représenté comme un objet dans le schéma JSON, le schéma est simplifié pour n'être que cet objet.

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.

Copier après la connexion

Outils dynamiques

Les outils peuvent être personnalisés avec une fonction de préparation, qui est appelée à chaque étape pour modifier la définition de l'outil ou omettre l'outil de cette étape.

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)

Copier après la connexion

Messages et historique des discussions

Accéder aux messages

Les messages échangés lors de l'exécution d'un agent sont accessibles via les méthodes all_messages() et new_messages() sur les objets RunResult et StreamedRunResult.

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

Copier après la connexion

Réutilisation des messages

Les messages peuvent être transmis au paramètre message_history pour poursuivre les conversations sur plusieurs exécutions d'agent. Lorsqu'un message_history est défini et non vide, une nouvelle invite système n'est pas générée.

Format des messages

Le format du message est indépendant du modèle, permettant aux messages d'être utilisés dans différents agents ou avec le même agent utilisant différents modèles.

Débogage et surveillance

Feu de bois pydantique

PydanticAI s'intègre à Pydantic Logfire, une plateforme d'observabilité qui vous permet de surveiller et de déboguer l'ensemble de votre application. Logfire peut être utilisé pour :

Débogage en temps réel : Pour voir ce qui se passe dans votre application en temps réel.
Surveillance des performances des applications : Utilisation de requêtes SQL et de tableaux de bord.

Pour utiliser PydanticAI avec Logfire, installez avec le groupe facultatif logfire : pip install 'pydantic-ai[logfire]'. Vous devez ensuite configurer un projet Logfire et authentifier votre environnement.

Installation et configuration

Installation

PydanticAI peut être installé en utilisant 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.

Copier après la connexion

Une installation slim est également disponible pour utiliser des modèles spécifiques, par exemple :

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!

Copier après la connexion

Intégration de Logfire

Pour utiliser PydanticAI avec Logfire, installez-le avec le groupe facultatif 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.

Copier après la connexion

Exemples

Des exemples sont disponibles dans un package séparé :

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'

Copier après la connexion

Tests et évaluation

Tests unitaires

Les tests unitaires vérifient que le code de votre application se comporte comme prévu. Pour PydanticAI, suivez ces stratégies :

Utilisez pytest comme harnais de test.
Utilisez TestModel ou FunctionModel à la place de votre modèle actuel.
Utilisez Agent.override pour remplacer votre modèle dans la logique de votre application.
Définissez ALLOW_MODEL_REQUESTS=False globalement pour éviter les appels accidentels vers des modèles non testés.

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

Copier après la connexion

Évaluations

Les évaluations sont utilisées pour mesurer les performances du LLM et ressemblent plus à des benchmarks qu'à des tests unitaires. Les évaluations se concentrent sur la mesure des performances du LLM pour une application spécifique. Cela peut être fait via des tests de bout en bout, des tests synthétiques autonomes, en utilisant des LLM pour évaluer les LLM, ou en mesurant les performances des agents en production.

Exemples de cas d'utilisation

PydanticAI peut être utilisé dans une grande variété de cas d'utilisation :

Roulette : Simulation d'une roue de roulette à l'aide d'un agent avec une dépendance entière et un résultat booléen.
Application de chat : Création d'une application de chat avec plusieurs exécutions, transmission des messages précédents à l'aide de message_history.
Agent de support bancaire : Création d'un agent de support pour une banque à l'aide d'outils, de l'injection de dépendances et de réponses structurées.
Prévisions météo : Création d'une application qui renvoie des prévisions météorologiques en fonction du lieu et de la date à l'aide d'outils de fonction et de dépendances.
Génération SQL : Génération de requêtes SQL à partir des invites utilisateur, avec validation à l'aide du validateur de résultat.

Conclusion

PydanticAI offre un cadre robuste et flexible pour développer des applications d'IA en mettant fortement l'accent sur la sécurité des types et la modularité. L'utilisation de Pydantic pour la validation et la structuration des données, couplée à son système d'injection de dépendances, en fait un outil idéal pour créer des applications d'IA fiables et maintenables. Grâce à sa large prise en charge LLM et à son intégration transparente avec des outils tels que Pydantic Logfire, PydanticAI permet aux développeurs de créer efficacement des projets puissants et prêts pour la production, basés sur l'IA.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!