Career Copilot

01.Objective

Google ADK v2 shipped its graph workflow API recently. I wanted to push it end-to-end on a real use case, in the spirit of LangGraph.

Why a graph

• Explicit topology: nodes, edges, parallel branches, conditional routing, and sync points are first-class.
• Testable units: each node can be unit-tested, swapped, or traced on its own.
• Typed state: contracts at every step instead of free-form prompt parsing.
• Predictable cost: you know upfront which steps run, in what order, and what model serves each one.

02.Use case

Upload a CV and a job description. An agent pipeline runs the full analysis and produces an output tailored to whichever side of the hiring loop is asking, recruiter or candidate.

What it does

• Shared preprocessing: the CV and the job description are parsed in parallel into typed objects (skills, years of experience, achievements, required vs preferred skills, seniority). All branching happens after parsing.
• Recruiter flow: scores how well the candidate fits the role (fit, borderline, or no fit), then either drafts a LinkedIn outreach quoting a real achievement from the CV, or explains the gaps and suggests adjacent roles.
• Candidate flow: live company research via Tavily MCP (funding, culture, recent news), CV positioning suggestions tailored to the job ad, and an interview prep bundle with probable questions and talking points.

Why we need a graph

• Shared work, divergent outputs: both flows reuse the same parsed CV and JD, then produce completely different artefacts. A single big-prompt agent would need a sprawling conditional prompt; a graph makes the split natural.
• Parallelism halves latency: parsing the CV and the JD concurrently, then on the candidate side running research and CV positioning in parallel. A linear chain would force these steps to run one after the other.
• Branching has to be deterministic: routing on the fit verdict (fit or borderline → outreach, no_fit → gap explainer) cannot depend on an LLM re-parsing its own output. A FunctionNode router decides in pure Python on a typed verdict.
• Why this domain fits: small enough to ship as a side project, real enough to expose failure modes, rich enough to need every one of the affordances above.

03.Agentic architecture

The graph encodes a deliberate separation between deterministic control flow and LLM-driven generation. Routing and synchronization stay in pure Python; only generation crosses an LLM boundary, and every output is constrained by a Pydantic schema before it leaves the node.

How the graph is wired

• Control flow stays out of the LLMs
  • Two FunctionNode routers (mode_router, verdict_router) decide branches in pure Python on already-typed state.
  • JoinNodes wait for parallel branches to complete before firing downstream.
  • No LLM ever decides which branch fires next, which removes a whole class of failure modes from the critical path.
• Eight specialised agents, one schema each
  • Each agent declares its own output_schema: ParsedCV, ParsedJD, FitVerdict, OutreachDraft, GapReport, CompanyIntelligence, CVOptimizationBundle, InterviewPrepBundle.
• Two branch topologies on the same parsed inputs
  • Recruiter branch: linear chain with one conditional split on the fit verdict (fit / borderline → outreach, no_fit → gap report).
  • Candidate branch: research and CV positioning fan out in parallel, synchronize through a JoinNode, then feed interview prep.
  • The candidate-side parallelism is safe because the two passes have no data dependency on each other.
• One escape hatch for tool use
  • ADK currently disallows combining output_schema with tools on gpt-5.4-mini.
  • The Research Agent uses the Tavily MCP toolset and emits its CompanyIntelligence payload as a JSON string.
  • The handler validates it with model_validate_json at the API boundary, preserving the typed-state invariant at the edge.

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,
        ),
    ],
)

The agents

Eight LLM agents, two FunctionNode routers, two JoinNodes. Every agent runs OpenAI gpt-5.4-mini via LiteLlm with low reasoning effort by default. The Outreach Writer escalates to medium reasoning for sharper, less generic copy. Each agent's output is a Pydantic schema enforced by ADK's output_schema, except the Research Agent which uses Tavily tools and validates its JSON output at the API boundary.

04.System design

The agentic graph above only ships if it is exposed through a stateless API and reachable from a UI. Two clean layers do that without leaking complexity into the graph.

Backend

• FastAPI exposes the graph: two stateless POST endpoints (/v1/analyze, /v1/extract-pdf). Auto-generated OpenAPI schema and native Pydantic integration.
• Async end-to-end: FastAPI plus the ADK async runner so parallel branches in the graph actually run in parallel.
• Container-ready: single Dockerfile, uv lockfile. Deploys identically on Cloud Run, Fly.io, or 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

• Thin Next.js 15 client: collect inputs, POST to /v1/analyze, render the typed response. All business logic stays in the backend graph.
• TanStack Query for the analyze call: useMutation drives the flow; Sonner toasts surface the structured 502 / 422 errors from the backend.
• Zod-validated forms: a single analyzeInputSchema for both modes. Paste text or upload a PDF (routed through /v1/extract-pdf first).

05.Observability

Every request, traced end-to-end with Langfuse.

An LLM system you cannot trace is a system you cannot fix. Every /v1/analyze call opens a parent agent observation in Langfuse. Sub-agent and tool calls nest under it as child spans, with inputs, outputs, latency and token counts captured. Mode, model, version and input sizes are propagated as trace attributes for fast filtering in the Langfuse 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