Agierende KI mit LangChain – Teil 7: Erstellung mehrerer Agenten

Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Agentic AI mit LangChain – Teil 7: Aufbau von Multi-Agent-Systemen mit LangGraph. Der Schwerpunkt liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den man ohne Rätseln über die Absicht direkt in ein Repository einfügen kann. In der Übersichtsphase sollten Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Warum Multi-Agent-Systeme wichtig sind

Beim Bearbeiten des Abschnitts „Warum mehrere Agenten wichtig sind“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Erstellen Sie Zwischenkontrollpunkte nach kostspieligen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneute Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut versucht.

In diesem Artikel verwendete technische Konzepte

Beim Arbeiten an den in dieser Phase verwendeten technischen Konzepten sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach kostspieligen Schritten. Das System sollte bei einem Neuanlauf eines späteren Knotens nicht denselben LLM-Aufruf erneut berechnen.

Projektübersicht: Ein Finanzassistenten-Agent

Beim Arbeiten an der Phase „Projektübersicht A – Finanzen“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach teuren Schritten einen Kontrollpunkt an. Das Wiederaufnehmen des Prozesses sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut versucht. Beim Arbeiten an der Phase „Projektübersicht A – Finanzen“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

1. Systemkonfiguration

Die Phase der Systemkonfiguration funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf. Halten Sie den Zustand der Graphen flach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Störungen beim Wiederaufnehmen nach Unterbrechungen.

OPENAI_API_KEY="<Your OpenAI API Key>"
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Projektstruktur

Die Phase „2 Projektstruktur“ funktioniert am besten, wenn sie als messbarer Rahmen betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Behandeln Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Arbeiten ab. Halten Sie den Zustand der Graphen flach und typisiert – verschachtelte Strukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Arbeit.

project/
├── tools/
│   ├── __init__.py
│   ├── financial_search.py # fetch stock prices for companies
│   ├── python_REPL.py      # async wrapper of PythonREPL()
|   └── wikipedia_search.py # fetch financial summary of companies
│
├── graphs/            # implement subagents and multi-agents
│   ├── agent.py       # implement research and analyst in a single agent
│   ├── sub_agents.py  # implement research and analyst sub agents
│   ├── multi_agent_supervisor.py # implement supervisor multi-agents
│   └── multi_agent_swarm.py # implement swamp multi-agents
│
├── config.py           # pydantic BaseConfig
│── utils.py            # utility for printing and executing agents
├── .env                # environment variable definition
├── .gitignore
└── requirements.txt     # package requirements

3. Werkzeuge des Finanzassistenten-Agenten

The 3 Tools of the stage funktionieren am besten, wenn sie als messbare Struktur betrachtet werden. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Stellen Sie Tools mit engen Schemata sowie klaren Kennzeichnungen für Nebeneffekte bereit. Die Betreiber müssen wissen, welche Aufrufe den Zustand verändern, bevor sie automatisch zustimmen. The 3 Tools of the stage funktionieren am besten, wenn sie als messbare Struktur betrachtet werden. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie den erfolgreichen Ablauf sowie den Wiederherstellungsprozess gemeinsam. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

3.1. Tool zur Suche in Stammdaten

Für die Phase „3 1 Stock Data“ sollten vor der Codeänderung Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Vorzuziehen sind kleine, testbare Einheiten statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Pipeline-System. Authentifizieren Sie sich am Gateway und erteilen Sie erneut Berechtigungen auf der Datenebene. Ein alleinigesBearer-Token stellt keine Trennlinie zwischen verschiedenen Nutzern dar.

# define LangChain tool to retrieve stock data
@tool
async def stock_data_tool(
        company_ticker: Annotated[str, "The ticker symbol of the company to retrieve their stock performance data."],
        num_days: Annotated[int, "The number of business days of stock data required to respond to the user query."]
) -> str:

        # fetch stock data asynchronously from yfinance
        # check full code for _fetch_stock_sync function
        df = await asyncio.to_thread(_fetch_stock_sync, company_ticker, f"{num_days}d")

        return (f"Successfully executed the stock performance data retrieval tool to retrieve "
                f"the last *{num_days} days* of data for company **{company_ticker}**:\n\n{df.to_markdown()}")

3.2. Python-Code-Ausführungstool

In der Phase 3 2 des Python-Code-Abschnitts sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor der Codeänderung definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Trennen Sie den Client-Aufbau vom Nachrichtenzyklus, damit Provider ausgetauscht werden können, ohne den Zustandsautomaten der Konversation umschreiben zu müssen.

@tool
async def python_repl_tool(
        code: Annotated[str, "The python code to execute to generate your chart."],
) -> str:

    # wrap LLM generated code and execute it by python.REPL()
    # repl = PythonREPL().
    clean_code = textwrap.dedent(code).strip()
    result = await asyncio.to_thread(repl.run, clean_code)

    return (
        f"Successfully executed the Python REPL tool.\n\n"
        f"Python code executed:\n```python\n{code}\n```\n\n"
        f"Code output:\n```\n{result}```"
    )

3.3. Wikipedia-Suchwerkzeug

Für die 3 3 Wikipedia-Suchphase sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Erhalten Sie Zeiten sowie Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen auf. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Authentifizieren Sie sich am Gateway und erteilen Sie erneut Berechtigungen auf der Datenebene. Ein alleiniges Trägertoken stellt keine Trennlinie zwischen verschiedenen Nutzern dar. Für die 3 3 Wikipedia-Suchphase sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen.

@tool()
async def wikipedia_tool(
        query: Annotated[str, "The Wikipedia search to execute to find key summary information."],
) -> str:
    """Use this to search Wikipedia for factual information asynchronously."""
    headers = {"User-Agent": "MyLangChainAgent/1.0 (contact@example.com)"}

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:

        # Step 1: Search Wikipedia for matching pages
        search_url = "https://en.wikipedia.org/w/api.php"
        search_params = {
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        }
        response = await client.get(search_url, params=search_params)
        search_data = response.json()
        results = search_data.get("query", {}).get("search", [])
        title = results[0]["title"]

        # Step 2: Fetch the page extract/summary
        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
        summary_res = await client.get(summary_url)
        summary_data = summary_res.json()
        summary = summary_data.get("extract", "No summary available.")

        return f"Successfully executed:\nWikipedia summary: {summary}"

4. Swarm Multi-Agents

Beim Arbeiten an der Stufe 4 „Swarm Multi-Agents“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Erstellen Sie Zwischenprüfungen nach kostspieligen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneuten Aufrufe derselben LLM-Funktion veranlassen, wenn ein Operator einen späteren Knoten erneut ausführt.

4.1. Architektur von Swarm Multi-Agents

Beim Arbeiten an der 4 1-Architektur der Phase sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneute Abrechnung für denselben LLM-Aufruf veranlassen, wenn ein Operator einen späteren Knoten erneut versucht.

4.2. Implementierung von Swarm-Multi-Agenten

Beim Arbeiten an der Phase „4 2 Swarm Multi-agents“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsame Umgebungen übergeht. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Prozesses sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Arbeiten an der Phase „4 2 Swarm Multi-agents“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# Create a handoff tool for researcher -> analyst
transfer_to_analyst = create_handoff_tool(
    agent_name="analyst",
    description="Transfer user to the analyst assistant, who can create visualizations of provided data.",
)

# create research agent as a subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool, transfer_to_analyst],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# define the checkpointer
checkpointer = InMemorySaver()

# create swarm agents
swarm_agent = create_swarm(
    agents=[research_agent, analyst_agent],
    default_active_agent="researcher"
).compile(checkpointer=checkpointer)

5. Supervisor Multi-Agents

Die Phase „5 Supervisor Multi-Agents“ funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein gelungenes Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf. Halten Sie den Zustand des Graphen einfach und typisiert. Verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Störungen nach Unterbrechungen.

5.1. Architektur von Supervisor Multi-Agents

Die 5-1-Architektur von Stage funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie einen erfolgreichen Fall, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand der Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

5.2. Implementierung von Supervisor-Multi-Agenten

Die 5 2 Supervisor Multi-agents-Phase funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsame Umgebungen übergeht. Halten Sie den Graphenzustand strukturiert und typisiert – eingebettete Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen des Ablaufs. Die 5 2 Supervisor Multi-agents-Phase funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie den erfolgreichen Ablauf sowie den Wiederherstellungsprozess gemeinsam. Wiederholversuche, menschliche Kontrollen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# create a checkpointer
checkpointer = InMemorySaver()

# create research subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# Create the supervisor multi-agent graph and compile it
supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, analyst_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research and data collection tasks to this agent\n"
        "- an analyst agent. Assign the creation of visualizations via code to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="last_message"
).compile(checkpointer=checkpointer)

6. Abschließende Zusammenfassung

Zur Phase des 6. Abschlusszusammenfassens sollten Eingabedaten, Verantwortliche für die jeweiligen Schritte sowie Abbruchkriterien definiert werden, bevor Code geändert wird. Die Mitarbeiter sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Vorzuziehen sind kleine, testbare Einheiten statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Bei Schritten, die Geld ausgeben oder Produktionsdaten ändern, ist menschliche Freigabe erforderlich. Kompilierzeitbezogene Verbindungen bedeuten nicht automatisch vollständige Geschäftsabläufe.

Operative Checkliste

Die Phase der operativen Checkliste funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlfall sowie eine Notiz zur Rücksetzung.

Halten Sie die Konfiguration außerhalb des Anwendungscode. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreiber überprüfen können, ohne den gesamten Graphen durchlesen zu müssen.

Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Blob-Strukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen dazu, dass die Fortsetzung nach Unterbrechungen nicht möglich ist.

Fügen Sie immer dann, wenn das Budget es zulässt, einen Smoke-Test hinzu, der den kritischen Pfad in CI mit Fixtures und nicht mit live genutzten, bezahlten APIs testet.

Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Blob-Strukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen dazu, dass die Fortsetzung nach Unterbrechungen nicht möglich ist.