Stop calling functions: my first LangGraph graph, where state does the calling

Three parts in, JobCopilot had layers that worked but couldn't talk to each other. Part 1 gave me a structured-output layer — ParseOk/ParseError as data. Part 2 gave me retrieval — build_context() pulling the top-8 resume chunks for a job description. The pipeline was one line on paper; in code it was nested calls, and the awkward questions had no home: who retries when the model drifts into prose? Who stops the retries? Where does the score live between "parsed" and "shown to the user"?

I did what a React dev does in a new language: I nested functions. The retry loop closed over a closure that closed over a callback, and I couldn't tell you who owned what. Rewriting the flow as a graph — my first with LangGraph — made the nesting disappear. Not because a framework absorbed it, but because the mental model changed: functions stop calling each other; state flows through a graph, and nodes return patches of that state.

It survived everything I threw at it — after I fixed one wire that made every job come back "error", no matter how perfectly the model answered.

1. The state is a pydantic model; nodes return patches

If the graph is the wiring, state is everything a run knows. I made it one pydantic model — deliberately the same discipline as part 1:

# state.py — the whole run in one model
from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, Field

THRESHOLD = 60   # score at or above this = "worth applying"
MAX_CALLS = 3    # 1 first try + 2 repairs — the budget from part 1


class AgentState(BaseModel):
    """One run of the copilot, end to end."""

    jd_text: str = ""
    context: str = ""              # resume chunks, from part 2
    last_raw: str = ""             # what the model last said, verbatim
    error: str | None = None       # latest parse failure, if any
    attempt: int = 0               # how many LLM calls so far

    match_score: int | None = None
    missing_skills: list[str] = Field(default_factory=list)
    verdict: Literal["match", "no_match", "error"] | None = None

The rule that makes a graph a graph: a node never calls another node. It receives state, does one job, and returns a dict with only what changed — a patch. LangGraph merges the patch and hands state to the next node. retrieve fills context; judge fills the score fields; nobody knows about anybody else.

Fields without a custom reducer are last-writer-wins — I proved it to myself before trusting it:

def note_a(state: AgentState) -> dict:
    return {"missing_skills": ["Python"]}

def note_b(state: AgentState) -> dict:
    return {"missing_skills": ["Terraform"]}

# a -> b in that order: missing_skills == ["Terraform"] — b overwrote a

For my state that's correct by design: every field is overwrite-on-purpose, so zero reducers. If a list should accumulate across nodes, opt in with Annotated:

from typing import Annotated


def merge_lists(left: list[str] | None, right: list[str] | None) -> list[str]:
    if left is None:
        return right or []
    if right is None:
        return left
    return list(left) + [x for x in right if x not in left]


class AccumState(BaseModel):
    missing_skills: Annotated[list[str], merge_lists] = Field(default_factory=list)

# two nodes appending "Python" then "Terraform" -> ["Python", "Terraform"]

And why pydantic for state instead of the TypedDict every tutorial starts with? A patch is data from somewhere else, and at merge time nobody type-checks a plain dict:

# TypedDict state, merged by dict spread — type-checkers scream, Python shrugs:
>>> {**{"attempt": 0, "verdict": None}, "verdict": 21, "attempt": "two"}
{"attempt": "two", "verdict": 21}          # garbage, happily stored

# pydantic model:
>>> PS(attempt="two", verdict=21)
pydantic.ValidationError: attempt — Input should be a valid integer...

Same lesson as part 1: shape is a contract, and contracts live in pydantic, not in comments.

2. The graph is a handful of declared edges

Here is the entire graph.py — wiring, not logic:

# graph.py
from langgraph.graph import END, START, StateGraph

from nodes import Chat, fail_node, judge_node, retrieve_node, route_after_parse
from state import AgentState


def build_graph(chat: Chat):
    g = StateGraph(AgentState)

    g.add_node("retrieve", retrieve_node)
    g.add_node("judge", lambda s: judge_node(s, chat))  # closure: node gets the LLM client
    g.add_node("fail", fail_node)

    g.add_edge(START, "retrieve")
    g.add_edge("retrieve", "judge")

    g.add_conditional_edges(
        "judge",
        route_after_parse,          # returns "ok" | "retry" | "give_up"
        {"ok": END, "retry": "judge", "give_up": "fail"},
    )
    g.add_edge("fail", END)

    return g.compile()

Coming from React, this shape clicked instantly: it's a component tree. Components don't import each other to call each other — data flows down a declared structure, and the structure decides what happens next.

The lambda is deliberate: the graph is built once with a concrete LLM client, and no node ever imports a provider. That's what makes the graph testable — hand it a fake chat and the whole flow runs with zero network. Which is exactly how I caught the bug.

3. All routing logic in one boring function — and the wire that said "error" to everything

add_conditional_edges takes a router: a pure function from state to a string, plus a mapping from strings to destinations. Mine is twelve lines and holds the entire loop policy of the agent:

# nodes.py
from state import MAX_CALLS


def route_after_parse(state: AgentState) -> str:
    """All loop logic in one boring function."""
    if state.error is None:
        return "ok"
    if state.attempt < MAX_CALLS:
        return "retry"
    return "give_up"

Compare with my first wiring attempt, written because the router concept felt foreign:

g.add_edge("retrieve", "judge")
g.add_edge("judge", "fail")     # "temporarily — I'll wire the real branches next"

Then I ran it — with a fake chat returning a perfect JSON verdict. Result: verdict = "error". Every single job. The parse was flawless, the score valid, and the graph marched it into fail_node anyway: I had declared exactly one outgoing edge from judge, and a graph never "falls through". Nothing raised, nothing warned — the state machine politely followed the only road I built:

>>> # plain edge judge -> fail, no conditional edges, perfect JSON from the model
>>> app.invoke(AgentState(jd_text="junior"))
verdict = "error"     # a valid parse, executed through the wrong edge

The fix was the conditional edges above, plus two habits that came with it:

• Route values are data, not flow control. If the router returns a key that's not in the mapping, LangGraph doesn't guess — it throws KeyError at runtime. I know because my typo did exactly that. An unmapped route crashes loudly; a wrongly-wired edge fails silently. I'll take the crash.
• Loops are declared, not nested. "retry": "judge" makes a cycle in the graph. The loop exists as an edge, its budget lives in MAX_CALLS, and its stop condition is one pure function I can unit-test without LLM, network, or shame.

The shift, internalized: in nested functions, control flow was who calls whom — invisible at a glance. In the graph, it's data in the edges — visible in one screenful.

4. Closing the repair loop with state, not recursion

Last piece: judge_node, the only node that talks to the model. It reads state to decide which prompt to send — first pass or repair — and returns a patch either way:

# nodes.py
from schemas import ParseError, ParseOk, parse_match
from state import AgentState, THRESHOLD

MATCH_PROMPT = """You are a skeptical technical recruiter.
Compare the job description to the candidate's resume excerpt.
Reply with ONLY a JSON object:
{{"match_score": <int 0-100>, "missing_skills": [<string, ...>]}}

JOB DESCRIPTION:
{jd}

RESUME EXCERPT:
{context}
"""

FIX_PROMPT = """Your previous reply could not be parsed.

PREVIOUS REPLY:
{previous}

PARSER ERROR:
{error}

Reply with ONLY a valid JSON object:
{{"match_score": <int 0-100>, "missing_skills": [<string, ...>]}}
"""


def judge_node(state: AgentState, chat: Chat) -> dict:
    """One LLM call, one parse, a patch of state. Loops are the graph's job now."""
    if state.error is None:
        prompt = MATCH_PROMPT.format(jd=state.jd_text, context=state.context)
    else:
        prompt = FIX_PROMPT.format(previous=state.last_raw, error=state.error)

    raw = chat(prompt)
    result = parse_match(raw)          # the part-1 contract, unchanged

    patch: dict = {"last_raw": raw, "attempt": state.attempt + 1}
    if isinstance(result, ParseOk):
        score = result.data.match_score
        patch.update(
            match_score=score,
            missing_skills=result.data.missing_skills,
            verdict="match" if score >= THRESHOLD else "no_match",
            error=None,
        )
    else:
        patch.update(error=result.message)
    return patch


def fail_node(state: AgentState) -> dict:
    """The model never produced valid JSON. Say so — as state, not a stack trace."""
    return {"verdict": "error"}

Because build_graph takes chat as a parameter, the end-to-end test is a fake answer card — call one drifts into prose, call two complies:

def flaky_llm():
    calls = {"n": 0}

    def chat(prompt: str) -> str:
        calls["n"] += 1
        if calls["n"] == 1:
            return "Sure! The match is quite good, I'd say around 70 out of 100."
        return '{"match_score": 70, "missing_skills": ["Terraform"]}'

    return chat


app = build_graph(flaky_llm())
final = app.invoke(AgentState(jd_text="Senior Frontend Engineer (React, TS)"))

verdict: match | score: 70 | attempts: 2 | error: None

The prose landed in last_raw, parse_match filled error, the router said "retry", and the second call got FIX_PROMPT with the model's own prose quoted back at it — the part-1 trick, now driven by edges instead of a while-loop. When the model never complies:

verdict: error | error: no JSON object found in model output | attempts: 3

No exception slicing through three stack frames. fail_node writes verdict: "error" and the graph ends — for an agent run overnight over forty postings, a run that ends with a recorded failure beats ten runs that crashed and took the batch with them. Part 1's "errors as data", promoted from one function to the shape of the whole system.

Wrapping up

The first graph, in one sentence: state is a pydantic contract, nodes return patches, a router returns strings, loops live in edges.

Three things this React-dev-learning-Python took from it:

• Patches, not calls. The merge is last-writer-wins by default; accumulators are opt-in via Annotated. Every field in my state is overwrite-on-purpose — zero reducers, but verified behavior before trusting it.
• Routing is a pure function. Twelve lines, no LLM, unit-testable — and an unmapped return value crashes with a KeyError. That's the framework refusing to guess, and I've come to love it.
• Giving up is a node, not a throw. fail writes verdict: "error" and the graph ends cleanly. The batch survives; the log tells the truth.

And the bug that taught me most: a graph executes exactly the edges you declared — my single judge → fail wire turned perfect parses into errors with zero pushback from the framework. A state machine has no common sense. That's the feature.

Next Tuesday, part 4: evals. The judge says 72 — but is 72 right? Fixtures of job descriptions with expected score ranges, a negative case that must score LOW (a Python-backend posting should not match a frontend resume), and the first time JobCopilot tells me I'm wrong.

Follow if you want to watch the rest of the copilot get built — including the parts that break. Build log + daily code: t.me/novamind_hub. Part 1, in case you missed it: Stop prompting for valid JSON: an LLM output layer that does not break.

Do you build agent graphs with a framework or hand-roll the loop? And what's the first thing your graph did that you didn't expect?

Comments

[Lycore Development]

The comparison with nested closures resonates; making retry count live in state rather than in a call stack is what makes the budget testable. Did you find it easy to unit-test individual nodes as plain functions that take state and return a patch? I'd also be curious whether you checkpoint state between nodes, since that would let you resume a failed run instead of re-paying for the model calls.