Pydantic AI & Mastra — Type-Safe Agents Python & TypeScript

TL;DR — En 2026, les frameworks agents qui gagnent sont ceux qui ressemblent au code que tu écris déjà, pas ceux qui inventent un mini-langage parallèle. Pydantic AI (Python) est devenu le standard "type-safe" : tu écris des BaseModel pour les outputs, des RunContext[Deps] pour l'injection de dépendances, et l'agent renvoie des objets typés — pas du JSON à parser. Mastra (TS) joue le même rôle côté JavaScript : workflows + agents + RAG + memory + integrations, avec un déploiement natif Cloudflare/Vercel et un DX proche de Next.js. Tous deux remplacent avantageusement LangChain (verbeux, abstractions fuyantes) sur 80% des cas, et complètent LangGraph (qui reste roi sur les graphes vraiment complexes). Pour un dev PHP/TS qui pivote freelance AI : Mastra est ton meilleur premier framework. Pour un projet Python prod-grade avec eval continu : Pydantic AI.

🧠 Mental model

   ┌──────────────────────────────────────────────────────────────┐
   │                    Type-safe agent (cible 2026)              │
   │                                                              │
   │   ┌──────────────┐                                           │
   │   │  Deps (DI)   │   ──── injecté ────┐                      │
   │   │ db, http,    │                    │                      │
   │   │ vector, kv   │                    ▼                      │
   │   └──────────────┘            ┌──────────────┐               │
   │                               │              │               │
   │   ┌──────────────┐            │   Agent      │   ─────►      │
   │   │  System prompt│  ──────►  │   (LLM call) │               │
   │   │ + tools(typed)│           │              │  Result[T]    │
   │   │ + output_type │           │              │  (typed!)     │
   │   └──────────────┘            └──────┬───────┘               │
   │                                      │                       │
   │                                      ▼                       │
   │                              ┌──────────────┐                │
   │                              │   Evals      │                │
   │                              │   (built-in) │                │
   │                              └──────────────┘                │
   └──────────────────────────────────────────────────────────────┘

       LangChain (2023-24)     →     LangGraph (2024-25)    →     Pydantic AI / Mastra (2026)
       imperatif + chains            graphs + state           types + DI + evals
       "ça marche en demo"           "ça marche en prod"      "ça marche, c'est testable, ça compile"

Analogie pour dev PHP/TS :

LangChain ≈ "framework PHP des années 2010" — beaucoup d'abstractions, beaucoup de magie, beaucoup de surprises en prod.
LangGraph ≈ "state machine + workflow engine" — puissant, verbeux, pour les vrais besoins de graphe.
Pydantic AI ≈ "FastAPI pour les agents" — tu déclares des types, le framework s'occupe du plumbing.
Mastra ≈ "Next.js pour les agents" — convention over configuration, batteries included, déploiement edge first-class.

L'évolution est la même que côté web : Symfony 1 → Symfony 2 → API Platform. On gagne en sécurité de type, en testabilité, en cohérence d'écosystème.

🛠️ Code minimal

Pydantic AI — agent typé en 25 lignes

python

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from dataclasses import dataclass
import httpx

@dataclass
class Deps:
    http: httpx.AsyncClient
    siret_api_key: str

class CompanyInfo(BaseModel):
    siren: str
    name: str
    naf_code: str
    employee_band: str
    risk_flag: bool

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    deps_type=Deps,
    output_type=CompanyInfo,
    system_prompt="Tu enrichis une entreprise française à partir d'un SIRET/SIREN.",
)

@agent.tool
async def lookup_sirene(ctx: RunContext[Deps], siret: str) -> dict:
    """Lookup INSEE Sirene API."""
    r = await ctx.deps.http.get(
        f"https://api.insee.fr/entreprises/sirene/V3/siret/{siret}",
        headers={"Authorization": f"Bearer {ctx.deps.siret_api_key}"},
    )
    return r.json()

# Usage
async with httpx.AsyncClient() as h:
    deps = Deps(http=h, siret_api_key="...")
    result = await agent.run("Enrichis 12345678900012", deps=deps)
    print(result.output.name)  # type CompanyInfo, autocomplete IDE

Mastra — agent TS avec memory + RAG

typescript

import { Mastra, Agent } from "@mastra/core";
import { createTool } from "@mastra/core/tools";
import { LibSQLStore, LibSQLVector } from "@mastra/libsql";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";

const memory = new LibSQLStore({ url: process.env.TURSO_URL! });
const vector = new LibSQLVector({ connectionUrl: process.env.TURSO_URL! });

const searchCompany = createTool({
  id: "searchCompany",
  description: "Recherche une entreprise FR par SIRET ou nom",
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ siren: z.string(), name: z.string(), naf: z.string() }),
  execute: async ({ context }) => {
    const r = await fetch(`https://api.insee.fr/.../${context.query}`);
    return r.json();
  },
});

const enricher = new Agent({
  name: "company-enricher",
  instructions: "Enrichis une entreprise FR à partir d'un SIRET ou nom.",
  model: anthropic("claude-sonnet-4-6"),
  tools: { searchCompany },
  memory: { store: memory, vector },
});

export const mastra = new Mastra({ agents: { enricher } });
// Usage: await mastra.getAgent("enricher").generate("Enrichis 12345678900012");

Trois choses non-négociables : (1) output_type (Pydantic AI) / outputSchema (Mastra) — sinon retour à l'ère du JSON-parsing fragile, (2) DI explicite (deps_type / context), pas de singletons globaux, (3) tests unitaires sur les tools en isolation — c'est la valeur principale du type-safe : tester sans appeler le LLM.

🎬 Cas d'usage concrets

Scénario 1 — Refactor d'un agent LangChain Python vers Pydantic AI (compta SaaS)

Contexte : SaaS compta FR, 35 devs, agent LangChain en prod depuis 2024 pour classer les écritures comptables. 1200 lignes de code, 4 fichiers de chains imbriquées, débug-hostile. Le tech lead veut migrer.

Cible : Pydantic AI, ~400 lignes, output typé CategorizedEntry, eval continu via Logfire.

Migration en 3 phases (12 jours freelance) :

Audit & dataset : 250 écritures historiques + golden labels. Eval baseline LangChain : 78% accuracy.
Rewrite : 1 Agent Pydantic AI, 6 tools (PCG lookup, TVA lookup, similar entries vector search, audit log, etc.), output CategorizedEntry.
Cutover progressif : 10% traffic Pydantic AI → 50% → 100% sur 3 semaines. Monitoring side-by-side.

Résultat :

Accuracy : 78% → 87% (golden tests + meilleur prompt + injection PCG).
LOC : -68% (1200 → 380).
Coût tokens : -22% (moins de prompts de "stitching" entre chains).
Latence p50 : 4,1s → 2,3s.

ROI mission freelance : 12j × 1300€ = 15,6k€. Économie tokens annuelle : ~18k€/an. Productivité dev (debug + nouveaux features) : x3 selon le tech lead.

Scénario 2 — Agent commercial e-commerce en Mastra TS dans monorepo Next.js

Contexte : marketplace FR (mode/déco), 20 devs, monorepo Turborepo (Next.js 15 + tRPC + Drizzle + Postgres). Le head of product veut un chatbot pré-vente : conseille un produit, vérifie stock, propose des alternatives, escalade vers humain si panier >300€.

Cible : Mastra agent intégré directement dans le monorepo (packages/ai-agent/), exposé via une route App Router Next.js, déployé Cloudflare Workers (edge, latence <100ms en Europe).

Pourquoi Mastra et pas Pydantic AI ici : monorepo TS, équipe TS, drizzle ORM partagé, déploiement edge — la friction Python est rédhibitoire. Mastra réutilise les types Drizzle, les composants UI Tailwind, les composables Next.

Stack :

@mastra/core : agent + tools.
@mastra/memory + Turso (libSQL) pour la mémoire utilisateur.
@mastra/rag : index produits dans Turso vector.
Tools : searchProducts, checkStock, escalateToHuman, applyPromoCode.
HITL : tool escalateToHuman qui ouvre un ticket Crisp avec le contexte.

Résultat :

+12% taux de conversion mobile (mesuré A/B test 4 semaines).
-25% tickets SAV pré-vente (l'agent répond aux 80% courantes).
Coût : 0,012€ par session moyenne. <1% du panier moyen.

ROI freelance : 18j × 1200€ = 21,6k€. Gain marge annuelle estimé : 280k€.

Scénario 3 — Agent compta Pydantic AI avec eval continu (cabinet EC)

Contexte : cabinet 8 EC, 200 clients, veut un agent qui pré-classe les factures fournisseurs scannées (OCR → catégorisation PCG) avant validation EC.

Cible : Pydantic AI + Logfire pour eval continu en prod (chaque écriture est notée a posteriori par l'EC lors de la validation, le score remonte dans Logfire). Quand l'accuracy descend sous 90% sur une rolling window 7j, alerte Slack au tech lead du cabinet.

Pourquoi Pydantic AI : eval natif via pydantic_evals, intégration Logfire one-liner, output strict BookkeepingProposal que le front EC peut afficher sans parse défensif.

🛠️ Exemple end-to-end — Mastra TS "Assistant juridique" avec memory + RAG + tools + déploiement Cloudflare Workers

Objectif : assistant juridique pour PME (questions droit commercial, contrats, baux) avec mémoire utilisateur, RAG sur Légifrance + jurisprudence, tools "rédige clause", "vérifie compatibilité RGPD". Déployé Cloudflare Workers, monorepo Turborepo, Next.js front.

typescript

// packages/legal-agent/src/agent.ts
import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import { createTool } from "@mastra/core/tools";
import { Memory } from "@mastra/memory";
import { LibSQLStore, LibSQLVector } from "@mastra/libsql";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";
import { embed } from "ai";
import { openai } from "@ai-sdk/openai";

// ------------------- Memory + RAG storage (Turso libSQL) -------------------
const store = new LibSQLStore({
  url: process.env.TURSO_DATABASE_URL!,
  authToken: process.env.TURSO_AUTH_TOKEN!,
});
const vector = new LibSQLVector({
  connectionUrl: process.env.TURSO_DATABASE_URL!,
  authToken: process.env.TURSO_AUTH_TOKEN!,
});

const memory = new Memory({
  storage: store,
  vector,
  options: {
    lastMessages: 20,
    semanticRecall: { topK: 5, messageRange: 2 },
    workingMemory: {
      enabled: true,
      template: `# Profil client
- Forme juridique:
- Secteur:
- Effectif:
- CA:
- Préoccupations RGPD:
- Contrats en cours:`,
    },
  },
});

// ------------------- Tools -------------------
const searchLegifrance = createTool({
  id: "searchLegifrance",
  description: "Recherche dans Legifrance (codes, lois, décrets, jurisprudence).",
  inputSchema: z.object({
    query: z.string().describe("Requête juridique en français"),
    code: z.enum(["civil", "commerce", "travail", "consommation", "all"]).default("all"),
    maxResults: z.number().min(1).max(20).default(5),
  }),
  outputSchema: z.object({
    results: z.array(z.object({
      reference: z.string(),
      title: z.string(),
      excerpt: z.string(),
      url: z.string().url(),
      lastModified: z.string(),
    })),
  }),
  execute: async ({ context }) => {
    const r = await fetch("https://api.piste.gouv.fr/dila/legifrance/lf-engine-app/consult/code", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.PISTE_TOKEN}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        recherche: { search: context.query, fond: context.code === "all" ? "ALL" : `CODE_${context.code.toUpperCase()}` },
        pageSize: context.maxResults,
      }),
    });
    const data = await r.json();
    return {
      results: data.results.map((d: any) => ({
        reference: d.id,
        title: d.title,
        excerpt: d.preview,
        url: `https://www.legifrance.gouv.fr${d.url}`,
        lastModified: d.modDate,
      })),
    };
  },
});

const searchJurisprudence = createTool({
  id: "searchJurisprudence",
  description: "Recherche sémantique dans la jurisprudence indexée (Cour de cassation + Conseil d'État).",
  inputSchema: z.object({
    question: z.string(),
    jurisdiction: z.enum(["cassation", "conseil-etat", "all"]).default("all"),
    topK: z.number().default(3),
  }),
  outputSchema: z.object({
    cases: z.array(z.object({
      ref: z.string(),
      date: z.string(),
      court: z.string(),
      summary: z.string(),
      relevance: z.number(),
    })),
  }),
  execute: async ({ context }) => {
    const { embedding } = await embed({
      model: openai.embedding("text-embedding-3-large"),
      value: context.question,
    });
    const results = await vector.query({
      indexName: "jurisprudence",
      queryVector: embedding,
      topK: context.topK,
      filter: context.jurisdiction !== "all"
        ? { jurisdiction: context.jurisdiction }
        : undefined,
    });
    return {
      cases: results.map((r) => ({
        ref: r.metadata?.ref as string,
        date: r.metadata?.date as string,
        court: r.metadata?.court as string,
        summary: r.metadata?.summary as string,
        relevance: r.score,
      })),
    };
  },
});

const draftClause = createTool({
  id: "draftClause",
  description:
    "Rédige une clause contractuelle FR. À utiliser uniquement APRÈS searchLegifrance pour ancrer la rédaction.",
  inputSchema: z.object({
    clauseType: z.enum([
      "non-concurrence",
      "confidentialite",
      "propriete-intellectuelle",
      "force-majeure",
      "resiliation",
      "garantie",
      "limitation-responsabilite",
      "donnees-personnelles",
    ]),
    contractType: z.enum(["prestation", "vente", "travail", "bail", "partenariat", "autre"]),
    contextSummary: z.string().describe("Résumé du contexte client (2-3 phrases)"),
    referencesLegifrance: z.array(z.string()).describe("IDs Legifrance des références à ancrer"),
  }),
  outputSchema: z.object({
    clauseText: z.string(),
    legalBasis: z.array(z.string()),
    warnings: z.array(z.string()),
    needsLawyerReview: z.boolean(),
  }),
  execute: async ({ context }) => {
    // Cette tool délègue à un LLM secondaire avec un prompt très contraint
    // (pour éviter que l'agent principal "improvise" du droit)
    // Implémentation simplifiée pour l'exemple
    return {
      clauseText: `[Clause ${context.clauseType} - rédigée selon ${context.referencesLegifrance.join(", ")}]`,
      legalBasis: context.referencesLegifrance,
      warnings: [
        "Cette clause est une proposition automatisée. Validation par un avocat obligatoire avant signature.",
      ],
      needsLawyerReview: true,
    };
  },
});

const checkRgpdCompatibility = createTool({
  id: "checkRgpdCompatibility",
  description: "Vérifie la compatibilité RGPD d'un traitement décrit.",
  inputSchema: z.object({
    treatmentDescription: z.string(),
    dataCategories: z.array(z.enum([
      "identite", "contact", "professionnelle", "financiere",
      "localisation", "sensible-sante", "sensible-opinion", "mineurs",
    ])),
    legalBasis: z.enum(["consentement", "contrat", "obligation-legale",
      "interet-vital", "mission-publique", "interet-legitime"]),
    transferOutsideEu: z.boolean(),
  }),
  outputSchema: z.object({
    compatible: z.boolean(),
    issues: z.array(z.string()),
    recommendations: z.array(z.string()),
    aipdRequired: z.boolean(),
  }),
  execute: async ({ context }) => {
    const issues: string[] = [];
    const recs: string[] = [];
    let aipdRequired = false;

    if (context.dataCategories.includes("sensible-sante") ||
        context.dataCategories.includes("sensible-opinion")) {
      aipdRequired = true;
      if (context.legalBasis !== "consentement" && context.legalBasis !== "obligation-legale") {
        issues.push("Données sensibles: base légale 'consentement' ou 'obligation légale' requise.");
      }
    }
    if (context.dataCategories.includes("mineurs")) {
      aipdRequired = true;
      recs.push("Mineurs <15 ans: consentement du titulaire de l'autorité parentale.");
    }
    if (context.transferOutsideEu) {
      recs.push("Transfert hors UE: clauses contractuelles types (CCT 2021) ou BCR obligatoires.");
    }
    return {
      compatible: issues.length === 0,
      issues,
      recommendations: recs,
      aipdRequired,
    };
  },
});

// ------------------- Agent principal -------------------
export const legalAgent = new Agent({
  name: "legal-assistant",
  instructions: `Tu es assistant juridique pour dirigeants de PME françaises.
RÈGLES STRICTES:
- Tu ne donnes JAMAIS de conseil juridique définitif sans citer Legifrance.
- Avant toute rédaction de clause, tu appelles searchLegifrance.
- Pour les questions RGPD, tu appelles checkRgpdCompatibility.
- Tu maintiens le profil client dans working memory (forme juridique, secteur, etc.).
- Tu rappelles systématiquement: "validation avocat recommandée pour décisions à enjeu".
- Tu réponds en français, ton professionnel mais accessible (vouvoiement).
- Si la question est hors droit FR (droit US, fiscal expert, etc.), tu dis "hors périmètre" et orientes.`,
  model: anthropic("claude-sonnet-4-6"),
  tools: {
    searchLegifrance,
    searchJurisprudence,
    draftClause,
    checkRgpdCompatibility,
  },
  memory,
});

export const mastra = new Mastra({
  agents: { legalAgent },
  storage: store,
});

typescript

// apps/web/app/api/chat/route.ts (Next.js 15 + Cloudflare Workers via OpenNext)
import { mastra } from "@org/legal-agent";

export const runtime = "edge";

export async function POST(req: Request) {
  const { messages, userId, threadId } = await req.json();
  const agent = mastra.getAgent("legalAgent");

  const stream = await agent.stream(messages, {
    resourceId: userId,        // mémoire par utilisateur
    threadId: threadId,        // historique conversationnel
    maxSteps: 8,
  });
  return stream.toDataStreamResponse();
}

toml

# wrangler.toml (Cloudflare Workers)
name = "legal-agent-prod"
main = "dist/_worker.js"
compatibility_date = "2026-05-01"
compatibility_flags = ["nodejs_compat"]

[vars]
NODE_ENV = "production"

# Secrets injectés via `wrangler secret put`:
# TURSO_DATABASE_URL, TURSO_AUTH_TOKEN, ANTHROPIC_API_KEY, OPENAI_API_KEY, PISTE_TOKEN

[observability]
enabled = true

[[durable_objects.bindings]]
name = "MEMORY_DO"
class_name = "MemoryDurableObject"

typescript

// packages/legal-agent/src/__tests__/agent.test.ts
import { describe, it, expect, vi } from "vitest";
import { legalAgent } from "../agent";

describe("legalAgent", () => {
  it("ne rédige jamais une clause sans avoir lookup Legifrance d'abord", async () => {
    const calls: string[] = [];
    const spied = vi.spyOn(legalAgent as any, "callTool")
      .mockImplementation(async (name: string, args: any) => {
        calls.push(name);
        if (name === "searchLegifrance") return { results: [{ reference: "L1234-1", title: "...", excerpt: "...", url: "...", lastModified: "" }] };
        return {};
      });
    await legalAgent.generate("Rédige une clause de non-concurrence pour mon CDI directeur commercial");
    const draftIdx = calls.indexOf("draftClause");
    const searchIdx = calls.indexOf("searchLegifrance");
    expect(searchIdx).toBeGreaterThanOrEqual(0);
    expect(searchIdx).toBeLessThan(draftIdx);
  });

  it("force needsLawyerReview=true sur les clauses de non-concurrence", async () => {
    // ... test que toutes les clauses sensibles sortent avec le flag
  });
});

Ce que cet exemple démontre :

Memory + working memory + semantic recall : l'agent retient le profil client (working memory templatée) ET fait du semantic recall sur les conversations passées.
RAG sur jurisprudence est un tool, pas une chaîne préprocessing aveugle. Le LLM décide quand interroger.
Guardrail métier : draftClause exige des referencesLegifrance non-vides, et marque needsLawyerReview: true. Le LLM ne peut pas court-circuiter.
Déploiement edge : Cloudflare Workers + Turso (libSQL) = latence <80ms en Europe pour 95% des requêtes. Coût ridicule (~5€/mois pour 50k requêtes).
Testable sans LLM : les tools sont des createTool purs, on les unit-teste isolément. L'orchestration agent se teste avec un mock LLM ou contre un dataset eval.

🔬 Sous le capot — la couche SDK Anthropic que tu dois maîtriser

Pydantic AI et Mastra sont des abstractions au-dessus du SDK Anthropic. En entretien senior comme en prod, tu dois savoir ce qu'ils génèrent en dessous — parce que c'est là que vivent le coût, la latence, les retries, et 90% des incidents. Un staff engineer ne traite pas le framework comme une boîte noire : il sait quand descendre d'un cran.

Mental model : le framework te donne le DX (types, DI, evals). Le SDK te donne le contrôle (caching, timeouts, retries, streaming, observabilité du usage). Tu utilises le framework par défaut, mais tu dois pouvoir reconstruire l'agent en SDK natif quand le besoin l'exige (perf extrême, garde-fous custom, modèle non supporté par le framework).

Les patterns prod non-négociables (Python, SDK Anthropic natif)

python

import asyncio
from anthropic import AsyncAnthropic
from anthropic import RateLimitError, APIStatusError, APITimeoutError

# AsyncAnthropic pour un serveur : ne JAMAIS bloquer l'event loop.
# max_retries gère 429/500/529 avec backoff exponentiel automatiquement.
client = AsyncAnthropic(max_retries=4, timeout=30.0)

SYSTEM = "Tu enrichis une entreprise française à partir d'un SIRET/SIREN."
TOOLS = [
    {
        "name": "lookup_sirene",
        "description": "Lookup INSEE Sirene par SIRET. À appeler quand l'utilisateur fournit un SIRET/SIREN.",
        "input_schema": {
            "type": "object",
            "properties": {"siret": {"type": "string"}},
            "required": ["siret"],
        },
    }
]

async def enrich(siret: str) -> dict:
    try:
        resp = await client.messages.create(
            model="claude-opus-4-8",
            max_tokens=4096,
            thinking={"type": "adaptive"},        # adaptive sur 4.7/4.8 — PAS budget_tokens (400)
            output_config={"effort": "medium"},   # low | medium | high | xhigh | max
            system=[
                # cache_control sur le préfixe STABLE (system + tools).
                # Render order: tools -> system -> messages. Le préfixe stable se cache,
                # la question volatile reste après le dernier breakpoint.
                {"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}},
            ],
            tools=TOOLS,
            messages=[{"role": "user", "content": f"Enrichis {siret}"}],
        )
        # Logguer le usage = la seule façon de connaître ton coût réel.
        u = resp.usage
        log_cost(
            input=u.input_tokens,
            output=u.output_tokens,
            cache_read=u.cache_read_input_tokens,      # ~0.1x du prix input
            cache_write=u.cache_creation_input_tokens, # ~1.25x (TTL 5 min)
        )
        return parse(resp)
    except RateLimitError as e:
        # 429 — le SDK a déjà retenté max_retries fois. Ici on dégrade (queue, fallback).
        raise Degraded("rate limited", retry_after=e.response.headers.get("retry-after"))
    except APITimeoutError:
        raise Degraded("timeout — réduis max_tokens ou passe en streaming")
    except APIStatusError as e:
        raise Degraded(f"api {e.status_code}: {e.type}")

Ce que ce code dit sur ta séniorité :

AsyncAnthropic sur un serveur, jamais le client sync — sinon tu bloques l'event loop FastAPI/NestJS à chaque appel. Pydantic AI fait ça pour toi avec agent.run ; tu dois savoir pourquoi.
Exceptions typées (RateLimitError / APIStatusError / APITimeoutError / OverloadedError), jamais du string-matching sur le message d'erreur. Chaque code HTTP a sa classe.
max_retries + timeout par appel : le SDK retente 429/500/529 en backoff exponentiel. Tu fixes le plafond, pas la boucle.
Prompt caching via cache_control sur le préfixe stable (system + tools). Vérifie usage.cache_read_input_tokens > 0 sur les appels répétés ; s'il reste à 0, un invalidateur silencieux casse le préfixe (timestamp dans le system prompt, JSON non trié, set de tools qui varie).
thinking={"type": "adaptive"} + output_config.effort sur Opus 4.7/4.8 — la forme budget_tokens est supprimée et renvoie un 400. Sonnet 4.6 supporte adaptive ; Haiku non.
Streaming dès que max_tokens est gros (>~16k) : sinon timeout HTTP du SDK. Sur 128k de sortie (Opus), le streaming est obligatoire.
asyncio.gather pour les tool calls parallèles — si l'agent émet 3 tool calls indépendants, tu les exécutes en parallèle, pas en série.

Quand descendre du framework vers le SDK natif

Situation	Reste sur le framework	Descends en SDK natif
CRUD agent métier, output typé	✅ Pydantic AI / Mastra
Prompt caching fin (multi-breakpoint, pre-warming)	DX limité	✅ `cache_control` placé à la main
Modèle/feature pas encore exposé par le framework	bloqué	✅ SDK suit la release same-day
Tool calls parallèles à très haut débit	dépend du framework	✅ `asyncio.gather` explicite
Garde-fou de sécurité custom (HITL, approval gates)	possible mais verbeux	✅ boucle manuelle
Structured outputs natifs	`output_type` / `outputSchema`	✅ `client.messages.parse()` + Pydantic/Zod, ou `output_config.format`

Règle du staff engineer : tu codes l'agent au framework pour la vélocité, mais tu gardes une trappe d'échappement documentée vers le SDK pour les 5% de cas où le framework te bride. Les deux partagent le même modèle mental (system + tools + messages), donc la migration d'un tool isolé du framework vers le SDK est locale, pas un rewrite.

Structured outputs natifs — `messages.parse()` plutôt que du JSON-prompting maison

Ce que output_type (Pydantic AI) génère en dessous en 2026, ce n'est pas un prompt « réponds en JSON » avec un json.loads défensif derrière. C'est la contrainte serveur (output_config.format avec un JSON Schema), exposée par le SDK via client.messages.parse() qui valide la réponse contre ton schéma et te rend un objet typé — la même garantie que output_type, mais à la main quand tu descends au SDK.

python

from anthropic import AsyncAnthropic
from pydantic import BaseModel

client = AsyncAnthropic(max_retries=4, timeout=30.0)

class CompanyInfo(BaseModel):
    siren: str
    name: str
    naf_code: str
    employee_band: str
    risk_flag: bool

resp = await client.messages.parse(
    model="claude-opus-4-8",
    max_tokens=4096,
    thinking={"type": "adaptive"},
    messages=[{"role": "user", "content": "Enrichis 12345678900012"}],
    output_format=CompanyInfo,          # ou output_config={"format": {...}} en API brute
)
company: CompanyInfo = resp.parsed     # objet typé, validé serveur-side — pas de json.loads

Trois pièges que le output_type du framework te cache (et que l'entretien creuse) :

Pas de prefill ni de citations. Les structured outputs sont incompatibles avec le prefill de l'assistant (de toute façon supprimé → 400 sur Opus 4.6+/Fable 5) et avec les citations (400). Si ton agent RAG cite ses sources, tu ne peux pas combiner output_config.format et citations dans le même appel — tu sépares en deux passes.
stop_reason: "refusal" court-circuite le schéma. Si un classifier de sécurité refuse, l'output ne valide pas ton schéma. Branche sur stop_reason AVANT de lire resp.parsed, sinon None-pointer en prod.
Première requête = compilation du schéma. Un schéma neuf paie un coût one-shot (mis en cache 24h). Un schéma généré dynamiquement par requête tue ce cache — fige le schéma comme tu figes le préfixe cacheable.

🎯 Patterns courants

Pattern A — Output type strict + fallback parsing

output_type=MyModel (Pydantic AI) ou output: { schema } (Mastra). Le framework retry automatiquement si le LLM produit une output invalide (max 2-3 retries). N'ajoute jamais de try/except json.loads.

Pattern B — Deps via dataclass / context

Toutes les dépendances I/O (http client, DB, vector store, KV) passent par Deps/context. Aucun singleton global. Test = swap Deps.

Pattern C — Working memory templatée

Plutôt que de "tout retenir", utilise un template structuré (# Profil client - Forme: ...). L'agent remplit les champs, et tu lis le profil en TypeScript dans l'UI.

Pattern D — Tools "narrow"

Préfère 8 tools très spécifiques (searchLegifrance, searchJurisprudence, draftClause, checkRgpd...) à 2 tools génériques (legalSearch, legalGenerate). Le LLM choisit mieux, le test est isolé.

Pattern E — Stream + structured output

Mastra streame du texte ET des partial JSON simultanément. UX premium : l'utilisateur voit "je cherche dans Legifrance..." en temps réel.

Pattern F — Eval-driven development

Avant de coder l'agent, écris 30 cas (question + output attendu). Mesure baseline avec Pydantic Evals / Mastra Evals. Optimise jusqu'à >X%. Sans dataset eval, tu ne sais pas si ton refactor améliore ou casse.

Pattern G — Logfire / OpenTelemetry one-line

Pydantic AI : logfire.configure(); logfire.instrument_pydantic_ai(). Toutes les traces (LLM call, tool call, retries, tokens, $) apparaissent dans le dashboard. Mastra : MASTRA_TELEMETRY_ENABLED=true + OTLP exporter.

Pattern H — Multi-step avec max_steps

maxSteps (Mastra) / result = await agent.run(..., usage_limits=UsageLimits(request_limit=10)) (Pydantic AI). Évite les boucles infinies de tool calls. Cible 5-10 max selon use case.

🔄 Versions & écosystème 2026

Composant	Version mai 2026	Notes
Pydantic AI	0.5.x	API stable, multi-LLM, evals natifs, Logfire intégré
Pydantic Evals	0.5.x	Dataset-based eval, LLM-judge, run en CI
Logfire	3.x	OTel-based, dashboard hosted, free tier généreux
Mastra	0.5.x	Workflows, agents, RAG, memory, integrations, MCP client/server
Mastra CLI	0.5.x	`mastra dev`, `mastra deploy`, scaffold projets
AI SDK (Vercel)	5.x	Base de Mastra côté provider abstraction
Turso (libSQL)	GA	Edge DB + vector index, gratuit jusqu'à 500 DB
Cloudflare Workers	2025-12 runtime	Node.js compat majoritaire, streaming SSE OK

Comparatif synthétique 2026 :

Critère	Pydantic AI	Mastra	LangChain	LangGraph
Langage	Python	TypeScript	Py + TS	Py + TS
Type-safety	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐
LOC pour use case typique	300-500	400-600	800-1500	600-1000
Eval natif	✅	✅	partiel	-
RAG natif	via tools	✅ (`@mastra/rag`)	✅	via composants
Memory	via deps	✅ (Memory class)	partiel	✅ (checkpointer)
Multi-agent	basique	workflows	✅	✅ (graphs)
Streaming	✅	✅	✅	✅
Déploiement edge	-	✅ (CF, Vercel)	difficile	difficile
Courbe d'apprentissage	basse	basse	élevée	élevée
Prod-readiness	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐

Migration LangChain → Pydantic AI (tips) :

Chains → 1 Agent avec tools. La plupart des "chains" intermédiaires deviennent des tools.
PydanticOutputParser → output_type=MyModel direct.
AgentExecutor → agent.run(...) (synchrone) ou agent.run_stream(...) (stream).
Memory ConversationBufferMemory → message history passée en argument, ou Logfire pour la persistance.
Tokens count : result.usage() au lieu de callbacks.

Migration LangGraph → Mastra workflows :

StateGraph → Workflow Mastra.
Edges conditionnelles → .then((res) => ...) + branches.
Checkpointer → storage Mastra.
Streaming intermédiaire → events Mastra workflow.

⚠️ Pitfalls

Output type trop libéral. output_type=str annule tout l'intérêt. Vise toujours un BaseModel/Zod object.
Trop de retries silencieux. Pydantic AI retry jusqu'à 3× par défaut sur output invalide. En prod, observe les retries — c'est un signal de mauvais prompt.
Memory non purgée. Working memory + 20 derniers messages = 5-10k tokens contexte sur chaque call. Audit les conversations longues. Compression périodique nécessaire.
Tools qui appellent un autre LLM "en cachette". Ça fait gonfler le coût opaquement. Trace via Logfire/OTel.
Cloudflare Workers limites. 30s CPU max (modes payants). Si un workflow Mastra dépasse, split en plusieurs invocations via queue (Cloudflare Queues, Inngest).
Deps oubliée dans tests. Sans mocks de Deps, tu hit la vraie API en CI → flaky tests + facture. Toujours mock.
Mastra MCP client vs server confusion. Mastra peut consommer des MCP servers (client) ET exposer ses tools en MCP (server). Ne pas mélanger les deux usages dans le même projet sans schéma clair.
Eval dataset périmé. Si le LLM model upgrade (Sonnet 4.6 → Opus 4.8), tes labels "human gold" peuvent devenir obsolètes. Re-curate trimestriellement. Et attention : sur Opus 4.7/4.8 le tokenizer change (~1×–1.35× vs Sonnet/anciens modèles) — re-baseline tes budgets de tokens et max_tokens via count_tokens, ne réutilise pas les chiffres mesurés sur l'ancien modèle.
Working memory template trop long. >2k tokens dans le template = chaque turn paie ces tokens. Garde le template <500 tokens, mets le verbose dans semantic recall.
Pas de fallback model. Si Anthropic rate-limit (rare mais arrive) ou refuse (stop_reason: "refusal" sur les classifiers Fable 5), tout casse si tu n'as pas de fallback. Côté SDK Anthropic natif, le pattern propre est le paramètre serveur fallbacks (betas: ["server-side-fallback-2026-06-01"] + fallbacks: [{ model: "claude-opus-4-8" }]) ; côté framework, déclare un modèle de secours explicite plutôt qu'un singleton mono-provider.

💰 Pricing / ROI client

Mission type : agent Pydantic AI production-grade (use case métier)

Item	Jours	TJM	Total
Discovery + dataset eval (50 cas)	3	1300€	3 900€
Implémentation agent + tools	7	1300€	9 100€
Memory + RAG + intégrations	4	1200€	4 800€
Eval + Logfire + monitoring	3	1300€	3 900€
Déploiement + handover + 2 sem support	3	1200€	3 600€
Total	20		25 300€

Mission type : agent Mastra TS dans monorepo Next.js + déploiement edge

Item	Jours	TJM	Total
Discovery + intégration monorepo	2	1300€	2 600€
Implémentation agent + tools + RAG	8	1200€	9 600€
UI streaming + working memory UX	3	1200€	3 600€
Déploiement Cloudflare/Vercel + Turso	2	1300€	2 600€
Eval + observability	2	1300€	2 600€
Handover	1	1200€	1 200€
Total	18		22 200€

Cas concret migration LangChain → Pydantic AI (compta SaaS) : 12j, 15,6k€. Économies tokens 18k€/an + productivité dev x3 + accuracy +9pts. Argument client : tu rentabilises en 11 mois rien que sur les tokens.

Range FR 2026 :

POC Pydantic AI / Mastra (1 use case) : 6-12k€
Production (eval + obs + deploy) : 20-35k€
Migration LangChain → Pydantic AI : 12-25k€ selon complexité
Migration LangGraph → Mastra : 15-30k€

Argument vente : "vous payez la migration une fois, vous économisez tokens + maintenance + recrutement (le code est compréhensible par n'importe quel dev Python/TS, pas seulement un specialist LangChain)". Cet argument fait mouche auprès des CTO qui galèrent à recruter de l'AI eng senior.

🧪 Testing / Eval

python

# Pydantic AI: eval dataset + LLM judge
from pydantic_evals import Dataset, Case
from pydantic_evals.evaluators import LLMJudge, IsInstance

dataset = Dataset(
    cases=[
        Case(name="non-concurrence-cdi", inputs="Rédige une clause de non-concurrence pour mon directeur commercial",
             expected_output_contains=["L1121-1", "contrepartie financière", "durée"]),
        Case(name="rgpd-newsletter", inputs="Je veux envoyer une newsletter à mes clients sans consentement explicite",
             expected_keywords=["consentement", "opt-in", "non conforme"]),
        # ... 30+ cases
    ],
    evaluators=[
        IsInstance(type_=LegalResponse),
        LLMJudge(rubric="La réponse cite Legifrance et précise 'avocat recommandé'. Score 0-10."),
    ],
)

# Run en CI
report = await dataset.evaluate(legal_agent)
assert report.average_score("llm-judge") > 8.0

typescript

// Mastra: tests unitaires tools + integration agent
import { describe, it, expect } from "vitest";
import { createTool } from "@mastra/core/tools";
import { searchLegifrance } from "../tools";

describe("searchLegifrance", () => {
  it("renvoie 5 résultats max par défaut", async () => {
    const result = await searchLegifrance.execute({ context: { query: "non-concurrence", code: "travail" } });
    expect(result.results.length).toBeLessThanOrEqual(5);
    expect(result.results[0]).toMatchObject({ reference: expect.any(String), url: expect.stringMatching(/^https/) });
  });
});

// Eval via @mastra/evals
import { evaluate } from "@mastra/evals";
const evalResult = await evaluate({
  agent: legalAgent,
  dataset: "./datasets/legal-50-cases.jsonl",
  metrics: ["answer-relevancy", "faithfulness", "toxicity"],
});
expect(evalResult.aggregate["answer-relevancy"]).toBeGreaterThan(0.85);

Métriques senior à tracker :

Output validity rate : % d'outputs qui valident output_type sans retry. Cible >95%.
Tool selection accuracy : sur 50 prompts, le bon tool est appelé. Cible >92%.
Faithfulness (RAG) : la réponse est fidèle aux sources retrouvées. Cible >90%.
Latence p50 / p95 : Logfire ou Mastra observability.
Coût $ par session : track via Logfire/OTel attributes.

🔁 Quand utiliser / éviter

Utilise Pydantic AI quand :

Tu écris en Python, équipe déjà type-hint friendly (FastAPI, etc.).
Tu as besoin d'outputs typés strictement (downstream code en a besoin).
Tu veux eval natif + observabilité Logfire one-liner.
Tu remplaces un projet LangChain verbeux et fragile.

Évite Pydantic AI quand :

Le projet est en TypeScript (utilise Mastra).
Tu as besoin d'un graphe d'agents très complexe avec cycles et state distribué (utilise LangGraph).
Tu fais du prototypage 1-fichier 100 lignes (la simple anthropic/openai SDK suffit).

Utilise Mastra quand :

Tu es en TypeScript dans un monorepo Next.js/Turborepo.
Tu veux déploiement edge (Cloudflare, Vercel) natif.
Tu as besoin de workflows + agents + RAG + memory d'un seul tenant.
Tu construis un produit user-facing (chat, assistant) avec streaming UX.

Évite Mastra quand :

Tu es en Python (utilise Pydantic AI).
Tu construis un agent SWE code-gen complexe (MetaGPT ou Claude Code SDK).
Tu as besoin d'inférence locale offline (Mastra suppose API providers).

Hybride pragmatique : back office en Python + Pydantic AI (eval, jobs cron, intégrations DB), front user-facing en TypeScript + Mastra (streaming UX, edge). Communication via API REST/MCP. C'est le pattern qu'on voit gagner en 2026 chez les startups FR sérieuses.

🏋️ Exercices

Progression : "implémenter" → "rendre prod-grade" → "casser puis réparer" → "défendre le chiffre". Fais-les dans l'ordre, chacun construit sur le précédent.

Exercice 1 — Agent typé testable sans LLM (échauffement)

Objectif : écrire un agent Pydantic AI avec output_type=BaseModel, 1 tool injecté via Deps, et un test unitaire qui valide le tool sans appeler le LLM. Indice/Solution : déclare deps_type=Deps (dataclass avec un faux http), un tool @agent.tool async def lookup(...), et un test qui appelle directement lookup.function(ctx, ...) avec un Deps mocké (httpx mock ou stub). Le piège : si tu testes via agent.run, tu paies le LLM et le test est flaky. La valeur du type-safe, c'est précisément de tester le tool en isolation.

Exercice 2 — Eval-driven avant de coder

Objectif : avant d'optimiser un prompt, monter un dataset de 30 cas avec pydantic_evals, mesurer une baseline, puis prouver qu'un changement de prompt améliore (et ne régresse pas) le score. Indice/Solution : Dataset(cases=[Case(...)], evaluators=[IsInstance(...), LLMJudge(rubric=...)]). Mesure report.average_score("llm-judge") AVANT et APRÈS. Le piège senior : un LLM-judge non déterministe — fixe-le sur un modèle stable (claude-haiku-4-5 pour le coût, ou claude-sonnet-4-6 pour la qualité du jugement) et run 3× pour estimer la variance du juge lui-même, sinon tu confonds bruit du juge et amélioration réelle.

Exercice 3 — Descendre au SDK pour le prompt caching (prod-grade)

Objectif : prendre l'agent de l'exercice 1, le réécrire en SDK Anthropic natif (AsyncAnthropic), et faire passer usage.cache_read_input_tokens de 0 à >0 sur le 2ᵉ appel identique. Indice/Solution : mets cache_control: {"type": "ephemeral"} sur le dernier bloc system (le préfixe stable = tools + system). Vérifie usage.cache_creation_input_tokens au 1ᵉʳ appel puis cache_read_input_tokens au 2ᵉ. Si le read reste à 0 : tu as un invalidateur silencieux — un datetime.now() dans le system, un json.dumps sans sort_keys=True, ou un set de tools réordonné. Diffe les bytes des deux prompts rendus pour le trouver. Bonus : le préfixe doit dépasser le minimum cacheable (4096 tokens sur Opus 4.8) sinon ça ne cache pas, silencieusement.

Exercice 4 — Casser l'agent puis le durcir (failure modes)

Objectif : provoquer délibérément 3 pannes — (a) rate-limit 429, (b) tool qui timeout, (c) output qui ne valide pas le schéma — et faire en sorte que l'agent dégrade proprement sur chacune. Indice/Solution : (a) baisse max_retries à 0 et spamme l'API → attrape RateLimitError, lis le header retry-after, mets en queue. (b) un tool qui await asyncio.sleep(60) avec un timeout=5 par appel → APITimeoutError. (c) un prompt qui pousse le LLM hors schéma → observe les retries du framework (max 2-3) puis le fail. Le livrable senior : un tableau "panne → signal observable → action de dégradation", pas juste un try/except qui avale tout. Un retry silencieux qui boucle, c'est un incident qui se cache.

Exercice 5 — Tool calls parallèles + budget de coût (scale)

Objectif : un agent qui doit interroger 5 sources indépendantes par requête. Le faire en parallèle (asyncio.gather) plutôt qu'en série, et tenir un budget de coût par session loggé via usage. Indice/Solution : si le LLM émet 5 tool calls indépendants, exécute-les avec asyncio.gather(*[tool(x) for x in calls]) — la latence p50 passe de ~5× à ~1× le tool le plus lent. Pour le budget : agrège input_tokens + output_tokens + cache_* de chaque appel, convertis au prix (claude-opus-4-8 = 5$/25$ par M tok input/output), et coupe la session au-delà d'un plafond. Piège : les tool calls parallèles partagent un préfixe cacheable — envoie 1 appel, attends le 1ᵉʳ token streamé (qui écrit le cache), PUIS lance les 4 autres, sinon les 5 paient le write plein.

Exercice 6 — Défendre le chiffre (architecte)

Objectif : un CTO te dit "migrer notre agent LangChain vers Pydantic AI coûte 15k€, justifie le ROI au mois près". Construis le calcul défendable. Indice/Solution : décompose en (1) économie tokens/an mesurée sur le dataset eval (moins de prompts de stitching), (2) productivité dev (debug + features) — chiffrable en jours/an évités, (3) réduction du risque de recrutement (code lisible par n'importe quel dev Python vs specialist LangChain). Le piège : ne sors PAS un seul gros chiffre — sors un break-even ("rentabilisé en 11 mois rien que sur les tokens") avec les hypothèses explicites (volume d'appels/mois, % de réduction tokens, TJM). Un chiffre sans hypothèses traçables ne survit pas à une question de suivi.

Exercice 7 — Le fallback model qui ne casse pas la prod (résilience)

Objectif : ton agent tourne sur claude-opus-4-8. Construis un chemin de secours qui survit à (a) un 529 OverloadedError répété et (b) un stop_reason: "refusal" sur une tâche métier légitime (faux positif d'un classifier) — sans laisser tomber la requête. Indice/Solution : deux mécanismes distincts, ne les confonds pas. Pour le 529 overload : le SDK retente déjà en backoff (max_retries), au-delà tu bascules sur un modèle moins chargé (claude-sonnet-4-6) ou tu mets en queue — c'est de la résilience d'infra. Pour le refusal : c'est un HTTP 200 avec stop_reason: "refusal" (pré-output : content vide, non facturé ; mid-stream : le partiel est facturé, jette-le). Le pattern propre côté SDK Anthropic natif est le paramètre serveur fallbacks (betas=["server-side-fallback-2026-06-01"] + fallbacks=[{"model": "claude-opus-4-8"}]) qui re-sert la requête sur le modèle de secours dans le même appel, avec repricing crédit automatique. Côté framework, tu déclares un modèle de secours explicite plutôt qu'un singleton mono-provider. Le piège senior : ne traite pas un 529 comme un refusal (le fallback serveur ne se déclenche que sur les déclines de policy, pas sur les rate-limits/overloads), et branche toujours sur stop_reason avant de lire resp.content[0] — sinon index-error sur chaque refus.

🎤 En entretien

Q : Pydantic AI / Mastra vs LangChain vs LangGraph — quand tu choisis quoi ? R : Type-safe framework (Pydantic AI en Python, Mastra en TS) sur 80% des cas — output typé, DI, evals natifs, code lisible. LangGraph reste roi sur les vrais graphes d'agents avec cycles et state distribué. LangChain : surtout en legacy à migrer. Le critère décisif est l'écosystème (TS edge → Mastra ; Python eval-continu → Pydantic AI), pas le hype.

Q : Le framework abstrait l'appel LLM — qu'est-ce qu'il génère en dessous, et quand descends-tu au SDK ? R : En dessous c'est client.messages.create avec system + tools + messages. Je descends au SDK pour le prompt caching multi-breakpoint, les tool calls parallèles haut débit, un modèle/feature pas encore exposé, ou un garde-fou de sécurité custom. Je garde le framework par défaut mais avec une trappe d'échappement documentée — même modèle mental, donc migration locale d'un tool, pas un rewrite.

Q : Comment tu maîtrises le coût et la latence d'un agent en prod ? R : Prompt caching sur le préfixe stable (system+tools) vérifié via usage.cache_read_input_tokens, effort calibré par route (low pour les tâches simples, high/xhigh pour l'agentique), tool calls parallèles via asyncio.gather, streaming dès que max_tokens est gros, et logging systématique de usage pour connaître le coût réel par session — pas une estimation. Observabilité one-liner via Logfire/OTel.

Q : budget_tokens pour le thinking — comment tu le configures sur les modèles 2026 ? R : Piège classique. Sur Opus 4.7/4.8 la forme thinking={"type":"enabled","budget_tokens":N} est supprimée et renvoie un 400. On utilise thinking={"type":"adaptive"} + output_config.effort (low → max). Le modèle décide quand et combien penser ; effort contrôle la profondeur. Sonnet 4.6 supporte adaptive ; Haiku non.

Q : Ton agent commence à retenter en boucle / appeler un LLM caché dans un tool — comment tu le détectes ? R : Les deux gonflent le coût silencieusement. Je trace via Logfire/OTel : retries d'output invalide (signal de mauvais prompt, pas une feature), et tools qui appellent un LLM secondaire (coût opaque). Un usage_limits/maxSteps borne les boucles infinies, et chaque retry doit être observable — un retry silencieux est un incident qui se cache.

Q : output_type=MyModel — qu'est-ce que ça génère côté API, et quelles incompatibilités tu dois connaître ? R : Pas un prompt « réponds en JSON » + json.loads défensif, mais la contrainte serveur output_config.format (JSON Schema), exposée par client.messages.parse() qui valide et rend un objet typé. Trois pièges : incompatible avec le prefill assistant (de toute façon 400 sur Opus 4.6+/Fable 5) et avec les citations (400) ; un stop_reason: "refusal" court-circuite le schéma (branche sur stop_reason avant de lire parsed) ; et un schéma figé profite du cache de compilation 24h, un schéma généré par requête le tue.

🔗 Liens

Pydantic AI — https://ai.pydantic.dev
Pydantic Evals — https://ai.pydantic.dev/evals/
Logfire — https://pydantic.dev/logfire
Mastra — https://mastra.ai
Mastra docs — https://mastra.ai/docs
Mastra examples — https://github.com/mastra-ai/mastra/tree/main/examples
AI SDK Vercel — https://sdk.vercel.ai
Turso (libSQL) — https://turso.tech
Cloudflare Workers — https://developers.cloudflare.com/workers/
OpenNext (Next.js → CF Workers) — https://opennext.js.org
Legifrance API (PISTE) — https://piste.gouv.fr
LangChain → Pydantic AI migration — https://ai.pydantic.dev/migration/langchain/
LangGraph → Mastra workflows — https://mastra.ai/docs/workflows/migration-from-langgraph
"Death of LangChain?" debate 2025 — https://hamel.dev/blog/posts/langchain-rip/

Pydantic AI & Mastra — Type-Safe Agents Python & TypeScript ​

🧠 Mental model ​

🛠️ Code minimal ​

Pydantic AI — agent typé en 25 lignes ​

Mastra — agent TS avec memory + RAG ​

🎬 Cas d'usage concrets ​

Scénario 1 — Refactor d'un agent LangChain Python vers Pydantic AI (compta SaaS) ​

Scénario 2 — Agent commercial e-commerce en Mastra TS dans monorepo Next.js ​

Scénario 3 — Agent compta Pydantic AI avec eval continu (cabinet EC) ​

🛠️ Exemple end-to-end — Mastra TS "Assistant juridique" avec memory + RAG + tools + déploiement Cloudflare Workers ​

🔬 Sous le capot — la couche SDK Anthropic que tu dois maîtriser ​

Les patterns prod non-négociables (Python, SDK Anthropic natif) ​

Quand descendre du framework vers le SDK natif ​

Structured outputs natifs — messages.parse() plutôt que du JSON-prompting maison ​

🎯 Patterns courants ​

Pattern A — Output type strict + fallback parsing ​

Pattern B — Deps via dataclass / context ​

Pattern C — Working memory templatée ​

Pattern D — Tools "narrow" ​

Pattern E — Stream + structured output ​

Pattern F — Eval-driven development ​

Pattern G — Logfire / OpenTelemetry one-line ​

Pattern H — Multi-step avec max_steps ​

🔄 Versions & écosystème 2026 ​

⚠️ Pitfalls ​

💰 Pricing / ROI client ​

🧪 Testing / Eval ​

🔁 Quand utiliser / éviter ​

🏋️ Exercices ​

Exercice 1 — Agent typé testable sans LLM (échauffement) ​

Exercice 2 — Eval-driven avant de coder ​

Exercice 3 — Descendre au SDK pour le prompt caching (prod-grade) ​

Exercice 4 — Casser l'agent puis le durcir (failure modes) ​

Exercice 5 — Tool calls parallèles + budget de coût (scale) ​

Exercice 6 — Défendre le chiffre (architecte) ​

Exercice 7 — Le fallback model qui ne casse pas la prod (résilience) ​

🎤 En entretien ​

🔗 Liens ​

Pydantic AI & Mastra — Type-Safe Agents Python & TypeScript

🧠 Mental model

🛠️ Code minimal

Pydantic AI — agent typé en 25 lignes

Mastra — agent TS avec memory + RAG

🎬 Cas d'usage concrets

Scénario 1 — Refactor d'un agent LangChain Python vers Pydantic AI (compta SaaS)

Scénario 2 — Agent commercial e-commerce en Mastra TS dans monorepo Next.js

Scénario 3 — Agent compta Pydantic AI avec eval continu (cabinet EC)

🛠️ Exemple end-to-end — Mastra TS "Assistant juridique" avec memory + RAG + tools + déploiement Cloudflare Workers

🔬 Sous le capot — la couche SDK Anthropic que tu dois maîtriser

Les patterns prod non-négociables (Python, SDK Anthropic natif)

Quand descendre du framework vers le SDK natif

Structured outputs natifs — `messages.parse()` plutôt que du JSON-prompting maison

🎯 Patterns courants

Pattern A — Output type strict + fallback parsing

Pattern B — Deps via dataclass / context

Pattern C — Working memory templatée

Pattern D — Tools "narrow"

Pattern E — Stream + structured output

Pattern F — Eval-driven development

Pattern G — Logfire / OpenTelemetry one-line

Pattern H — Multi-step avec max_steps

🔄 Versions & écosystème 2026

⚠️ Pitfalls

💰 Pricing / ROI client

🧪 Testing / Eval

🔁 Quand utiliser / éviter

🏋️ Exercices

Exercice 1 — Agent typé testable sans LLM (échauffement)

Exercice 2 — Eval-driven avant de coder

Exercice 3 — Descendre au SDK pour le prompt caching (prod-grade)

Exercice 4 — Casser l'agent puis le durcir (failure modes)

Exercice 5 — Tool calls parallèles + budget de coût (scale)

Exercice 6 — Défendre le chiffre (architecte)

Exercice 7 — Le fallback model qui ne casse pas la prod (résilience)

🎤 En entretien

🔗 Liens