Flock – Our new open-source AI Agent Framework

Flock is a framework for orchestrating LLM-powered agents. It leverages a declarative approach where you simply specify what each agent needs as input and what it produces as output—without having to write lengthy, brittle prompts. Under the hood, Flock transforms these declarations into robust workflows, using cutting-edge components such as Temporal and DSPy to handle fault tolerance, state management, and error recovery.

We chose „Flock“ (English for „a group of birds“) because it beautifully encapsulates the essence of AI agent orchestration, much like how a group of birds seamlessly collaborates in flight. We decided on ducks because we love them and they perfectly symbolize the unity and coordination we aim to achieve.

Why we built Flock

We embarked on Flock’s journey some time ago, driven by the need for a more capable and flexible AI agent framework. Existing solutions on the market did not fully meet our requirements for comprehensive AI agent development. Faced with these constraints, we decided to build something new that aligns perfectly with our aspirations. Today, while other AI agent frameworks are evolving, we still see room for improvement.

A Different Approach: Declarative AI Agents

Flock takes a fundamentally different approach. Instead of fighting against the non-deterministic nature of LLMs, we’ve built a framework that embraces it while providing the engineering guardrails necessary for production systems.

At the core of Flock is the FlockAgent, a declarative base class that represents a radical departure from traditional prompt-based or state machine-based approaches:

agent = FlockAgent(
    name="content_analyzer",
    input="url: str | The webpage to analyze",
    output="title: str, headings: list[str], type: Literal['news', 'blog']",
    tools=[web_content_scraper],
    use_cache=True
)

This declarative approach provides several immediate benefits:

• Automatic Caching: Expensive operations (like web scraping) can be cached automatically
• Type Safety: By leveraging Python’s type hints and Pydantic, we get automatic validation at system boundaries
• Clear Contracts: Input and output specifications serve both as documentation and runtime constraints
• Tool Integration: External capabilities can be cleanly integrated without breaking the AI agent paradigm

The Power of Callables

One of Flock’s key innovations is making almost every aspect of an agent configurable through callables. From the implementation:

class FlockAgent(BaseModel, ABC):
    description: str | Callable[..., str] = Field(
        "", description="A human-readable description of the agent."
    )
    input: str | Callable[..., str] | None = Field(
        None,
        description="A comma-separated list of input keys..."
    )
    output: str | Callable[..., str] | None = Field(
        None,
        description="A comma-separated list of output keys..."
    )

This isn’t just flexibility for flexibility’s sake. It enables:

• Dynamic runtime adaptation based on context
• Clean handling of edge cases
• Complex logic that remains testable
• Clear separation of concerns

Platform out of the box

Looking at Flock’s implementation reveals a serious commitment to providing a complete platform for managing agents:

Lifecycle Management

Each agent has clear lifecycle hooks for initialization, evaluation, termination, and error handling. This makes it easy to manage resources, implement logging, and handle failures gracefully:

async def run(self, inputs: dict[str, Any]) -> dict[str, Any]:
    try:
        await self.initialize(inputs)
        result = await self.evaluate(inputs)
        await self.terminate(inputs, result)
        return result
    except Exception as run_error:
        await self.on_error(run_error, inputs)
        raise

Observability

Built-in OpenTelemetry integration provides production-grade observability out of the box:

with tracer.start_as_current_span("agent.evaluate") as span:
    span.set_attribute("agent.name", self.name)
    span.set_attribute("inputs", str(inputs))

Temporal Support

Flock includes native support for Temporal workflows enabling robust distributed execution:

async def run_temporal(self, inputs: dict[str, Any]) -> dict[str, Any]:
    client = await Client.connect("localhost:7233", namespace="default")
    agent_data = self.to_dict()
    return await run_activity(
        client,
        self.name,
        run_flock_agent_activity,
        {"agent_data": agent_data, "inputs": inputs_data},
    )

Serialization

Robust serialization support ensures agents can be safely persisted and transmitted:

def to_dict(self) -> dict[str, Any]:
    def convert_callable(obj: Any) -> Any:
        if callable(obj) and not isinstance(obj, type):
            return cloudpickle.dumps(obj).hex()
        # ... recursive handling of complex structures

Get started & join the flock!

Getting started with Flock is easy! Visit our GitHub Repository for detailed setup instructions, quick start guides, and a variety of examples for different use cases. Whether you’re new to AI agents or an experienced pro, our resources will help you build, run, and manage your AI agents effortlessly.

Join the flock and be part of something special! We created this project out of a passion for making AI agent development accessible and effective for all. As an open-source initiative, Flock embodies our belief in collaboration and community-driven progress. We invite you to contribute, offer feedback, and share your ideas. Your input can help shape the future of Flock. While we’re still in the early stages of development, we have exciting plans. The possibilities are endless, and the future is bright. Come, be a part of our growing community, and help us take Flock to new heights!