From 0fa711a7d9f4e70e96dc670f3ebf44674df7e650 Mon Sep 17 00:00:00 2001 From: "wenqiang.xu" Date: Thu, 9 Apr 2026 18:51:48 +0800 Subject: [PATCH] docs: add LangGraph demo, LangGraph Integration section, and AI Framework Integration to README - demo/demo-langgraph/: LangGraph workflow demo (langgraph_mq9_demo.py + README) showing two Agent nodes communicating asynchronously via mq9 mailboxes - docs/python.md: add ## LangGraph Integration section with code example - README.md: add AI Framework Integration section (LangChain + LangGraph), extend mq9 description, add demo-multi-agent and demo-langgraph table rows Co-Authored-By: Claude Sonnet 4.6 --- README.md | 51 +++++- demo/demo-langgraph/README.md | 89 +++++++++ demo/demo-langgraph/langgraph_mq9_demo.py | 212 ++++++++++++++++++++++ docs/python.md | 37 ++++ 4 files changed, 388 insertions(+), 1 deletion(-) create mode 100644 demo/demo-langgraph/README.md create mode 100644 demo/demo-langgraph/langgraph_mq9_demo.py diff --git a/README.md b/README.md index a53b966..beab69c 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ RobustMQ is a single-binary broker that natively supports MQTT, Kafka, NATS, AMQ ## mq9: AI Agent mailbox protocol -mq9 gives every agent a durable mailbox. Messages persist until TTL expires — senders and receivers do not need to be online simultaneously. +mq9 gives every agent a durable mailbox. Messages persist until TTL expires — senders and receivers do not need to be online simultaneously. It is purpose-built for multi-agent systems where tasks, results, and signals must survive disconnections and be delivered in priority order. | Concept | Description | |---------|-------------| @@ -145,6 +145,42 @@ await client.SendAsync(mailbox.MailId, "hello"u8.ToArray(), Priority.Normal); --- +## AI Framework Integration + +mq9 integrates with popular AI orchestration frameworks. Each Agent node gets a persistent +mailbox — tasks and results survive disconnections and are delivered in priority order. + +### LangChain + +[`langchain-mq9`](langchain-mq9/) provides six LangChain tools covering all mq9 protocol +operations. Drop them into any LangChain Agent or LangGraph node. + +```python +from langchain_mq9 import Mq9Toolkit + +tools = Mq9Toolkit(server="nats://demo.robustmq.com:4222").get_tools() +# CreateMailboxTool, CreatePublicMailboxTool, SendMessageTool, +# GetMessagesTool, ListMessagesTool, DeleteMessageTool +``` + +### LangGraph + +Each LangGraph node can own a private mq9 mailbox and communicate with other nodes +asynchronously — the graph edges carry state, while mq9 carries the payloads. + +```python +from langchain_mq9 import CreateMailboxTool, SendMessageTool + +async def node_writer(state): + create = CreateMailboxTool(server="nats://demo.robustmq.com:4222") + mail_id = await create._arun(ttl=120) + ... +``` + +Full examples: [demo/demo-langchain-mq9/](demo/demo-langchain-mq9/) · [demo/demo-langgraph/](demo/demo-langgraph/) + +--- + ## SDK documentation | Language | Docs | Demo | @@ -156,6 +192,8 @@ await client.SendAsync(mailbox.MailId, "hello"u8.ToArray(), Priority.Normal); | C# | [docs/csharp.md](docs/csharp.md) | [demo/demo-csharp/](demo/demo-csharp/) | | Rust | [docs/rust.md](docs/rust.md) | [demo/demo-rust/](demo/demo-rust/) | | langchain-mq9 | [docs/langchain-mq9.md](docs/langchain-mq9.md) | [demo/demo-langchain-mq9/](demo/demo-langchain-mq9/) | +| Multi-agent (Python ↔ Go) | — | [demo/demo-multi-agent/](demo/demo-multi-agent/) | +| LangGraph workflow | — | [demo/demo-langgraph/](demo/demo-langgraph/) | --- @@ -189,6 +227,15 @@ cd demo/demo-csharp && dotnet run # langchain-mq9 cd demo/demo-langchain-mq9 && pip install -r requirements.txt && python demo.py + +# Multi-agent (Python + Go, two terminals) +cd demo/demo-multi-agent && go run agent_b.go # terminal 1 +cd demo/demo-multi-agent && python agent_a.py # terminal 2 + +# LangGraph workflow +cd demo/demo-langgraph && pip install robustmq langchain-core langgraph langchain-openai +export OPENAI_API_KEY=sk-... +python langgraph_mq9_demo.py ``` --- @@ -212,6 +259,8 @@ demo/ demo-rust/ # Rust standalone demo demo-csharp/ # C# standalone demo demo-langchain-mq9/ # langchain-mq9 demo + demo-multi-agent/ # cross-language multi-agent demo (Python + Go) + demo-langgraph/ # LangGraph workflow demo VERSION # Canonical version (currently 0.3.5) ``` diff --git a/demo/demo-langgraph/README.md b/demo/demo-langgraph/README.md new file mode 100644 index 0000000..4617fd0 --- /dev/null +++ b/demo/demo-langgraph/README.md @@ -0,0 +1,89 @@ +# LangGraph + mq9 Demo + +Two LangGraph Agent nodes communicate asynchronously via mq9 mailboxes. + +## Scenario + +``` +node_writer → generates a draft paragraph (gpt-4o-mini) + → sends draft to node_reviewer's mailbox via mq9 +node_reviewer → receives draft from mq9 + → produces a review comment + → sends review back to node_writer's mailbox +node_writer → receives review from mq9 + → produces the final revised paragraph +``` + +mq9 acts as the async transport layer between nodes: each node has its own private mailbox; +messages are durable and priority-aware. Nodes do not need to be running at the same time. + +## Prerequisites + +- Python 3.10+ +- Access to `nats://demo.robustmq.com:4222` (public RobustMQ service), or run your own RobustMQ instance +- OpenAI API key + +## Install + +```bash +cd demo/demo-langgraph +pip install robustmq langchain-core langgraph langchain-openai +``` + +## Run + +```bash +export OPENAI_API_KEY=sk-... +python langgraph_mq9_demo.py +``` + +## Expected output + +``` +[node_reviewer] mailbox: m-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx +[node_writer] mailbox: m-yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy +[node_writer] draft: +AI agents are increasingly capable of communicating across distributed systems ... + +[node_writer] sent draft to reviewer (m-xxxxxxxx-...) +[node_reviewer] review: +Consider adding a concrete example of how store-first semantics enable ... + +[node_reviewer] sent review to writer (m-yyyyyyyy-...) +[node_writer] final: +AI agents are increasingly capable of communicating across distributed systems. +For instance, store-first semantics allow an agent to publish a task and shut down ... + +============================================================ +DRAFT: + AI agents are increasingly capable of communicating across distributed systems ... + +REVIEW: + Consider adding a concrete example of how store-first semantics enable ... + +FINAL: + AI agents are increasingly capable of communicating across distributed systems. + For instance, store-first semantics allow an agent to publish a task and shut down ... +============================================================ +``` + +## What this demo shows + +| Concept | Where | +|---------|-------| +| Each LangGraph node owns a private mq9 mailbox | `node_writer` and `node_reviewer` each call `client.create()` | +| Async handoff between nodes via mq9 | writer sends draft; reviewer subscribes, reads, replies | +| Store-first semantics | reviewer can start after writer has already published | +| Priority-ordered delivery | draft sent as `NORMAL`, review sent back as `HIGH` | +| LangChain LLM inside a LangGraph node | `ChatOpenAI(model="gpt-4o-mini")` called inside each node | + +## Using `langchain-mq9` tools instead + +If you prefer LangChain tool wrappers rather than calling the mq9 client directly, +the `langchain-mq9` package provides drop-in tools for all protocol operations: + +```bash +pip install langchain-mq9 +``` + +See [docs/langchain-mq9.md](../../docs/langchain-mq9.md) and [demo/demo-langchain-mq9/](../demo-langchain-mq9/). diff --git a/demo/demo-langgraph/langgraph_mq9_demo.py b/demo/demo-langgraph/langgraph_mq9_demo.py new file mode 100644 index 0000000..e2c6448 --- /dev/null +++ b/demo/demo-langgraph/langgraph_mq9_demo.py @@ -0,0 +1,212 @@ +""" +LangGraph + mq9 demo — two Agent nodes communicate asynchronously via mq9 mailboxes. + +Workflow: + node_writer → generates a short paragraph + → sends it to node_reviewer's mailbox via mq9 + node_reviewer → receives the paragraph + → produces a review comment + → sends it back to node_writer's mailbox + node_writer → receives the review, finalises the response + +mq9 acts as the async transport layer between nodes: +each node has its own private mailbox; messages are durable and priority-aware. + +Run: + cd demo/demo-langgraph + pip install robustmq langchain-core langgraph langchain-openai + export OPENAI_API_KEY=... + python langgraph_mq9_demo.py +""" + +from __future__ import annotations + +import asyncio +import json +from typing import TypedDict + +from langchain_openai import ChatOpenAI +from langgraph.graph import END, StateGraph + +from robustmq.mq9 import Client, Priority + +# demo.robustmq.com is a public RobustMQ service for testing. Replace with your own server address if needed. +SERVER = "nats://demo.robustmq.com:4222" +LLM = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) + + +# --------------------------------------------------------------------------- +# Shared state +# --------------------------------------------------------------------------- + + +class WorkflowState(TypedDict): + writer_mail_id: str + reviewer_mail_id: str + draft: str + review: str + final: str + + +# --------------------------------------------------------------------------- +# Nodes +# --------------------------------------------------------------------------- + + +async def node_writer(state: WorkflowState) -> WorkflowState: + """Generate a draft paragraph and send it to node_reviewer via mq9.""" + async with Client(server=SERVER) as client: + # Create writer's mailbox on first call + if not state.get("writer_mail_id"): + mb = await client.create(ttl=300) + state["writer_mail_id"] = mb.mail_id + print(f"[node_writer] mailbox: {mb.mail_id}") + + # If we already have a review, read it and produce the final version + if state.get("review"): + response = LLM.invoke( + f"You wrote this draft:\n{state['draft']}\n\n" + f"A reviewer said:\n{state['review']}\n\n" + "Revise the paragraph incorporating the feedback. " + "Return only the revised paragraph." + ) + state["final"] = response.content + print(f"[node_writer] final:\n{state['final']}") + return state + + # First pass: generate a draft + response = LLM.invoke( + "Write a short paragraph (3–4 sentences) about the future of AI agents " + "communicating asynchronously." + ) + state["draft"] = response.content + print(f"[node_writer] draft:\n{state['draft']}") + + # Send draft to reviewer + payload = json.dumps({ + "draft": state["draft"], + "reply_to": state["writer_mail_id"], + }).encode() + + await client.send(state["reviewer_mail_id"], payload, priority=Priority.NORMAL) + print(f"[node_writer] sent draft to reviewer ({state['reviewer_mail_id']})") + + return state + + +async def node_reviewer(state: WorkflowState) -> WorkflowState: + """Receive the draft from mq9, produce a review, send it back.""" + async with Client(server=SERVER) as client: + # Create reviewer's mailbox on first call + if not state.get("reviewer_mail_id"): + mb = await client.create(ttl=300) + state["reviewer_mail_id"] = mb.mail_id + print(f"[node_reviewer] mailbox: {mb.mail_id}") + # Return immediately so node_writer can send to this mailbox + return state + + # Receive the draft from mq9 + received: list[dict] = [] + + async def on_message(msg): + received.append(json.loads(msg.payload)) + + sub = await client.subscribe(state["reviewer_mail_id"], on_message) + await asyncio.sleep(1.0) # drain buffered messages + await sub.unsubscribe() + + if not received: + print("[node_reviewer] no messages yet, waiting...") + return state + + item = received[0] + draft = item["draft"] + reply_to = item["reply_to"] + + # Generate review + response = LLM.invoke( + f"Review the following paragraph and provide one specific, constructive " + f"suggestion to improve it:\n\n{draft}" + ) + review = response.content + state["review"] = review + print(f"[node_reviewer] review:\n{review}") + + # Send review back to writer + await client.send(reply_to, review.encode(), priority=Priority.HIGH) + print(f"[node_reviewer] sent review to writer ({reply_to})") + + return state + + +async def node_writer_finalise(state: WorkflowState) -> WorkflowState: + """Read the review from mq9 and produce the final version.""" + async with Client(server=SERVER) as client: + # Read review from writer's mailbox + received: list[str] = [] + + async def on_review(msg): + received.append(msg.payload.decode()) + + sub = await client.subscribe(state["writer_mail_id"], on_review) + await asyncio.sleep(1.0) + await sub.unsubscribe() + + if not received: + print("[node_writer] no review received yet") + return state + + state["review"] = received[0] + + return await node_writer(state) + + +# --------------------------------------------------------------------------- +# Graph +# --------------------------------------------------------------------------- + + +def build_graph() -> StateGraph: + g = StateGraph(WorkflowState) + + g.add_node("setup_reviewer", node_reviewer) # creates reviewer mailbox + g.add_node("writer_draft", node_writer) # writes draft, sends to reviewer + g.add_node("reviewer_review", node_reviewer) # reads draft, sends review back + g.add_node("writer_finalise", node_writer_finalise) # reads review, writes final + + g.set_entry_point("setup_reviewer") + g.add_edge("setup_reviewer", "writer_draft") + g.add_edge("writer_draft", "reviewer_review") + g.add_edge("reviewer_review", "writer_finalise") + g.add_edge("writer_finalise", END) + + return g.compile() + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +async def main() -> None: + graph = build_graph() + + initial: WorkflowState = { + "writer_mail_id": "", + "reviewer_mail_id": "", + "draft": "", + "review": "", + "final": "", + } + + result = await graph.ainvoke(initial) + + print("\n" + "=" * 60) + print("DRAFT:\n", result["draft"]) + print("\nREVIEW:\n", result["review"]) + print("\nFINAL:\n", result["final"]) + print("=" * 60) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/docs/python.md b/docs/python.md index d8fe41e..8d91230 100644 --- a/docs/python.md +++ b/docs/python.md @@ -177,4 +177,41 @@ async def run_agent() -> None: if __name__ == "__main__": asyncio.run(run_agent()) + +## LangGraph Integration + +The `langchain-mq9` tools work directly inside LangGraph nodes — each node can create a +mailbox, send messages, and receive replies without any extra wiring. + +See the full working example in [demo/demo-langgraph/](../demo/demo-langgraph/) and the +tool reference in [docs/langchain-mq9.md](langchain-mq9.md). + +```python +from langchain_mq9 import CreateMailboxTool, SendMessageTool, GetMessagesTool +from langgraph.graph import StateGraph, END +from typing import TypedDict + +SERVER = "nats://demo.robustmq.com:4222" + +class State(TypedDict): + mail_id: str + messages: str + +async def node_inbox(state: State) -> State: + create = CreateMailboxTool(server=SERVER) + state["mail_id"] = await create._arun(ttl=120) + return state + +async def node_read(state: State) -> State: + get = GetMessagesTool(server=SERVER) + state["messages"] = await get._arun(mail_id=state["mail_id"], limit=10) + return state + +g = StateGraph(State) +g.add_node("inbox", node_inbox) +g.add_node("read", node_read) +g.set_entry_point("inbox") +g.add_edge("inbox", "read") +g.add_edge("read", END) +graph = g.compile() ```