AutoGen : Systemes multi-agents pour RAG

AutoGen est le framework multi-agents de Microsoft qui permet de creer des conversations entre plusieurs agents IA. Contrairement aux approches mono-agent, AutoGen orchestre des equipes d'agents specialises qui collaborent pour resoudre des problemes complexes.

Prerequis : Consultez les fondamentaux du RAG et notre guide sur l'orchestration d'agents RAG.

Warum AutoGen für RAG?

Vorteile des Multi-Agenten-Ansatzes

Anwendungsfälle Multi-Agenten RAG

• Recherche approfondie : Un agent cherche, un autre valide, un troisieme synthetise
• Analyse de documents : Agents specialises par type de document
• Q&A complexe : Decomposition et recomposition des reponses
• Validation croisee : Verification mutuelle entre agents

ROI du multi-agents

• +40% de precision sur les questions complexes
• -60% d'hallucinations grace a la validation croisee
• Tracabilite : Chaque agent documente son raisonnement

Architecture AutoGen

Concepts fondamentaux

ARCHITECTURE AUTOGEN

┌──────────────────────────────────────────────────────────┐
│                      GroupChat                           │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │ Agent 1  │  │ Agent 2  │  │ Agent 3  │              │
│  │Recherche │  │ Analyse  │  │Synthese  │              │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘              │
│       │             │             │                      │
│       └─────────────┼─────────────┘                      │
│                     │                                    │
│              ┌──────┴──────┐                            │
│              │  Manager    │                            │
│              │ (Routeur)   │                            │
│              └─────────────┘                            │
└──────────────────────────────────────────────────────────┘

Flux:
1. User → Manager
2. Manager → Agent specialise
3. Agents communiquent entre eux
4. Manager → User (reponse finale)

Installation et configuration

DEVELOPERpython
# Installation
# pip install pyautogen

from autogen import ConversableAgent, AssistantAgent, UserProxyAgent
from autogen import GroupChat, GroupChatManager
import os

# Configuration LLM
config_list = [
    {
        "model": "gpt-4o",
        "api_key": os.environ["OPENAI_API_KEY"]
    }
]

llm_config = {
    "config_list": config_list,
    "temperature": 0.7,
    "timeout": 120
}

RAG multi-agents avec AutoGen

Agent de recherche (Retriever)

DEVELOPERpython
from typing import List, Dict
import chromadb
from chromadb.utils import embedding_functions

class RAGRetriever:
    """Gestionnaire de retrieval pour les agents."""

    def __init__(self, collection_name: str = "documents"):
        self.client = chromadb.Client()
        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.environ["OPENAI_API_KEY"],
            model_name="text-embedding-3-small"
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn
        )

    def search(self, query: str, n_results: int = 5) -> List[Dict]:
        """Recherche des documents pertinents."""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )

        documents = []
        for i, doc in enumerate(results["documents"][0]):
            documents.append({
                "content": doc,
                "metadata": results["metadatas"][0][i] if results["metadatas"] else {},
                "distance": results["distances"][0][i] if results["distances"] else 0
            })

        return documents

    def add_documents(self, documents: List[str], metadatas: List[Dict] = None):
        """Ajoute des documents a la collection."""
        ids = [f"doc_{i}" for i in range(len(documents))]
        self.collection.add(
            documents=documents,
            metadatas=metadatas or [{}] * len(documents),
            ids=ids
        )

# Initialisation
retriever = RAGRetriever()

# Fonction de recherche pour les agents
def search_documents(query: str) -> str:
    """Recherche des documents et retourne le contexte."""
    results = retriever.search(query, n_results=5)

    if not results:
        return "Aucun document pertinent trouve."

    context = "Documents trouves:\n\n"
    for i, doc in enumerate(results, 1):
        context += f"[{i}] {doc['content'][:500]}...\n"
        context += f"    Score: {1 - doc['distance']:.2f}\n\n"

    return context

Creation des agents specialises

DEVELOPERpython
# Agent de recherche
research_agent = AssistantAgent(
    name="Researcher",
    system_message="""Tu es un agent de recherche specialise.

    Ton role:
    1. Analyser la question de l'utilisateur
    2. Formuler des requetes de recherche pertinentes
    3. Utiliser la fonction search_documents pour trouver des informations
    4. Evaluer la pertinence des resultats
    5. Reformuler la recherche si necessaire

    Tu dois toujours justifier tes choix de recherche.
    Si les resultats ne sont pas pertinents, essaie d'autres formulations.
    """,
    llm_config=llm_config
)

# Agent d'analyse
analyst_agent = AssistantAgent(
    name="Analyst",
    system_message="""Tu es un agent d'analyse critique.

    Ton role:
    1. Examiner les documents fournis par le chercheur
    2. Identifier les informations cles et les faits
    3. Detecter les contradictions ou incoherences
    4. Evaluer la fiabilite des sources
    5. Organiser les informations de maniere logique

    Tu dois etre rigoureux et signaler les incertitudes.
    """,
    llm_config=llm_config
)

# Agent de synthese
writer_agent = AssistantAgent(
    name="Writer",
    system_message="""Tu es un agent de redaction et synthese.

    Ton role:
    1. Prendre les analyses fournies
    2. Rediger une reponse claire et structuree
    3. Citer les sources utilisees
    4. Adapter le niveau de detail au contexte
    5. S'assurer que la reponse est complete

    Tu dois produire des reponses professionnelles et bien formatees.
    """,
    llm_config=llm_config
)

# Agent proxy utilisateur
user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config=False,
    llm_config=llm_config
)

Configuration du GroupChat

DEVELOPERpython
from autogen import register_function

# Enregistrer la fonction de recherche
register_function(
    search_documents,
    caller=research_agent,
    executor=user_proxy,
    name="search_documents",
    description="Recherche des documents dans la base de connaissances."
)

# Creer le GroupChat
groupchat = GroupChat(
    agents=[user_proxy, research_agent, analyst_agent, writer_agent],
    messages=[],
    max_round=15,
    speaker_selection_method="auto"
)

# Manager du GroupChat
manager = GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

def query_rag_team(question: str) -> str:
    """Interroge l'equipe d'agents RAG."""
    groupchat.messages = []

    result = user_proxy.initiate_chat(
        manager,
        message=f"""Question de l'utilisateur: {question}

        Processus:
        1. Researcher: recherche les documents pertinents
        2. Analyst: analyse et verifie les informations
        3. Writer: redige la reponse finale

        Commencez par la recherche.
        """
    )

    for msg in reversed(groupchat.messages):
        if msg.get("name") == "Writer":
            return msg.get("content", "")

    return "Aucune reponse generee."

Patterns avances

Pattern 1: Validation croisee

DEVELOPERpython
class CrossValidationRAG:
    """RAG avec validation croisee entre agents."""

    def __init__(self, llm_config: dict):
        self.searcher_1 = AssistantAgent(
            name="Searcher_Primary",
            system_message="Tu recherches des documents de maniere exhaustive.",
            llm_config=llm_config
        )

        self.searcher_2 = AssistantAgent(
            name="Searcher_Secondary",
            system_message="Tu recherches avec des formulations alternatives.",
            llm_config=llm_config
        )

        self.validator = AssistantAgent(
            name="Validator",
            system_message="""Tu compares les resultats des deux chercheurs.
            1. Identifier les informations communes (haute confiance)
            2. Signaler les contradictions
            3. Fusionner les resultats uniques
            4. Attribuer un score de confiance""",
            llm_config=llm_config
        )

        self.synthesizer = AssistantAgent(
            name="Synthesizer",
            system_message="Tu produis la reponse finale basee sur les resultats valides.",
            llm_config=llm_config
        )

    def query(self, question: str) -> dict:
        """Execute une requete avec validation croisee."""
        results_1 = self._search_with_agent(self.searcher_1, question)
        results_2 = self._search_with_agent(self.searcher_2, question)

        validated = self._validate(results_1, results_2)
        response = self._synthesize(question, validated)

        return {
            "response": response,
            "confidence": validated["confidence"],
            "sources_primary": len(results_1),
            "sources_secondary": len(results_2)
        }

    def _validate(self, results_1: list, results_2: list) -> dict:
        """Valide et fusionne les resultats."""
        common = []
        unique_1 = []
        unique_2 = []

        confidence = len(common) / max(len(results_1), len(results_2), 1)

        return {
            "common": common,
            "unique_1": unique_1,
            "unique_2": unique_2,
            "confidence": confidence
        }

Pattern 2: Agents specialises par domaine

DEVELOPERpython
class DomainSpecialistRAG:
    """RAG avec agents specialises par domaine."""

    def __init__(self, llm_config: dict):
        self.llm_config = llm_config

        self.specialists = {
            "technical": self._create_specialist(
                "Technical_Expert",
                "Tu es expert en documentation technique, code et architecture."
            ),
            "legal": self._create_specialist(
                "Legal_Expert",
                "Tu es expert en documents juridiques et conformite."
            ),
            "financial": self._create_specialist(
                "Financial_Expert",
                "Tu es expert en documents financiers et comptables."
            ),
            "general": self._create_specialist(
                "General_Expert",
                "Tu geres les questions generales et transverses."
            )
        }

        self.router = AssistantAgent(
            name="Router",
            system_message="""Analyse la question et determine le domaine:
            - technical: code, architecture, APIs, infrastructure
            - legal: contrats, RGPD, conformite, licences
            - financial: budgets, factures, comptabilite
            - general: autre

            Reponds uniquement avec le domaine (un mot).""",
            llm_config=llm_config
        )

    def _create_specialist(self, name: str, expertise: str) -> AssistantAgent:
        return AssistantAgent(
            name=name,
            system_message=f"""{expertise}

            Tu dois:
            1. Rechercher dans ta base de connaissances specialisee
            2. Fournir des reponses precises avec sources
            3. Utiliser le vocabulaire technique appropriate
            4. Signaler si la question sort de ton domaine""",
            llm_config=self.llm_config
        )

    def route_and_query(self, question: str) -> dict:
        """Route la question vers le bon specialiste."""
        routing_result = self.router.generate_reply(
            messages=[{"role": "user", "content": question}]
        )
        domain = routing_result.strip().lower()

        specialist = self.specialists.get(domain, self.specialists["general"])

        response = specialist.generate_reply(
            messages=[{"role": "user", "content": question}]
        )

        return {
            "domain": domain,
            "specialist": specialist.name,
            "response": response
        }

Pattern 3: Debat entre agents

DEVELOPERpython
class DebateRAG:
    """RAG avec debat structure entre agents."""

    def __init__(self, llm_config: dict):
        self.proposer = AssistantAgent(
            name="Proposer",
            system_message="""Tu proposes une reponse initiale basee sur les documents.
            Tu dois etre confiant mais ouvert aux critiques.""",
            llm_config=llm_config
        )

        self.critic = AssistantAgent(
            name="Critic",
            system_message="""Tu critiques constructivement la reponse proposee.
            1. Identifier les faiblesses ou lacunes
            2. Questionner les sources
            3. Proposer des alternatives
            4. Ne pas critiquer pour critiquer""",
            llm_config=llm_config
        )

        self.arbiter = AssistantAgent(
            name="Arbiter",
            system_message="""Tu arbitres le debat entre Proposer et Critic.
            1. Evaluer les arguments des deux cotes
            2. Trancher les desaccords
            3. Produire la reponse finale consensuelle
            4. Integrer les meilleures contributions de chacun""",
            llm_config=llm_config
        )

    def debate_and_answer(self, question: str, context: str, rounds: int = 2) -> dict:
        """Lance un debat structure et produit une reponse."""
        debate_history = []

        proposal = self.proposer.generate_reply(
            messages=[{
                "role": "user",
                "content": f"Context: {context}\n\nQuestion: {question}\n\nPropose une reponse."
            }]
        )
        debate_history.append({"agent": "Proposer", "content": proposal})

        for round_num in range(rounds):
            critique = self.critic.generate_reply(
                messages=[{
                    "role": "user",
                    "content": f"Proposition actuelle: {proposal}\n\nCritique cette reponse."
                }]
            )
            debate_history.append({"agent": "Critic", "content": critique})

            proposal = self.proposer.generate_reply(
                messages=[{
                    "role": "user",
                    "content": f"Critique recue: {critique}\n\nAmeliore ta proposition."
                }]
            )
            debate_history.append({"agent": "Proposer", "content": proposal})

        final_answer = self.arbiter.generate_reply(
            messages=[{
                "role": "user",
                "content": f"""Historique du debat:
                {self._format_history(debate_history)}

                Produis la reponse finale a la question: {question}"""
            }]
        )

        return {
            "answer": final_answer,
            "debate_rounds": len(debate_history),
            "history": debate_history
        }

    def _format_history(self, history: list) -> str:
        return "\n\n".join([f"[{h['agent']}]: {h['content']}" for h in history])

Gestion de la memoire

Memoire partagee entre agents

DEVELOPERpython
from typing import Dict, Any
import json

class SharedMemory:
    """Memoire partagee entre agents."""

    def __init__(self):
        self.facts = {}
        self.decisions = []
        self.sources = {}
        self.confidence_scores = {}

    def add_fact(self, key: str, value: Any, source: str, confidence: float):
        """Ajoute un fait a la memoire."""
        self.facts[key] = value
        self.sources[key] = source
        self.confidence_scores[key] = confidence

    def get_context(self) -> str:
        """Retourne le contexte pour les agents."""
        context = "Faits etablis:\n"
        for key, value in self.facts.items():
            conf = self.confidence_scores.get(key, 0)
            context += f"- {key}: {value} (confiance: {conf:.0%})\n"
        return context

    def add_decision(self, decision: str, reasoning: str):
        """Enregistre une decision."""
        self.decisions.append({
            "decision": decision,
            "reasoning": reasoning
        })

    def to_json(self) -> str:
        return json.dumps({
            "facts": self.facts,
            "decisions": self.decisions,
            "sources": self.sources
        }, indent=2)

memory = SharedMemory()

def research_with_memory(agent: AssistantAgent, query: str) -> str:
    """Recherche avec memoire partagee."""
    context = memory.get_context()

    response = agent.generate_reply(
        messages=[{
            "role": "user",
            "content": f"""Contexte connu:
            {context}

            Nouvelle question: {query}

            Recherche et ajoute de nouveaux faits a la memoire."""
        }]
    )

    return response

Monitoring et observabilite

Traces des conversations

DEVELOPERpython
import logging
from datetime import datetime
from typing import List, Dict

class ConversationTracer:
    """Trace les conversations multi-agents."""

    def __init__(self):
        self.traces = []
        self.logger = logging.getLogger("autogen_tracer")

    def trace_message(self, sender: str, receiver: str, content: str, metadata: dict = None):
        """Enregistre un message."""
        trace = {
            "timestamp": datetime.now().isoformat(),
            "sender": sender,
            "receiver": receiver,
            "content_length": len(content),
            "content_preview": content[:200],
            "metadata": metadata or {}
        }
        self.traces.append(trace)
        self.logger.info(f"{sender} -> {receiver}: {content[:100]}...")

    def get_summary(self) -> dict:
        """Resume de la conversation."""
        agents = set()
        for trace in self.traces:
            agents.add(trace["sender"])
            agents.add(trace["receiver"])

        return {
            "total_messages": len(self.traces),
            "agents_involved": list(agents),
            "duration_seconds": self._calculate_duration(),
            "average_message_length": self._avg_message_length()
        }

    def _calculate_duration(self) -> float:
        if len(self.traces) < 2:
            return 0
        start = datetime.fromisoformat(self.traces[0]["timestamp"])
        end = datetime.fromisoformat(self.traces[-1]["timestamp"])
        return (end - start).total_seconds()

    def _avg_message_length(self) -> float:
        if not self.traces:
            return 0
        return sum(t["content_length"] for t in self.traces) / len(self.traces)

Couts et performance

Estimation des couts

Optimisations

DEVELOPERpython
# 1. Limiter les rounds
groupchat = GroupChat(
    agents=agents,
    max_round=10
)

# 2. Caching des recherches
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
    return search_documents(query)

# 3. Modeles differents par agent
cheap_config = {"model": "gpt-4o-mini", "temperature": 0}
expensive_config = {"model": "gpt-4o", "temperature": 0.7}

research_agent = AssistantAgent("Researcher", llm_config={"config_list": [cheap_config]})
writer_agent = AssistantAgent("Writer", llm_config={"config_list": [expensive_config]})

Checklist d'implementation

• Agents bien definis avec roles clairs
• Fonction de recherche enregistree
• GroupChat configure avec limite de rounds
• Gestion des erreurs et timeouts
• Memoire partagee si necessaire
• Tracing pour le debug
• Tests avec differents types de questions
• Monitoring des couts

Conclusion

AutoGen permet de construire des systemes RAG sophistiques avec collaboration multi-agents. La cle est de bien definir les roles et responsabilites de chaque agent, et de gerer efficacement la communication entre eux.

Pour aller plus loin

• LangGraph : Workflows complexes
• CrewAI : Equipes d'agents
• Function calling RAG

Besoin d'un RAG multi-agents ? Ailog propose des solutions RAG avec orchestration intelligente d'agents specialises. Architecture robuste et scalable.