Pydantic AI Middleware

Jeden KI-Aufruf abfangen, transformieren und schützen

Leichtgewichtige Middleware-Bibliothek mit 7 Lifecycle-Hooks für Pydantic AI. Parallele Ausführung, async Guardrails und Kostentracking.

Installation

pip install pydantic-ai-middleware

Pydantic AI Middleware bietet 7 Lifecycle-Hooks zum Abfangen jeder Phase eines Agent-Laufs: before_run, after_run, before_model_request, before_tool_call, after_tool_call, on_tool_error und on_run_error. Mehrere Middleware-Hooks werden parallel mit konfigurierbaren Aggregationsstrategien ausgeführt (first_response, all_responses, majority_vote). Async Guardrails unterstützen drei Timing-Modi — BLOCKING (vor dem Fortfahren warten), CONCURRENT (neben dem Agenten ausführen) und ASYNC_POST (Fire-and-Forget nach Abschluss).

Features

Schnellstart

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

Anwendungsfälle

Hook-Lebenszyklus

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]