Examples¶

Learn by example! Explore working code samples demonstrating Flock's features and patterns.

🎯 Example Categories¶

Core Examples¶

Production-ready demonstrations of core features

Explore the numbered example folders in this repository:

01 — The Declarative Way (examples/01-the-declarative-way) — minimal and focused
02 — The Blackboard (examples/02-the-blackboard) — architecture overview
03 — The Dashboard (examples/03-the-dashboard) — real-time monitoring

Component Examples¶

Learn to build custom components and engines

05 — Custom Engines (examples/05-engines) — deterministic logic engines
06 — Agent Components (examples/06-agent-components) — per-agent behavior patterns
07 — Orchestrator Components (examples/07-orchestrator-components) — global coordination patterns

These examples show how to extend Flock with custom logic:

Custom Engines - Replace LLM calls with deterministic logic: - emoji_mood_engine.py - Pattern-based mood detection - potion_batch_engine.py - Batch processing rules

Agent Components - Add per-agent behavior: - cheer_meter_component.py - Track agent-specific metrics - plot_twist_component.py - Dynamic agent state

Orchestrator Components - Global coordination: - quest_tracker_component.py - Real-time quest scoring system - kitchen_monitor_component.py - Restaurant performance monitoring

Feature Examples¶

Focused examples for specific capabilities

Feature-focused examples are integrated into the folders above (e.g., dashboard edge cases). Additional feature demos may be added over time.

Dashboard Examples¶

Interactive dashboard demonstrations

Check out examples/03-the-dashboard to explore:

Declarative Pizza - Single-agent dashboard demo
Edge Cases - Multi-agent cascades and filtering
Real-time Updates - WebSocket streaming

Claude's Workshop¶

🎓 Interactive learning course from beginner to expert

Complete hands-on workshop: examples/03-claudes-workshop

Beginner Track: - Lesson 01: Code Detective - Your first agent - Lesson 02: Band Formation - Multi-agent chaining

Intermediate Track: - Lesson 03: Quality Gates - Conditional consumption - Lesson 04: Debate Club - Feedback loops - Lesson 05: Debugging Detective - Unified tracing

Advanced Track: - Lesson 06: Secret Agent Network - Visibility controls - Lesson 07: News Agency - Parallel execution

Expert Track: - Lesson 08: The Matchmaker - JoinSpec correlation - Lesson 09: Batch Optimizer - BatchSpec patterns - Lesson 10: Smart Factory - Combined features

Architecture Track: - Lesson 11: Performance Monitor - Orchestrator components - Lesson 12: Confidence Booster - Agent components - Lesson 13: Regex Matcher - Custom engines

Each lesson is self-contained with detailed comments and runnable code!

🚀 Running Examples¶

Prerequisites¶

# Install Flock with all features
pip install "flock-core[all]"

# Set your API key
export OPENAI_API_KEY="sk-..."
export DEFAULT_MODEL="openai/gpt-4.1"

Run an Example¶

# Clone the repository
git clone https://github.com/whiteducksoftware/flock.git
cd flock

# Run a minimal example
python examples/01-the-declarative-way/01_declarative_pizza.py

# Run with dashboard
python examples/03-the-dashboard/01_declarative_pizza.py

📚 Example Highlights¶

1. Hello Flock (Minimal)¶

What it demonstrates: Basic agent creation and execution

from flock import Flock, flock_type
from pydantic import BaseModel

@flock_type
class Greeting(BaseModel):
    name: str

@flock_type
class Response(BaseModel):
    message: str

flock = Flock("openai/gpt-4.1")

greeter = (
    flock.agent("greeter")
    .consumes(Greeting)
    .publishes(Response)
)

async def main():
    await flock.invoke(
        greeter,
        Greeting(name="World")
    )

asyncio.run(main())

Run it locally: python examples/01-the-declarative-way/01_declarative_pizza.py

2. Dashboard Edge Cases¶

What it demonstrates: Agent cascades, filtering, and real-time updates

Run: python examples/03-the-dashboard/02-dashboard-edge-cases.py

3. Dashboard Demo¶

What it demonstrates: Real-time agent monitoring

Features: - Live agent status updates - Artifact flow visualization - Streaming LLM outputs - WebSocket communication

await orchestrator.serve(
    dashboard=True,
    port=8344
)

Run: python examples/03-the-dashboard/01_declarative_pizza.py

4. Feedback Prevention¶

What it demonstrates: Preventing agent feedback loops using built-in safeguards

Default safeguard: prevent_self_trigger=True (agents won’t re-trigger on their own outputs)
Use label/tenant visibility for finer control (see Visibility guide)

5. Visibility Controls¶

What it demonstrates: Access control patterns

Examples for: - Public - All agents can access - Private - Only producing agent can access - Tenant - Multi-tenant isolation - Label - Fine-grained control - Time - Temporal constraints

See: Docs → User Guides → Visibility Controls

💡 Example Patterns¶

Pattern: Parallel Batching¶

# Publish multiple items
for item in items:
    await flock.publish(item)

# Process all in parallel
await flock.run_until_idle()

Used in: Blog review, pizza ordering

Pattern: Conditional Consumption¶

agent.consumes(
    Review,
    where=lambda r: r.score >= 9  # Only high scores
)

Used in: Dashboard edge cases

Pattern: Multi-Type Agents¶

agent.consumes(
    [RequestType, FeedbackType],  # Multiple inputs
    where=lambda x: x.user_id == current_user
)

Used in: Feedback prevention

Pattern: Unified Tracing¶

async with flock.traced_run("workflow_name"):
    await flock.publish(input_data)
    await flock.run_until_idle()

Used in: Most showcase examples

🎓 Learning Path¶

New to Flock? We recommend this order:

Hello Flock - Understand basics
Pizza Ordering - Learn type contracts
Blog Review - Master multi-agent flows
Dashboard Demo - Add observability
Feedback Prevention - Handle edge cases

🔧 Customization Examples¶

Custom Component¶

from flock import Component, Agent

class LoggingComponent(Component):
    async def on_pre_consume(self, agent: Agent, artifacts):
        print(f"Agent {agent.name} processing {len(artifacts)} artifacts")

agent.add_component(LoggingComponent())

See full example →

Custom Engine¶

from flock import Engine

class CustomEngine(Engine):
    async def evaluate(self, agent: Agent, artifacts):
        # Your custom evaluation logic
        return CustomOutput(...)

flock.engine = CustomEngine()