Wie man LangGraph installiert und benutzt: Eine Schritt-für-Schritt-Anleitung

Einführung

LangGraph ist eine leistungsstarke Bibliothek zum Erstellen von zustandsbasierten, mehraktorischen Anwendungen mit Large Language Models (LLMs). Dieser Leitfaden führt Sie durch den Prozess der Verwendung von LangGraph, von der Installation bis zur Erstellung komplexer KI-Agenten mit detaillierten Codebeispielen.

1. Installation und Setup

Als Erstes installieren wir LangGraph und seine Abhängigkeiten:

pip install langgraph langchain langchain_openai tavily-python

Als Nächstes richten Sie Ihre Umgebungsvariablen ein:

import os
os.environ["OPENAI_API_KEY"] = "Ihr-OPENAI-API-Schlüssel"
os.environ["TAVILY_API_KEY"] = "Ihr-TAVILY-API-Schlüssel"

2. Erstellung einer einfachen LangGraph-Anwendung

Erstellen wir einen einfachen Agenten, der das Internet durchsuchen und Fragen beantworten kann.

Schritt 1: Notwendige Module importieren

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage, FunctionMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
import json
from typing import TypedDict, Annotated, List

Schritt 2: Zustand und Werkzeuge definieren

class State(TypedDict):
    messages: Annotated[List[HumanMessage | AIMessage | FunctionMessage], "Die Nachrichten im Gespräch"]

tools = [TavilySearchResults(max_results=1)]
tool_executor = ToolExecutor(tools)
model = ChatOpenAI(temperature=0).bind_tools(tools)

Schritt 3: Agenten- und Aktionsknoten erstellen

def agent_node(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

def action_node(state):
    messages = state['messages']
    last_message = messages[-1]
    action = last_message.additional_kwargs["function_call"]
    result = tool_executor.invoke(action)
    return {"messages": [FunctionMessage(content=str(result), name=action["name"])]}

def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    if isinstance(last_message, AIMessage) and "function_call" in last_message.additional_kwargs:
        return "continue"
    return "end"

Schritt 4: Grafikstruktur erstellen

workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("action", action_node)
workflow.set_entry_point("agent")

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

Schritt 5: Grafik kompilieren und ausführen

app = workflow.compile()
inputs = {"messages": [HumanMessage(content="Wie ist das Wetter in San Francisco?")]}
result = app.invoke(inputs)

for message in result['messages']:
    print(f"{message.type}: {message.content}")

3. Fortgeschrittene Funktionen von LangGraph

Persistenz

LangGraph bietet integrierte Persistenzfunktionen. So verwenden Sie sie:

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# Führen Sie die Grafik mit Persistenz aus
result = app.invoke(inputs)

# Von einem Kontrollpunkt aus fortsetzen
checkpoint_id = result['checkpoint_id']
resumed_result = app.invoke(inputs, checkpoint_id=checkpoint_id)

Streaming

Um Ausgaben während ihrer Erzeugung zu streamen:

for event in app.stream(inputs):
    for key, value in event.items():
        if key == 'messages':
            print(f"Neue Nachricht: {value[-1].content}")

Menschliche Rückkopplung

Um menschliches Feedback einzubeziehen:

def human_approval_node(state):
    print("Aktueller Zustand:", state)
    approval = input("Genehmigen? (ja/nein): ")
    return {"approved": approval.lower() == "ja"}

workflow.add_node("human_approval", human_approval_node)
workflow.add_edge("agent", "human_approval")
workflow.add_conditional_edges(
    "human_approval",
    lambda x: "continue" if x["approved"] else "end",
    {
        "continue": "action",
        "end": END
    }
)

4. Erstellen einer komplexeren Anwendung mit LangGraph

Erstellen wir ein Mehragentensystem zur gemeinsamen Lösung von Aufgaben.

Schritt 1: Mehrere Agenten definieren

def researcher_agent(state):
    # Logik für die Recherche von Informationen
    return {"messages": [AIMessage(content="Rechercheergebnisse...")]}

def writer_agent(state):
    # Logik für das Schreiben von Inhalten basierend auf der Recherche
    return {"messages": [AIMessage(content="Geschriebene Inhalte...")]}

def editor_agent(state):
    # Logik für das Bearbeiten und Verfeinern von Inhalten
    return {"messages": [AIMessage(content="Bearbeitete Inhalte...")]}

Schritt 2: Eine komplexere Grafik erstellen

workflow = StateGraph(State)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("editor", editor_agent)
workflow.set_entry_point("researcher")

workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "editor")
workflow.add_conditional_edges(
    "editor",
    lambda x: "verfeinern" if "verfeinerung_erforderlich" in x['messages'][-1].content else "end",
    {
        "verfeinern": "writer",
        "end": END
    }
)

Schritt 3: Fortgeschrittene Funktionen implementieren

# Streaming-Unterstützung hinzufügen
app = workflow.compile()
for event in app.stream({"messages": [HumanMessage(content="Schreiben Sie einen Bericht über KI-Fortschritte")]}):
    print(f"Aktueller Schritt: {event['current_step']}")
    if 'messages' in event:
        print(f"Neueste Nachricht: {event['messages'][-1].content}")

# Persistenz hinzufügen
memory = SqliteSaver.from_conn_string("kollaborative_agenten.db")
app_with_persistence = workflow.compile(checkpointer=memory)

# Mit Persistenz ausführen
result = app_with_persistence.invoke({"messages": [HumanMessage(content="Schreiben Sie einen Bericht über KI-Fortschritte")]})
checkpoint_id = result['checkpoint_id']

# Von einem Kontrollpunkt aus fortsetzen
resumed_result = app_with_persistence.invoke({}, checkpoint_id=checkpoint_id)

5. Best Practices und Tipps für die Verwendung von LangGraph

1. Zustandsverwaltung: Halten Sie Ihren Zustandsobjekt sauber und gut strukturiert. Verwenden Sie Typ-Hinweise für Klarheit:

class DetailedState(TypedDict):
    messages: List[HumanMessage | AIMessage | FunctionMessage]
    research_data: dict
    draft_content: str
    edit_history: List[str]

1. Fehlerbehandlung: Implementieren Sie try-except-Blöcke in Ihren Knotenfunktionen:

def safe_researcher_agent(state):
    try:
        # Recherchelogik hier
        return {"messages": [AIMessage(content="Rechercheergebnisse...")]}
    except Exception as e:
        return {"messages": [AIMessage(content=f"Fehler bei der Recherche: {str(e)}")]}

1. Modulares Design: Zerlegen Sie komplexe Logik in kleinere, wiederverwendbare Funktionen:

def perform_research(query):
    # Recherchelogik
    pass

def analyze_results(research_data):
    # Analyselogik
    pass

def researcher_agent(state):
    query = state['messages'][-1].content
    research_data = perform_research(query)
    analysis = analyze_results(research_data)
    return {"messages": [AIMessage(content=analysis)], "research_data": research_data}

1. Testen: Erstellen Sie Unit-Tests für einzelne Knoten und Integrationstests für die gesamte Grafik:

import unittest

class TestResearcherAgent(unittest.TestCase):
    def test_researcher_agent(self):
        state = {"messages": [HumanMessage(content="Recherche KI")]}
        result = researcher_agent(state)
        self.assertIn("messages", result)
        self.assertIn("research_data", result)

if __name__ == '__main__':
    unittest.main()

1. Dokumentation: Fügen Sie ausführliche Docstrings zu Ihren Funktionen und Klassen hinzu:

def editor_agent(state: State) -> State:
    """
    Bearbeitet und verfeinert den Inhalt, der vom Writer-Agenten bereitgestellt wird.

    Args:
        state (State): Der aktuelle Zustand mit Nachrichten und Entwurf.

    Returns:
        State: Aktualisierter Zustand mit bearbeitetem Inhalt und potenziellen Verfeinerungsflaggen.
    """
    # Bearbeitungslogik hier
    pass

Zusammenfassung

Dieser Leitfaden bietet einen umfassenden Überblick über die Verwendung von LangGraph, von der grundlegenden Einrichtung bis zur Erstellung komplexer mehraktorischer Systeme. Durch das Befolgen dieser Schritte und bewährten Methoden können Sie LangGraph nutzen, um anspruchsvolle, zustandsbehaftete KI-Anwendungen zu erstellen, die eine Vielzahl von Aufgaben und Szenarien bewältigen können. Experimentieren Sie mit verschiedenen Grafikstrukturen und Knotenimplementierungen, um den besten Ansatz für Ihren spezifischen Anwendungsfall zu finden.