Getting started

How this chapter was written

Recipes for the part of GraphARC you touch first: a typed state model, nodes that declare what they write, edges that decide what runs next, and the machinery that stops a run before it costs you something. Every snippet was executed against this repo's .venv, and every block labelled Output is pasted from that run — not predicted, not tidied. Exactly one snippet is an exception: the real-model call near the end, labelled as unrun because it would spend money. tests/test_cookbook_basics.py reproduces every recipe and asserts these exact strings, so the page cannot rot quietly. Verified against grapharc 0.1.0, Python 3.14.6, langgraph 1.2.9, langchain-core 1.5.1, pydantic 2.13.4. Each snippet is a complete file; nothing carries over between recipes.

How do I install it?

Not on PyPI — git clone is the install path. Clone, cd, and uv sync --group dev. Check that it took with uv run grapharc --version, which prints grapharc 0.1.0. Everything below uses only the base install — no API key, no network, no optional extras.

git clone https://github.com/CodeGraphContext/GraphARC
cd GraphARC
uv sync --group dev

$ uv run grapharc --version
grapharc 0.1.0

How do I build and run my first graph?

Four moving parts: a state schema, a function, edges wiring it between START and END, and compile().invoke(). A node takes the state model (not a dict) and returns a dict of updates; it never assigns to state, and the returned dict is the only way anything leaves a node. On imports: GraphARC, GraphARCState, Budget, BudgetExceeded, BudgetMeter and WritePermissionError are on the top-level grapharc package; START and END are not — they come from grapharc.runtime.graph, along with StateTypeError, GraphCycleError, GraphRoutingError, MissingRunContextError, AsyncNodeError and RunContext.

from grapharc import GraphARC, GraphARCState
from grapharc.runtime.graph import END, START


class State(GraphARCState):
    question: str
    answer: str = ""


def answer(state: State) -> dict:
    return {"answer": f"42 (you asked: {state.question})"}


g = GraphARC(State, name="hello")
g.add_node("answer", answer, writes={"answer"})
g.add_edge(START, "answer")
g.add_edge("answer", END)

print(g.compile().invoke({"question": "meaning of life"}))

{'question': 'meaning of life', 'answer': '42 (you asked: meaning of life)'}

Why is a field missing from the result dict?

invoke() returns a plain dict of channel values, and a field that was never in the input and never written by a node is not one — its schema default is not materialised for you. Inside the graph your nodes always see a fully constructed State, defaults and all, because the model is rebuilt on the way into every node. The gap is only at the exit: LangGraph hands back the channel dict, not a model. State(**result) is the one-line fix, and it is worth doing at the boundary of your program anyway: it is also the moment your state model's own validators finally run.

result = g.compile().invoke({"question": "?"})
print(result)
print(State(**result))

{'question': '?', 'answer': '42'}
question='?' answer='42' untouched='default never written'

How do I say what a node is allowed to write?

writes= is an allowlist, not documentation. A write outside it raises WritePermissionError and the update never reaches state. A node that writes nothing still needs the parameter: writes=set(); the check is on the keys of the returned dict, so returning None is the idiomatic "I wrote nothing" and is always legal. Note the asymmetry in when the two failures land: a typo naming a field the schema does not have is caught at add_node, because the schema is known then; a node writing a real field it did not declare can only be caught when it actually returns.

def write_draft(state: State) -> dict:
    return {"draft": "hello", "published": "hello"}   # published is not declared


g = GraphARC(State, name="perms")
g.add_node("write_draft", write_draft, writes={"draft"})
g.add_edge(START, "write_draft")
g.add_edge("write_draft", END)

try:
    g.compile().invoke({})
except WritePermissionError as exc:
    print(exc)

try:
    GraphARC(State, name="perms").add_node("x", write_draft, writes={"publised"})
except WritePermissionError as exc:
    print(exc)

node 'write_draft' wrote undeclared fields ['published']; declared writes: ['draft']
node 'x' declares writes to unknown state fields: ['publised']

How do I stop a node from mutating state behind my back?

You do not have to: every node gets a deep copy, so in-place mutation of a nested model is invisible to everyone else. Pydantic passes nested models by reference, so without the copy a node could rewrite state.report.title and have it stick — a write no writes= declaration approved and no trace recorded. The copy makes the returned dict the only channel out. The mutation is not an error, just local; if you want the change to land, return it.

def sneak(state: State) -> dict:
    state.report.title = "rewritten in place"   # not a declared write
    return {"seen": state.report.title}


result = g.compile().invoke({"report": Report(title="original")})
print("node saw: ", result["seen"])
print("state has:", result["report"].title)

node saw:  rewritten in place
state has: original

How do I get a bad value rejected before it reaches the next node?

You already have it. Every value a node returns is validated against the declared type of its field at the node boundary, and the validated value is what gets written, raising StateTypeError otherwise. Constraints carried in the annotation (Annotated[int, Field(ge=0)]) bite too. The error names the node, the field, the declared type, the type that arrived and the value, because when this fires at 3am the question is always "which node did it". Writing "7" into an int field is accepted — Pydantic's default coercion — but what lands is the integer 7: a schema that says int means the result holds an int.

try:
    one_node(lambda s: {"count": "seven"}, writes={"count"}).invoke({})
except StateTypeError as exc:
    print(exc)

out = one_node(lambda s: {"count": "7"}, writes={"count"}).invoke({})
print(repr(out["count"]))

node 'n' wrote 'count' with a value the state schema rejects: expected int, got str ('seven'); Input should be a valid integer, unable to parse string as an integer
node 'n' wrote 'retries' with a value the state schema rejects: expected int, got int (-1); Input should be greater than or equal to 0
7

Does my state model's own @field_validator run when a node writes?

No. This is the sharpest edge in the kernel. Write-time validation is built per field from that field's annotation, and the state model itself is never constructed — only the one field's value is — so a @field_validator declared on the state model has nothing to run against. Direct construction does run it; a write from the last node before END does not; add a node downstream and the model is rebuilt on the way in, so it does. The rule to hold in your head: writes= is GraphARC's; the types are Pydantic's, and only the annotation half of Pydantic. So prefer Annotated[...] constraints (Field(ge=0), Field(pattern=...), Field(min_length=1), Literal[...], StringConstraints(to_lower=True)) for anything you want enforced per write, or push the invariant into a nested model — the annotation is that model, so validating the write constructs it and its validators run.

class Article(BaseModel):
    slug: str

    @field_validator("slug")
    @classmethod
    def must_be_lower(cls, v: str) -> str:
        if v != v.lower():
            raise ValueError("slug must be lowercase")
        return v


class State(GraphARCState):
    article: Article | None = None


def publish(state: State) -> dict:
    return {"article": {"slug": "NOT-LOWER"}}

ValidationError on direct construction
result: {'slug': 'NOT-LOWER'}
ValidationError when the next node receives the state

How do I stop a graph that will not stop itself?

Give it a Budget. max_iterations is charged once per node execution and checked before every node, so a router that never says stop still stops. Crossing a budget raises BudgetExceeded; it does not return partial state — a truncated answer that looks like a real one is worse than an exception — which means the caller, not the graph, decides what to do about it. For a partial result, either catch BudgetExceeded and read the checkpoint, or build the stop condition into the graph so the run ends through END with a recorded reason. The budget is the ceiling, not the plan.

g = GraphARC(State, name="runaway", budget=Budget(max_iterations=5))
g.add_node("spin", spin, writes={"spins"})
g.add_edge(START, "spin")
g.add_conditional_edge("spin", again, {"go": "spin", "stop": END})

compiled = g.compile()
try:
    compiled.invoke({})
except BudgetExceeded as exc:
    print("stopped:", exc.reason)

print("iterations charged:", compiled.last_run.meter.iterations)

stopped: max_iterations reached (5/5)
iterations charged: 5

How do I cap tokens and wall-clock time?

Same Budget; neither max_tokens nor max_seconds needs your nodes to cooperate. GraphARC installs a usage callback for the duration of every node, so any chat model invoked on that thread charges this run's meter, and the token ceiling is enforced at on_llm_end rather than waiting for the node to return — the LangChain "Error in MeterCallbackHandler.on_llm_end callback" line is that mechanism, not a second failure. max_seconds is delivered as an interrupt into the running node — SIGALRM on the main thread, PyThreadState_SetAsyncExc otherwise — which is why a node parked in time.sleep(5.0) dies at 0.25s. Budget also carries max_concurrency. Budgets are per-invoke, not per-thread: resuming a thread starts a fresh meter, and you can override with invoke(..., budget=Budget(...)).

try:
    one_node(talk, Budget(max_tokens=5)).invoke({})
except BudgetExceeded as exc:
    print(exc.reason)

t0 = time.monotonic()
try:
    one_node(nap, Budget(max_seconds=0.25)).invoke({})
except BudgetExceeded as exc:
    print(exc.reason.split(" (")[0])
print(f"gave up after under a second: {time.monotonic() - t0 < 1.0}")

Error in MeterCallbackHandler.on_llm_end callback: BudgetExceeded('max_tokens reached (10/5)')
max_tokens reached (10/5)
max_seconds reached while node 'n' was running
gave up after under a second: True

How do I forbid cycles until I actually need one?

dag=True. Cycles are something you earn, not something you start with. Static edges can only be judged together, so a cycle built from them waits for compile() and GraphCycleError names the cycle it found (a -> b -> a) rather than saying "graph is cyclic". Conditional and fan-out edges are refused at add_* time, because in dag mode there is no version of them that would be legal later.

g = GraphARC(State, name="pipeline", dag=True)
g.add_node("a", bump, writes={"n"})
g.add_node("b", bump, writes={"n"})
g.add_edge(START, "a")
g.add_edge("a", "b")
g.add_edge("b", "a")        # accepted here...
g.add_edge("b", END)

try:
    g.compile()             # ...and refused here
except GraphCycleError as exc:
    print(exc)

graph 'pipeline' is dag=True but has a cycle: a -> b -> a
graph 'pipeline' is dag=True: conditional edges are not allowed

How do I write a loop that ends for a reason instead of running out of road?

Route with code, and record why you stopped in state. ProgressGuard gives you three independent brakes — a target, a no-progress window, and a round cap — and assess() returns the first one that trips as a StopReason. In the example the target of 10 findings was never met, but the loop stopped anyway at round 4 with no_progress on the record, and the 50-iteration budget was never touched: the budget is the thing that catches your bug, not the thing that ends your run. Routers are ordinary Python functions over typed state — no model output is ever consulted to pick an edge, so a model writing ROUTE TO: all_verified into a state field is writing a string, not choosing a branch.

GUARD = ProgressGuard(target=10, max_rounds=8, max_empty_rounds=2)


def decide(state: State) -> dict:
    stop = GUARD.assess(
        round=state.round,
        total_findings=state.findings,
        empty_rounds=state.empty_rounds,
    )
    return {"termination_reason": stop.value if stop else None}


def route(state: State) -> str:
    return "stop" if state.termination_reason else "again"


g.add_conditional_edge("decide", route, {"again": "search", "stop": END})

{'round': 4, 'findings': 3, 'empty_rounds': 2, 'termination_reason': 'no_progress'}
stopped because: no_progress
iterations used: 8 of 50

How do I see what actually happened?

Hand the graph a TraceRecorder. Every node execution writes a start line and then either an end or an error line, as JSONL. start carries identity only (run, thread, attempt, graph, node, step, timestamp) and is written before the node body, so it exists even when the node never returns; end adds state_delta (exactly the validated update applied), duration_ms and tokens; error adds duration_ms and error (repr() of the exception, so the type is preserved) and has no state_delta, because a node that raised wrote nothing. start and end share a step number — the pair is the node execution — so step numbers do not order the file; read events in file order (the recorder appends under a lock). The grapharc trace, grapharc metrics and grapharc viz commands read this exact file, which is why the dashboard and the audit trail cannot disagree.

trace = TraceRecorder(tmp / "trace.jsonl")

g = GraphARC(State, name="counter", trace=trace)
...
VOLATILE = {"ts", "run_id", "thread_id", "duration_ms"}
for event in trace.read_events():
    print({k: v for k, v in event.model_dump(exclude_none=True).items() if k not in VOLATILE})

{'attempt': 1, 'graph': 'counter', 'node': 'load', 'phase': 'start', 'step': 1}
{'attempt': 1, 'graph': 'counter', 'node': 'load', 'phase': 'end', 'step': 1, 'state_delta': {'items': ['a', 'b', 'c']}, 'tokens': 0}
{'attempt': 1, 'graph': 'counter', 'node': 'count', 'phase': 'start', 'step': 2}
{'attempt': 1, 'graph': 'counter', 'node': 'count', 'phase': 'error', 'step': 2, 'error': "ValueError('the counter is not implemented yet')"}

How do I see what a run spent, from inside a node?

Give your node a second parameter. Any node function with two or more parameters receives the RunContext as the second one, and the rule is detected from the function signature, so you never register which form you wrote. ctx.meter.tokens is populated without the node charging anything by hand — the usage callback did it. RunContext also carries run_id, thread_id, attempt and graph, which is what you want when a node writes to something outside the graph and needs to stamp it with the run that produced it.

def talk(state: State, ctx: RunContext) -> dict:      # two parameters => you get ctx
    model = ScriptedChatModel(responses=["a scripted reply"])
    reply = str(model.invoke([HumanMessage(content="hi")]).content)
    return {"reply": reply, "spent_so_far": ctx.meter.tokens}

{'reply': 'a scripted reply', 'spent_so_far': 5}
iterations: 1
tokens:     5

How do I resume a run that died halfway?

Compile with a checkpointer, then call invoke(None, thread_id=...) to resume that thread from its last checkpoint — input=None means "resume". get_state("t1").next tells you what will run, and in the example exactly one fetch happened across both attempts. Crash-safe resume is LangGraph's, not GraphARC's: the checkpointer passed to compile() goes straight through. What GraphARC adds is trace continuity — after a resume, step numbers continue from the thread's history and attempt increments, so (thread_id, step) stays unique across attempts and one logical thread is stitchable across restarts.

compiled = (
    GraphARC(State, name="fetch_save")
    .add_node("fetch", fetch, writes={"data"})
    .add_node("save", save, writes={"saved"})
    .add_edge(START, "fetch")
    .add_edge("fetch", "save")
    .add_edge("save", END)
    .compile(checkpointer=SqliteSaver(conn))
)

print("next to run:", compiled.get_state("t1").next)
result = compiled.invoke(None, thread_id="t1")     # input=None means "resume"

attempt 1  step 1  fetch  start
attempt 1  step 1  fetch  end
attempt 1  step 2  save   start
attempt 1  step 2  save   error
attempt 2  step 3  save   start
attempt 2  step 3  save   end

Does that survive a real process kill?

Partly. LangGraph's default checkpoint durability is "async" — the checkpoint for a completed step is written in the background while the next step starts — and CompiledGraphARC.invoke() does not expose the durability parameter to change it. After a SIGKILL the run resumed and produced exactly one result, but fetch ran twice: no checkpoint reached disk saying "fetch is done, start at save", so the resume replayed from the beginning. What to do: make nodes idempotent, or make the expensive step's effect externally deduplicated (content-addressed path, upsert on a key, check before you fetch) — the right answer regardless of durability mode, since a node can also be re-run by a retry or a deliberate resume from an earlier checkpoint.

    def save(state: State) -> dict:
        if os.environ.get("KILL_ME"):
            os.kill(os.getpid(), signal.SIGKILL)
        return {"saved": True}

child killed by signal: 9
a checkpoint that resumes at 'save': False
resumed: True
times fetch ran: 2

Why can't I just call .inner.invoke()?

Because a run with no budget and no trace is not a GraphARC run, and failing silently into one is the bug this prevents — raw LangGraph entry points raise MissingRunContextError. .inner is an inspection escape hatch, not an execution one: compiled.inner.get_graph().draw_mermaid() works fine, as does anything else that reads; only the entry points that would run nodes fail closed. GraphARC already wraps get_state, get_state_history and update_state on the compiled object, so you rarely need .inner at all — and the wrapped update_state type-checks your values and, when you pass as_node=, applies that node's write allowlist.

try:
    compiled.inner.invoke({})       # raw LangGraph entry point
except MissingRunContextError as exc:
    print(exc)

node 'bump' executed without a GraphARC run context; drive the graph via CompiledGraphARC.invoke()/stream()/ainvoke()/astream() — raw LangGraph entry points would silently bypass budgets and traces

How do I fan out across workers without one of them sinking the run?

add_fanout_edge dispatches parallel Sends; run_guarded turns a worker's crash or hang into data. Use both — the first alone gives you parallelism with a shared fate. The pieces: input_schema=Shard types the payload a worker receives (a fan-out worker is called with the Send payload, not the graph state, which is why worker never reads state); a reducer on the target field, Annotated[list[WorkerResult], operator.add], is what lets three parallel workers all write results without clobbering each other; run_guarded runs the job on a daemon thread joined with a timeout, so a raised exception becomes WorkerResult(ok=False, error=repr(exc)) and a hang becomes ok=False, error="timeout after 0.2s"; Budget(max_concurrency=2) caps how many workers run at once. Two of three workers died — one crashing, one hanging — and the run still produced an answer with both failures attributed by cause.

def worker(shard: Shard) -> dict:
    def job() -> list[dict]:
        if shard.fail:
            raise RuntimeError("shard parser blew up")
        if shard.hang:
            time.sleep(shard.hang)
        return [{"word": w} for w in shard.words]

    return {"results": [run_guarded(job, worker=shard.name, timeout_seconds=0.2)]}


g = GraphARC(State, name="fanout", budget=Budget(max_concurrency=2))
g.add_node("worker", worker, writes={"results"}, input_schema=Shard)
g.add_fanout_edge("prepare", dispatch)

counted:  2
failed:   w1: RuntimeError('shard parser blew up')
failed:   w2: timeout after 0.2s

run died: shard parser blew up
the fan-out dispatcher on node 'prepare' routed to Send(node='wroker', ...), but 'wroker' is not a node of graph 'unguarded'; LangGraph drops an unknown Send target and the run continues as if the routing had not happened. Valid Send targets: 'prepare', 'worker' — END is not one, because a Send has to name a node that runs

Can my nodes be async?

Yes — and if a graph has any async def node you must use ainvoke/astream, which it will tell you via AsyncNodeError. The rejection happens before any node runs, so a graph mixing sync and async nodes cannot half-execute; plain LangGraph would run every sync node first and then fail on the first coroutine. Sync nodes stay legal under ainvoke() — LangGraph runs each on a worker thread.

print(asyncio.run(compiled.ainvoke({})))

try:
    compiled.invoke({})
except AsyncNodeError as exc:
    print(exc)

{'reply': 'from an async node'}
graph 'async' has async nodes ['fetch']; use ainvoke(), because invoke() has no event loop to await them on

How do I swap the scripted model for a real one?

Everything above used grapharc.testing.ScriptedChatModel, a real BaseChatModel that replays a fixed list of strings and reports plausible usage metadata — so budgets, traces and the usage callback are all exercised for free. Swapping in a real backend is a one-line change at the point of construction via grapharc.gateway.get_model. What you can check without spending anything is how a spec resolves, with describe, which never constructs a model and never touches a credential; a mistyped backend is rejected rather than folded into a model name, so opnerouter/... fails there instead of failing much later as a nonsense Claude CLI call. Once a real model is in a node, nothing else on the page changes: the usage callback charges the run's meter whether the model is scripted or real, including calls made deep inside library code your node merely called — the whole reason max_tokens is worth setting.

from grapharc.gateway import get_model

# Claude Code CLI: uses your Claude subscription, no API key. Text completion only —
# bind_tools and with_structured_output raise NotImplementedError on this backend.
model = get_model("claude-cli/claude-sonnet-5")

# OpenRouter: needs OPENROUTER_API_KEY and the `openrouter` extra
# (uv sync --extra openrouter). Tool calling, structured output, streaming, async.
model = get_model("openrouter/anthropic/claude-haiku-4.5")

# OpenAI directly: needs OPENAI_API_KEY and the `openai` extra.
model = get_model("openai/gpt-4o-mini")

# Ollama: a model on this machine. No key, no bill — the one real backend you
# can run without an account. Needs the `ollama` extra and a pulled model.
model = get_model("ollama/llama3.1")

{'spec': 'claude-cli/claude-sonnet-5', 'backend': 'claude-cli', 'model': 'claude-sonnet-5'}
{'spec': 'openrouter/anthropic/claude-haiku-4.5', 'backend': 'openrouter', 'model': 'anthropic/claude-haiku-4.5'}

What this page did not cover

Deliberate omissions, so you know they exist rather than discovering them: stream() / astream() / astream_events() — the same discipline, incremental output, and .inner.stream() fails closed exactly like .inner.invoke(); returning a langgraph Command from a node for dynamic goto routing, whose update goes through the same allowlist and type checks and whose destination is checked against the graph's nodes; update_state for human-in-the-loop edits between steps; and the model gateway's retry policies, spend meters and cost ceilings, verification (verify_claim), memory, and the tool harness — other pages.

Source file: /home/shashank/Desktop/GraphARC/docs/cookbook/01-basics.md (1512 lines, 20 sections). Companion test asserting these outputs: /home/shashank/Desktop/GraphARC/tests/test_cookbook_basics.py.