AutoGen: Multiagentensysteme für RAG

AutoGen ist das Multi-Agenten-Framework von Microsoft, mit dem Konversationen zwischen mehreren KI-Agenten erstellt werden können. Im Gegensatz zu Einzelagenten-Ansätzen orchestriert AutoGen Teams spezialisierter Agenten, die zusammenarbeiten, um komplexe Probleme zu lösen.

Voraussetzungen: Lesen Sie die Grundlagen des RAG und unseren Leitfaden zur RAG-Agenten-Orchestrierung.

Hinweis (Juli 2026): Microsofts AutoGen befindet sich nun im Wartungsmodus (nur noch Sicherheitsfixes). Die aktive Weiterentwicklung erfolgt über den Community-Fork AG2 (pip install ag2), der die in diesem Leitfaden gezeigte klassische API beibehält; Microsoft verweist neue Projekte zudem auf das Microsoft Agent Framework. Die folgenden Konzepte bleiben gültig.

Warum AutoGen für RAG?

Vorteile des Multi-Agenten-Ansatzes

Anwendungsfälle Multi-Agenten RAG

• Vertiefte Recherche: Ein Agent sucht, ein anderer validiert, ein dritter synthetisiert
• Dokumentenanalyse: Spezialisierte Agenten je nach Dokumenttyp
• Komplexe Q&A: Zerlegung und Zusammensetzung der Antworten
• Kreuzvalidierung: Gegenseitige Überprüfung zwischen Agenten

ROI des Multi-Agenten-Ansatzes

• +40% Genauigkeit bei komplexen Fragen
• -60% Halluzinationen dank Kreuzvalidierung
• Nachvollziehbarkeit: Jeder Agent dokumentiert sein Reasoning

AutoGen-Architektur

Grundlegende Konzepte

ARCHITEKTUR AUTOGEN

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

Ablauf:
1. User → Manager
2. Manager → spezialisierter Agent
3. Agenten kommunizieren untereinander
4. Manager → User (finale Antwort)

Installation et configuration

DEVELOPERpython
# Installation (klassische v0.2-API, die in diesem Leitfaden verwendet wird)
# pip install "pyautogen~=0.2"   # oder: pip install ag2

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

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

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

Multi-Agenten-RAG mit AutoGen

Recherche-Agent (Retriever)

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

class RAGRetriever:
    """Retrieval-Manager für die Agenten."""

    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]:
        """Sucht relevante Dokumente."""
        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):
        """Fügt der Collection Dokumente hinzu."""
        ids = [f"doc_{i}" for i in range(len(documents))]
        self.collection.add(
            documents=documents,
            metadatas=metadatas or [{}] * len(documents),
            ids=ids
        )

# Initialisierung
retriever = RAGRetriever()

# Suchfunktion für die Agenten
def search_documents(query: str) -> str:
    """Sucht Dokumente und liefert den Kontext zurück."""
    results = retriever.search(query, n_results=5)

    if not results:
        return "Kein relevantes Dokument gefunden."

    context = "Gefundene Dokumente:\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

Erstellung der spezialisierten Agenten

DEVELOPERpython
# Recherche-Agent
research_agent = AssistantAgent(
    name="Researcher",
    system_message="""Du bist ein spezialisierter Recherche-Agent.

    Deine Aufgabe:
    1. Die Frage des Nutzers analysieren
    2. Relevante Suchanfragen formulieren
    3. Die Funktion search_documents nutzen, um Informationen zu finden
    4. Die Relevanz der Ergebnisse bewerten
    5. Die Suche bei Bedarf neu formulieren

    Du musst deine Suchentscheidungen stets begründen.
    Sind die Ergebnisse nicht relevant, versuche andere Formulierungen.
    """,
    llm_config=llm_config
)

# Analyse-Agent
analyst_agent = AssistantAgent(
    name="Analyst",
    system_message="""Du bist ein kritischer Analyse-Agent.

    Deine Aufgabe:
    1. Die vom Researcher gelieferten Dokumente prüfen
    2. Die wichtigsten Informationen und Fakten identifizieren
    3. Widersprüche oder Ungereimtheiten erkennen
    4. Die Verlässlichkeit der Quellen bewerten
    5. Die Informationen logisch strukturieren

    Du musst gründlich vorgehen und Unsicherheiten kennzeichnen.
    """,
    llm_config=llm_config
)

# Synthese-Agent
writer_agent = AssistantAgent(
    name="Writer",
    system_message="""Du bist ein Agent für Redaktion und Synthese.

    Deine Aufgabe:
    1. Die gelieferten Analysen aufnehmen
    2. Eine klare, strukturierte Antwort verfassen
    3. Die verwendeten Quellen zitieren
    4. Den Detailgrad an den Kontext anpassen
    5. Sicherstellen, dass die Antwort vollständig ist

    Du musst professionelle und gut formatierte Antworten liefern.
    """,
    llm_config=llm_config
)

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

Konfiguration des GroupChat

DEVELOPERpython
from autogen import register_function

# Suchfunktion registrieren
register_function(
    search_documents,
    caller=research_agent,
    executor=user_proxy,
    name="search_documents",
    description="Sucht Dokumente in der Wissensdatenbank."
)

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

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

def query_rag_team(question: str) -> str:
    """Befragt das RAG-Agententeam."""
    groupchat.messages = []

    result = user_proxy.initiate_chat(
        manager,
        message=f"""Frage des Nutzers: {question}

        Ablauf:
        1. Researcher: sucht relevante Dokumente
        2. Analyst: analysiert und prüft die Informationen
        3. Writer: verfasst die finale Antwort

        Beginnt mit der Recherche.
        """
    )

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

    return "Keine Antwort generiert."

Fortgeschrittene Patterns

Pattern 1: Kreuzvalidierung

DEVELOPERpython
class CrossValidationRAG:
    """RAG mit Kreuzvalidierung zwischen Agenten."""

    def __init__(self, llm_config: dict):
        self.searcher_1 = AssistantAgent(
            name="Searcher_Primary",
            system_message="Du recherchierst Dokumente umfassend.",
            llm_config=llm_config
        )

        self.searcher_2 = AssistantAgent(
            name="Searcher_Secondary",
            system_message="Du recherchierst mit alternativen Formulierungen.",
            llm_config=llm_config
        )

        self.validator = AssistantAgent(
            name="Validator",
            system_message="""Du vergleichst die Ergebnisse der beiden Rechercheure.
            1. Gemeinsame Informationen identifizieren (hohe Konfidenz)
            2. Widersprüche melden
            3. Einzigartige Ergebnisse zusammenführen
            4. Einen Konfidenzwert vergeben""",
            llm_config=llm_config
        )

        self.synthesizer = AssistantAgent(
            name="Synthesizer",
            system_message="Du erstellst die finale Antwort auf Basis der validierten Ergebnisse.",
            llm_config=llm_config
        )

    def query(self, question: str) -> dict:
        """Führt eine Anfrage mit Kreuzvalidierung aus."""
        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:
        """Validiert und führt die Ergebnisse zusammen."""
        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: Fachlich spezialisierte Agenten

DEVELOPERpython
class DomainSpecialistRAG:
    """RAG mit fachlich spezialisierten Agenten."""

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

        self.specialists = {
            "technical": self._create_specialist(
                "Technical_Expert",
                "Du bist Experte für technische Dokumentation, Code und Architektur."
            ),
            "legal": self._create_specialist(
                "Legal_Expert",
                "Du bist Experte für juristische Dokumente und Compliance."
            ),
            "financial": self._create_specialist(
                "Financial_Expert",
                "Du bist Experte für Finanz- und Buchhaltungsdokumente."
            ),
            "general": self._create_specialist(
                "General_Expert",
                "Du kümmerst dich um allgemeine und bereichsübergreifende Fragen."
            )
        }

        self.router = AssistantAgent(
            name="Router",
            system_message="""Analysiere die Frage und bestimme den Bereich:
            - technical: Code, Architektur, APIs, Infrastruktur
            - legal: Verträge, DSGVO, Compliance, Lizenzen
            - financial: Budgets, Rechnungen, Buchhaltung
            - general: sonstiges

            Antworte ausschließlich mit dem Bereich (ein Wort).""",
            llm_config=llm_config
        )

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

            Du musst:
            1. In deiner spezialisierten Wissensdatenbank recherchieren
            2. Präzise Antworten mit Quellen liefern
            3. Das passende Fachvokabular verwenden
            4. Melden, wenn die Frage nicht in deinen Bereich fällt""",
            llm_config=self.llm_config
        )

    def route_and_query(self, question: str) -> dict:
        """Leitet die Frage an den passenden Spezialisten weiter."""
        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: Debatte zwischen Agenten

DEVELOPERpython
class DebateRAG:
    """RAG mit strukturierter Debatte zwischen Agenten."""

    def __init__(self, llm_config: dict):
        self.proposer = AssistantAgent(
            name="Proposer",
            system_message="""Du schlägst eine erste Antwort auf Basis der Dokumente vor.
            Du sollst überzeugt, aber offen für Kritik sein.""",
            llm_config=llm_config
        )

        self.critic = AssistantAgent(
            name="Critic",
            system_message="""Du übst konstruktive Kritik an der vorgeschlagenen Antwort.
            1. Schwachstellen oder Lücken identifizieren
            2. Quellen hinterfragen
            3. Alternativen vorschlagen
            4. Nicht Kritik um der Kritik willen üben""",
            llm_config=llm_config
        )

        self.arbiter = AssistantAgent(
            name="Arbiter",
            system_message="""Du moderierst die Debatte zwischen Proposer und Critic.
            1. Die Argumente beider Seiten bewerten
            2. Meinungsverschiedenheiten entscheiden
            3. Die finale konsensuelle Antwort erstellen
            4. Die besten Beiträge beider Seiten integrieren""",
            llm_config=llm_config
        )

    def debate_and_answer(self, question: str, context: str, rounds: int = 2) -> dict:
        """Startet eine strukturierte Debatte und erstellt eine Antwort."""
        debate_history = []

        proposal = self.proposer.generate_reply(
            messages=[{
                "role": "user",
                "content": f"Kontext: {context}\n\nFrage: {question}\n\nSchlage eine Antwort vor."
            }]
        )
        debate_history.append({"agent": "Proposer", "content": proposal})

        for round_num in range(rounds):
            critique = self.critic.generate_reply(
                messages=[{
                    "role": "user",
                    "content": f"Aktueller Vorschlag: {proposal}\n\nKritisiere diese Antwort."
                }]
            )
            debate_history.append({"agent": "Critic", "content": critique})

            proposal = self.proposer.generate_reply(
                messages=[{
                    "role": "user",
                    "content": f"Erhaltene Kritik: {critique}\n\nVerbessere deinen Vorschlag."
                }]
            )
            debate_history.append({"agent": "Proposer", "content": proposal})

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

                Erstelle die finale Antwort auf die Frage: {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])

Speicherverwaltung

Gemeinsamer Speicher zwischen Agenten

DEVELOPERpython
from typing import Dict, Any
import json

class SharedMemory:
    """Gemeinsamer Speicher zwischen Agenten."""

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

    def add_fact(self, key: str, value: Any, source: str, confidence: float):
        """Fügt dem Speicher einen Fakt hinzu."""
        self.facts[key] = value
        self.sources[key] = source
        self.confidence_scores[key] = confidence

    def get_context(self) -> str:
        """Gibt den Kontext für die Agenten zurück."""
        context = "Etablierte Fakten:\n"
        for key, value in self.facts.items():
            conf = self.confidence_scores.get(key, 0)
            context += f"- {key}: {value} (Konfidenz: {conf:.0%})\n"
        return context

    def add_decision(self, decision: str, reasoning: str):
        """Protokolliert eine Entscheidung."""
        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 mit gemeinsamem Speicher."""
    context = memory.get_context()

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

            Neue Frage: {query}

            Recherchiere und füge neue Fakten zum Speicher hinzu."""
        }]
    )

    return response

Monitoring und Observability

Konversations-Traces

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

class ConversationTracer:
    """Protokolliert Multi-Agenten-Konversationen."""

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

    def trace_message(self, sender: str, receiver: str, content: str, metadata: dict = None):
        """Protokolliert eine Nachricht."""
        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:
        """Zusammenfassung der Konversation."""
        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)

Kosten und Performance

Kostenschätzung

Optimierungen

DEVELOPERpython
# 1. Runden begrenzen
groupchat = GroupChat(
    agents=agents,
    max_round=10
)

# 2. Caching der Suchanfragen
from functools import lru_cache

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

# 3. Unterschiedliche Modelle je 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]})

Umsetzungs-Checkliste

• Agenten mit klaren Rollen gut definiert
• Suchfunktion registriert
• GroupChat mit Rundenlimit konfiguriert
• Fehler- und Timeout-Behandlung
• Gemeinsamer Speicher bei Bedarf
• Tracing für das Debugging
• Tests mit unterschiedlichen Fragetypen
• Kostenmonitoring

Fazit

AutoGen ermöglicht den Aufbau ausgereifter RAG-Systeme mit Multi-Agenten-Zusammenarbeit. Der Schlüssel liegt darin, Rollen und Verantwortlichkeiten jedes Agenten klar zu definieren und die Kommunikation zwischen ihnen effizient zu steuern.

Weiterführende Ressourcen

• LangGraph: Komplexe Workflows
• CrewAI: Spezialisierte Agententeams
• Function Calling RAG

Brauchen Sie ein Multi-Agenten-RAG? Ailog bietet RAG-Lösungen mit intelligenter Orchestrierung spezialisierter Agenten. Robuste und skalierbare Architektur.