LangGraph in der Praxis: Zustände, Knoten, Kanten und fünf Agentenmuster

Zusammenfassung: Wo wir aufgehört haben

In der vorherigen Folge haben wir die konzeptionellen Grundlagen erläutert: Ein KI-Agent verbindet ein „Gehirn“ (das LLM) mit einem Orchesterator, der Werkzeuge und Speicher koordiniert – alles wird dabei durch eine Art Überwachungssystem abgerundet. Wir haben außerdem gesehen, dass LangGraph dies in einen Graphen umwandelt: Knoten enthalten Arbeitseinheiten, Kanten steuern den Datenfluss dazwischen, und ein gemeinsames State-Objekt bewegt sich durch das gesamte System und wird bei jedem Schritt von Reducers zusammengeführt.

Dieses Mal gehen wir von der Theorie zur Umsetzung über. Wenn Sie diesen Artikel gelesen haben, werden Sie in der Lage sein:

• Den Zustand mithilfe von Annotation zu deklarieren, einschließlich benutzerdefinierter Reducer
• Knoten mithilfe von regulären sowie bedingten Kanten miteinander zu verbinden

Implementieren Sie alle fünf Workflow-Muster aus dem vorherigen Beitrag – Prompt-Chaining, Routing, Parallelisierung, Orchestrator-Worker sowie Evaluator-Optimizer – in funktionierendem JavaScript.

Führen Sie für jedes Muster ein Diagramm an, das genau zeigt, wie die einzelnen Komponenten miteinander verbunden sind.

Der gesamte untenstehende Code basiert auf @langchain/langgraph sowie reinem JavaScript (die gleichen Muster gelten auch für TypeScript; fügen Sie lediglich Typparameter zu Annotation<T> hinzu).

npm install @langchain/langgraph @langchain/core @langchain/openai

1. Zustand und Annotationen – Das gemeinsame Notebook

Bevor ein einzelner Knoten ausgeführt werden kann, muss LangGraph die Struktur der durch den Graphen fließenden Daten verstehen. Annotation dient dazu, diese Struktur zu beschreiben.

import { Annotation, StateGraph, START, END } from "@langchain/langgraph";

const WorkflowState = Annotation.Root({
  query: Annotation(),
  searchResults: Annotation(),
  draft: Annotation(),
  finalAnswer: Annotation(),
});

Jeder Schlüssel, den Sie innerhalb von Annotation.Root({...}) definieren, wird zu einem Feld im gemeinsamen Zustandsobjekt. Jeder Knoten kann von diesen Feldern lesen, und wenn er zurückkehrt, muss er nur die Felder mitliefern, die er tatsächlich aktualisiert – LangGraph kümmert sich darum, dieses teilweise Ergebnis wieder mit dem vollständigen Zustand zu verschmelzen.

Reducers: Entscheidung darüber, wie Updates zusammengeführt werden

Sofern Sie nicht anders angeben, ersetzt der Rückgabewert eines Knotens das entsprechende Feld vollständig. Das funktioniert gut für etwas wie draft, bei dem nur die neueste Version von Bedeutung ist. Bei Dingen wie einer anwachsenden Liste von Nachrichten führt das Überschreiben jedoch dazu, dass alles, was zuvor vorhanden war, gelöscht wird. Hier kommt ein Reducer ins Spiel – es handelt sich dabei um eine Funktion, die definiert, wie der vorhandene Wert mit einem eingehenden Update zusammengeführt werden soll.

const AgentState = Annotation.Root({
  // Default behavior: last write wins
  draft: Annotation(),

  // Custom reducer: append instead of overwrite
  messages: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
});

Sobald dies implementiert ist, führt das Merger-Verhalten dazu, dass, wenn nodeA { messages: ["hi"] } und nodeB { messages: ["hello"] } zurückgibt, diese Werte zu ["hi", "hello"] zusammengeführt werden, anstatt dass einer den anderen überschreibt. Dieses Mechanismus ist entscheidend, um parallele Abläufe sicher auszuführen (siehe Muster 3 unten) – zwei Knoten können gleichzeitig auf denselben Zustands Schlüssel schreiben, ohne in Konflikt zu geraten.

2. Knoten – Arbeitseinheiten

Ein Knoten ist nichts anderes als eine Funktion – synchron oder asynchron –, die den aktuellen Zustand erhält und einen teilweisen Update dafür zurückgibt.

async function draftNode(state) {
  const response = await llm.invoke(`Write a short draft about: ${state.query}`);
  return { draft: response.content };
}

Man registriert ihn im Graphen mithilfe von addNode:

const graph = new StateGraph(WorkflowState)
  .addNode("draft", draftNode);

Beachten Sie, dass der Knoten selbst keine Kenntnis von der umgebenden Graphstruktur hat. Er verbraucht einfach den Zustand und erzeugt eine Aktualisierung. Das Verbinden von Knoten ist ausschließlich Aufgabe der Kanten.

3. Kanten – Die Verbindung des Flusses

Kanten bestimmen die Ausführungsreihenfolge. LangGraph unterstützt drei Arten:

Normale Kanten – ein fester, bedingungsloser Pfad von einem Knoten zum anderen.

graph.addEdge(START, "draft");   // entry point: where the graph starts
graph.addEdge("draft", "polish");
graph.addEdge("polish", END);    // exit point: where the graph stops

Bedingte Kanten – eine Routing-Funktion prüft den aktuellen Zustand und entscheidet, welcher Knoten als Nächstes ausgeführt wird.

graph.addConditionalEdges("classify", (state) => {
  if (state.category === "billing") return "handleBilling";
  if (state.category === "technical") return "handleTechnical";
  return "handleGeneral";
});

Parallelen (Fan-out-)Kanten – wenn ein einziger Knoten mehrere ausgehende normale Kanten hat, werden alle ihre Ziele gleichzeitig ausgeführt.

graph.addEdge("orchestrator", "workerA");
graph.addEdge("orchestrator", "workerB");
graph.addEdge("orchestrator", "workerC");

Jeder Graph muss mindestens eine Kante enthalten, die von START ausgeht, sowie einen Pfad, der letztendlich END erreicht – andernfalls weiß LangGraph nicht, wo die Ausführung beginnen oder enden soll.

Nun, da Zustände, Knoten und Kanten behandelt wurden, verfügen Sie über alle Bausteine, die zur Erstellung der fünf zuvor vorgestellten Workflow-Muster erforderlich sind.

4. Muster 1 – Prompt-Verkettung

Konzept: Eine große Aufgabe in eine lineare Kette von LLM-Aufrufen aufteilen, wobei jeder Schritt auf dem vorherigen aufbaut.

START → outline → draft → polish → END

const ChainState = Annotation.Root({
  topic: Annotation(),
  outline: Annotation(),
  draft: Annotation(),
  final: Annotation(),
});

async function outlineNode(state) {
  const res = await llm.invoke(`Create a short outline for: ${state.topic}`);
  return { outline: res.content };
}async function draftNode(state) {
  const res = await llm.invoke(`Write a draft based on this outline:\n${state.outline}`);
  return { draft: res.content };
}async function polishNode(state) {
  const res = await llm.invoke(`Polish and tighten this draft:\n${state.draft}`);
  return { final: res.content };
}const chainGraph = new StateGraph(ChainState)
  .addNode("outline", outlineNode)
  .addNode("draft", draftNode)
  .addNode("polish", polishNode)
  .addEdge(START, "outline")
  .addEdge("outline", "draft")
  .addEdge("draft", "polish")
  .addEdge("polish", END)
  .compile();const result = await chainGraph.invoke({ topic: "the history of tea" });
console.log(result.final);

Jeder Knoten liest hier nur das Feld ein, das vom vorherigen Knoten geschrieben wurde – es gibt weder Verzweigungen noch Schleifen, nur ein gerader Ablauf von Anfang bis Ende.

5. Muster 2 – Routing

Konzept: Zuerst wird die eingehende Eingabe klassifiziert, anschließend an einen der mehreren speziell dafür vorgesehenen Verarbeitungsweg weitergeleitet.

┌──> handleBilling   ──┐
START → classify ───┼──> handleTechnical ─┼──> END
                    └──> handleGeneral   ──┘

const RouteState = Annotation.Root({
  message: Annotation(),
  category: Annotation(),
  response: Annotation(),
});

async function classifyNode(state) {
  const res = await llm.invoke(
    `Classify this message as "billing", "technical", or "general". Only reply with one word.\n${state.message}`
  );
  return { category: res.content.trim().toLowerCase() };
}async function handleBilling(state) {
  return { response: `[Billing team] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleTechnical(state) {
  return { response: `[Tech support] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleGeneral(state) {
  return { response: `[General desk] ${await llm.invoke(state.message).then(r => r.content)}` };
}const routeGraph = new StateGraph(RouteState)
  .addNode("classify", classifyNode)
  .addNode("handleBilling", handleBilling)
  .addNode("handleTechnical", handleTechnical)
  .addNode("handleGeneral", handleGeneral)
  .addEdge(START, "classify")
  .addConditionalEdges("classify", (state) => {
    if (state.category === "billing") return "handleBilling";
    if (state.category === "technical") return "handleTechnical";
    return "handleGeneral";
  })
  .addEdge("handleBilling", END)
  .addEdge("handleTechnical", END)
  .addEdge("handleGeneral", END)
  .compile();

Die Funktion, die an addConditionalEdges übergeben wird, muss den Namen des nächsten zu besuchenden Knotens zurückgeben – das ist der Mechanismus, der es dem Graphen ermöglicht, dynamisch einen Weg auszuwählen, anstatt stets denselben Pfad zu folgen.

6. Muster 3 – Parallelisierung

Konzept: Mehrere LLM-Aufrufe gleichzeitig gegen verschiedene Teile einer Aufgabe starten und anschließend die Ergebnisse kombinieren.

┌──> summarizeDocA ──┐
START → split ──┼──> summarizeDocB ──┼──> merge → END
                └──> summarizeDocC ──┘

const ParallelState = Annotation.Root({
  documents: Annotation(),
  summaries: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  finalSummary: Annotation(),
});

function splitNode(state) {
  return {}; // no-op entry point, just fans out to the summarizers below
}async function summarizeDocA(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[0]}`);
  return { summaries: [res.content] };
}
async function summarizeDocB(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[1]}`);
  return { summaries: [res.content] };
}
async function summarizeDocC(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[2]}`);
  return { summaries: [res.content] };
}async function mergeNode(state) {
  const res = await llm.invoke(`Combine these summaries into one:\n${state.summaries.join("\n---\n")}`);
  return { finalSummary: res.content };
}const parallelGraph = new StateGraph(ParallelState)
  .addNode("split", splitNode)
  .addNode("summarizeDocA", summarizeDocA)
  .addNode("summarizeDocB", summarizeDocB)
  .addNode("summarizeDocC", summarizeDocC)
  .addNode("merge", mergeNode)
  .addEdge(START, "split")
  .addEdge("split", "summarizeDocA")
  .addEdge("split", "summarizeDocB")
  .addEdge("split", "summarizeDocC")
  .addEdge("summarizeDocA", "merge")
  .addEdge("summarizeDocB", "merge")
  .addEdge("summarizeDocC", "merge")
  .addEdge("merge", END)
  .compile();

Achten Sie auf den Reducer, der mit summaries verbunden ist: Ohne ihn würde die letzte abgeschlossene parallele Spur einfach die vorherigen Ergebnisse überschreiben, anstatt alle drei Ausgaben sicher im Array anzusammeln.

7. Muster 4 – Orchestrator-Worker

Konzept: Ein einziger Aufruf der LLM teilt die gesamte Aufgabe in kleinere Unteraufgaben auf und weist jede davon an einen Worker-Node zu; schließlich wird alles in einem letzten Schritt zu einem Ergebnis zusammengefügt. Dies ähnelt stark der Parallelisierung, doch hier entscheidet der Orchestrator im Laufzeitbetrieb darüber, wie viele Worker gestartet werden sollen und was jeder von ihnen tun soll, anstatt dass dies im Voraus festgelegt wird.

START → orchestrator ──> worker(s), one per subtask ──> compile → END

const OrchestratorState = Annotation.Root({
  task: Annotation(),
  subtasks: Annotation(),
  sections: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  report: Annotation(),
});

async function orchestratorNode(state) {
  const res = await llm.invoke(
    `Break this task into 3-5 short subtasks, one per line:\n${state.task}`
  );
  const subtasks = res.content.split("\n").filter(Boolean);
  return { subtasks };
}async function workerNode(state) {
  // In a real app you'd invoke one worker per subtask (e.g. via Send / map-reduce);
  // here's the simple per-subtask logic each worker runs:
  const results = await Promise.all(
    state.subtasks.map((subtask) => llm.invoke(`Write a section about: ${subtask}`))
  );
  return { sections: results.map((r) => r.content) };
}async function compileNode(state) {
  const res = await llm.invoke(`Compile these sections into one report:\n${state.sections.join("\n\n")}`);
  return { report: res.content };
}const orchestratorGraph = new StateGraph(OrchestratorState)
  .addNode("orchestrator", orchestratorNode)
  .addNode("worker", workerNode)
  .addNode("compile", compileNode)
  .addEdge(START, "orchestrator")
  .addEdge("orchestrator", "worker")
  .addEdge("worker", "compile")
  .addEdge("compile", END)
  .compile();

Falls Sie eine echte dynamische Verteilung benötigen – das heißt, die Anzahl der Worker-Branches ändert sich tatsächlich von Ausführung zu Ausführung – bietet LangGraph.js die Send-API an, mit der Sie zur Ausführungszeit eine Worker-Aufruf pro Unteraufgabe erstellen können. Dieses Mechanismus verdient eine eigene ausführliche Erklärung, doch die hier gezeigte Einrichtung deckt bereits den Großteil der alltäglichen Anforderungen ab, wie zum Beispiel das Erstellen von Berichten oder das Bauen von Forschungsagenten.

8. Muster 5 – Evaluator-Optimierer

Konzept: Ein Knoten erzeugt eine Ausgabe, ein zweiter Knoten überprüft sie, und eine bedingte Kante leitet die Ausführung zurück zum Ersteller, bis die Ausgabe akzeptiert wird. In diesem Muster wird LangGraphs Unterstützung für Schleifen tatsächlich genutzt.

START → generate → evaluate ──(fail)──> generate  (loop)
                       │
                     (pass)
                       ↓
                      END

const LoopState = Annotation.Root({
  prompt: Annotation(),
  code: Annotation(),
  feedback: Annotation(),
  passed: Annotation(),
  attempts: Annotation({
    reducer: (current, update) => current + update,
    default: () => 0,
  }),
});

async function generateNode(state) {
  const context = state.feedback ? `\nPrevious feedback: ${state.feedback}` : "";
  const res = await llm.invoke(`Write code for: ${state.prompt}${context}`);
  return { code: res.content, attempts: 1 };
}async function evaluateNode(state) {
  const res = await llm.invoke(
    `Review this code for bugs. Reply "PASS" if it's correct, otherwise explain what's wrong:\n${state.code}`
  );
  const passed = res.content.trim().startsWith("PASS");
  return { passed, feedback: passed ? null : res.content };
}const loopGraph = new StateGraph(LoopState)
  .addNode("generate", generateNode)
  .addNode("evaluate", evaluateNode)
  .addEdge(START, "generate")
  .addEdge("generate", "evaluate")
  .addConditionalEdges("evaluate", (state) => {
    if (state.passed || state.attempts >= 4) return END; // safety cap on retries
    return "generate";
  })
  .compile();

Beachten Sie, dass das Feld attempts einen Reducer verwendet, der die laufende Summe erhöht anstatt sie zu ersetzen – dies dient auch als Schutzmaßnahme. Ohne eine Obergrenze für die Wiederholungen könnte ein Evaluator, der die Ausgabe niemals genehmigt, den Loop endlos am Laufen halten.

9. Kompilieren und Ausführen von irgendwelchen dieser Muster

Egal, welches Muster Sie erstellen, der Abschluss erfolgt auf dieselbe Weise:

const app = graph.compile();
const finalState = await app.invoke({ /* initial state fields */ });

Durch Aufruf von .compile() wird überprüft, ob auf jeden Knoten tatsächlich zugegriffen werden kann, es wird bestätigt, dass START und END ordnungsgemäß miteinander verbunden sind, und die Beschreibung Ihres Graphen wird in ein ausführbares Objekt umgewandelt. .invoke() führt den Graphen anschließend von Anfang bis Ende aus und gibt den resultierenden Zustand zurück – dies ist derselbe Ausführungszyklus, der konzeptionell bereits früher in dieser Serie beschrieben wurde.

Zusammenfassung

Durch diesen Leitfaden wurden die Konzepte aus der Theorie in lauffähigen Code umgewandelt: Annotation.Root zur Definition des Zustands, gewöhnliche Funktionen als Knoten, addEdge und addConditionalEdges zur Steuerung des Flusses sowie Reducer, um zu verhindern, dass gleichzeitige oder wiederholte Aktualisierungen sich gegenseitig überschreiben. Mit diesen vier Elementen erweisen sich alle fünf behandelten Workflow-Muster – Kettung, Routing, Parallelisierung, Orchestrator-Worker und Evaluator-Optimierer – als unterschiedliche Kombinationen derselben zugrundeliegenden Werkzeuge.

Der nächste Teil dieser Serie wird externe Tools einbeziehen, Speicher hinzufügen, damit der Zustand zwischen Sitzungen erhalten bleibt, sowie einen Schritt einführen, bei dem eine Person Aktionen vor deren Ausführung überprüft und genehmigt – die notwendigen Komponenten, um aus diesen Graphen echte Agenten zu machen.