black-langcube 0.3.0

A LangGraph-based extension framework for complex workflow applications, enabling the integration of various AI models and tools into a cohesive system.

Black LangCube

A LangGraph-based extension framework designed to facilitate the development of complex applications by providing a structured way to define and manage workflows.

🚀 Features

• BaseGraph Framework: Foundational interface for constructing, compiling, and executing stateful workflow graphs
• Data Structures: Pydantic models for scientific article metadata, search strategies, outlines, and more
• LLM Nodes: Pre-built nodes for common language model operations
• Helper Utilities: Token counting, result processing, file management, and workflow utilities
• Subgraph System: Modular subworkflows for translation, output generation, and specialized tasks
• Extensible Architecture: Easy to extend with custom nodes and workflows

📦 Installation

From PyPI (when published):

pip install black_langcube

Development Installation:

git clone https://github.com/cerna-kostka/black-langcube.git
cd black-langcube
pip install -e .

With optional dependencies:

pip install black_langcube[dev,examples]

🏗️ Core Components

BaseGraph

The foundation for building stateful workflow graphs using LangGraph:

from black_langcube.graf.graph_base import BaseGraph, GraphState

class MyCustomGraph(BaseGraph):
    def __init__(self, user_message, folder_name, language):
        super().__init__(MyGraphState, user_message, folder_name, language)
        self.build_graph()
    
    def build_graph(self):
        # Add nodes and edges to your workflow
        self.add_node("my_node", my_node_function)
        self.add_edge(START, "my_node")
        self.add_edge("my_node", END)
    
    @property
    def workflow_name(self):
        return "my_custom_graph"

LLMNode

A base class for defining nodes that interact with language models:

from black_langcube.llm_modules.LLMNodes.LLMNode import LLMNode

class MyCustomNode(LLMNode):
    def generate_messages(self):
        return [
            ("system", "You are a helpful assistant"),
            ("human", self.state.get("user_input", ""))
        ]

    def execute(self, extra_input=None):
        result, tokens = self.run_chain(extra_input)
        return {"output": result, "tokens": tokens}

Data Structures

Pydantic models for structured data handling:

from black_langcube.data_structures.data_structures import Article, Strategies, Outline

# Use pre-defined data structures
article = Article(topic="AI Research", language="English")
strategies = Strategies(strategy1="Search academic papers", strategy2="Analyze trends")

LLM Nodes

Pre-built nodes for language model operations:

from black_langcube.llm_modules.LLMNodes.LLMNode import LLMNode

class MyCustomNode(LLMNode):
    def generate_messages(self):
        return [
            ("system", "You are a helpful assistant"),
            ("human", self.state.get("user_input", ""))
        ]
    
    def execute(self, extra_input=None):
        result, tokens = self.run_chain(extra_input)
        return {"output": result, "tokens": tokens}

📚 Architecture

The library is organized into several key modules:

• graf/: Core graph classes and workflow definitions
• data_structures/: Pydantic models for data validation
• llm_modules/: Language model integration and node definitions
• helper_modules/: Utility functions and helper classes
• messages/: Message formatting and composition utilities
• prompts/: Prompt templates and configurations
• format_instructions/: Output formatting utilities

🛠️ Usage Examples

Basic Workflow

from black_langcube.graf.graph_base import BaseGraph, GraphState
from langgraph.graph import START, END

class SimpleWorkflow(BaseGraph):
    def __init__(self, message, folder, language):
        super().__init__(GraphState, message, folder, language)
        self.build_graph()
    
    def build_graph(self):
        def process_message(state):
            return {"result": f"Processed: {state['messages'][-1].content}"}
        
        self.add_node("process", process_message)
        self.add_edge(START, "process")
        self.add_edge("process", END)
    
    @property
    def workflow_name(self):
        return "simple_workflow"

# Usage
workflow = SimpleWorkflow("Hello, world!", "output", "English")
result = workflow.run()

Using Subgraphs

from black_langcube.graf.subgrafs.translator_en_subgraf import TranslatorEnSubgraf

# Translation subgraph
translator = TranslatorEnSubgraf(config, subfolder="translations")
result = translator.run(extra_input={
    "translation_input": "Bonjour le monde",
    "language": "French"
})

🔧 Configuration

The library uses environment variables for configuration. Create a .env file:

OPENAI_API_KEY=your_openai_api_key_here

# optional: LangChain configuration
LANGCHAIN_API_KEY=your_langchain_api_key_here
LANGCHAIN_TRACING_V2=true

📖 Examples

See the examples/ directory for complete working examples:

• Basic Graph: Simple workflow with custom nodes
• Translation Pipeline: Multi-language processing workflow
• Scientific Article Processing: Complex multi-step analysis pipeline
• Custom Data Structures: Extending the framework with your own models

🧪 Development

Setting up development environment:

git clone https://github.com/cerna-kostka/black-langcube.git
cd black-langcube
pip install -e .[dev]

Running tests:

pytest

Code formatting:

black .
isort .

Parallel Fan-Out (Scatter-Gather)

BaseGraph exposes add_parallel_nodes for wiring an intra-graph fan-out: a single node dispatches to multiple branches that run concurrently (via LangGraph's Send API), and a merge node aggregates their results.

State setup

Use operator.add (or any reducer) with Annotated so that concurrent branches can each append to the same list field without overwriting each other:

import operator
from typing import Annotated
from black_langcube.graf.graph_base import GraphState

class FanOutState(GraphState):
    topic: str
    branch_results: Annotated[list, operator.add]  # reducer – each branch appends
    merged_summary: str

Graph wiring

from langgraph.graph import START, END
from black_langcube.graf.graph_base import BaseGraph

class MyFanOutGraph(BaseGraph):
    def __init__(self, topic, folder, language="English"):
        super().__init__(FanOutState, topic, folder, language)
        self._topic = topic
        self._build()

    def _build(self):
        def prepare(state):
            return {}                                   # fan-out source

        def branch_a(state):
            return {"branch_results": [f"A: {state['topic']}"]}

        def branch_b(state):
            return {"branch_results": [f"B: {state['topic']}"]}

        def merge(state):
            return {"merged_summary": " | ".join(state["branch_results"])}

        self.add_node("prepare", prepare)
        self.add_node("branch_a", branch_a)
        self.add_node("branch_b", branch_b)
        self.add_node("merge", merge)

        self.add_edge(START, "prepare")
        # Wire fan-out → concurrent branches → merge
        self.add_parallel_nodes("prepare", ["branch_a", "branch_b"], "merge")
        self.add_edge("merge", END)

    @property
    def workflow_name(self):
        return "my_fanout"

A custom router_fn can be supplied to control what state each branch receives:

from langgraph.types import Send

def router(state):
    return [
        Send("branch_a", {**state, "mode": "fast"}),
        Send("branch_b", {**state, "mode": "thorough"}),
    ]

self.add_parallel_nodes("prepare", ["branch_a", "branch_b"], "merge", router_fn=router)

Pipeline-level parallelism

To run independent graph instances simultaneously, use run_parallel_pipeline:

import asyncio
from black_langcube import run_parallel_pipeline

graph_a = MyFanOutGraph("topic A", "output/a")
graph_b = MyFanOutGraph("topic B", "output/b")

results = asyncio.run(run_parallel_pipeline([graph_a, graph_b]))
# results["status"]           → "completed" | "partial_failure"
# results["parallel_results"] → [result_a, result_b]

See src/black_langcube/examples/parallel_fanout_workflow.py for a fully working end-to-end example.

• Python 3.9+
• LangChain >= 0.3.24
• LangGraph >= 0.3.7
• Pydantic >= 2.0.0
• OpenAI API access

🤝 Contributing

This is a work in progress and contributions are welcome! Please feel free to:

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests if applicable
5. Submit a pull request

📄 License

MIT License (MIT)

⚠️ Note

This library is intended to be used within a larger application context. The code is provided as-is and is actively being improved. Take it with a grain of salt and feel free to contribute improvements!

🔗 Links

• LangGraph Documentation
• LangChain Documentation
• Examples and Tutorials