Career Copilot

01.Objectif

Google ADK v2 vient de sortir son API workflow basée sur un graphe. Je voulais la pousser de bout en bout sur un cas d'usage réel, dans l'esprit de LangGraph.

Pourquoi un graphe

• Topologie explicite: nodes, edges, branches parallèles, routage conditionnel et points de synchro sont first-class.
• Unités testables: chaque node peut être testé unitairement, swappé, ou tracé indépendamment.
• State typé: des contrats à chaque étape au lieu de parsing de prompts en free-form.
• Coût prédictible: on sait à l'avance quelles étapes tournent, dans quel ordre, et quel modèle sert chacune.

02.Cas d'usage

Upload un CV et une description de poste. Un pipeline d'agents fait l'analyse complète et produit une sortie adaptée au côté de la boucle de recrutement qui pose la question, recruteur ou candidat.

Ce que ça fait

• Préprocessing partagé: le CV et l'offre sont parsés en parallèle en objets typés (compétences, années d'expérience, réalisations, requis vs préférés, séniorité). Tout le branching arrive après le parsing.
• Flux recruteur: score le fit du candidat (fit, borderline, ou no fit), puis soit drafte un message LinkedIn citant une vraie réalisation du CV, soit explique les gaps et suggère des postes adjacents.
• Flux candidat: recherche live sur l'entreprise via Tavily MCP (levée, culture, news récentes), suggestions de positionnement CV adaptées à l'offre, et un pack de prep d'entretien avec questions probables et talking points.

Pourquoi on a besoin d'un graphe

• Travail partagé, sorties divergentes: les deux flux réutilisent le même CV et la même offre parsés, puis produisent des artefacts complètement différents. Un agent single big-prompt aurait besoin d'un prompt conditionnel tentaculaire ; un graphe rend la séparation naturelle.
• Le parallélisme divise la latence par deux: parsing du CV et de l'offre en concurrence, puis côté candidat la recherche et le positionnement CV en parallèle. Une chaîne linéaire forcerait ces étapes à tourner les unes après les autres.
• Le branching doit être déterministe: le routage sur le fit verdict (fit ou borderline → outreach, no_fit → gap explainer) ne peut pas dépendre d'un LLM qui re-parse sa propre sortie. Un FunctionNode router décide en pure Python sur un verdict typé.
• Pourquoi ce domaine colle: assez petit pour shipper en side project, assez réel pour exposer les failure modes, assez riche pour avoir besoin de toutes les affordances ci-dessus.

03.Architecture agentique

Le graphe encode une séparation délibérée entre control flow déterministe et génération LLM. Routage et synchronisation restent en pure Python ; seule la génération franchit une frontière LLM, et chaque sortie est contrainte par un schema Pydantic avant de quitter le node.

Comment le graphe est câblé

• Le control flow reste hors des LLMs
  • Deux FunctionNode routers (mode_router, verdict_router) décident des branches en pure Python sur du state déjà typé.
  • Les JoinNodes attendent que les branches parallèles se complètent avant de fire en aval.
  • Aucun LLM ne décide quelle branche fire ensuite, ce qui supprime une classe entière de failure modes du chemin critique.
• Huit agents spécialisés, un schema chacun
  • Chaque agent déclare son propre output_schema : ParsedCV, ParsedJD, FitVerdict, OutreachDraft, GapReport, CompanyIntelligence, CVOptimizationBundle, InterviewPrepBundle.
• Deux topologies de branches sur les mêmes inputs parsés
  • Branche recruteur : chaîne linéaire avec un split conditionnel sur le fit verdict (fit / borderline → outreach, no_fit → gap report).
  • Branche candidat : recherche et positionnement CV fan-out en parallèle, synchronisent via un JoinNode, puis alimentent l'interview prep.
  • Le parallélisme côté candidat est safe parce que les deux passes n'ont aucune dépendance de données entre elles.
• Une échappatoire pour le tool use
  • ADK ne permet actuellement pas de combiner output_schema avec des tools sur gpt-5.4-mini.
  • Le Research Agent utilise le toolset Tavily MCP et émet son payload CompanyIntelligence en JSON string.
  • Le handler le valide avec model_validate_json à la frontière API, préservant l'invariant typed-state au bord.

root_agent = Workflow(
    name="career_copilot",
    edges=[
        (
            "START",
            (cv_parser_agent, jd_parser_agent),  # parallel
            parse_join,
            mode_router,
            {
                "RECRUITER": fit_analyzer_agent,
                "CANDIDATE": (research_agent, cv_optimizer_agent),
            },
        ),
        (
            fit_analyzer_agent,
            verdict_router,
            {"OUTREACH": outreach_writer_agent, "GAP": gap_explainer_agent},
        ),
        (
            (research_agent, cv_optimizer_agent),
            candidate_join,
            interview_prep_agent,
        ),
    ],
)

Les agents

Huit agents LLM, deux routers FunctionNode, deux JoinNodes. Chaque agent tourne sur OpenAI gpt-5.4-mini via LiteLlm avec un reasoning effort low par défaut. L'Outreach Writer monte à medium pour générer des messages plus précis. La sortie de chaque agent est un schema Pydantic appliqué par output_schema d'ADK, sauf le Research Agent qui utilise les outils Tavily et valide son JSON à la frontière API.

04.Conception système

Le graphe agentique ci-dessus ne ship que s'il est exposé via une API stateless et atteignable depuis une UI. Deux couches propres font ça sans laisser fuir la complexité dans le graphe.

Backend

• FastAPI expose le graphe: deux endpoints POST stateless (/v1/analyze, /v1/extract-pdf). Schema OpenAPI auto-généré et intégration Pydantic native.
• Async de bout en bout: FastAPI plus le runner async d'ADK, donc les branches parallèles du graphe tournent vraiment en parallèle.
• Container-ready: un seul Dockerfile, lockfile uv. Déploie identique sur Cloud Run, Fly.io ou HuggingFace Spaces.

# Discriminated union: clients get one of three typed response
# shapes, picked by the requested mode.
AnalyzeResponse = Union[
    RecruiterFitResponse,    # recruiter mode, fit / borderline
    RecruiterNoFitResponse,  # recruiter mode, no_fit
    CandidateResponse,       # candidate mode
]


@router.post("/v1/analyze", response_model=AnalyzeResponse)
async def analyze(request: AnalyzeRequest) -> AnalyzeResponse:
    """Run the agent graph against a CV + JD pair, return the typed result."""

    # ADK Workflow graph runs end-to-end. Parallel branches actually
    # execute in parallel thanks to the async runner.
    state = await run_agent(root_agent, initial_state)

    # Pydantic schemas at every node boundary guarantee state is typed
    # where it matters; we just pick the right response builder.
    return (
        _build_recruiter_response(state)
        if request.mode == "recruiter"
        else _build_candidate_response(state)
    )

Frontend

• Client Next.js 15 thin: collecte les inputs, POST vers /v1/analyze, render la réponse typée. Toute la business logic reste dans le graphe backend.
• TanStack Query pour l'appel analyze: useMutation drive le flow ; les toasts Sonner surface les erreurs structurées 502 / 422 du backend.
• Forms validés par Zod: un seul analyzeInputSchema pour les deux modes. Coller du texte ou uploader un PDF (qui passe par /v1/extract-pdf d'abord).

05.Observabilité

Chaque requête, tracée de bout en bout avec Langfuse.

Un système LLM qu'on ne peut pas tracer est un système qu'on ne peut pas réparer. Chaque appel /v1/analyze ouvre une observation parente dans Langfuse. Les appels sub-agents et tool calls s'imbriquent dessous comme des spans enfants, avec inputs, outputs, latence et token counts capturés. Mode, modèle, version et tailles d'inputs sont propagés en attributs de trace pour un filtrage rapide dans l'UI.

@router.post("/analyze", response_model=AnalyzeResponse)
async def analyze(request: AnalyzeRequest) -> AnalyzeResponse:
    langfuse = get_client()
    with langfuse.start_as_current_observation(
        name=f"analyze.{request.mode}",
        as_type="agent",
        input={"mode": request.mode, "cv_text": ..., "jd_text": ...},
    ) as span, propagate_attributes(
        trace_name=f"career-copilot.analyze.{request.mode}",
        tags=["analyze", request.mode],
        metadata={
            "mode": request.mode,
            "model": PRIMARY_MODEL,
            "version": VERSION,
            "cv_chars": str(len(request.cv_text)),
            "jd_chars": str(len(request.jd_text)),
        },
    ):
        state = await run_agent(root_agent, initial_state)
        response = _build_response(state)
        span.update(output=response.model_dump())
        return response