Aller au contenu principal

LangGraph ⇄ Pulse

Intermédiaire ⏱ 15 min langgraph · python · nœud de graphe

LangGraph possède le graphe d'agent/décision ; Pulse est le moteur de flux temps réel qu'il appelle. Le repo livre l'exemple fonctionnel (examples/langgraph-pulse-bridge/), validé en live contre langgraph 1.2.9.

Direction A — Pulse comme nœud dans le graphe (le cas courant)​

START → enrich → [pulse_score] → decide → END

pulse_score est un nœud LangGraph normal qui délègue la décision temps réel à Pulse :

POST /api/pulse/x/order-scoring/in             (envoyer l'événement)
GET  /api/pulse/events/order-scoring.score.out (lire le verdict)

1 · Déployez le step Pulse​

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

2 · Lancez le graphe​

pip install langgraph requests
python graph.py

Tout ce qui touche Pulse vit dans un seul fichier, pulse_step.py — une fonction score_order(order) que vous pouvez exécuter et tester sans LangGraph (python pulse_step.py).

3 · Ce qui se passe (validé en live dans le repo)​

Le nœud pulse_score appelle le stage streaming filter amount >= 1000 de Pulse, et decide route sur le verdict :

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

Remplacez le stage score de l'app par des fenêtres, keyBy, joins, un stage llm/mcp — le graphe n'est pas touché ; le même nœud renvoie une intelligence plus riche.

Moins de latence : le canal duplex du SDK​

Pour une décision synchrone dans un nœud de graphe, pulse-py offre un WebSocket corrélé au lieu de POST-puis-lecture — un aller-retour, pas de 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 pilote le graphe​

L'inverse (pulse_to_langgraph.py) : abonnez-vous au flux SSE vivant de Pulse (GET /api/pulse/events/stream) et invoquez le graphe compilé par événement émis par Pulse — LangGraph cesse de poller et réagit à l'instant où une fenêtre se ferme ou un incident est signalé. Validé en live dans le repo : avec le consommateur en marche, seule la commande qualifiante déclenche un run de graphe.

Ensuite​