AgentFlow

Framework Python open-core pour orchestrer des agents IA spécialisés via un système Kanban event-driven.

Pas de LangChain. Pas de CrewAI. Un Board Kanban comme machine à états, un Event Bus qui découple tout, et des agents qui exécutent — le tout testable sans LLM.

Fonctionnalités

• 5 providers LLM : Claude, OpenAI, Gemini, Mistral, Ollama (local/offline)
• Orchestration Kanban : flux tiré, WIP limits, event bus pub/sub
• Policies pluggables : MaxCost, Timeout, Deadlock, MaxBlocked
• Parallelisme : builders en parallèle via asyncio.gather
• Observabilité : logs structurés JSON, 9 métriques Kanban, cost tracking
• CLI riche : commandes run, report, list, status — affichage rich ou texte
• HTTP entrypoint : API REST FastAPI (optionnel)
• LLM-agnostique : zéro dépendance structurante (pas de SDK LLM)

Installation

# Minimal
pip install agentflow

# Avec affichage CLI enrichi (tableaux colorés)
pip install "agentflow[rich]"

# Avec entrypoint HTTP
pip install "agentflow[http]"

# Tout inclus
pip install "agentflow[all]"

# Depuis les sources
git clone https://github.com/ppaino/agentflow
cd agentflow
pip install -e ".[dev]"

Démarrage rapide

CLI

# Claude (défaut)
export ANTHROPIC_API_KEY=sk-ant-...
agentflow run "Créer une API REST CRUD pour gérer des produits"

# Gemini
export GOOGLE_API_KEY=AIza...
agentflow run "Implémenter un système de cache Redis" --provider gemini --model gemini-1.5-flash

# Mistral
export MISTRAL_API_KEY=...
agentflow run "Refactoriser le module auth" --provider mistral --model mistral-small-latest

# Ollama (local, zéro API key)
# Prérequis : ollama serve && ollama pull llama3
agentflow run "Écrire des tests unitaires pour auth.py" --provider ollama --model llama3

# Dry-run (décomposition uniquement, sans exécution)
agentflow run "Construire un pipeline CI/CD" --dry

# Avec budget et parallélisme
agentflow run "Migrer la base de données" --max-cost 1.50 --parallel

# Lister les sessions passées
agentflow list

# Voir le statut d'une session
agentflow status ./agentflow_sessions/abc123.json

# Rapport complet
agentflow report ./agentflow_sessions/abc123.json

Python API

import asyncio
from agentflow.agents.architect import ArchitectAgent
from agentflow.agents.builder import BuilderAgent
from agentflow.agents.registry import AgentRegistry
from agentflow.agents.reviewer import ReviewerAgent
from agentflow.core.board import Board, BoardConfig, WIPLimits
from agentflow.core.events import EventBus
from agentflow.core.policies import MaxCostPolicy, PolicyEngine, TimeoutPolicy
from agentflow.llm.claude import ClaudeProvider
from agentflow.orchestrator import Orchestrator

async def main():
    llm = ClaudeProvider(api_key="sk-ant-...", model="claude-haiku-4-5-20251001")

    # 1. Décomposer le brief en cards
    architect = ArchitectAgent("architect-01", llm)
    cards = await architect.decompose("Créer une API REST pour gérer des utilisateurs")

    # 2. Créer le board Kanban
    board = Board(
        brief="Créer une API REST pour gérer des utilisateurs",
        config=BoardConfig(wip_limits=WIPLimits(in_progress=3, review=2)),
    )
    for card in cards:
        board.add_card(card)

    # 3. Monter les agents
    registry = AgentRegistry()
    registry.register(BuilderAgent("builder-01", llm))
    registry.register(ReviewerAgent("reviewer-01", llm))

    # 4. Policies de sécurité
    engine = PolicyEngine()
    engine.add_rule(MaxCostPolicy(max_cost_usd=2.0))
    engine.add_rule(TimeoutPolicy(max_duration_seconds=300))

    # 5. Orchestrer
    orch = Orchestrator(board, registry, EventBus(), policies=engine, parallel=True)
    final_board = await orch.run()

    print(f"Done: {final_board.metrics['cards_done']}/{final_board.metrics['cards_total']}")

asyncio.run(main())

Changer de provider

# OpenAI
from agentflow.llm.openai import OpenAIProvider
llm = OpenAIProvider(api_key="sk-...", model="gpt-4o-mini")

# Gemini
from agentflow.llm.gemini import GeminiProvider
llm = GeminiProvider(api_key="AIza...", model="gemini-1.5-flash")

# Mistral
from agentflow.llm.mistral import MistralProvider
llm = MistralProvider(api_key="...", model="mistral-small-latest")

# Ollama (local)
from agentflow.llm.ollama import OllamaProvider
llm = OllamaProvider(model="llama3")

# Ollama distant
llm = OllamaProvider(model="mistral", host="http://192.168.1.10:11434")

Configuration via variables d'environnement

# Clés API
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...
GOOGLE_API_KEY=AIza...
MISTRAL_API_KEY=...
OLLAMA_HOST=http://localhost:11434   # optionnel

# Session
AGENTFLOW_PROVIDER=gemini            # claude | openai | gemini | mistral | ollama
AGENTFLOW_MODEL=gemini-1.5-flash
AGENTFLOW_MAX_COST_USD=5.0
AGENTFLOW_TIMEOUT=600
AGENTFLOW_PARALLEL=true              # builders en parallèle

# Kanban
AGENTFLOW_WIP_PROGRESS=3
AGENTFLOW_WIP_REVIEW=2
AGENTFLOW_MAX_RETRIES=3

# Output
AGENTFLOW_OUTPUT_DIR=./sessions
AGENTFLOW_LOG_LEVEL=INFO
AGENTFLOW_LOG_JSON=true

Entrypoint HTTP (FastAPI)

pip install "agentflow[http]"
uvicorn agentflow.http.app:app --reload

Endpoints disponibles :

# Lancer une session via HTTP
curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{"brief": "Créer une API REST", "provider": "ollama", "model": "llama3"}'

Architecture

Entry Points (CLI, Python API, HTTP)
    ↓
Orchestrateur Kanban (Python pur — ZÉRO appel LLM)
  → Board (machine à états) + Event Bus (pub/sub) + Policy Engine
    ↓
Agent Layer
  → Architect (décompose) + Builder×N (exécute) + Reviewer (valide)
    ↓
LLM Abstraction
  → Claude | OpenAI | Gemini | Mistral | Ollama
    ↓
Telemetry → Structured logs, métriques Kanban, cost tracking

Transitions d'état Kanban

BACKLOG → IN_PROGRESS → REVIEW → DONE
               ↓           ↓
            BLOCKED     IN_PROGRESS (retry)
               ↓
            BACKLOG (unblock)

Métriques Kanban

AgentFlow exporte 9 métriques par session :

Tests

# Tests unitaires (zéro appel LLM)
pytest tests/unit/

# Tous les tests
pytest --cov=agentflow --cov-report=term-missing

# Tests d'intégration réels (nécessite clés API)
pytest -m slow

Providers LLM

Licence

MIT