{"id":2473,"date":"2025-09-17T15:02:58","date_gmt":"2025-09-17T13:02:58","guid":{"rendered":"https:\/\/davidka.net\/ru\/?p=2473"},"modified":"2025-09-22T02:14:05","modified_gmt":"2025-09-22T00:14:05","slug":"ai-agent-with-mcp","status":"publish","type":"post","link":"https:\/\/de.davidka.net\/de\/2025\/09\/17\/ai-agent-with-mcp\/","title":{"rendered":"Praktische Anleitung zum Erstellen von KI-Agenten mit LangGraph und MCP"},"content":{"rendered":"\n

Dieser Artikel ist eine praktische Anleitung f\u00fcr 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\u00f6sung realer Probleme.<\/p>\n\n\n\n

Ziel:<\/strong> Zwei Projekte von Grund auf zu erstellen:<\/p>\n\n\n\n

    \n
  1. Ein Klassifizierungs-Agent:<\/strong> Ein mehrstufiger Agent mit verwaltetem Zustand, aber ohne externe Werkzeuge.<\/li>\n\n\n\n
  2. Ein Assistenz-Agent:<\/strong> Ein vollwertiger Agent mit Zugriff auf das Dateisystem und die Websuche \u00fcber das MCP-Protokoll, aufgebaut auf zyklischer Logik.<\/li>\n<\/ol>\n\n\n\n

    Wir werden Best Practices behandeln: Konfigurationsmanagement, Modellauswahl und Fehlerbehandlung, um robuste Systeme zu erstellen.<\/p>\n\n\n\n

    Kurzer \u00dcberblick \u00fcber die Konzepte: Der Agent und die MCP-Br\u00fccke<\/h3>\n\n\n\n

    Bevor wir in den Code eintauchen, lassen Sie uns zwei Konzepte festhalten:<\/p>\n\n\n\n

      \n
    • KI-Agent:<\/strong> Ein Programm, das um eine \u201eSchlussfolgerungs-Aktions-Schleife\u201c herum aufgebaut ist. Es erh\u00e4lt eine Aufgabe, entscheidet mit Hilfe eines LLM, was als n\u00e4chstes zu tun ist (z. B. ein Werkzeug aufrufen), f\u00fchrt die Aktion aus und wiederholt den Zyklus, bis die Aufgabe abgeschlossen ist.<\/li>\n\n\n\n
    • MCP (Model Context Protocol):<\/strong> Ein Standard, der als Br\u00fccke<\/strong> zwischen der Logik des Agenten und externen Werkzeugen fungiert. Er erm\u00f6glicht es dem Agenten, auf einheitliche Weise mit Dateien, APIs oder der Suche zu arbeiten, ohne sich um deren Implementierungsdetails k\u00fcmmern zu m\u00fcssen.<\/li>\n<\/ul>\n\n\n\n

      Teil 1: Einrichtung einer robusten Umgebung<\/h3>\n\n\n\n

      Schritt 1: Virtuelle Umgebung und Abh\u00e4ngigkeiten<\/h4>\n\n\n\n

      Erstellen und aktivieren Sie eine virtuelle Umgebung. Erstellen Sie dann eine requirements.txt<\/code>-Datei:<\/p>\n\n\n\n

      # Kern-Frameworks\nlangchain\nlanggraph\n\n# Modell-Adapter\nlangchain-openai\nlangchain-google-genai\nlangchain-mistralai\nlangchain-community # F\u00fcr Ollama\n\n# Werkzeuge und Protokolle\nlangchain-mcp-adapters\nmcp\nollama\n\n# Hilfsprogramme\npython-dotenv\ntenacity # F\u00fcr robuste Fehlerbehandlung<\/code><\/pre>\n\n\n\n
      Installieren Sie die Abh\u00e4ngigkeiten:<\/p>\n\n\n\n
      pip install -r requirements.txt```\n\n#### Schritt 2: Konfiguration der API-Schl\u00fcssel\n\nErstellen Sie eine `.env`-Datei, um Ihre Schl\u00fcssel zu speichern:<\/code><\/pre>\n\n\n\n
      OPENAI_API_KEY=“sk-\u2026“
      GOOGLE_API_KEY=“AIzaSy\u2026“
      MISTRAL_API_KEY=“\u2026“
      BRAVE_API_KEY=“\u2026“ # F\u00fcr das Websuch-Werkzeug \u00fcber MCP<\/p>\n\n\n\n
      #### Schritt 3: Das \u201eModell-Factory\u201c-Muster\n\nUm flexibel zwischen Cloud- und lokalen Modellen wechseln zu k\u00f6nnen, ohne den Code des Agenten zu \u00e4ndern, verwenden wir das Factory-Muster.<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      llm_factory.py<\/h1>\n\n\n\n
      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<\/p>\n\n\n\n
      load_dotenv()<\/p>\n\n\n\n
      class ModelProvider(Enum):
      OPENAI = „openai“
      GEMINI = „gemini“
      MISTRAL_API = „mistral_api“
      OLLAMA = „ollama“<\/p>\n\n\n\n
      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\u00e4uft
      # docker exec -it ollama ollama pull mistral
      return ChatOllama(model=model_name or „mistral“, temperature=0)
      raise ValueError(f“Unbekannter Modellanbieter: {provider}“)<\/p>\n\n\n\n
      Anwendungsbeispiel<\/h1>\n\n\n\n
      if name<\/strong> == „main<\/strong>„:
      # local_llm = get_llm(ModelProvider.OLLAMA)
      openai_llm = get_llm(ModelProvider.OPENAI)
      response = openai_llm.invoke(„Erkl\u00e4ren Sie das Konzept von RAG in drei S\u00e4tzen.“)
      print(response.content)<\/p>\n\n\n\n
      ### Teil 2: Projekt 1 \u2014 Agent zur Klassifizierung von Stellenanzeigen\n\nDieser 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.\n\n#### Schritt 1: Definition des Zustands\n\nDer Zustand ist der \u201eSpeicher\u201c unseres Graphen, der von einem Knoten zum n\u00e4chsten weitergegeben wird.<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      vacancy_classifier.py<\/h1>\n\n\n\n
      from typing import TypedDict, Dict<\/p>\n\n\n\n
      class ClassificationState(TypedDict):
      „““Zustand f\u00fcr 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<\/p>\n\n\n\n
      #### Schritt 2: Implementierung der Graph-Knoten\n\nJeder Knoten ist eine Funktion, die den Zustand entgegennimmt, ihren Teil der Arbeit erledigt und den aktualisierten Zustand zur\u00fcckgibt.<\/code><\/pre>\n\n\n\n
      python
      import asyncio
      import json
      from langchain_core.prompts import ChatPromptTemplate
      from llm_factory import get_llm, ModelProvider<\/p>\n\n\n\n
      class VacancyClassifierAgent:
      def init<\/strong>(self):
      self.llm = get_llm(ModelProvider.OPENAI, model_name=“gpt-4o-mini“)<\/p>\n\n\n\n
      async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:\n    """Knoten 1: Bestimmt die Art der Anstellung."""\n    prompt = ChatPromptTemplate.from_messages([\n        ("system", "Bestimmen Sie die Art der Anstellung. Die Antwort muss 'projektbasiert' oder 'festangestellt' sein."),\n        ("human", "Stellenbeschreibung:\\n\\n{description}")\n    ])\n    chain = prompt | self.llm\n    result = await chain.ainvoke({"description": state["description"]})\n\n    state["job_type"] = result.content.strip()\n    state["classification_log"].append("Art der Anstellung bestimmt.")\n    return state\n\nasync def _classify_category(self, state: ClassificationState) -> ClassificationState:\n    """Knoten 2: Bestimmt die Berufskategorie."""\n    # Kategorien k\u00f6nnen aus einer Datei oder Datenbank geladen werden\n    categories = ["Python-Entwickler", "Designer", "Marketingspezialist", "3D-Animator"]\n    prompt = ChatPromptTemplate.from_messages([\n        ("system", f"W\u00e4hlen Sie die am besten geeignete Kategorie aus der Liste: {', '.join(categories)}."),\n        ("human", "Stellenbeschreibung:\\n\\n{description}")\n    ])\n    chain = prompt | self.llm\n    result = await chain.ainvoke({"description": state["description"]})\n\n    state["category"] = result.content.strip()\n    state["classification_log"].append("Kategorie bestimmt.")\n    return state\n\nasync def _classify_search_type(self, state: ClassificationState) -> ClassificationState:\n    """Knoten 3: Bestimmt das Suchziel."""\n    prompt = ChatPromptTemplate.from_messages([\n        ("system", "Bestimmen Sie das Ziel des Autors. Die Antwort muss 'sucht Job' oder 'sucht Auftragnehmer' sein."),\n        ("human", "Stellenbeschreibung:\\n\\n{description}")\n    ])\n    chain = prompt | self.llm\n    result = await chain.ainvoke({"description": state["description"]})\n\n    state["search_type"] = result.content.strip()\n    state["classification_log"].append("Suchziel bestimmt.")\n    return state<\/code><\/pre>\n\n\n\n
      #### Schritt 3: Zusammenbau und Ausf\u00fchrung des Graphen\n\nWir bauen die Knoten zu einem einzigen Arbeitsablauf zusammen.<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      \u2026 Fortsetzung der Klasse VacancyClassifierAgent \u2026<\/h1>\n\n\n\n
      from langgraph.graph import StateGraph, END<\/p>\n\n\n\n
      def build_graph(self):\n    """Baut den Zustandsgraphen zusammen."""\n    workflow = StateGraph(ClassificationState)\n\n    workflow.add_node("job_type_classifier", self._classify_job_type)\n    workflow.add_node("category_classifier", self._classify_category)\n    workflow.add_node("search_type_classifier", self._classify_search_type)\n\n    workflow.set_entry_point("job_type_classifier")\n    workflow.add_edge("job_type_classifier", "category_classifier")\n    workflow.add_edge("category_classifier", "search_type_classifier")\n    workflow.add_edge("search_type_classifier", END)\n\n    return workflow.compile()<\/code><\/pre>\n\n\n\n
      async def main():
      agent = VacancyClassifierAgent()
      graph = agent.build_graph()<\/p>\n\n\n\n
      description = "Wir suchen einen erfahrenen Python-Entwickler, der unser Team in Vollzeit bei der Arbeit an einem Fintech-Projekt unterst\u00fctzt."\n\ninitial_state = ClassificationState(\n    description=description,\n    job_type="", category="", search_type="",\n    classification_log=[]\n)\n\nfinal_state = await graph.ainvoke(initial_state)\n\nprint("--- Klassifizierungsergebnis ---")\nprint(json.dumps(final_state, indent=2, ensure_ascii=False))<\/code><\/pre>\n\n\n\n
      if name<\/strong> == „main<\/strong>„:
      asyncio.run(main())<\/p>\n\n\n\n
      ### Teil 3: Projekt 2 \u2014 Assistenz-Agent mit Werkzeugen (MCP)\n\nDieser Agent demonstriert eine **zyklische Logik**, bei der er wiederholt Werkzeuge aufrufen kann, um eine Aufgabe zu l\u00f6sen.\n\n#### Schritt 1: Konfigurationsmanagement\n\nF\u00fcr Agenten, die mit der Au\u00dfenwelt interagieren, ist eine robuste Konfiguration unerl\u00e4sslich.<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      mcp_agent_config.py<\/h1>\n\n\n\n
      from dataclasses import dataclass, field
      import os
      from llm_factory import ModelProvider<\/p>\n\n\n\n
      @dataclass
      class AgentConfig:
      workdir: str = „.\/agent_workdir“
      model_provider: ModelProvider = ModelProvider.OLLAMA<\/p>\n\n\n\n
      def __post_init__(self):\n    """Validierung nach der Initialisierung."""\n    os.makedirs(self.workdir, exist_ok=True)<\/code><\/pre>\n\n\n\n
      #### Schritt 2: Definition des Zustands f\u00fcr den Dialog\n\nDer Zustand speichert nun den Nachrichtenverlauf.<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      mcp_agent.py<\/h1>\n\n\n\n
      from typing import TypedDict, Annotated, Sequence
      from langchain_core.messages import BaseMessage
      import operator<\/p>\n\n\n\n
      class AgentState(TypedDict):
      messages: Annotated[Sequence[BaseMessage], operator.add]<\/p>\n\n\n\n
      #### Schritt 3: Implementierung des zyklischen Graphen\n\nDer Graph besteht aus zwei Hauptknoten und einer bedingten Kante, die die \u201eSchlussfolgerungs-Aktions-Schleife\u201c erzeugt.<\/code><\/pre>\n\n\n\n
      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<\/p>\n\n\n\n
      class MCPAgent:
      def init<\/strong>(self, config: AgentConfig):
      self.config = config
      self.llm = get_llm(config.model_provider)
      self.tools = []
      self.tool_executor = None<\/p>\n\n\n\n
      async def setup_tools(self):\n    """Initialisiert Werkzeuge \u00fcber MCP."""\n    mcp_config = {\n        "filesystem": {\n            "command": "npx", "args": ["-y", "@modelcontextprotocol\/server-filesystem", self.config.workdir],\n            "transport": "stdio"\n        },\n        # F\u00fcgen Sie brave-search hinzu, wenn Sie einen BRAVE_API_KEY haben\n    }\n    mcp_client = MultiServerMCPClient(mcp_config)\n    self.tools = await mcp_client.get_tools()\n    self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])\n\n    # Binden der Werkzeuge an das Modell\n    self.llm = self.llm.bind_tools(self.tools)\n\ndef _should_continue(self, state: AgentState):\n    """Bedingte Kante: entscheidet, ob ein Werkzeug aufgerufen werden soll."""\n    last_message = state['messages'][-1]\n    if not last_message.tool_calls:\n        return "end"\n    return "continue"\n\ndef _call_model(self, state: AgentState):\n    """Knoten 1: Ruft das LLM auf, um eine Entscheidung zu treffen."""\n    response = self.llm.invoke(state['messages'])\n    return {"messages": [response]}\n\ndef _call_tool(self, state: AgentState):\n    """Knoten 2: F\u00fchrt den Werkzeugaufruf aus."""\n    last_message = state['messages'][-1]\n    tool_call = last_message.tool_calls[0]\n\n    action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}\n    response = self.tool_executor.invoke(action)\n\n    return {"messages": [response]}\n\ndef build_graph(self):\n    workflow = StateGraph(AgentState)\n    workflow.add_node("agent", self._call_model)\n    workflow.add_node("action", self._call_tool)\n\n    workflow.set_entry_point("agent")\n    workflow.add_conditional_edges(\n        "agent",\n        self._should_continue,\n        {"continue": "action", "end": END}\n    )\n    workflow.add_edge("action", "agent")\n\n    return workflow.compile()<\/code><\/pre>\n\n\n\n
      #### Schritt 4: Ausf\u00fchrung und Interaktion<\/code><\/pre>\n\n\n\n
      python<\/p>\n\n\n\n
      \u2026 Fortsetzung von mcp_agent.py \u2026<\/h1>\n\n\n\n
      import asyncio
      from langchain_core.messages import HumanMessage
      from tenacity import retry, stop_after_attempt, wait_fixed<\/p>\n\n\n\n
      @retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
      async def run_agent_task(graph, task):
      „““F\u00fchrt eine Aufgabe mit Fehlerbehandlung aus.“““
      return await graph.ainvoke({„messages“: [HumanMessage(content=task)]})<\/p>\n\n\n\n
      async def main():
      config = AgentConfig(model_provider=ModelProvider.OPENAI) # oder OLLAMA
      agent = MCPAgent(config)
      await agent.setup_tools()
      graph = agent.build_graph()<\/p>\n\n\n\n
      task = "Erstelle eine Datei namens 'hallo.txt' im Arbeitsverzeichnis und schreibe 'Hallo, Welt!' hinein."\nresult = await run_agent_task(graph, task)\n\nprint("\\n--- Endg\u00fcltige Antwort des Agenten ---")\nprint(result['messages'][-1].content)<\/code><\/pre>\n\n\n\n
      if name<\/strong> == „main<\/strong>„:
      asyncio.run(main())
      `` Hier haben wir den<\/code>tenacity`-Dekorator f\u00fcr Robustheit hinzugef\u00fcgt \u2013 wenn der Agentenaufruf aufgrund eines vor\u00fcbergehenden Netzwerkfehlers fehlschl\u00e4gt, wird er automatisch wiederholt.<\/p>\n\n\n\n
      Fazit<\/h3>\n\n\n\n
      Wir haben zwei Arten von Agenten unter Verwendung moderner Praktiken erstellt:<\/p>\n\n\n\n
        \n
      • Ein linearer Graph<\/strong> eignet sich hervorragend f\u00fcr Aufgaben mit einer klaren Abfolge von Schritten, wie z. B. ETL-Prozesse oder mehrstufige Analysen.<\/li>\n\n\n\n
      • Ein zyklischer Graph<\/strong> ist die Grundlage f\u00fcr die Erstellung interaktiver Assistenten und autonomer Agenten, die in der Lage sind, komplexe Probleme mit Werkzeugen zu l\u00f6sen.<\/li>\n<\/ul>\n\n\n\n
        Die vorgestellten Architekturmuster \u2013 die Modell-Factory, das Konfigurationsmanagement, die Trennung der Logik in Knoten und die Verwendung von Zustandsgraphen \u2013 sind die Grundlagen f\u00fcr den Aufbau skalierbarer und robuster KI-Systeme.<\/p>\n","protected":false},"excerpt":{"rendered":"
        Dieser Artikel ist eine praktische Anleitung f\u00fcr 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\u00f6sung realer Probleme.<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"neve_meta_sidebar":"","neve_meta_container":"","neve_meta_enable_content_width":"","neve_meta_content_width":0,"neve_meta_title_alignment":"","neve_meta_author_avatar":"","neve_post_elements_order":"","neve_meta_disable_header":"","neve_meta_disable_footer":"","neve_meta_disable_title":"","_themeisle_gutenberg_block_has_review":false,"footnotes":""},"categories":[],"tags":[],"class_list":["post-2473","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/posts\/2473","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/comments?post=2473"}],"version-history":[{"count":0,"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/posts\/2473\/revisions"}],"wp:attachment":[{"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/media?parent=2473"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/categories?post=2473"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/de.davidka.net\/de\/wp-json\/wp\/v2\/tags?post=2473"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}