Architectures Multi-Agents et Orchestration avec Agno

# Le superviseur décide quel agent appeler
SUPERVISOR_PROMPT = """Tu es un manager d''équipe IA.
Tu as 3 agents spécialisés :
- "researcher" : cherche des informations sur le web
- "analyst" : analyse des données et produit des statistiques
- "writer" : rédige des textes et rapports

Pour chaque demande, décide quel agent appeler et dans quel ordre.
Retourne le plan sous forme JSON."""

# Pipeline séquentiel simple
def run_pipeline(query: str) -> str:
    # Étape 1 : Recherche
    raw_data = research_agent.invoke(query)

    # Étape 2 : Extraction des faits clés
    facts = extraction_agent.invoke(raw_data)

    # Étape 3 : Analyse
    analysis = analysis_agent.invoke(facts)

    # Étape 4 : Rédaction du rapport
    report = writing_agent.invoke(analysis)

    return report

def debate(question: str, max_rounds: int = 3) -> str:
    proposal = proposer_agent.invoke(question)

    for round in range(max_rounds):
        critique = critic_agent.invoke(
            f"Analyse cette proposition et identifie les faiblesses :\n{proposal}"
        )

        if "APPROUVÉ" in critique:
            return proposal

        proposal = proposer_agent.invoke(
            f"Question : {question}\n"
            f"Ta proposition précédente : {proposal}\n"
            f"Critique reçue : {critique}\n"
            f"Améliore ta proposition."
        )

    return proposal  # Retourner la dernière version

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.team.team import Team
from agno.team.mode import TeamMode

# 1. Créer les agents spécialisés
researcher = Agent(
    name="Chercheur",
    role="Spécialiste en recherche d''informations",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=[
        "Tu recherches des informations fiables et récentes.",
        "Cite toujours tes sources.",
    ],
)

analyst = Agent(
    name="Analyste",
    role="Expert en analyse de données",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=[
        "Tu analyses les données et identifies les tendances.",
        "Présente tes résultats avec des chiffres précis.",
    ],
)

writer = Agent(
    name="Rédacteur",
    role="Rédacteur de contenu",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=[
        "Tu rédiges des textes clairs et bien structurés.",
        "Utilise un ton professionnel mais accessible.",
    ],
)

# 2. Créer l''équipe
team = Team(
    name="Équipe Recherche & Analyse",
    mode=TeamMode.coordinate,  # Le leader délègue et synthétise
    model=Claude(id="claude-sonnet-4-5"),
    members=[researcher, analyst, writer],
    instructions=[
        "Coordonne les membres pour produire un rapport complet.",
        "Le chercheur trouve les infos, l''analyste les analyse,",
        "et le rédacteur produit le rapport final.",
    ],
    show_members_responses=True,
    markdown=True,
)

# 3. Lancer l''équipe
team.print_response(
    "Analyse les tendances de l''IA en 2025 et rédige un rapport",
    stream=True,
)

# Agents spécialisés par domaine
technical_agent = Agent(
    name="Support Technique",
    role="Expert technique",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=["Tu résous les problèmes techniques et bugs."],
)

billing_agent = Agent(
    name="Support Facturation",
    role="Expert facturation",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=["Tu gères les questions de paiement et facturation."],
)

general_agent = Agent(
    name="Support Général",
    role="Agent généraliste",
    model=Claude(id="claude-sonnet-4-5"),
    instructions=["Tu réponds aux questions générales."],
)

# Mode route : redirige vers le bon spécialiste
support_team = Team(
    name="Support Client",
    mode=TeamMode.route,
    model=Claude(id="claude-sonnet-4-5"),
    members=[technical_agent, billing_agent, general_agent],
    instructions=[
        "Analyse la requête du client.",
        "Redirige vers le support technique pour les bugs,",
        "vers la facturation pour les paiements,",
        "vers le support général pour le reste.",
    ],
)

support_team.print_response(
    "Mon paiement a été débité deux fois",
    stream=True,
)
# → Automatiquement redirigé vers billing_agent

# Mode broadcast : tous les agents reçoivent la même question
debate_team = Team(
    name="Panel d''Experts",
    mode=TeamMode.broadcast,
    model=Claude(id="claude-sonnet-4-5"),
    members=[optimist_agent, pessimist_agent, pragmatist_agent],
    instructions=["Synthétise les différentes perspectives des membres."],
    markdown=True,
)

debate_team.print_response(
    "L''IA va-t-elle remplacer les développeurs ?",
    stream=True,
)
# → Chaque agent donne sa perspective, le leader synthétise

# Mode tasks : décompose en étapes séquentielles
project_team = Team(
    name="Pipeline de Contenu",
    mode=TeamMode.tasks,
    model=Claude(id="claude-sonnet-4-5"),
    members=[researcher, analyst, writer],
    instructions=[
        "Décompose le projet en tâches séquentielles.",
        "Assigne chaque tâche au membre le plus qualifié.",
    ],
    max_iterations=10,  # Limite le nombre d''itérations
    markdown=True,
)

project_team.print_response(
    "Crée un guide complet sur le fine-tuning de LLMs",
    stream=True,
)

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.memory import MemoryManager
from agno.models.anthropic import Claude

# Base de données pour la persistance
db = SqliteDb(db_file="tmp/agents.db")

# Le MemoryManager extrait automatiquement les faits importants
memory_manager = MemoryManager(
    model=Claude(id="claude-sonnet-4-5"),
    db=db,
)

agent = Agent(
    model=Claude(id="claude-sonnet-4-5"),
    db=db,
    memory_manager=memory_manager,
    enable_agentic_memory=True,       # L''agent gère sa mémoire
    add_history_to_context=True,      # Historique dans le contexte
    num_history_runs=5,               # 5 dernières conversations
    markdown=True,
)

# L''agent se souvient entre les sessions
agent.print_response("Je préfère les réponses courtes et en français", stream=True)
# → Le MemoryManager stocke cette préférence

# Session suivante...
agent.print_response("Explique le RAG", stream=True)
# → Réponse courte, en français (il a retenu la préférence !)

def reflexion_node(state: dict) -> dict:
    """L''agent évalue sa propre réponse."""
    score_response = llm.invoke(
        f"Évalue cette réponse (qualité 1-10, points à améliorer) :\n"
        f"Question : {state[''query'']}\n"
        f"Réponse : {state[''response'']}"
    )

    if "score: " in score_response.content:
        score = int(score_response.content.split("score: ")[1][0])
        if score < 7:
            state["needs_improvement"] = True
        else:
            state["needs_improvement"] = False

    return state

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.os import AgentOS

# Vos agents existants
research_agent = Agent(
    name="Research",
    model=Claude(id="claude-sonnet-4-5"),
    tools=[search_web, get_weather],
    markdown=True,
)

analysis_agent = Agent(
    name="Analysis",
    model=Claude(id="claude-sonnet-4-5"),
    tools=[analyze_data],
    markdown=True,
)

# Déployer en une ligne
agent_os = AgentOS(
    agents=[research_agent, analysis_agent],
    authorization=True,  # Authentification activée
)

# Servir sur le port 7777
agent_os.serve(port=7777)