Pydantic AI Middleware

Przechwytuj, transformuj i chroń każde wywołanie AI

Lekka biblioteka middleware z 7 hookami cyklu życia dla Pydantic AI. Obsługuje równoległe wykonanie, asynchroniczne guardrails i śledzenie kosztów.

Instalacja

pip install pydantic-ai-middleware

Pydantic AI Middleware udostępnia 7 hooków cyklu życia pozwalających przechwycić każdy etap uruchomienia agenta: before_run, after_run, before_model_request, before_tool_call, after_tool_call, on_tool_error i on_run_error. Wiele hooków middleware wykonuje się równolegle z konfigurowalnymi strategiami agregacji (first_response, all_responses, majority_vote). Asynchroniczne guardrails wspierają trzy tryby — BLOCKING (czekaj przed kontynuacją), CONCURRENT (uruchom obok agenta) i ASYNC_POST (odpal-i-zapomnij po zakończeniu).

Funkcje

Szybki Start

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?")

Przypadki Użycia

Cykl Życia Hooków

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]