vs-graph

JSON-driven LangGraph wrapper for building agentic workflows — define graph topology in JSON, implement nodes and edges in Python, and get a fully wired, observable, checkpointed execution graph.

---

Overview

vs-graph wraps LangGraph and eliminates the boilerplate of building agentic graphs: wiring nodes, routing edges, setting up checkpointers, and propagating trace events. You declare the graph structure in a JSON file, implement node logic by extending VsBaseNode, and wire routing logic by extending VsBaseConditionalEdge or VsDynamicFanOutEdge. The library handles everything else.

The @node and @edge decorators self-register implementations at import time. The @graph decorator on a graph class specifies which packages to auto-import so no manual import list is required.

---

The Problem It Solves

Without vs-graph

# Manual LangGraph wiring — every graph requires this boilerplate
from langgraph.graph import StateGraph, START, END

sg = StateGraph(MyState)
sg.add_node("classifier", classifier_node.invoke)
sg.add_node("executor", executor_node.invoke)
sg.add_node("responder", responder_node.invoke)
sg.add_edge(START, "classifier")
sg.add_conditional_edges("classifier", routing_fn, {"execute": "executor", "default": "responder"})
sg.add_edge("executor", "responder")
sg.add_edge("responder", END)
checkpointer = MemorySaver()
graph = sg.compile(checkpointer=checkpointer)

Every graph has this same boilerplate. Adding a node requires updating both code and wiring.

With vs-graph

# graph topology is in JSON
# Python only contains business logic

@graph(nodes=["myapp.graph.nodes"], edges=["myapp.graph.edges"])
class MyGraph(VsBaseGraph[MyState]):
    def __init__(self):
        with open("my_graph.json") as f:
            graph_json = json.load(f)
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()

graph = MyGraph()
result = await graph.invoke({"trace_id": "", "error": None, "user_input": "hello"})

---

How It All Fits Together

@graph(nodes=[...], edges=[...])
class MyGraph(VsBaseGraph[MyState])
    └── __init__: load JSON, call build()

build()
    ├── _auto_import_packages()     # walks node/edge packages, triggers @node/@edge registration
    ├── _instantiate_nodes()        # VsNodeRegistry.create() for each JSON node definition
    ├── _instantiate_edges()        # VsEdgeRegistry.create() for conditional/dynamic_fan_out
    ├── _validate_graph()           # checks all node ids, edge targets, and exit nodes
    ├── _load_checkpointer()        # reads config.ini [graph] section
    └── _build_langgraph()          # compiles StateGraph with checkpointer

invoke(state)
    ├── auto-generates trace_id if not set
    └── StateGraph.ainvoke(state)
            └── per node: VsBaseNode.invoke()
                    ├── trace: "node started"
                    ├── execute(state)          ← your business logic
                    └── trace: "node completed" / "node failed"
            └── per conditional / dynamic_fan_out edge: async VsBaseEdge.route()
                    ├── _evaluate(state) / get_sends(state)   ← your routing logic
                    └── trace: "routing to next_node" / "fanning out to N branch(es)"
            (direct and fan_out edges are plain LangGraph edges — no wrapper, no trace)

---

Graph JSON

The JSON file declares the graph topology. Python classes are referenced by the name field in each node/edge definition — matching the name passed to @node(name="...") or @edge(name="...").

{
  "version": "1.0",
  "name": "my_graph",
  "description": "Routes a user message to a tool run or a plain chat reply.",
  "entry_node": "classifier",
  "exit_nodes": ["responder"],
  "nodes": [
    { "id": "classifier", "name": "classifier_node", "config": {},
      "description": "Decides whether the message needs a tool or just an answer." },
    { "id": "executor",   "name": "executor_node",   "config": {},
      "description": "Runs the tool the classifier picked." },
    { "id": "responder",  "name": "responder_node",  "config": {},
      "description": "Turns whatever happened into a reply the user can read." }
  ],
  "edges": [
    {
      "type": "conditional",
      "source": "classifier",
      "edge_class": "classifier_routing_edge",
      "description": "Tool-worthy intents go to the executor, everything else answers directly.",
      "route": {
        "execute": "executor",
        "default": "responder"
      }
    },
    { "type": "direct", "source": "executor", "target": "responder" }
  ]
}

JSON Fields

Field	Required	Description
name	Yes	Graph name — used in logs and error messages
version	No	Version string (default "1.0")
description	No	What this graph is for. Documentation only — read it and you know the workflow without opening Python.
entry_node	Yes	id of the node that receives the initial state
exit_nodes	Yes	List of node ids that connect to END
nodes	Yes	Array of node definitions
nodes[].id	Yes	Unique node id within this graph — used in edge source/target fields
nodes[].name	Yes	Registry name — must match the @node(name="...") decorator value
nodes[].config	No	Arbitrary dict passed to the node constructor as self.config
nodes[].description	No	Why this node exists. Available as self.description and attached to every trace event the node publishes.
edges	No	Array of edge definitions
edges[].type	Yes	direct, fan_out, conditional, or dynamic_fan_out
edges[].source	Yes	Node id this edge originates from
edges[].target	For direct	Target node id
edges[].targets	For fan_out	Array of target node ids (minimum 2)
edges[].edge_class	For conditional, dynamic_fan_out	Registry name matching @edge(name="...")
edges[].route	For conditional	Map of route keys to target node ids — must include "default"
edges[].config	No	Arbitrary dict passed to the edge constructor as self.config
edges[].description	No	Why this edge exists. Available as self.description and attached to every trace event the edge publishes.

Edge Types

Type	Python class needed	Description
direct	No	Fixed single target — fully declared in JSON
fan_out	No	Fixed parallel targets — fully declared in JSON
conditional	Yes — extends VsBaseConditionalEdge	_evaluate(state) returns a route key; key is looked up in route map
dynamic_fan_out	Yes — extends VsDynamicFanOutEdge	get_sends(state) returns a list of Send objects for runtime branching

---

State

All graph state must extend VsBaseGraphState:

from typing import Optional
from vs_graph.schema.vs_base_graph_state import VsBaseGraphState

class MyState(VsBaseGraphState):
    user_input: str
    intent: Optional[str]
    response: Optional[str]

VsBaseGraphState is a TypedDict providing:

Field	Type	Description
trace_id	str	Auto-generated per invoke() call if not set. Links all trace events for this execution.
error	Optional[VsGraphError]	Populated by the framework when a node raises VsUserException. Contains message, cause, stack_trace.
current_message	Optional[str]	Convenience field for the current user-facing message. Not set by the framework.

---

Nodes

Extend VsBaseNode and implement execute. Decorate with @node(name="...") to self-register in VsNodeRegistry.

from vs_graph.node.vs_base_node import VsBaseNode
from vs_graph.decorator.vs_node_decorator import node
from vs_graph.exception.vs_graph_exceptions import VsUserException, VsNodeExecutionException

@node(name="classifier_node")
class ClassifierNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        if not state.get("user_input"):
            raise VsUserException("Input is required.")
        state["intent"] = "execute"
        await self.trace(state["trace_id"], "intent classified", data={"intent": state["intent"]})
        return state

@node decorator

@node(name="classifier_node")   # registers under key "classifier_node"
@node                           # registers under the class name

The name value must match the "name" field in the JSON node definition.

What invoke() does (called by the graph, not you)

VsBaseNode.invoke() wraps your execute():

1. Logs and publishes trace: "node '{id}' started"
2. Calls await self.execute(state)
3. On success: logs and publishes trace: "node '{id}' completed" with duration_seconds
4. On VsUserException: stores error in state["error"] and returns state (graph continues)
5. On VsNodeExecutionException: logs error and re-raises (graph stops)
6. On any other exception: logs error and re-raises (graph stops)

Publishing trace events

async def execute(self, state: MyState) -> dict:
    await self.trace(
        state["trace_id"],
        "retrieved context",
        data={"num_chunks": 8},
    )
    return state

self.trace() is a no-op if no trace_publisher was passed to the graph.

Node constructor args

VsBaseNode.__init__ receives:

Parameter	Source
node_id	The id field from the JSON node definition
config	The config dict from the JSON node definition
description	The description field from the JSON node definition
trace_publisher	The publisher passed to VsBaseGraph.__init__

Access them as self.node_id, self.config, self.description, self.trace_publisher.

---

Prebuilt Nodes

Some nodes are the same in every agentic graph. vs-graph ships them so you don't rewrite them each time:

• Shipped defaults — fully implemented and self-registered. VsHeuristicInputGuardNode (as vs_guard_node) is one: drop the registry name into your JSON, configure it, done. No Python.
• Half-built — orchestration, tracing, and skip logic done; you implement one method (VsHydrationNode.load, VsModerationInputGuardNode.moderate). Subclass, @node(name="..."), reference it in JSON.

VsHydrationNode

The typical first node in a conversational graph. LangGraph loads the checkpoint for the current thread_id into state before the graph runs — but a checkpoint only exists while it's warm. New conversation, or Redis TTL expired after 30 days of silence, means an empty state. VsHydrationNode rebuilds it from your durable store (ctx-resolver, DB, …) so the agent keeps its memory across gaps the checkpointer can't cover.

execute() is already written:

1. is_hydrated(state) → if true, the checkpoint was warm, return {} and move on
2. otherwise await load(state) and return the delta — LangGraph checkpoints it from there

You implement load:

from vs_graph.node.vs_hydration_node import VsHydrationNode
from vs_graph.decorator.vs_node_decorator import node

@node(name="hydration_node")
class ConversationHydrationNode(VsHydrationNode[MyState]):

    hydration_keys = ["history", "resolved_context"]

    async def load(self, state: MyState) -> dict:
        cid = state["conversation_id"]
        return {
            "history": await conversation_store.get_history(cid),
            "resolved_context": await ctx_resolver.resolve(cid),
        }

Member	Kind	Description
load(state)	Abstract — you implement	Fetch prior context from the durable store (key off conversation_id / user_id in state) and return it as a state delta.
hydration_keys	Class attr	State keys that, once all non-None, mean hydration already happened. Set this and you never touch is_hydrated.
is_hydrated(state)	Override (optional)	Default: all hydration_keys are non-None. Override for custom "is the state warm?" logic.
execute(state)	Provided — don't override	The skip-or-load template above.

Emits node '{id}' hydrated or node '{id}' already hydrated trace events.

Input Guards

Run before the graph does anything expensive. A blocked message raises VsUserException, which the graph stores in state["error"] and carries to your responder — so the user gets a clean "that's not allowed" instead of a crash or a wasted LLM bill.

Two flavours, both extending VsInputGuardNode:

Node	I/O	Implement	Use for
VsHeuristicInputGuardNode	none	nothing — self-registered as vs_guard_node, configure patterns in JSON; subclass only for rules()	Cheap fast-fail — obvious junk, banned strings, prompt-injection tells
VsModerationInputGuardNode	one call	moderate(text)	A real moderation model — hate, self-harm, sexual, violence, …

Common shape:

Member	Kind	Description
check(state, text)	Abstract on VsInputGuardNode; each flavour implements it	Returns a VsGuardDecision.
guard_text(state)	Override (optional)	What text to inspect. Default: state["current_message"].
execute(state)	Provided — don't override	Runs check, traces guard passed / guard blocked, raises VsUserException on block.

VsGuardDecision(allowed, reason=None, category=None, scores={}) — reason is what the user sees.

Heuristic — VsHeuristicInputGuardNode (shipped default)

This one is a default node: vs-graph imports it, it self-registers under vs_guard_node, and it's ready the moment you import vs_graph. Nothing to subclass, nothing to write.

Why use it: it's the cheapest guard you can put in front of a graph — pure regex, zero I/O, runs in microseconds. Catch the obvious stuff (banned strings, prompt-injection tells like "ignore previous instructions", oversized payloads) here and you never pay for the LLM call, the retrieval, or the moderation round-trip on garbage input. Put it first, chain a VsModerationInputGuardNode after it for the cases regex can't judge.

How to use it: add it to the graph JSON as your entry_node (or right after hydration) and pass patterns in config — no Python:

{
  "id": "guard",
  "name": "vs_guard_node",
  "config": {
    "patterns": ["ignore (all )?previous instructions", "\\bBEGIN\\s+SYSTEM\\s+PROMPT\\b"],
    "block_reason": "That request can't be processed."
  }
}

Subclass only when you need dynamic rules():

from vs_graph.node.vs_heuristic_input_guard_node import VsHeuristicInputGuardNode
from vs_graph.decorator.vs_node_decorator import node

@node(name="length_guard")
class LengthGuard(VsHeuristicInputGuardNode[MyState]):

    def rules(self, state, text):
        if len(text) > 8000:
            yield "Message is too long."

Moderation

Subclass and implement moderate() — return a VsModerationResult. check() does the thresholding (threshold, blocked_categories).

VsModerationResult(scores={category: float}, flagged=None, raw=None).

OpenAI (omni-moderation-latest — scores come back ready-made):

from openai import AsyncOpenAI
from vs_graph.node.vs_moderation_input_guard_node import VsModerationInputGuardNode
from vs_graph.schema.vs_moderation_result import VsModerationResult
from vs_graph.decorator.vs_node_decorator import node

_openai = AsyncOpenAI()

@node(name="moderation_input_guard")
class OpenAiModerationGuard(VsModerationInputGuardNode[MyState]):

    threshold = 0.5

    async def moderate(self, text: str) -> VsModerationResult:
        resp = await _openai.moderations.create(
            model="omni-moderation-latest", input=text,
        )
        result = resp.results[0]
        return VsModerationResult(
            scores=dict(result.category_scores),
            flagged=result.flagged,
            raw=result.model_dump(),
        )

Claude (no moderation endpoint — ask Claude to score it, parse the JSON):

import json
import anthropic
from vs_graph.node.vs_moderation_input_guard_node import VsModerationInputGuardNode
from vs_graph.schema.vs_moderation_result import VsModerationResult
from vs_graph.decorator.vs_node_decorator import node

_claude = anthropic.AsyncAnthropic()

_CATS = ["hate", "harassment", "self_harm", "sexual", "violence"]
_SYS = (
    "You are a content-moderation classifier. Score the user message from 0.0 to 1.0 "
    "for each of these categories: " + ", ".join(_CATS) + ". "
    'Reply with ONLY a JSON object mapping each category to its float score.'
)

@node(name="moderation_input_guard")
class ClaudeModerationGuard(VsModerationInputGuardNode[MyState]):

    threshold = 0.6

    async def moderate(self, text: str) -> VsModerationResult:
        msg = await _claude.messages.create(
            model="claude-opus-5",          # claude-haiku-4-5 is plenty for this, your call
            max_tokens=256,
            system=_SYS,
            messages=[{"role": "user", "content": text}],
        )
        raw = next(b.text for b in msg.content if b.type == "text")
        scores = {k: float(v) for k, v in json.loads(raw).items()}
        return VsModerationResult(scores=scores, raw=raw)

Chain them in the JSON — heuristic first (free), moderation next — by pointing the heuristic guard's edge at the moderation guard.

---

Edges

Direct edge

No Python class. Declared fully in JSON:

{ "type": "direct", "source": "executor", "target": "responder" }

Fan-out edge

No Python class. Declares fixed parallel targets in JSON (minimum 2):

{
  "type": "fan_out",
  "source": "splitter",
  "targets": ["worker_a", "worker_b", "worker_c"]
}

All targets receive the same state and execute in parallel via LangGraph's native fan-out.

Conditional edge

Extend VsBaseConditionalEdge and implement _evaluate. Decorate with @edge(name="..."):

from vs_graph.edge.vs_base_conditional_edge import VsBaseConditionalEdge
from vs_graph.decorator.vs_edge_decorator import edge

@edge(name="classifier_routing_edge")
class ClassifierRoutingEdge(VsBaseConditionalEdge[MyState]):

    def _evaluate(self, state: MyState) -> str:
        if state.get("intent") == "execute":
            return "execute"
        return "default"

_evaluate returns a route key. The base class's async route() wrapper looks it up in the route map from JSON, publishes the routing trace, and returns the target node id. If the key is not in the map, it falls back to "default". A "default" entry in the JSON route map is required.

JSON:

{
  "type": "conditional",
  "source": "classifier",
  "edge_class": "classifier_routing_edge",
  "route": {
    "execute": "executor",
    "default": "responder"
  }
}

Dynamic fan-out edge

Extend VsDynamicFanOutEdge and implement get_sends. Returns a list of LangGraph Send objects — each specifies a target node and a custom state for that branch:

from vs_graph.edge.vs_dynamic_fan_out_edge import VsDynamicFanOutEdge
from vs_graph.decorator.vs_edge_decorator import edge
from langgraph.types import Send

@edge(name="document_fan_out_edge")
class DocumentFanOutEdge(VsDynamicFanOutEdge[MyState]):

    def get_sends(self, state: MyState) -> list:
        return [
            Send("process_document", {**state, "document": doc})
            for doc in state["documents"]
        ]

JSON:

{
  "type": "dynamic_fan_out",
  "source": "splitter",
  "edge_class": "document_fan_out_edge"
}

@edge decorator

@edge(name="classifier_routing_edge")   # registers under key "classifier_routing_edge"
@edge                                   # registers under the class name

The name value must match the "edge_class" field in the JSON edge definition.

---

@graph Decorator — Auto-Import Packages

Instead of manually importing every node and edge module, use @graph on your graph class to declare which packages contain your implementations. The library scans and imports them automatically at build() time, triggering @node and @edge registration.

from vs_graph.decorator.vs_graph_decorator import graph
from vs_graph.vs_base_graph import VsBaseGraph

@graph(
    nodes=["myapp.graph.nodes"],
    edges=["myapp.graph.edges"],
)
class MyGraph(VsBaseGraph[MyState]):

    def __init__(self):
        with open("my_graph.json") as f:
            graph_json = json.load(f)
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()

Both nodes and edges accept a list of package paths. Each package is walked recursively — all modules within it are imported. You can specify multiple packages:

@graph(
    nodes=["myapp.graph.nodes", "shared.common_nodes"],
    edges=["myapp.graph.edges"],
)
class MyGraph(VsBaseGraph[MyState]):
    ...

Without @graph, node and edge modules must be imported manually before build() is called:

import myapp.graph.nodes.classifier_node  # noqa
import myapp.graph.nodes.executor_node    # noqa

@graph with no packages, or subclasses without the decorator, are both valid — auto-import is skipped.

---

Registries

VsNodeRegistry

VsNodeRegistry is a static registry. @node populates it at import time.

Method	Description
register(name, node_class)	Called by @node — raises ValueError if name already registered
get(name)	Returns the class — raises ValueError if not found
create(name, node_id, config, **kwargs)	Instantiates the node class
is_registered(name)	Returns True if name is registered
list_nodes()	Returns sorted list of registered names
clear()	Removes all registered nodes — for testing only

VsEdgeRegistry

Same pattern as VsNodeRegistry for edge classes.

Method	Description
register(name, edge_class)	Called by @edge — raises ValueError if name already registered
get(name)	Returns the class — raises ValueError if not found
create(name, source, **kwargs)	Instantiates the edge class
is_registered(name)	Returns True if name is registered
list_edges()	Returns sorted list of registered names
clear()	Removes all registered edges — for testing only

---

Checkpointer

Configured via config.ini. Provides LangGraph thread-level state persistence across invoke() calls. The thread key defaults to state["trace_id"]; override it with config={"configurable": {"thread_id": ...}} on invoke().

[graph]
checkpointer = memory

[graph]
checkpointer             = redis
redis_url                = redis://localhost:6379
checkpoint_key_prefix    = vs:checkpoint:
checkpoint_ttl_seconds   = 3600

Key	Default	Description
graph.checkpointer	memory	Checkpointer type: memory or redis
graph.redis_url	redis://localhost:6379	Redis connection URL (redis only)
graph.checkpoint_key_prefix	vs:checkpoint:	Key prefix for stored checkpoints (redis only)
graph.checkpoint_ttl_seconds	None	TTL for checkpoint keys in seconds (redis only)

Custom checkpointers can be registered via VsCheckpointerFactory:

from vs_graph.checkpointer.vs_checkpointer_factory import VsCheckpointerFactory

VsCheckpointerFactory.register(
    "postgres",
    lambda cfg: MyPostgresCheckpointer(config=cfg)
)

The factory lambda receives the same cfg dict that _load_checkpointer builds from config.ini.

---

Trace Publisher

Pass a VsBaseTracePublisher to the graph constructor to stream execution events to any external sink (queue, websocket, database):

from vs_common.publisher.vs_base_trace_publisher import VsBaseTracePublisher
from vs_common.schema.vs_trace_level import VsTraceLevel
from typing import Any, Dict, Optional

class MyTracePublisher(VsBaseTracePublisher):

    async def publish(
        self,
        trace_id: str,
        message: str,
        level: VsTraceLevel = VsTraceLevel.INFO,
        data: Optional[Dict[str, Any]] = None,
    ) -> None:
        await my_queue.push({
            "trace_id": trace_id,
            "message": message,
            "level": level.value,
            "data": data or {},
        })

graph = MyGraph(graph_json, state_class=MyState, trace_publisher=MyTracePublisher())

The publisher is automatically propagated to every node and edge instance. You do not pass it to nodes or edges directly.

Events published automatically by the framework:

Event	Source
"node '{id}' started"	VsBaseNode.invoke()
"node '{id}' completed" with duration_seconds	VsBaseNode.invoke()
"node '{id}' user error"	VsBaseNode.invoke() on VsUserException
"node '{id}' failed"	VsBaseNode.invoke() on exception
"edge '{source}' routing to '{next_node}'" with route_key, next_node, duration_seconds	VsBaseConditionalEdge.route()
"edge '{source}' fanning out to {n} branch(es)" with branch_count, targets, duration_seconds	VsDynamicFanOutEdge.route()

If a node or edge has a description in the JSON, it is merged into the data payload of every trace event that node or edge publishes — the framework's own started/completed events and anything you publish from execute(). Whoever is watching the stream sees what ran and why, without opening the code:

{
  "trace_id": "…",
  "message": "node 'guard' started",
  "data": { "description": "Blocks junk input before we pay for the LLM call." }
}

An explicit "description" key in your own data dict wins over the JSON one.

Direct and static fan_out edges are pure topology — they compile to plain LangGraph edges and emit no trace events. The routing wrappers are async and await the trace publish before returning the next node, so the routing event is always ordered ahead of the next node's "started" event.

---

Exceptions

Exception	When to raise	Effect
VsUserException	Error is safe to surface to the user	Stored in state["error"] — graph continues to next node
VsNodeExecutionException	Internal node failure	Logged and re-raised — graph stops
VsGraphBuildException	Invalid build state	Raised during build()
VsEdgeEvaluationException	Edge routing failure	Raised from edge evaluate()

from vs_graph.exception.vs_graph_exceptions import VsUserException, VsNodeExecutionException

async def execute(self, state: MyState) -> dict:
    if not state.get("user_input"):
        raise VsUserException("Please provide an input.")

    try:
        result = await call_external_service()
    except Exception as e:
        raise VsNodeExecutionException(f"External service failed: {e}") from e

    state["result"] = result
    return state

state["error"] shape when VsUserException is raised:

{
  "message": "Please provide an input.",
  "cause": null,
  "stack_trace": null
}

---

Full Example

import json
from typing import Optional

from vs_graph.vs_base_graph import VsBaseGraph
from vs_graph.schema.vs_base_graph_state import VsBaseGraphState
from vs_graph.node.vs_base_node import VsBaseNode
from vs_graph.edge.vs_base_conditional_edge import VsBaseConditionalEdge
from vs_graph.decorator.vs_node_decorator import node
from vs_graph.decorator.vs_edge_decorator import edge
from vs_graph.decorator.vs_graph_decorator import graph
from vs_graph.exception.vs_graph_exceptions import VsUserException


# --- State ---

class MyState(VsBaseGraphState):
    user_input: str
    intent: Optional[str]
    response: Optional[str]


# --- Nodes ---

@node(name="classifier_node")
class ClassifierNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        if not state.get("user_input"):
            raise VsUserException("Input is required.")
        state["intent"] = "search" if "find" in state["user_input"] else "chat"
        await self.trace(state["trace_id"], "classified", data={"intent": state["intent"]})
        return state


@node(name="responder_node")
class ResponderNode(VsBaseNode[MyState]):

    async def execute(self, state: MyState) -> dict:
        state["response"] = f"Handling intent: {state['intent']}"
        return state


# --- Edges ---

@edge(name="classifier_routing_edge")
class ClassifierRoutingEdge(VsBaseConditionalEdge[MyState]):

    def _evaluate(self, state: MyState) -> str:
        return state.get("intent", "default")


# --- Graph ---

@graph(nodes=["myapp.graph.nodes"], edges=["myapp.graph.edges"])
class MyGraph(VsBaseGraph[MyState]):

    def __init__(self):
        graph_json = {
            "version": "1.0",
            "name": "my_graph",
            "entry_node": "classifier",
            "exit_nodes": ["responder"],
            "nodes": [
                {"id": "classifier", "name": "classifier_node", "config": {}},
                {"id": "responder",  "name": "responder_node",  "config": {}},
            ],
            "edges": [
                {
                    "type": "conditional",
                    "source": "classifier",
                    "edge_class": "classifier_routing_edge",
                    "route": {"search": "responder", "chat": "responder", "default": "responder"}
                }
            ]
        }
        super().__init__(graph_json=graph_json, state_class=MyState)
        self.build()


# --- Run ---

import asyncio

async def main():
    g = MyGraph()
    result = await g.invoke({
        "trace_id": "",
        "error": None,
        "current_message": None,
        "user_input": "find me a product",
        "intent": None,
        "response": None,
    })
    print(result["response"])  # Handling intent: search

asyncio.run(main())

---

Class Reference

---

VsBaseGraph

Abstract base class for all graphs. Extend it and call build() in __init__.

Constructor:

Parameter	Type	Description
graph_json	Dict[str, Any]	Parsed graph JSON — validated on construction. Its optional description is exposed as self.description.
state_class	Type[S]	TypedDict subclass used as LangGraph state type
trace_publisher	Optional[VsBaseTracePublisher]	Publisher injected into all nodes and edges

Methods:

Method	Description
build()	Auto-imports packages, instantiates nodes/edges, validates, loads checkpointer, compiles LangGraph. Call once at startup.
async invoke(state, config=None)	Runs the compiled graph. Auto-generates trace_id if not set. Sets the checkpointer thread_id to state["trace_id"] unless you pass config={"configurable": {"thread_id": ...}} — an explicit thread_id wins, but a configurable dict without one still gets trace_id merged in.

---

VsHydrationNode

Prebuilt first node. Rebuilds state from a durable store when the checkpoint is cold (new thread or expired TTL). See Prebuilt Nodes.

Member	Kind	Description
load(state)	Abstract	Fetch prior context from your store, return a state delta.
hydration_keys	Class attr (List[str])	Keys that mean "already hydrated" when all non-None.
is_hydrated(state)	Optional override	Default checks hydration_keys.
execute(state)	Provided	Skip-or-load template — do not override.

---

VsInputGuardNode / VsHeuristicInputGuardNode / VsModerationInputGuardNode

Prebuilt guard nodes. Block bad input before the graph spends anything. See Input Guards.

Member	Where	Description
check(state, text)	Abstract on base; implemented by each flavour	Returns VsGuardDecision.
guard_text(state)	Base, overridable	Text to inspect. Default state["current_message"].
execute(state)	Base, provided	Runs check, traces, raises VsUserException on block.
patterns / rules(state, text)	VsHeuristicInputGuardNode	Regex list (from config.patterns or a subclass attr) / generator of block reasons. No I/O. Self-registers as vs_guard_node.
moderate(text)	VsModerationInputGuardNode, abstract	Call a moderation model, return VsModerationResult.
threshold / blocked_categories	VsModerationInputGuardNode	Score cutoff / category allowlist for blocking.

VsGuardDecision(allowed, reason, category, scores) and VsModerationResult(scores, flagged, raw) live in vs_graph.schema.

---

VsBaseNode

Abstract base class for all nodes.

Constructor parameters (injected by VsNodeRegistry.create()):

Parameter	Type	Description
node_id	str	Node id from JSON
config	Optional[Dict[str, Any]]	Node config dict from JSON
description	Optional[str]	Node description from JSON
trace_publisher	Optional[VsBaseTracePublisher]	Propagated from the graph

Instance attributes:

Attribute	Description
self.node_id	The node id from JSON
self.config	The config dict from JSON
self.description	The node description from JSON — auto-attached to every trace event
self.trace_publisher	The graph-level trace publisher
self.logger	Logger instance keyed to the class name

Abstract method:

Method	Signature	Description
execute	async execute(state: S) -> dict	Implement your node logic here. Return the updated state dict.

Helper method:

Method	Signature	Description
trace	async trace(trace_id, message, data=None, level=INFO)	Publishes a trace event, merging self.description into data. No-op if no publisher.

---

VsBaseConditionalEdge

Abstract base class for conditional routing edges.

Constructor parameters:

Parameter	Type	Description
source	str	Source node id
route	Dict[str, str]	Map of route keys to target node ids — must include "default". Stored as self.route_map.
config	Optional[Dict[str, Any]]	Edge config dict from JSON
description	Optional[str]	Edge description from JSON — auto-attached to every trace event
trace_publisher	Optional[VsBaseTracePublisher]	Propagated from the graph

Abstract method:

Method	Signature	Description
_evaluate	_evaluate(state: S) -> str	Return a route key. route() looks it up in route_map; falls back to "default".

Routing function (registered with LangGraph — you do not call it):

Method	Signature	Description
route	async route(state: S) -> str	Calls _evaluate, resolves the key against route_map (falling back to "default"), publishes the routing trace, returns the target node id. Raises if there is no default.

---

VsDynamicFanOutEdge

Abstract base class for runtime fan-out edges.

Abstract method:

Method	Signature	Description
get_sends	get_sends(state: S) -> List	Return a list of langgraph.types.Send objects. Each Send specifies a target node and a custom state dict for that branch.

Routing function (registered with LangGraph — you do not call it):

Method	Signature	Description
route	async route(state: S) -> List	Calls get_sends, publishes a fan-out trace with branch_count and targets, returns the Send list.

---

VsCheckpointerFactory

Static factory for checkpointer registration and creation.

Method	Description
register(name, factory_fn)	Register a custom checkpointer. factory_fn receives a config dict.
create(checkpointer_type, config)	Instantiate and return the checkpointer.

Built-in types: "memory", "redis".

---

Running Tests

./run_tests.sh