Zum Inhalt springen
> 💻 🧠 Code 1001 > Praktische Anleitung zum Erstellen von KI-Agenten mit LangGraph und MCP

Praktische Anleitung zum Erstellen von KI-Agenten mit LangGraph und MCP

  • von

Dieser Artikel ist eine praktische Anleitung für Entwickler zur Erstellung autonomer KI-Agenten in Python. Wir werden die Theorie, was LangChain und LangGraph sind, nicht wiederholen. Stattdessen konzentrieren wir uns auf Code, Architektur und die Lösung realer Probleme.

Ziel: Zwei Projekte von Grund auf zu erstellen:

  1. Ein Klassifizierungs-Agent: Ein mehrstufiger Agent mit verwaltetem Zustand, aber ohne externe Werkzeuge.
  2. Ein Assistenz-Agent: Ein vollwertiger Agent mit Zugriff auf das Dateisystem und die Websuche über das MCP-Protokoll, aufgebaut auf zyklischer Logik.

Wir werden Best Practices behandeln: Konfigurationsmanagement, Modellauswahl und Fehlerbehandlung, um robuste Systeme zu erstellen.

Kurzer Überblick über die Konzepte: Der Agent und die MCP-Brücke

Bevor wir in den Code eintauchen, lassen Sie uns zwei Konzepte festhalten:

  • KI-Agent: Ein Programm, das um eine „Schlussfolgerungs-Aktions-Schleife“ herum aufgebaut ist. Es erhält eine Aufgabe, entscheidet mit Hilfe eines LLM, was als nächstes zu tun ist (z. B. ein Werkzeug aufrufen), führt die Aktion aus und wiederholt den Zyklus, bis die Aufgabe abgeschlossen ist.
  • MCP (Model Context Protocol): Ein Standard, der als Brücke zwischen der Logik des Agenten und externen Werkzeugen fungiert. Er ermöglicht es dem Agenten, auf einheitliche Weise mit Dateien, APIs oder der Suche zu arbeiten, ohne sich um deren Implementierungsdetails kümmern zu müssen.

Teil 1: Einrichtung einer robusten Umgebung

Schritt 1: Virtuelle Umgebung und Abhängigkeiten

Erstellen und aktivieren Sie eine virtuelle Umgebung. Erstellen Sie dann eine requirements.txt-Datei:

# Kern-Frameworks
langchain
langgraph

# Modell-Adapter
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # Für Ollama

# Werkzeuge und Protokolle
langchain-mcp-adapters
mcp
ollama

# Hilfsprogramme
python-dotenv
tenacity # Für robuste Fehlerbehandlung

Installieren Sie die Abhängigkeiten:

pip install -r requirements.txt```

#### Schritt 2: Konfiguration der API-Schlüssel

Erstellen Sie eine `.env`-Datei, um Ihre Schlüssel zu speichern:

OPENAI_API_KEY=“sk-…“
GOOGLE_API_KEY=“AIzaSy…“
MISTRAL_API_KEY=“…“
BRAVE_API_KEY=“…“ # Für das Websuch-Werkzeug über MCP

#### Schritt 3: Das „Modell-Factory“-Muster

Um flexibel zwischen Cloud- und lokalen Modellen wechseln zu können, ohne den Code des Agenten zu ändern, verwenden wir das Factory-Muster.

python

llm_factory.py

import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama

load_dotenv()

class ModelProvider(Enum):
OPENAI = „openai“
GEMINI = „gemini“
MISTRAL_API = „mistral_api“
OLLAMA = „ollama“

def get_llm(provider: ModelProvider, model_name: str = None):
„““Eine Factory zum Erstellen von LLM-Instanzen.“““
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or „gpt-4o-mini“, temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or „gemini-1.5-flash“, temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or „mistral-large-latest“, temperature=0)
elif provider == ModelProvider.OLLAMA:
# Stellen Sie sicher, dass Ollama mit dem erforderlichen Modell läuft
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or „mistral“, temperature=0)
raise ValueError(f“Unbekannter Modellanbieter: {provider}“)

Anwendungsbeispiel

if name == „main„:
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke(„Erklären Sie das Konzept von RAG in drei Sätzen.“)
print(response.content)

### Teil 2: Projekt 1 — Agent zur Klassifizierung von Stellenanzeigen

Dieser Agent demonstriert, wie man LangGraph verwendet, um einen **linearen Graphen** mit verwaltetem Zustand zu erstellen. Er nimmt eine Stellenbeschreibung entgegen und klassifiziert sie sequenziell nach drei Parametern.

#### Schritt 1: Definition des Zustands

Der Zustand ist der „Speicher“ unseres Graphen, der von einem Knoten zum nächsten weitergegeben wird.

python

vacancy_classifier.py

from typing import TypedDict, Dict

class ClassificationState(TypedDict):
„““Zustand für den Klassifizierungs-Agenten.“““
description: str # Quelltext
job_type: str # Art der Anstellung (projektbasiert/festangestellt)
category: str # Berufsbezeichnung
search_type: str # Ziel (sucht Job/Auftragnehmer)
classification_log: list # Debug-Protokoll

#### Schritt 2: Implementierung der Graph-Knoten

Jeder Knoten ist eine Funktion, die den Zustand entgegennimmt, ihren Teil der Arbeit erledigt und den aktualisierten Zustand zurückgibt.

python
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider

class VacancyClassifierAgent:
def init(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name=“gpt-4o-mini“)

async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
    """Knoten 1: Bestimmt die Art der Anstellung."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Bestimmen Sie die Art der Anstellung. Die Antwort muss 'projektbasiert' oder 'festangestellt' sein."),
        ("human", "Stellenbeschreibung:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["job_type"] = result.content.strip()
    state["classification_log"].append("Art der Anstellung bestimmt.")
    return state

async def _classify_category(self, state: ClassificationState) -> ClassificationState:
    """Knoten 2: Bestimmt die Berufskategorie."""
    # Kategorien können aus einer Datei oder Datenbank geladen werden
    categories = ["Python-Entwickler", "Designer", "Marketingspezialist", "3D-Animator"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"Wählen Sie die am besten geeignete Kategorie aus der Liste: {', '.join(categories)}."),
        ("human", "Stellenbeschreibung:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["category"] = result.content.strip()
    state["classification_log"].append("Kategorie bestimmt.")
    return state

async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
    """Knoten 3: Bestimmt das Suchziel."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Bestimmen Sie das Ziel des Autors. Die Antwort muss 'sucht Job' oder 'sucht Auftragnehmer' sein."),
        ("human", "Stellenbeschreibung:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["search_type"] = result.content.strip()
    state["classification_log"].append("Suchziel bestimmt.")
    return state
#### Schritt 3: Zusammenbau und Ausführung des Graphen

Wir bauen die Knoten zu einem einzigen Arbeitsablauf zusammen.

python

… Fortsetzung der Klasse VacancyClassifierAgent …

from langgraph.graph import StateGraph, END

def build_graph(self):
    """Baut den Zustandsgraphen zusammen."""
    workflow = StateGraph(ClassificationState)

    workflow.add_node("job_type_classifier", self._classify_job_type)
    workflow.add_node("category_classifier", self._classify_category)
    workflow.add_node("search_type_classifier", self._classify_search_type)

    workflow.set_entry_point("job_type_classifier")
    workflow.add_edge("job_type_classifier", "category_classifier")
    workflow.add_edge("category_classifier", "search_type_classifier")
    workflow.add_edge("search_type_classifier", END)

    return workflow.compile()

async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()

description = "Wir suchen einen erfahrenen Python-Entwickler, der unser Team in Vollzeit bei der Arbeit an einem Fintech-Projekt unterstützt."

initial_state = ClassificationState(
    description=description,
    job_type="", category="", search_type="",
    classification_log=[]
)

final_state = await graph.ainvoke(initial_state)

print("--- Klassifizierungsergebnis ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))

if name == „main„:
asyncio.run(main())

### Teil 3: Projekt 2 — Assistenz-Agent mit Werkzeugen (MCP)

Dieser Agent demonstriert eine **zyklische Logik**, bei der er wiederholt Werkzeuge aufrufen kann, um eine Aufgabe zu lösen.

#### Schritt 1: Konfigurationsmanagement

Für Agenten, die mit der Außenwelt interagieren, ist eine robuste Konfiguration unerlässlich.

python

mcp_agent_config.py

from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider

@dataclass
class AgentConfig:
workdir: str = „./agent_workdir“
model_provider: ModelProvider = ModelProvider.OLLAMA

def __post_init__(self):
    """Validierung nach der Initialisierung."""
    os.makedirs(self.workdir, exist_ok=True)
#### Schritt 2: Definition des Zustands für den Dialog

Der Zustand speichert nun den Nachrichtenverlauf.

python

mcp_agent.py

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]

#### Schritt 3: Implementierung des zyklischen Graphen

Der Graph besteht aus zwei Hauptknoten und einer bedingten Kante, die die „Schlussfolgerungs-Aktions-Schleife“ erzeugt.

python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig

class MCPAgent:
def init(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None

async def setup_tools(self):
    """Initialisiert Werkzeuge über MCP."""
    mcp_config = {
        "filesystem": {
            "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
            "transport": "stdio"
        },
        # Fügen Sie brave-search hinzu, wenn Sie einen BRAVE_API_KEY haben
    }
    mcp_client = MultiServerMCPClient(mcp_config)
    self.tools = await mcp_client.get_tools()
    self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])

    # Binden der Werkzeuge an das Modell
    self.llm = self.llm.bind_tools(self.tools)

def _should_continue(self, state: AgentState):
    """Bedingte Kante: entscheidet, ob ein Werkzeug aufgerufen werden soll."""
    last_message = state['messages'][-1]
    if not last_message.tool_calls:
        return "end"
    return "continue"

def _call_model(self, state: AgentState):
    """Knoten 1: Ruft das LLM auf, um eine Entscheidung zu treffen."""
    response = self.llm.invoke(state['messages'])
    return {"messages": [response]}

def _call_tool(self, state: AgentState):
    """Knoten 2: Führt den Werkzeugaufruf aus."""
    last_message = state['messages'][-1]
    tool_call = last_message.tool_calls[0]

    action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
    response = self.tool_executor.invoke(action)

    return {"messages": [response]}

def build_graph(self):
    workflow = StateGraph(AgentState)
    workflow.add_node("agent", self._call_model)
    workflow.add_node("action", self._call_tool)

    workflow.set_entry_point("agent")
    workflow.add_conditional_edges(
        "agent",
        self._should_continue,
        {"continue": "action", "end": END}
    )
    workflow.add_edge("action", "agent")

    return workflow.compile()
#### Schritt 4: Ausführung und Interaktion

python

… Fortsetzung von mcp_agent.py …

import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
„““Führt eine Aufgabe mit Fehlerbehandlung aus.“““
return await graph.ainvoke({„messages“: [HumanMessage(content=task)]})

async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # oder OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()

task = "Erstelle eine Datei namens 'hallo.txt' im Arbeitsverzeichnis und schreibe 'Hallo, Welt!' hinein."
result = await run_agent_task(graph, task)

print("\n--- Endgültige Antwort des Agenten ---")
print(result['messages'][-1].content)

if name == „main„:
asyncio.run(main())
`` Hier haben wir dentenacity`-Dekorator für Robustheit hinzugefügt – wenn der Agentenaufruf aufgrund eines vorübergehenden Netzwerkfehlers fehlschlägt, wird er automatisch wiederholt.

Fazit

Wir haben zwei Arten von Agenten unter Verwendung moderner Praktiken erstellt:

  • Ein linearer Graph eignet sich hervorragend für Aufgaben mit einer klaren Abfolge von Schritten, wie z. B. ETL-Prozesse oder mehrstufige Analysen.
  • Ein zyklischer Graph ist die Grundlage für die Erstellung interaktiver Assistenten und autonomer Agenten, die in der Lage sind, komplexe Probleme mit Werkzeugen zu lösen.

Die vorgestellten Architekturmuster – die Modell-Factory, das Konfigurationsmanagement, die Trennung der Logik in Knoten und die Verwendung von Zustandsgraphen – sind die Grundlagen für den Aufbau skalierbarer und robuster KI-Systeme.

Schreibe einen Kommentar

Deine E-Mail-Adresse wird nicht veröffentlicht. Erforderliche Felder sind mit * markiert