Pydantic AI: Ein praktischer Einstieg mit Beispielen

Beim Entwickeln KI-gestützter Anwendungen kommt es schnell zu unstrukturierten Ausgaben, Typkonflikten und Problemen mit der Zuverlässigkeit in der Produktion. Klassische Ansätze, LLMs in Python-Apps zu integrieren, fehlen oft Struktur und Validierung, die Produktionssysteme brauchen. Pydantic AI löst das, indem es Pydantics Datenvalidierung mit einem Agent-Framework für LLM-Interaktionen kombiniert.

In diesem Tutorial baust du strukturierte KI-Agenten, die validierte Ausgaben liefern und Typensicherheit wahren. Wir richten Agents mit Systemprompts ein, erstellen Funktionstools, die LLMs im Dialog aufrufen können, und implementieren valide, strukturierte Ausgaben. Außerdem lernst du, wie du Abhängigkeiten über Agent-Komponenten hinweg verwaltest und Antworten für Echtzeitanwendungen streamst. Am Ende hast du praktische Erfahrung darin, KI-Anwendungen zu entwickeln, die komplexe Workflows beherrschen.

Wenn du einen praxisnahen Kurs zum Aufbau agentischer KI-Systeme suchst, empfehle ich den Building Multi-Agent Systems with LangGraph Kurs. 

Was ist Pydantic AI?

Pydantic AI ist ein Python-Agent-Framework, das LLM-Anwendungen Struktur und Typensicherheit gibt. Anders als einfache LLM-Integrationen, die reinen Text zurückgeben, behandelt Pydantic AI KI-Interaktionen als strukturierte Gespräche mit validierten Ein- und Ausgaben – ideal, um verlässliche Anwendungen zu bauen.

Wichtige Funktionen und Vorteile:

• Validierung strukturierter Ausgaben — Deine KI-Antworten entsprechen automatisch Pydantic-Modellen, vermeiden Parsing-Fehler und sichern Datenkonsistenz
• Funktionstools — LLMs können während des Gesprächs deine Python-Funktionen aufrufen und so auf echte Daten und Berechnungen zugreifen
• Typensicherheit — Vollständige Typunterstützung bedeutet weniger Laufzeitfehler und bessere IDE-Unterstützung für komplexe KI-Workflows
• Systemprompts — Definiere klare Anweisungen für deine KI-Agenten, die über alle Interaktionen hinweg konsistent bleiben
• Dependency Injection — Teile Kontext, Datenbankverbindungen und Nutzerpräferenzen sauber über alle Agent-Komponenten
• Mehrere Ausführungsmodi — Führe Agents synchron, asynchron aus oder streame Antworten in Echtzeit – je nach Bedarf
• Wiederverwendbare Agents — Erstelle Agents einmal und nutze sie in deiner Anwendung wieder, ähnlich wie FastAPI-Router

Dieser Ansatz macht aus unvorhersehbarer Textgenerierung verlässliche, typisierte Interaktionen, die sich nahtlos in bestehende Python-Anwendungen einfügen. Statt dich mit Prompt-Engineering und Output-Parsing herumzuschlagen, kannst du dich auf die Funktionen konzentrieren, die deine Nutzer wirklich brauchen.

Voraussetzungen

Bevor du loslegst, brauchst du Folgendes:

Erforderliche Kenntnisse:

• Python-Grundlagen: Vertrautheit mit Klassen und Funktionen
• Grundlagen der asynchronen Programmierung in Python
• Pydantic-Basics: Verständnis von Modellen, Feldvalidierung und Typannotationen
• LLM-Grundverständnis:  Grundlegendes Verständnis, wie Sprachmodelle funktionieren und was Prompts sind

Hilfreich, aber nicht zwingend:

• Agenten-Konzepte:  Erste Berührungspunkte mit KI-Agenten und Tool-Aufrufen, wir gehen aber darauf ein

Setup-Anforderungen:

• Python 3.9 oder höher
• OpenAI- und Anthropic-API-Schlüssel
• Grundkenntnisse im Terminal/Kommandozeile zur Paketinstallation

Wenn du neu bei Pydantic bist: Die offizielle Dokumentation deckt die Grundlagen gut ab. Für LLM-Grundlagen reicht ein Einsteiger-Guide zu ChatGPT oder ähnlichen Modellen, um mitzukommen.

Erste Schritte mit Pydantic AI

Im Lauf des Tutorials bauen wir einen Datenanalyse-Agenten, der uns hilft, die Verkaufsleistung über Regionen und Produkte hinweg zu verstehen. Der Agent startet einfach und wird nach und nach um Funktionen wie benutzerdefinierte Tools, strukturierte Ausgaben und Streaming erweitert.

Installation und Setup

Bevor wir mit dem Code loslegen, installieren wir Pydantic AI und richten die Umgebung ein:

pip install pydantic-aiBereitgestellt von

Du brauchst außerdem einen OpenAI- und einen Anthropic-API-Schlüssel. Setze sie als Umgebungsvariablen:

export OPENAI_API_KEY="your-api-key-here"export ANTHROPIC_API_KEY="your-api-key-here"Bereitgestellt von

> Wir verwenden sowohl OpenAI- als auch Anthropic-Modelle, da einige Funktionen von Pydantic AI jeweils nur von einem der Anbieter unterstützt werden.

Für unsere Datenanalyse-Beispiele nutzen wir außerdem pandas zur Datenverarbeitung:

pip install pandasBereitgestellt von

Erstelle deinen ersten Agenten

Starten wir mit dem einfachsten möglichen Agenten – einem, der grundlegende Fragen zu Verkaufsdaten beantwortet. In Pydantic AI ist ein Agent deine Hauptschnittstelle für LLM-Interaktionen (führe den folgenden Code in einem Python-Skript aus):

from pydantic_ai import Agent# Create a basic sales analysis agentsales_agent = Agent(   'openai:gpt-4o',   system_prompt=(       "You are a data analyst specializing in sales performance. "       "Provide clear, actionable insights based on the data provided."   ))# Run the agent with a simple questionresult = sales_agent.run_sync("What are the key metrics I should track for sales performance? Answer in two sentences.")print(result.output)Bereitgestellt von

Ausgabe:

Key metrics to track for sales performance include conversion rate, which measures the percentage of leads that result in sales, and average deal size, indicating the average revenue earned per transaction. Additionally, monitoring customer acquisition cost (CAC) and customer lifetime value (CLV) helps evaluate the efficiency and profitability of sales efforts.Bereitgestellt von

So erstellst du einen Agenten, der seine Rolle als Sales-Analyst versteht und passende Empfehlungen gibt. Der system_prompt gibt dem LLM Kontext zu den gewünschten Antworten, während run_sync() den Agenten ausführt und ein Ergebnis zurückgibt.

Hier funktioniert run_sync() nur in Python-Skripten. In einem Jupyter Notebook kannst du folgende Syntax verwenden:

result = await agent.run("...")print(result.output)Bereitgestellt von

LLM und Einstellungen ändern

Pydantic AI unterstützt mehrere LLM-Anbieter und erlaubt dir, das Modellverhalten anzupassen. So experimentierst du mit verschiedenen Optionen:

# Creating agents with different modelsclaude_agent = Agent('anthropic:claude-3-5-sonnet-20241022', system_prompt="You are a sales analyst.")gpt_mini_agent = Agent('openai:gpt-4o-mini', system_prompt="You are a sales analyst.")Bereitgestellt von

Jedes Modell bringt andere Stärken in deinen Analyse-Workflow. Claude-Modelle glänzen bei detailliertem Reasoning und komplexen Analysen – ideal, wenn du gründliche Einblicke aus Verkaufsdaten brauchst. GPT-4o-mini liefert schnellere Antworten bei geringeren Kosten – perfekt für kurze Fragen oder hohe Abfragevolumina.

Über die Modellauswahl hinaus kannst du das Verhalten mit Modelleinstellungen feinsteuern:

# Add model-specific settingsconfigured_agent = Agent(   'openai:gpt-4o',   system_prompt="You are a sales analyst.",   model_settings={       'temperature': 0.3,  # Fokussiertere, weniger kreative Antworten       'max_tokens': 500,   # Antwortlänge begrenzen   })Bereitgestellt von

Anpassbare Modelleinstellungen:

• temperature — Steuert Kreativität (0,0 = fokussiert, 1,0 = kreativ)
• max_tokens — Begrenzt die Antwortlänge
• top_p — Alternative zu temperature, um Zufälligkeit zu steuern

Für Datenanalyse-Aufgaben funktionieren niedrigere Temperaturwerte (0,1–0,3) meist besser, da du konsistente, faktenbasierte Antworten willst statt kreativer Ausschmückungen. Mit max_tokens kontrollierst du die Länge – nützlich für prägnante Zusammenfassungen oder bei API-Kosten.

Tools in Agents verwenden

Bisher kann unser Sales-Analyse-Agent nur mit den Informationen arbeiten, die wir im Prompt liefern. In der Praxis braucht Analyse oft externe Informationen, Berechnungen oder Anbindungen an Business-Systeme. Hier kommen Tools ins Spiel – sie geben deinem Agenten die Fähigkeit, während des Gesprächs zu handeln und Informationen zu sammeln.

Eingebaute Tools

Eingebaute Tools sind sofort nutzbare Funktionen der LLM-Anbieter. Pydantic AI unterstützt Websuche und Codeausführung, die direkt auf der Infrastruktur des Providers laufen.

Starten wir mit der Websuche – ideal, um aktuelle Marktdaten zu bekommen:

from pydantic_ai import Agent, WebSearchTool# Create an agent with web search abilitiesmarket_research_agent = Agent(   'anthropic:claude-3-5-sonnet-20241022',   builtin_tools=[WebSearchTool()],   system_prompt=(       "You are a sales analyst. Use web search to find current information. "       "Keep responses to 3-4 sentences maximum."   ))# Get current market informationresult = market_research_agent.run_sync(   "Search for 2025 laptop sales trends and average selling prices. Give me the main trends.")print(result.output)Bereitgestellt von

Let me search for current laptop sales trends.Based on the search results, here are the key laptop market trends for 2025:…Bereitgestellt von

Damit hebst du die Analyse von veralteten Infos auf ein Niveau mit Echtzeit-Marktdaten. Dein Agent kann die Performance deines Unternehmens mit aktuellen Branchentrends vergleichen.

Jetzt fügen wir Codeausführung für mathematische Analysen und Datenverarbeitung hinzu:

from pydantic_ai import Agent, CodeExecutionTool# Agent that can run calculations and data analysiscalculation_agent = Agent(   'anthropic:claude-sonnet-4-0',   builtin_tools=[CodeExecutionTool()],   system_prompt=(       "You are a data analyst. Use code to perform calculations and create visualizations. "       "Show your work and provide clear explanations."   ))# Analyze sales data with calculationsresult = calculation_agent.run_sync("""I have sales data for Q1 2025:- North region: 150 laptops at $1,500 each, 320 phones at $500 each- South region: 89 laptops at $1,500 each, 245 phones at $500 each - East region: 201 laptops at $1,500 each, 180 phones at $500 each- West region: 167 laptops at $1,500 each, 290 phones at $500 eachCalculate total revenue by region and create a simple chart showing the results.""")print(result.output)```plaintext## SummaryBased on the Q1 2025 sales data analysis, here are the key findings:**Total Revenue by Region:**- **West Region**: $395,500 (highest)- **East Region**: $391,500- **North Region**: $385,000- **South Region**: $256,000 (lowest)Bereitgestellt von

Mit Codeausführung gehst du über reine Textantworten hinaus zu echter Rechenanalyse. Dein Agent kann komplexe Berechnungen durchführen, Visualisierungen erstellen und Daten verarbeiten, was im Chat allein unmöglich wäre. Das Result-Objekt enthält sowohl die Analyse des Agents als auch eventuell erzeugte Dateien aus der Codeausführung, die du herunterladen kannst.

Eigene Tools schreiben

Eingebaute Tools sind hilfreich, aber oft brauchst du eigene Tools für deine spezifischen Systeme. Custom Tools sind Python-Funktionen, die dein Agent im Gespräch aufrufen kann.

Erstellen wir Tools für gängige Sales-Kennzahlen:

from pydantic_ai import Agent# Create an agent with custom business toolssales_agent = Agent(   'anthropic:claude-sonnet-4-0',   system_prompt=(       "You are a sales analyst. Use tools to calculate metrics and analyze business data. "       "Give concise answers in 3-4 sentences maximum."   ))@sales_agent.tool_plaindef calculate_conversion_rate(leads: int, sales: int) -> str:   """Calculate conversion rate from leads to sales."""   if leads == 0:       return "Cannot calculate conversion rate: no leads provided"     conversion_rate = (sales / leads) * 100   return f"Conversion rate: {conversion_rate:.2f}% ({sales} sales from {leads} leads)"@sales_agent.tool_plaindef calculate_average_order_value(total_revenue: float, total_orders: int) -> str:   """Calculate average order value."""   if total_orders == 0:       return "Cannot calculate AOV: no orders provided"     aov = total_revenue / total_orders   return f"Average Order Value: ${aov:.2f} (${total_revenue:,.0f} revenue from {total_orders} orders)"# Use the custom toolsresult = sales_agent.run_sync(   "Calculate conversion rate for 500 leads that resulted in 75 sales, and AOV for $180,000 revenue from 120 orders")print(result.output)Bereitgestellt von

Your conversion rate is 15.00%, meaning you're successfully converting 75 out of every 500 leads into sales. The average order value is $1,500.00, indicating strong revenue per transaction. These metrics suggest effective sales processes with high-value customers, though there's room to improve lead conversion rates.Bereitgestellt von

Diese Custom-Tools übernehmen wiederkehrende Sales-Berechnungen deines Unternehmens. Die Conversion-Rate misst die Marketingleistung, der Average Order Value zeigt Kaufverhalten. Eigene Tools verbinden deinen Agenten mit deiner Business-Logik und deinen Systemen.

Für fortgeschrittene Tool-Muster inklusive Dependency Injection und Fehlerbehandlung sieh dir die vollständige Tool-Dokumentation an.

Strukturierte Ausgaben nutzen

Bisher gaben unsere Agents Klartext zurück – gut zur Exploration, aber problematisch in Produktionsanwendungen. Wenn du Antworten programmatisch verarbeiten, bestimmte Datenpunkte parsen oder KI-Ausgaben mit anderen Systemen verbinden willst, brauchst du garantierte Struktur.

Angenommen, du willst Verkaufsdaten analysieren und immer bestimmte Kennzahlen zurückbekommen. Mit Freitext liefert die KI jedes Mal andere Formate. Mit strukturierten Ausgaben definierst du exakt, was du willst:

from pydantic import BaseModelfrom pydantic_ai import Agentclass SalesInsight(BaseModel):   total_revenue: float   best_performing_region: str   worst_performing_region: str   recommendation: str# Create an agent that returns structured analysisanalysis_agent = Agent(   'anthropic:claude-sonnet-4-0',   output_type=SalesInsight,   system_prompt="Analyze sales data and provide structured insights with clear recommendations.")# Get structured analysisresult = analysis_agent.run_sync("""Q1 2025 sales data:- North: $385,000 revenue (470 units)- South: $256,000 revenue (334 units) - East: $391,500 revenue (381 units)- West: $395,500 revenue (457 units)Analyze this data and provide insights.""")print(f"Total Revenue: ${result.output.total_revenue:,.0f}")print(f"Best Region: {result.output.best_performing_region}")print(f"Recommendation: {result.output.recommendation}")Bereitgestellt von

Total Revenue: $1,428,000Best Region: WestRecommendation: Focus on improving South region performance through targeted marketing and sales training, as it significantly underperforms with only $256,000 revenue. The West region's success ($395,500) should be analyzed and replicated across other regions. Additionally, investigate why East region has the lowest average selling price per unit ($1,028) compared to West's highest ($865) - there may be pricing strategy opportunities.Bereitgestellt von

Beachte den Parameter output_type=SalesInsight. Er zwingt den Agenten, ein SalesInsight-Objekt mit genau diesen vier Feldern zurückzugeben – nicht mehr, nicht weniger. Das Ergebnis kommt als result.output, automatisch validiert und typisiert.

Strukturierte Ausgaben funktionieren, indem dein Pydantic-Modell in ein JSON-Schema überführt wird, dem das LLM folgt. Die KI kann nicht vom Schema abweichen – falls doch, bittet Pydantic AI um einen neuen Versuch.

Für komplexere Analysen kannst du Modelle schachteln und Listen verwenden:

from typing import Listfrom pydantic import BaseModelclass RegionalMetrics(BaseModel):   region: str   revenue: float   units_sold: int   average_price: float   performance_rating: str  # "excellent", "good", "needs_improvement"class ComprehensiveAnalysis(BaseModel):   total_revenue: float   total_units: int   regional_breakdown: List[RegionalMetrics]   top_performer: str   areas_for_improvement: List[str]   quarterly_grade: str# Agent with detailed structured outputdetailed_agent = Agent(   'anthropic:claude-sonnet-4-0',   output_type=ComprehensiveAnalysis,   system_prompt=(       "Provide comprehensive sales analysis with detailed regional metrics. "       "Rate each region's performance and give an overall quarterly grade (A-F)."   ))result = detailed_agent.run_sync("""Analyze Q1 2025 performance:- North: $385k revenue, 470 units- South: $256k revenue, 334 units- East: $391.5k revenue, 381 units - West: $395.5k revenue, 457 units""")# Access structured datafor region in result.output.regional_breakdown:   print(f"{region.region}: {region.performance_rating} (${region.average_price:.2f} avg price)")print(f"Quarterly Grade: {result.output.quarterly_grade}")Bereitgestellt von

North: B ($819.15 avg price)South: C ($766.47 avg price)East: A ($1027.56 avg price)West: A- ($865.43 avg price)Quarterly Grade: B+Bereitgestellt von

So wird aus unvorhersehbarer Textgenerierung verlässliche Datenverarbeitung. Du kannst jede regionale Kennzahl in einer Datenbank speichern, Alarme anhand der Bewertungen auslösen oder automatisch Management-Dashboards erzeugen. Kein Regex-Parsing, kein Handling wechselnder Formate – nur saubere, validierte Daten.

Der Parameter output_type akzeptiert jedes Pydantic-Modell, Python-Dataclass, TypedDict oder einfache Typen wie int oder list[str]. Für Datenanalyse-Workflows bedeutet das: Du kannst verlässliche Pipelines bauen, in denen KI-Analysen direkt in deine bestehenden Systeme fließen.

Nachrichten und Chatverlauf in Pydantic-Agenten

Bei Datenanalyse-Agenten ziehen sich Gespräche oft über mehrere Interaktionen. Vielleicht startest du mit Sales-Trends, vertiefst dann einzelne Regionen oder vergleichst Zeiträume. Nachrichten und Chatverlauf halten den Kontext über diese Gespräche hinweg.

Auf den Nachrichtenverlauf zugreifen

Bei jedem Lauf protokolliert Pydantic AI das gesamte Gespräch. Du kannst diesen Verlauf einsehen, um zu verstehen, was passiert ist, oder später weitermachen:

from pydantic_ai import Agent# Create our sales analysis agentsales_agent = Agent(   'anthropic:claude-sonnet-4-0',   system_prompt="You are a sales analyst. Provide clear, concise analysis.")# First questionresult1 = sales_agent.run_sync("What are the main KPIs I should track for Q1 2025 sales?")print(result1.output)# Access the conversation historyall_messages = result1.all_messages()print(f"Total messages in conversation: {len(all_messages)}")# Just the new messages from this runnew_messages = result1.new_messages()print(f"New messages from this run: {len(new_messages)}")Bereitgestellt von

Here are the essential KPIs to track for Q1 2025 sales:...Total messages in conversation: 2New messages from this run: 2Bereitgestellt von

Die Methode all_messages() liefert alles – Systemprompts, Nutzerfragen und Agent-Antworten. new_messages() gibt nur das zurück, was im aktuellen Lauf passiert ist. Praktisch für längere Gespräche, wenn du Änderungen nachverfolgen willst.

Gespräche mit Nachrichtenverlauf fortsetzen

Um auf vorherigen Gesprächen aufzubauen, übergib den Nachrichtenverlauf an den nächsten Agent-Lauf. So „erinnert“ sich der Agent an das Vorherige:

# Continue the conversation from where we left offresult2 = sales_agent.run_sync(   "How should I calculate conversion rates for each of those KPIs?",   message_history=result1.all_messages())print(result2.output)# The agent now has context from both interactionsprint(f"Full conversation length: {len(result2.all_messages())}")# Ask a follow-up that references the entire conversationresult3 = sales_agent.run_sync(   "Which of these metrics would be most important for a monthly executive report?",   message_history=result2.all_messages())print(result3.output)Bereitgestellt von

Jeder neue Lauf baut auf dem bisherigen Kontext auf. Der Agent behält nicht nur deine Fragen im Kopf, sondern auch seine eigenen Antworten – das sorgt für ein schlüssiges, analytisches Gespräch.

Nachrichten als JSON speichern und laden

Für langfristige Speicherung oder den Austausch zwischen Systemen kannst du den Verlauf als JSON serialisieren:

import json# Get the conversation as JSONconversation_json = result3.all_messages_json()# Save to filewith open('sales_analysis_conversation.json', 'w') as f:   f.write(conversation_json.decode('utf-8'))print("Conversation saved to JSON file")# Load it back laterwith open('sales_analysis_conversation.json', 'r') as f:   loaded_conversation = f.read()# You can now use this loaded conversation in a new agent run# (Note: You'd need to convert back from JSON to message objects for actual use)print(f"Loaded conversation size: {len(loaded_conversation)} characters")Bereitgestellt von

Die Methode all_messages_json() liefert den Verlauf als JSON-Bytes – ideal zum Speichern in Datenbanken, Versenden über APIs oder zum Archivieren. Es gibt auch new_messages_json(), wenn du nur den letzten Abschnitt sichern willst.

So baust du persistente Analyse-Workflows. Du kannst jede Kundensitzung speichern, komplexe Analysen über mehrere Tage fortsetzen oder Gesprächskontext im Team teilen.

Mit Bildern und Dokumenten arbeiten

Reale Datenanalyse umfasst oft mehr als Text und Zahlen. Du musst vielleicht Diagramme aus Berichten analysieren, Daten aus PDFs extrahieren oder Visualisierungen interpretieren. Pydantic AI unterstützt Bild- und Dokumenteingaben, sodass deine Agents direkt mit diesen Quellen arbeiten können.

Bilder aus URLs analysieren

Wenn Diagramme oder Visualisierungen online liegen, kannst du den Agenten direkt darauf verweisen:

from pydantic_ai import Agent, ImageUrl# Create an agent that can analyze visual datavisual_agent = Agent(   'anthropic:claude-sonnet-4-0',   system_prompt="You are a data analyst who can interpret charts, graphs, and business documents.")# Analyze a sales chart from a URLresult = visual_agent.run_sync([   "What are the main trends shown in this sales chart? Give me 3 key takeaways.",   ImageUrl(url='https://example.com/q1-sales-chart.png')])print(result.output)Bereitgestellt von

Der Agent erkennt Diagrammtypen, liest Achsenwerte, erkennt Trends und liefert Business-Insights – wie ein Analyst, der dasselbe Chart betrachtet.

Mit lokalen Bildern arbeiten

Für lokal gespeicherte Bilder – etwa Screenshots aus BI-Tools oder aus Excel exportierte Charts – nutze BinaryContent:

from pathlib import Pathfrom pydantic_ai import Agent, BinaryContent# Load a local sales dashboard screenshotdashboard_path = Path('sales_dashboard_jan_2025.png')result = visual_agent.run_sync([   "Analyze this sales dashboard. What metrics need attention based on what you see?",   BinaryContent(       data=dashboard_path.read_bytes(),       media_type='image/png'   )])print(result.output)Bereitgestellt von

Das eignet sich ideal für automatisierte Reportings. Du kannst Dashboards per Skript screenshotten und die KI die begleitende Analyse schreiben lassen.

Dokumente aus URLs analysieren

Viele Geschäftsdokumente werden per URL geteilt – Berichte im Intranet, PDFs im Cloudspeicher oder öffentliche Studien, die deine Analyse stützen:

from pydantic_ai import Agent, DocumentUrl# Create an agent for document analysisdoc_agent = Agent(   'anthropic:claude-sonnet-4-0',   system_prompt="Extract and summarize business-relevant insights from documents.")# Analyze a quarterly report PDFresult = doc_agent.run_sync([   "Read this quarterly earnings report and summarize the revenue trends and outlook.",   DocumentUrl(url='https://company.com/reports/q1-2025-earnings.pdf')])print(result.output)Bereitgestellt von

Der Agent kann ganze Dokumente lesen, relevante Punkte extrahieren und Zusammenfassungen liefern, für die Menschen deutlich länger bräuchten.

Lokale Dokumente verarbeiten

Für Dokumente auf deinem System – etwa exportierte Reports, Verträge oder interne Präsentationen – verwende BinaryContent:

from pathlib import Pathfrom pydantic_ai import Agent, BinaryContent# Load a local Excel export saved as PDFreport_path = Path('monthly_sales_analysis_jan_2025.pdf')result = doc_agent.run_sync([   "Extract the key performance metrics from this monthly sales report and identify any concerning trends.",   BinaryContent(       data=report_path.read_bytes(),       media_type='application/pdf'   )])print(result.output)Bereitgestellt von

Das eröffnet starke Automatisierungsmöglichkeiten. Du kannst ganze Reportstapel verarbeiten, Rechnungsdaten extrahieren oder Vertragsklauseln analysieren – Aufgaben, die sonst manuelle Prüfung erfordern.

Wichtige Hinweise:

Unterschiedliche Modelle unterstützen unterschiedliche Dateiformate. Die meisten kommen mit gängigen Bildformaten (PNG, JPEG) und PDFs klar, prüfe aber die Doku deines Modells. Einige Modelle wie Googles Vertex AI können direkt auf Cloud-Storage-URLs zugreifen, andere laden Inhalte zuerst herunter.

Für Datenanalyse-Workflows schließt diese Multimodalität die Lücke zwischen visuellen Reports und KI-Insights – und macht die automatisierte Auswertung deiner täglichen Geschäftsdokumente möglich.

Streaming in Pydantic AI aktivieren

Bei komplexen Analysen willst du Ergebnisse sehen, während sie entstehen – statt auf die komplette Antwort zu warten. Streaming lässt dich die Analyse in Echtzeit mitverfolgen, besonders wertvoll bei detaillierten Fragen zur Sales-Performance.

Pydantic AI bietet integriertes Streaming über die Methode run_stream(). So funktioniert es:

import asynciofrom pydantic_ai import Agent# Create our sales analysis agentsales_agent = Agent(   'anthropic:claude-sonnet-4-0',   system_prompt=(       "You are a data analyst specializing in sales performance. "       "Provide detailed, actionable insights based on the data provided."   ))async def demo_streaming():   question = """   Analyze this sales scenario: Our company sold 1,200 laptops and 2,300 phones   last quarter across 4 regions. Laptops average $1,500 each, phones $500 each.   What are the key insights and recommendations?   """     async with sales_agent.run_stream(question) as result:       async for text in result.stream_text():           print(text, end='', flush=True)# Run the streaming analysisasyncio.run(demo_streaming())Bereitgestellt von

Die Streaming-Syntax verstehen:

Das async with öffnet eine Verbindung zum LLM, die während des Empfangs von Textblöcken offen bleibt – wie ein Telefonat, das du aufbaust, führst und wieder beendest.

Die async for-Schleife verarbeitet jeden Textblock bei Ankunft. Beachte: stream_text() generiert jedes Mal den kompletten Text neu – erst „Hello“, dann „Hello world“, dann „Hello world, how“. Das gibt vollen Kontext, ist für die Anzeige aber teils redundant.

Warum async und asyncio zusammengehören:

Eine LLM-Antwort braucht Zeit – meist 3–10 Sekunden. In dieser Zeit sendet das Modell alle 50–200 Millisekunden kleine Textstücke zurück.

Ohne async würde dein Programm beim Warten einfrieren. Mit async kann Pythons Event Loop zwischen Tasks wechseln. So bleibt dein Programm responsiv und kann sogar mehrere Gespräche parallel führen.

asyncio.run() startet das Ganze – es erstellt den Event Loop und koordiniert alles. Mit asyncio.run(demo_streaming()) bereitest du die Umgebung, in der deine Async-Funktionen pausieren und fortsetzen können, während sie auf LLM-Antworten warten.

Jetzt erstellen wir eine wiederverwendbare Funktion, die nur neue Text-Deltas zeigt – für ein besseres Streaming-Erlebnis:

from pydantic_ai.messages import PartDeltaEvent, TextPartDeltaasync def stream_agent_response(agent: Agent, task: str):   """   Stream a response from any Pydantic AI agent showing just new text deltas.   """   async def event_handler(ctx, event_stream):       async for event in event_stream:           if isinstance(event, PartDeltaEvent) and isinstance(               event.delta, TextPartDelta           ):               print(event.delta.content_delta, end="", flush=True)   # Use agent.run() with event_stream_handler for proper streaming   result = await agent.run(task, event_stream_handler=event_handler)   print()  # Add a newline when streaming is complete   return result# Usage exampleresponse = asyncio.run(stream_agent_response(   sales_agent,   "What are the top 3 metrics I should track for quarterly sales analysis?"))Bereitgestellt von

Die Streaming-Funktion aufgeschlüsselt:

Die Funktion stream_agent_response() sorgt für ein natürlicheres Streaming, indem sie nur neuen Text anzeigt. So funktioniert’s:

1. Event-Handler einrichten: Die innere Funktion event_handler() lauscht auf spezifische Events während der Agent-Verarbeitung. Sie filtert PartDeltaEvent mit TextPartDelta – das sind neue Textbausteine.
2. Delta-Inhalte: Statt jedes Mal den kompletten Text zu bekommen, liefert event.delta.content_delta nur die neuen Wörter oder Zeichen – wie eine Schreibmaschinen-Animation.
3. Agent ausführen: Wir nutzen agent.run() mit event_stream_handler. So ruft der Agent unseren Handler für jedes Streaming-Event auf und wir steuern die Anzeige fein granular.
4. Echtzeit-Ausgabe: print(event.delta.content_delta, end="", flush=True) zeigt jeden neuen Teil sofort und fließend an.

So bekommst du das Gefühl, der KI in Echtzeit beim Denken zuzuschauen – statt auf die Komplettantwort zu warten oder wiederholt regenerierten Text zu sehen. Wir nutzen diese Funktion in den nächsten Abschnitten, um Streaming-Antworten in unseren Beispielen zu zeigen.

Fazit

In diesem Tutorial hast du Pydantic AI anhand praktischer Datenanalyse-Beispiele umfassend kennengelernt. Du hast gelernt, Agents mit Systemprompts zu erstellen, eingebaute und eigene Tools einzubinden, Ausgaben mit Pydantic-Modellen zu strukturieren und Antworten in Echtzeit zu streamen. Zusammengenommen entstehen KI-Systeme, die eher analytische Partner sind als einfache Chatbots.

Die Anwendungsmöglichkeiten gehen weit über Sales-Analysen hinaus. Du kannst jetzt KI-Agenten bauen, die deine spezifischen Datenquellen lesen, deine Geschäftsregeln befolgen und Ergebnisse in Formaten liefern, die mit deinen bestehenden Systemen harmonieren. Ob automatisierte Report-Erstellung, Analyse von Kundenfeedback oder Verarbeitung finanzieller Daten – Pydantic AI liefert die Struktur und Zuverlässigkeit, die es für produktive Anwendungen braucht.

Wenn dich Multi-Agent-Systeme mit LangGraph interessieren, schau dir unbedingt unseren Hands-on-Kurs an.