feat: add LangGraph adapter with atomic fan-out reserve/settle and typed budget advisory (#33)

.github/workflows/ci.yml

@@ -64,8 +64,9 @@ jobs:
       # Install the framework extras so the adapter handler tests actually run
       # (instead of skipping). langchain-core covers the LangChain handler
       # tests; litellm covers the callback-swallowing regression tests (the
-      # CrewAI adapter tests stub crewai itself, so the heavy extra stays out).
-      - run: pip install -e ".[dev,langchain,litellm]"
+      # CrewAI adapter tests stub crewai itself, so the heavy extra stays out);
+      # langgraph covers the fan-out reserve/settle and advisory-channel tests.
+      - run: pip install -e ".[dev,langchain,litellm,langgraph]"
       - name: Pytest
         run: pytest -q
 

CHANGELOG.md

@@ -10,6 +10,16 @@ both packages adhere to [Semantic Versioning](https://semver.org/).
 
 ## Unreleased
 
+### Added (py)
+
+- **LangGraph adapter** (`floe-guard[langgraph]`, issue #33): `guarded_node`
+  wraps graph nodes with the reserve-before / settle-after contract, so a
+  `StateGraph` fan-out of parallel sub-agents holds one shared ceiling
+  atomically; `AdvisoryChannel` / `latest_advisory` expose the typed
+  `BudgetAdvisory` in graph state after each metered node, so a router can
+  downshift models on `near_limit` before the hard-stop. Ships with a
+  no-API-key example (`examples/langgraph_budget_aware.py`).
+
 ## py 0.2.0 / js 0.2.1 — 2026-07-10
 
 Everything the repo grew between the 0.1.0 uploads and this release ships here —

CONTRIBUTING.md

@@ -31,6 +31,7 @@ The optional adapters need their extras to run/import:
 pip install -e ".[crewai]"      # CrewAI adapter
 pip install -e ".[litellm]"     # LiteLLM adapter
 pip install -e ".[langchain]"   # LangChain adapter
+pip install -e ".[langgraph]"   # LangGraph adapter
 pip install -e ".[openai]"      # OpenAI adapter
 pip install -e ".[anthropic]"   # Anthropic adapter
 ```

README.md

@@ -13,7 +13,7 @@ $0.10 instead of $4,000. No account, no signup, no network, **no telemetry**.
 Runs in your process.
 
 Works with [CrewAI](#crewai) · [LiteLLM](#litellm) · [LangChain](#langchain) ·
-[OpenAI](#openai) · [Anthropic](#anthropic) ·
+[LangGraph](#langgraph) · [OpenAI](#openai) · [Anthropic](#anthropic) ·
 [Vercel AI SDK](#vercel-ai-sdk) — or any stack, via plain `check()` / `record()`.
 
 ```bash
@@ -221,6 +221,48 @@ llm.invoke("hello")            # checks budget before the call, records spend af
 The handler checks the budget on LLM start (raising `BudgetExceeded` aborts the
 call before it runs) and records token usage on LLM end.
 
+### LangGraph
+
+```bash
+pip install floe-guard[langgraph]
+```
+
+```python
+import operator
+from typing import Annotated
+from typing_extensions import TypedDict
+
+from floe_guard import BudgetGuard
+from floe_guard.integrations.langgraph import AdvisoryChannel, guarded_node
+
+class State(TypedDict):
+    results: Annotated[list, operator.add]
+    budget: AdvisoryChannel          # typed BudgetAdvisory, refreshed per call
+
+guard = BudgetGuard(limit_usd=0.10)
+
+@guarded_node(guard, estimated_cost=0.01)   # reserve() before, settle()/release() after
+def worker(state: State) -> dict:
+    response = my_llm_call(state)
+    return {"results": [response["text"]], "usage": {
+        "model": response["model"],
+        "prompt_tokens": response["prompt_tokens"],
+        "completion_tokens": response["completion_tokens"],
+    }}
+```
+
+`guarded_node` gives every branch of a `StateGraph` fan-out its own atomic
+slice of the ceiling (reserve-before / settle-after, the same contract the
+OpenAI and Anthropic adapters use), so N parallel sub-agents can't race one
+shared total. Pass `estimated_cost` to hold a conservative fixed slice on
+every call of that node (the `0.01` above); a node that omits it estimates
+from the guard's last settled cost instead, which is `0` on a fresh guard, so
+seed a cold-start fan-out explicitly. After each settled call it writes the guard's `BudgetAdvisory`
+into `state["budget"]`, so a router node can downshift to a cheaper model on
+`near_limit` *before* the hard-stop — see
+[`examples/langgraph_budget_aware.py`](examples/langgraph_budget_aware.py) for
+the full budget-aware router (no API key needed).
+
 ### OpenAI
 
 ```bash

examples/langgraph_budget_aware.py

@@ -0,0 +1,100 @@
+"""Context-aware budgeting in a LangGraph graph — the router tapers near the cap.
+
+No API key, no account, no network — a stub LLM returns fixed token usage. The
+LangGraph port of ``examples/budget_aware.py``: every worker node is wrapped
+with ``guarded_node`` (atomic reserve/settle, so this pattern survives a
+parallel fan-out unchanged), and each settled call refreshes a typed
+``BudgetAdvisory`` in the graph state. The router node reads
+``state["budget"].near_limit`` and downshifts to the cheap model, so the run
+finishes on budget instead of slamming into the hard-stop mid-task.
+
+The advisory is a *soft* signal you choose to act on; ``reserve()`` is still
+the hard guarantee on every guarded node.
+
+Run:  pip install floe-guard[langgraph]
+      python examples/langgraph_budget_aware.py
+"""
+
+from __future__ import annotations
+
+import operator
+from typing import Annotated
+
+from langgraph.graph import END, START, StateGraph
+from typing_extensions import TypedDict
+
+from floe_guard import BudgetGuard
+from floe_guard.integrations.langgraph import AdvisoryChannel, guarded_node
+
+FULL = ("gpt-4o", 1000, 1000)  # ~$0.0125 / call
+CHEAP = ("gpt-4o-mini", 1000, 1000)  # ~$0.0008 / call
+
+
+class State(TypedDict):
+    steps: Annotated[int, operator.add]
+    log: Annotated[list, operator.add]
+    budget: AdvisoryChannel
+
+
+def stub_llm(model: tuple[str, int, int]) -> dict[str, object]:
+    """A fake LLM call — no network, no key."""
+    name, prompt_tokens, completion_tokens = model
+    return {"model": name, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens}
+
+
+def make_worker(model: tuple[str, int, int]):
+    def worker(state: State) -> dict:
+        response = stub_llm(model)
+        return {
+            "steps": 1,
+            "log": [f"{response['model']}"],
+            # Report the call's usage; guarded_node settles it and refreshes
+            # state["budget"] with the guard's advisory.
+            "usage": response,
+        }
+
+    return worker
+
+
+def route(state: State) -> str:
+    """Downshift on the advisory; stop when not even a cheap call fits."""
+    adv = state.get("budget")
+    if adv is None:
+        return "full_step"  # first call — no signal yet
+    if adv.remaining_usd < 0.0008:
+        return END
+    if adv.near_limit:
+        return "cheap_step"
+    return "full_step"
+
+
+def main() -> None:
+    # Taper at 70% used so there's room to downshift before the ceiling.
+    guard = BudgetGuard(limit_usd=0.10, near_limit_bps=7000)
+    print(f"Budget ${guard.limit_usd:.2f} · taper at {guard.near_limit_bps / 100:.0f}% used\n")
+
+    graph = StateGraph(State)
+    graph.add_node("full_step", guarded_node(guard, make_worker(FULL), estimated_cost=0.0125))
+    graph.add_node("cheap_step", guarded_node(guard, make_worker(CHEAP), estimated_cost=0.0008))
+    graph.add_conditional_edges(START, route)
+    graph.add_conditional_edges("full_step", route)
+    graph.add_conditional_edges("cheap_step", route)
+
+    tapered = False
+    final = graph.compile().invoke({"steps": 0, "log": []}, {"recursion_limit": 200})
+
+    for step, model in enumerate(final["log"], start=1):
+        if model == CHEAP[0] and not tapered:
+            tapered = True
+            print("  [advisory] near_limit tripped → tapering to", model, "\n")
+        print(f"  step {step:>2}: {model}")
+
+    print(
+        f"\nFinished at step {final['steps']}. "
+        f"Final spend ${guard.spent_usd:.4f} (held under ${guard.limit_usd:.2f}), "
+        f"advisory read {final['budget'].used_bps / 100:.0f}% used."
+    )
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -10,7 +10,7 @@ readme = "README.md"
 requires-python = ">=3.10"
 license = { file = "LICENSE" }
 authors = [{ name = "Floe Labs" }]
-keywords = ["llm", "agents", "budget", "guardrail", "crewai", "litellm", "langchain", "cost", "openai", "anthropic"]
+keywords = ["llm", "agents", "budget", "guardrail", "crewai", "litellm", "langchain", "langgraph", "cost", "openai", "anthropic"]
 classifiers = [
     "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
@@ -27,6 +27,7 @@ dependencies = []
 litellm = ["litellm>=1.0"]
 crewai = ["crewai>=0.30", "litellm>=1.0"]
 langchain = ["langchain-core>=0.3"]
+langgraph = ["langgraph>=1.0"]
 openai = ["openai>=1.0"]
 anthropic = ["anthropic>=0.40"]
 dev = ["pytest>=7.0", "ruff>=0.4"]

src/floe_guard/integrations/__init__.py

@@ -5,4 +5,5 @@
     pip install floe-guard[litellm]
     pip install floe-guard[crewai]
     pip install floe-guard[langchain]
+    pip install floe-guard[langgraph]
 """