bricks-and-graphs 0.1.1
An collaborative agentic framework for designing dynamic AI agents that can work together to resolve complex problems.
Requires Python: >=3.12

# Bricks and Graphs

A collaborative agentic framework for designing dynamic AI agents that can work together to resolve complex problems.

[![PyPI version](https://badge.fury.io/py/bricks-and-graphs.svg)](https://badge.fury.io/py/bricks-and-graphs)
[![CI](https://github.com/wheredatalives/bricks-and-graphs/workflows/CI/badge.svg)](https://github.com/wheredatalives/bricks-and-graphs/actions)
[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/)
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

## Features

- **Library**: Core framework for building agentic decision graphs
- **Built-in Bricks**: Ready-to-use components for prompts, processing, and routing
- **Multi-format Support**: Handle JSON, YAML, Arrow, Pandas, Polars, and LiteLLM responses
- **CLI**: Command-line interface for running agentic graphs
- **API**: FastAPI-based web service for graph execution

## Installation

### From PyPI (Recommended)

```bash
# Install the latest stable version
pip install bricks-and-graphs

# Or with uv
uv add bricks-and-graphs
```

### From Source (Development)

```bash
# Clone the repository
git clone https://github.com/wheredatalives/bricks-and-graphs.git
cd bricks-and-graphs

# Install in development mode
uv sync --extra dev

# Install for production
uv pip install -e .
```

## Usage

### CLI
```bash
# Run with config file
bag --config config.yaml

# Get help
bag --help
```

### API Server
```bash
# Start the FastAPI server
bag-server

# Or run directly
uvicorn bag.api.main:app --reload
```

### Docker

#### Build and run locally
```bash
# Build the image
docker build -t bag:latest .

# Run the container
docker run -p 8000:8000 bag:latest

# Or use docker-compose
docker-compose up --build
```

#### Use pre-built image from GitHub Container Registry
```bash
# Pull and run the latest image
docker run -p 8000:8000 ghcr.io/wheredatalives/bricks-and-graphs:latest
```

### Library

#### Core Framework

**AgentContext** - Shared state across graph execution:
```python
from bag.core import AgentGraph, AgentBrick, BrickType, ExecutionContext

class StatefulBrick(AgentBrick):
    @property
    def brick_type(self) -> BrickType:
        return BrickType.PROCESSOR

    async def execute(self, context: ExecutionContext):
        # Access shared agent context with dict-style interface
        counter = context.agent_context.get("counter", 0)
        context.agent_context["counter"] = counter + 1

        # Store results for other nodes
        context.agent_context["results"] = {"value": counter}
        return {"processed": counter}

# Agent context persists across nodes and executions
graph = AgentGraph()
# ... add nodes ...
await graph.execute()  # First run
await graph.execute()  # Second run - context preserved
```

**LiteLLM Integration** - Built-in LLM support for all nodes:
```python
from bag.core import AgentGraph, GraphConfig, LiteLLMConfig, LiteLLMModelConfig

# Configure LLM models
litellm_config = LiteLLMConfig(
    models=[
        LiteLLMModelConfig(
            model="gpt-4",
            api_key="${OPENAI_API_KEY}",
            temperature=0.7,
            max_tokens=4096
        ),
        LiteLLMModelConfig(
            model="claude-3-5-sonnet-20241022",
            api_key="${ANTHROPIC_API_KEY}",
            temperature=0.7,
            max_tokens=8192
        )
    ],
    default_model="claude-3-5-sonnet-20241022",
    enable_fallback=True,
    fallback_order=["claude-3-5-sonnet-20241022", "gpt-4"]
)

# Create graph with LiteLLM
graph_config = GraphConfig(litellm_config=litellm_config)
graph = AgentGraph(config=graph_config)

# Use in bricks
class LLMBrick(AgentBrick):
    async def execute(self, context: ExecutionContext):
        # LiteLLM manager is available in context
        response = await context.litellm_manager.complete([
            {"role": "user", "content": "Hello, AI!"}
        ])
        return {"response": response.choices[0].message.content}
```

See `examples/litellm/` for complete configuration examples.

**AgentGraph** - Create and execute agent graphs:
```python
from bag.core import AgentGraph, AgentNode, AgentBrick

# Create a simple graph
graph = AgentGraph(name="MyGraph")
node = AgentNode(name="ProcessingNode")
graph.add_node(node)
```

#### Built-in Bricks

**PromptBrick** - Text and template management:
```python
from bag.bricks import PromptBrick

# Static prompt
system_prompt = PromptBrick(
    content="You are a helpful AI assistant.",
    role="system"
)

# Template-based prompt
user_prompt = PromptBrick(
    template="Analyze the following {data_type}: {content}",
    role="user",
    variables={"data_type": "JSON"}
)
```

**ProcessorBrick** - Multi-format data processing:
```python
from bag.bricks import (
    LiteLLMResponseProcessor,
    DataTransformProcessor,
    DataFormat
)

# Process LLM responses
llm_processor = LiteLLMResponseProcessor(
    extract_content=True,
    parse_json=True
)

# Transform between formats
transformer = DataTransformProcessor(
    input_format=DataFormat.JSON,
    output_format=DataFormat.PANDAS
)
```

**RouterBrick** - Intelligent routing decisions:
```python
from bag.bricks import ConditionalRouter, DataFieldRouter

# Route based on conditions
router = ConditionalRouter(
    conditions={
        "has_error": lambda d: "error" in d,
        "needs_retry": lambda d: d.get("attempts", 0) < 3
    },
    routes={
        "has_error": "error_handler",
        "needs_retry": "retry_node"
    },
    default_route="success_node"
)

# Route based on data fields
field_router = DataFieldRouter(
    field="action",
    field_routes={
        "create": "create_handler",
        "update": "update_handler",
        "delete": "delete_handler"
    }
)
```

#### Node Execution Flow

AgentNode provides two execution methods:

1. **`execute()`** - Runs all bricks in sequence (original behavior)
2. **`run()`** - Orchestrated execution with specific flow:
   - Collects and joins all PromptBricks into messages
   - Executes LLM completion if prompts exist (using LiteLLM manager)
   - Passes LLM response through ProcessorBricks in order
   - Routes final output through RouterBrick (if present)

```python
# Example of orchestrated node execution
node = AgentNode(node_id="analyzer")

# Add bricks in any order - run() handles the flow
node.add_brick(SystemPromptBrick("You are a helpful assistant"))
node.add_brick(UserPromptBrick("Analyze this text: {input}"))
node.add_brick(SentimentAnalyzerBrick())  # Processor
node.add_brick(SummarizerBrick())         # Another processor
node.add_brick(ContentRouterBrick())      # Router

# The run() method will:
# 1. Combine system + user prompts → ["system: ...", "user: ..."]
# 2. Call LLM with combined prompts → LLM response
# 3. Pass LLM response to SentimentAnalyzer → sentiment result
# 4. Pass sentiment result to Summarizer → summary
# 5. Pass summary to Router for decision → next node

routing_decision = await node.run(context)
```

See `examples/node_execution/` for complete working examples.

For detailed API documentation, see:
- [Core Framework API](src/bag/core/)
- [Built-in Bricks API](src/bag/bricks/)

## Documentation

- [Architecture Guide](docs/architecture.md) - System design and components
- [Configuration Guide](docs/configuration.md) - Complete configuration reference
- [Pipeline Configuration](docs/pipeline_configuration.md) - Building AI agent pipelines

## Examples

Explore working examples organized by topic:

- **[Basic](examples/basic/)** - Simple graphs and custom bricks
- **[LiteLLM Integration](examples/litellm/)** - Multi-model LLM configuration
- **[Node Execution](examples/node_execution/)** - Orchestrated execution flows
- **[Graph Configuration](examples/graph_configuration/)** - YAML-based graph definitions

## Development

### Setup
```bash
# Install dependencies
uv sync --dev

# Install pre-commit hooks
uv run pre-commit install
```

### Testing (MANDATORY - ALL CODE MUST BE TESTED)
```bash
# Run tests - REQUIRED before any commits
uv run pytest

# Run tests with coverage - MUST achieve ≥85% coverage
uv run pytest --cov=src --cov-report=html --cov-fail-under=85

# Quick test run during development
uv run pytest -q

# Test with branch coverage - REQUIRED for comprehensive testing
uv run pytest --cov=src --cov-branch --cov-report=term-missing
```

**🚨 CRITICAL**: This project enforces mandatory testing. Every function, class, and module MUST have corresponding tests. Code without tests will be rejected.

### Code Quality
```bash
# Format code
uv run black .
uv run isort .

# Lint code
uv run ruff check .

# Type checking
uv run mypy src

# Run all pre-commit hooks
uv run pre-commit run --all-files
```

### CI/CD

The project uses GitHub Actions for continuous integration and deployment:

- **Automated Testing**: Runs on every push and pull request
- **Code Quality**: Pre-commit hooks ensure consistent formatting
- **Coverage Enforcement**: Minimum 85% code coverage required
- **Docker Builds**: Multi-platform images built and pushed to GitHub Container Registry
- **Workflow**: Single unified CI workflow that runs tests first, then builds Docker images only if tests pass

## Project Structure

```
src/bag/
├── __init__.py          # Package initialization
├── core/                # Core framework components
│   ├── brick.py        # AgentBrick base class
│   ├── node.py         # AgentNode implementation
│   ├── graph.py        # AgentGraph with NetworkX
│   ├── types.py        # Type definitions
│   └── config_loader.py # YAML/JSON configuration
├── bricks/              # Built-in brick implementations
│   ├── prompt.py       # PromptBrick for text management
│   ├── processor.py    # ProcessorBrick for data handling
│   ├── router.py       # RouterBrick for graph routing
│   └── types.py        # Brick-specific types
├── cli/                 # Command-line interface
│   └── main.py         # CLI entry point
└── api/                 # FastAPI web service
    └── main.py         # API entry point

tests/                   # Test suite
├── core/               # Core framework tests
│   ├── test_brick.py   # AgentBrick tests
│   ├── test_node.py    # AgentNode tests
│   ├── test_graph.py   # AgentGraph tests
│   └── test_types.py   # Type definition tests
├── test_api.py         # API tests
└── test_cli.py         # CLI tests
```

## License

MIT License - see LICENSE file for details.