Blocks in Legion

Blocks are enhanced functions that provide a foundation for building complex workflows in Legion. They offer built-in features for validation, monitoring, and chain integration.

What are Blocks?

Blocks are decorated functions that provide:

  • Input/output validation using Pydantic models
  • Execution monitoring and logging
  • Async compatibility
  • Chain integration
  • Tagging for organization

Creating Blocks

Here's how to create and use blocks:

import logging
from typing import Any
from pydantic import BaseModel, Field
from legion import block, chain

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Define input/output schemas for type safety
class TextInput(BaseModel):
    text: str = Field(description="Input text to process")

class WordCountOutput(BaseModel):
    word_count: int = Field(description="Number of words in text")
    char_count: int = Field(description="Number of characters in text")

# Create a block using decorator syntax
@block(
    input_schema=TextInput,
    output_schema=WordCountOutput,
    tags=["text", "analysis"]
)
def count_words(input_data: TextInput) -> WordCountOutput:
    """Count words and characters in text."""
    text = input_data.text
    words = len(text.split())
    chars = len(text)
    return WordCountOutput(word_count=words, char_count=chars)

Block Features

  1. Schema Validation: Input and output validation using Pydantic
  2. Tagging: Organize blocks by functionality with tags
  3. Monitoring: Track execution and performance
  4. Chain Integration: Easily combine blocks into processing chains

Using Blocks

Blocks can be used individually or as part of chains:

# Using a block individually
input_data = TextInput(text="This is a test sentence.")
result = await count_words(input_data)
print(f"Word count: {result.word_count}, Character count: {result.char_count}")

Using Blocks in Chains

Blocks can be combined into chains for complex workflows:

# Define another block for sentiment analysis
class SentimentOutput(BaseModel):
    sentiment: str = Field(description="Detected sentiment (positive/negative/neutral)")
    confidence: float = Field(description="Confidence score of sentiment")

@block(
    input_schema=TextInput,
    output_schema=SentimentOutput,
    tags=["text", "sentiment", "nlp"]
)
async def analyze_sentiment(input_data: TextInput) -> SentimentOutput:
    """Analyze sentiment of text."""
    # Implementation details...
    return SentimentOutput(sentiment="positive", confidence=0.8)

# Create a chain that combines blocks
@chain
class TextAnalysisChain:
    """A chain that analyzes text by counting words and determining sentiment."""

    # List the blocks in processing order
    members = [
        count_words,
        analyze_sentiment
    ]

# Use the chain
text_analysis_chain = TextAnalysisChain()
chain_result = await text_analysis_chain.aprocess(
    TextInput(text="This is a great example!")
)
print(f"Chain results: {chain_result}")

Best Practices

  1. Define clear input and output schemas
  2. Use descriptive tags for organization
  3. Keep blocks focused on single responsibilities
  4. Use chains to combine blocks into workflows
  5. Leverage async for performance-critical operations
ToolsChains