FlowForge — Multi-Agent Workflow Automation Platform

A drag-and-drop canvas for composing multi-agent AI workflows with structured handoffs, conditional routing, and human-in-the-loop approval checkpoints.

Architecture

• Frontend: React + Vite + React Flow (drag-and-drop canvas)
• Backend: FastAPI + a minimal LangGraph-style workflow runner
• Schemas: Pydantic for all agent inputs/outputs
• Storage: PostgreSQL in real mode, SQLite in mock mode
• Agents: LLM-powered in real mode, deterministic stubs in mock mode

Features

• Canvas UI for connecting agents into a DAG
• Four built-in agent types: summarizer, classifier, extractor, router
• Conditional routing (router agent selects next node based on output)
• Human-in-the-loop approval nodes (workflow pauses, waits for decision)
• Run history with per-step inputs, outputs, and timing
• Mock mode — no API keys needed; agents return deterministic results

Quick Start

./scripts/setup.sh
cp .env.example .env        # MOCK_MODE=true by default

# Backend
cd backend && uvicorn main:app --reload --port 8000

# Frontend (new terminal)
cd frontend && npm install && npm run dev

Open http://localhost:5173. Drag agents onto the canvas, connect them, click Run.

How the workflow engine works

A workflow is a DAG of nodes where each node is an agent with a typed I/O schema. Execution is topological:

1. The runner finds all source nodes (no inputs)
2. For each ready node (all parents executed), invoke its agent
3. Pass the output as input to downstream nodes following edges
4. If the node is router, use its output field next to pick which outgoing edge to follow
5. If the node is human_approval, pause and persist state until approved via API

API

POST /workflows                  — create a workflow (nodes + edges)
POST /workflows/{id}/run         — start an execution
GET  /runs/{run_id}              — get run state (including pending approvals)
POST /runs/{run_id}/approve      — resolve a human-in-the-loop node

Agent schemas

Every agent declares Pydantic Input and Output classes so handoffs are validated at the edge:

class SummarizerInput(BaseModel):
    text: str
    max_words: int = 50

class SummarizerOutput(BaseModel):
    summary: str
    word_count: int

Tests

cd backend && pytest tests/