Pydantic AI Middleware

Intercepta, transforma y protege cada llamada de IA

Biblioteca middleware ligera con 7 hooks de ciclo de vida para Pydantic AI. Ejecución paralela, guardrails async y seguimiento de costos.

Instalación

pip install pydantic-ai-middleware

Pydantic AI Middleware proporciona 7 hooks de ciclo de vida para interceptar cada etapa de una ejecución de agente: before_run, after_run, before_model_request, before_tool_call, after_tool_call, on_tool_error y on_run_error. Múltiples hooks middleware se ejecutan en paralelo con estrategias de agregación configurables (first_response, all_responses, majority_vote). Los guardrails async soportan tres modos — BLOCKING (esperar antes de continuar), CONCURRENT (ejecutar junto al agente) y ASYNC_POST (disparar-y-olvidar tras completar).

Características

Inicio Rápido

from pydantic_ai import Agent
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware

class CostTracker(AgentMiddleware[None]):
    """Track token usage and costs across all agent runs."""

    async def after_run(self, prompt, output, deps, ctx=None):
        print(f"Tokens used: {ctx.usage.total_tokens}")
        return output

base_agent = Agent("openai:gpt-4o", instructions="You are a helpful assistant.")
agent = MiddlewareAgent(agent=base_agent, middleware=[CostTracker()])
result = await agent.run("Hello, how are you?")

Casos de Uso

Ciclo de Vida de Hooks

flowchart LR
    Input[Input] --> before_run[before_run]
    before_run --> before_model[before_model_request]
    before_model --> Agent[Agent]
    Agent -->|tool call| before_tool[before_tool_call]
    before_tool --> Tool[Tool]
    Tool --> after_tool[after_tool_call]
    after_tool --> Agent
    Tool -.->|error| on_error[on_tool_error]
    on_error -.-> Agent
    Agent -->|finish| after_run[after_run]
    after_run --> Output[Output]