Skip to main content

LangGraph ⇄ Pulse

Intermediate ⏱ 15 min langgraph · python · graph node

LangGraph owns the agent/decision graph; Pulse is the real-time stream engine it calls. The repo ships the working example (examples/langgraph-pulse-bridge/), validated live against langgraph 1.2.9.

Direction A — Pulse as a node inside the graph (the common one)​

START → enrich → [pulse_score] → decide → END

pulse_score is a normal LangGraph node that off-loads the real-time decision to Pulse:

POST /api/pulse/x/order-scoring/in            (send the event)
GET  /api/pulse/events/order-scoring.score.out (read the verdict)

1 · Deploy the Pulse step​

pulse server start --dev
mkdir app && cp pulse.yaml app/ && (cd app && pulse deploy .)   # stands up order-scoring

2 · Run the graph​

pip install langgraph requests
python graph.py

Everything that touches Pulse lives in one file, pulse_step.py — a single score_order(order) function you can run and unit-test without LangGraph (python pulse_step.py).

3 · What happens (validated live in the repo)​

The pulse_score node calls Pulse's streaming filter amount >= 1000 stage, and decide routes on the verdict:

amount= 2500  ->  high_value=True   decision=escalate
amount=  300  ->  high_value=False  decision=auto-approve

Swap the app's score stage for windows, keyBy, joins, an llm/mcp stage — the graph is untouched; the same node returns richer intelligence.

Lower latency: the SDK duplex channel​

For a synchronous decision inside a graph node, pulse-py offers a correlated WebSocket instead of POST-then-read — one round-trip, no polling:

# pip install "streamflow-pulse-client[duplex]"
from pulse_client import PulseClient
async with PulseClient("http://localhost:9090").duplex("Score") as ch:
    await ch.send({"orderId": "O-1", "amount": 2500}, correlation_id="O-1")
    verdict = await ch.recv()      # correlation_id == "O-1"

Direction B — Pulse drives the graph​

The reverse (pulse_to_langgraph.py): subscribe to Pulse's live SSE stream (GET /api/pulse/events/stream) and invoke the compiled graph per event Pulse emits — LangGraph stops polling and reacts the instant a window closes or an incident is flagged. Validated live in the repo: with the consumer running, only the qualifying order fires a graph run.

Next​