LangGraph Cross-Chain Communication Framework

A Python package extending LangGraph to enable cross-chain node communication - allowing nodes in different chains to call and communicate with each other directly.

🎯 Project Overview

This framework addresses a critical gap in current AI agent frameworks by enabling:

• Cross-Chain Node Calls: Direct communication between nodes in different chains (Chain1.Node2 → Chain2.Node3)
• Shared State Management: Share state between separate chain instances
• Dynamic Inter-Chain Workflows: Create flexible workflows that span multiple chains
• Modular Components: Build reusable chain components that can communicate seamlessly

💡 Innovation Gap Being Addressed

🚀 Quick Start

Installation

pip install langgraph-crosschain

Or install from source:

git clone https://github.com/yourusername/langgraph-crosschain.git
cd langgraph-crosschain
pip install -e .

Basic Example

from langgraph.graph import StateGraph
from langgraph_crosschain import ChainRegistry, CrossChainNode

# Create a registry to manage chains
registry = ChainRegistry()

# Define your chains
def analyzer_node(state):
    # Process data
    return {"analyzed": True, "result": state.get("data")}

def processor_node(state):
    # Call a node in another chain
    node = CrossChainNode("chain2", "processor", processor_node)
    result = node.call_remote("chain1", "analyzer", {"data": "test"})
    return {"processed": True, "result": result}

# Build your chains
chain1 = StateGraph(...)
chain2 = StateGraph(...)

# Register chains
registry.register("chain1", chain1.compile())
registry.register("chain2", chain2.compile())

# Now chains can communicate with each other!

📚 Core Components

ChainRegistry

Central registry for managing multiple chain instances:

from langgraph_crosschain import ChainRegistry

registry = ChainRegistry()
registry.register("my_chain", chain_instance)
chain = registry.get("my_chain")

CrossChainNode

Base class for nodes that can communicate across chains:

from langgraph_crosschain import CrossChainNode

node = CrossChainNode(
    chain_id="chain1",
    node_id="processor",
    func=my_node_function
)

# Call a node in another chain
result = node.call_remote("chain2", "analyzer", {"data": "test"})

# Broadcast to multiple chains
node.broadcast(["chain2", "chain3"], "receiver", {"data": "broadcast"})

MessageRouter

Handles routing of messages between chains:

from langgraph_crosschain import MessageRouter

router = MessageRouter()
# Messages are automatically routed by CrossChainNode

SharedStateManager

Manages shared state across multiple chains:

from langgraph_crosschain import SharedStateManager

state_manager = SharedStateManager()
state_manager.set("shared_data", {"key": "value"})
data = state_manager.get("shared_data")

# Subscribe to state changes
def on_change(value):
    print(f"State changed: {value}")

state_manager.subscribe("shared_data", on_change)

🏗️ Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Chain Registry                           │
│              (Manages all chain instances)                   │
└───────────────────┬─────────────────────────────────────────┘
                    │
        ┌───────────┴───────────┐
        │                       │
┌───────▼────────┐     ┌────────▼──────┐
│    Chain 1     │     │    Chain 2    │
│  ┌──────────┐  │     │  ┌──────────┐ │
│  │  Node A  │  │────►│  │  Node X  │ │
│  └──────────┘  │     │  └──────────┘ │
│  ┌──────────┐  │     │  ┌──────────┐ │
│  │  Node B  │  │◄────│  │  Node Y  │ │
│  └──────────┘  │     │  └──────────┘ │
└────────────────┘     └───────────────┘
        │                       │
        └───────────┬───────────┘
                    │
        ┌───────────▼────────────┐
        │   Message Router       │
        │  (Routes messages)     │
        └────────────────────────┘
                    │
        ┌───────────▼────────────┐
        │ Shared State Manager   │
        │  (Shared state store)  │
        └────────────────────────┘

📖 Use Cases

1. Multi-Agent Systems

Create specialized chains for different tasks that collaborate:

# Analytics chain
analytics_chain = build_analytics_chain()
registry.register("analytics", analytics_chain)

# Execution chain
execution_chain = build_execution_chain()
registry.register("execution", execution_chain)

# Chains can now call each other as needed

2. Modular Workflows

Break complex workflows into reusable, communicating components:

# Data ingestion chain
ingestion = build_ingestion_chain()

# Processing chain (calls ingestion)
processing = build_processing_chain()

# Output chain (calls processing)
output = build_output_chain()

3. Distributed Processing

Distribute workload across multiple specialized chains:

# Master chain coordinates work
master_node.broadcast(
    ["worker1", "worker2", "worker3"],
    "process_task",
    {"task": task_data}
)

🧪 Testing

Run the test suite:

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=langgraph_crosschain --cov-report=html

🔧 Development

Setup Development Environment

# Clone the repository
git clone https://github.com/yourusername/langgraph-crosschain.git
cd langgraph-crosschain

# Install in editable mode with dev dependencies
pip install -e ".[dev]"

# Install pre-commit hooks
pre-commit install

Code Quality

# Format code
black langgraph_crosschain tests

# Lint code
ruff check langgraph_crosschain tests

# Type checking
mypy langgraph_crosschain

📝 Examples

Check out the examples directory for complete working examples:

• basic_communication.py - Simple cross-chain communication
• shared_state.py - Using shared state between chains
• multi_agent_system.py - Building a multi-agent system
• distributed_workflow.py - Distributed processing example

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

• Built on top of LangGraph
• Inspired by the need for more modular AI agent architectures
• Thanks to all contributors!

🗺️ Roadmap

• Core cross-chain communication
• Shared state management
• Message routing
• Async/await support
• Distributed execution
• Event streaming
• Performance monitoring
• Web UI for visualization
• More examples and tutorials

---

Star ⭐ this repo if you find it useful!