neruva-langgraph

Drop-in LangGraph integration for Neruva agent memory + reasoning substrate. Wraps the canonical LangGraph plug points (graph checkpoints + cross-thread KV store) and — new in 0.2.0 — adds the auto-pilot surface so your graphs proactively use the right Neruva cognitive tool per user intent.

What's new in 0.2.0 — Auto-pilot

• NeruvaAutoPilot — framework-agnostic core (classify intent + reflect + extract via your callable LLM).
• classify_intent_node(auto_pilot, message_key="messages") — pre-built LangGraph node factory. Inject BEFORE your main node so every user turn auto-routes to the right cognitive tool (stored under state["neruva_intent"]).
• reflect_node(auto_pilot, last_n=10) — pre-built node that reflects over the last N messages and auto-writes durable records (decisions / facts / mistakes / open_questions) to substrate. Wire as the final node of your graph or call periodically.

Plus NeruvaCodeGraph (5 sub-ms code-navigation methods) and NeruvaCognitive (13 cognitive-primitive methods — counterfactual / ToM / schema lifting / EFE / continual K-gram / hierarchical / rule induction). Callable from any node body.

Pattern-C: substrate stays $0/call. Classification + reflection runs through YOUR caller LLM.

from langgraph.graph import StateGraph, START
from neruva_langgraph import (
    NeruvaCheckpointSaver,
    NeruvaAutoPilot,
    classify_intent_node,
    reflect_node,
)

ap = NeruvaAutoPilot(
    api_key="nv_...", namespace="my_app",
    llm_callable=lambda p: llm.invoke(p).content,
)

graph = StateGraph(MyState)
graph.add_node("classify", classify_intent_node(ap))
graph.add_node("main", my_main_node)
graph.add_node("reflect", reflect_node(ap, last_n=10))
graph.add_edge(START, "classify")
graph.add_edge("classify", "main")
graph.add_edge("main", "reflect")

app = graph.compile(checkpointer=NeruvaCheckpointSaver(api_key="nv_..."))

pip install neruva-langgraph

What's new in the substrate (v0.5.7, May 2026)

The substrate this adapter wraps has gained a lot since the last release. Existing code keeps working — new capabilities are just available.

• Deterministic replay — every query is bit-identical across reruns from the same seed. Replay any past graph state for audit or debugging.
• Typed-shape context — pull structured JSON from records with per-field citations. {question, shape: {field: type}} → typed result without an LLM at query time.
• Tenant-specific PII rules — register your custom ID formats (employee codes, patient codes, order IDs) from 3-5 examples. The substrate redacts them automatically. Sub-microsecond per span.
• Depth-unlimited nested-belief tracking — store and retrieve chains like Alice → Bob → Carol thinks X at any depth, with inner-position-swap rejection.
• Counterfactual rollouts — "what if action k had been a' instead?" Replay an action sequence with one step substituted.
• Active inference planning — score candidate action sequences by KL distance to a goal-marginal. Caller-owned dynamics.
• Continual K-gram learning — provable no-forgetting via integer-add commutativity; repeated train() calls accumulate.

NeruvaCheckpointSaver

Persists every graph checkpoint into Neruva Records. One namespace per thread_id, so each conversation gets its own replayable timeline. Drop into any StateGraph.compile():

from langgraph.graph import StateGraph
from neruva_langgraph import NeruvaCheckpointSaver

graph = builder.compile(
    checkpointer=NeruvaCheckpointSaver(api_key="nv_..."),
)
graph.invoke(
    {"input": "hello"},
    config={"configurable": {"thread_id": "user_alice"}},
)

Realistic example -- multi-turn StateGraph with persistent checkpoints

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from neruva_langgraph import NeruvaCheckpointSaver

class State(TypedDict):
    messages: list

def call_model(state: State) -> State:
    llm = ChatAnthropic(model="claude-opus-4-7")
    state["messages"].append(llm.invoke(state["messages"]))
    return state

builder = StateGraph(State)
builder.add_node("model", call_model)
builder.set_entry_point("model")
builder.add_edge("model", END)
graph = builder.compile(checkpointer=NeruvaCheckpointSaver())
graph.invoke({"messages": ["What did I tell you last week?"]},
             config={"configurable": {"thread_id": "user_alice"}})

NeruvaContextStore

LangGraph's BaseStore for cross-thread KV (e.g. user preferences, shared facts). Backed by Records + agent_recall for semantic search:

from neruva_langgraph import NeruvaContextStore

store = NeruvaContextStore(api_key="nv_...")
graph = builder.compile(
    checkpointer=NeruvaCheckpointSaver(),
    store=store,
)
store.put(("users", "alice"), "profile", {"city": "Toronto"})
hit = store.get(("users", "alice"), "profile")
print(hit)   # -> Item(value={'city': 'Toronto'}, ...)

Why Neruva instead of LangGraph's built-in checkpointers?

Get an API key - Docs - Status