Ein standardisiertes Siebenteiliges Layout für jeden LangGraph-Agenten.
Eine feste Dateistruktur für LangGraph-Agenten, von Importen und Zuständen über Tools, Knoten, Kanten, Zusammenbau und Aufruf bis hin zu Vorlagen sowie einer Erweiterung für mehrere Agenten.
Quellen: The LangGraph Mental Model: A standardized architecture guide for every agent you’ll ever build
Jeder, der bereits eine Weile mit LangGraph gearbeitet hat, kennt dieses seltsame Gefühl: Die Idee macht Sinn, doch der Code will einfach nicht entstehen. Ein Graph enthält Knoten, die Knoten erledigen Aufgaben, Kanten bestimmen, wohin es als Nächstes geht, und der Zustand speichert Informationen. Man könnte das in fünf Minuten an einer Tafel skizzieren. Doch wenn man dann eine leere Datei öffnet, kommt nichts zustande. Die Importe sind einem entfallen, die gewünschte Klasse hat keinen offensichtlichen Platz, und es scheinen fünf verschiedene Möglichkeiten zu geben, dasselbe zu implementieren.
Warum der Code schwieriger erscheint als das Konzept
Der zugrunde liegende Grund ist, dass LangGraph keine eigene Struktur vorgibt. Es stellt Ihnen leistungsstarke Bausteine zur Verfügung und überlässt die Anordnung Ihnen. Jedes Tutorial und jedes Repository strukturiert die Inhalte anders, wodurch sich Ihre Erinnerung niemals auf ein einziges Muster festlegen kann.
Die hier vorgeschlagene Lösung besteht darin, eine Standardstruktur auszuwählen und sie für jeden Agenten zu verwenden – egal ob es sich um einen kleinen Chatbot oder um einen Pipeline mit vielen zusammenarbeitenden Agenten handelt. Man kann sich das wie ein Muster vorstellen, das man sich leicht einprägen und ohne Anstrengung abrufen kann, ähnlich wie die Diagramme, die zeigen, was ein Agent bereits erledigt hat, nur dass sie sich auf Quellcode statt auf Laufzeitdaten beziehen.
Der gesamte Dateiinhalt auf einen Blick
Jede LangGraph-Anwendung kann als eine einzige Datei mit sieben Abschnitten in fester Reihenfolge von oben nach unten geschrieben werden:
langgraph_agent.py
│
├── MODULE 1: IMPORTS & CONFIGURATION
│ └── All your libraries, API keys, model setup
│
├── MODULE 2: STATE
│ └── The TypedDict that defines your agent's memory
│
├── MODULE 3: TOOLS (optional, but common)
│ └── Functions decorated with @tool that the LLM can call
│
├── MODULE 4: NODES
│ └── Functions that do the actual work at each graph step
│
├── MODULE 5: EDGES & ROUTING
│ └── Functions that decide what happens next
│
├── MODULE 6: GRAPH ASSEMBLY
│ └── Where you build, wire, and compile the graph
│
└── MODULE 7: ENTRYPOINT
└── The __main__ block or invoke() call that runs everything
Das ist der Rahmen. Wenn Sie ein Projekt beginnen, öffnen Sie eine Datei und geben Sie die sieben Abschnittsüberschriften als Kommentare ein, bevor Sie sie nacheinander ausfüllen. Der Rest dieses Textes behandelt jeden Abschnitt, erläutert dessen Zweck, die Namen, die Sie kennen müssen, sowie ein Muster zum Kopieren.
Abschnitt 1: Importe und Konfiguration
Zweck
Betrachten Sie diesen Abschnitt als Werkzeugkasten. Ein Tischler legt alle Werkzeuge vor dem Bau von etwas zurecht – hier teilen Sie Python mit, welche externen Funktionen Sie benötigen, und richten Sie das Modell ein, das der Agent verwenden wird.
Namen, die man kennen sollte
ChatOpenAIoderChatAnthropicist die Klassen für das Sprachmodell, das Gehirn des Agents. Erstellen Sie diese einmal hier und wenden Sie sie überall wieder an.StateGraphist die zentrale Klasse, die das Graphen erstellt – ähnlich wie ein Entwurfsblatt, auf dem Sie zeichnen.ENDist eine spezielle Konstante für den Endpunkt. Eine Kante, die darauf zeigt, beendet die Ausführung.MemorySaverist ein Checkpointer, der den Speicher über mehrere Gesprächsschritte hinweg aufrechterhält.
Template
# --- Standard Library ---
import os
from typing import TypedDict, Annotated, Literal
# --- LangChain Core ---
from langchain_openai import ChatOpenAI # or ChatAnthropic, ChatGroq, etc.
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
# --- LangGraph Core ---
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages # The message reducer
from langgraph.prebuilt import ToolNode # Pre-built node for tool execution
from langgraph.checkpoint.memory import MemorySaver
# --- Configuration ---
# Always name your model variable 'llm' - it's the standard in every node
llm = ChatOpenAI(
model="gpt-4o", # or "claude-3-5-sonnet-20241022", etc.
temperature=0, # 0 = deterministic; raise for creativity
api_key=os.environ.get("OPENAI_API_KEY")
)
Warum es so strukturiert ist
Die Importe sind in vier Gruppen unterteilt: die Standardbibliothek, LangChain (die Modellschicht), LangGraph (die Graphschicht) sowie die Konfiguration. Dies folgt der Schichtung der Bibliotheken selbst, wobei LangChain KI-Primitiven bereitstellt und LangGraph die Orchestrierung übernimmt. Durch die Trennung der Gruppen lässt sich schnell erkennen, was wohin gehört, wenn etwas nicht funktioniert.
Die nützlichste Gewohnheit hier ist es, das Modell llm zu nennen. Jede Knotenfunktion wird später auf diese Variable verweisen, und wenn ein Projekt model verwendet, während ein anderes chat verwendet, wirkt der Code nicht mehr einheitlich und ist schwieriger zu überfliegen.
Abschnitt 2: Zustand
Wofür er dient
State ist das Arbeitsgedächtnis des Agents, eine einzige Quelle der Wahrheit, die durch das gesamte Graph strömt. Jeder Knoten kann darin lesen und schreiben, während der Engine sein Lebenszyklus verwaltet. Von allem in LangGraph ist es vermutlich am wichtigsten, State richtig umzusetzen.
Ein Whiteboard im Büro ist ein gutes Beispiel. Jeder, der den Raum betritt – also jeder Knoten – kann lesen, was dort steht, und etwas hinzufügen, und das Whiteboard bleibt während des gesamten Meetings, also während des gesamten Graph-Ausführungsprozesses, unverändert.
Wichtige Begriffe
TypedDictist eine Python-Klasse, die die Struktur des States beschreibt, ähnlich wie ein Schema. Ihre State-Klasse erbt davon ab.Annotatedist ein Wrapper für Typhinweise. LangGraph verwendet ihn, um einem Feld einen Reducer zuzuweisen, also eine Funktion, die bestimmt, wie ein neuer Wert mit dem bereits vorhandenen verschmolzen wird.
add_messages ist der eingebaute Reducer für das Feld „Messages“. Anstatt die Liste jedes Mal zu ersetzen, wenn ein Knoten eine Nachricht hinzufügt, fügt er diese hinteran hinzu. Es handelt sich um den Reducer, den Sie am häufigsten verwenden werden.operator.add ist ein weiterer häufiger Reducer für Felder, die durch Addition akkumuliert werden sollen, wie beispielsweise eine Liste von Ergebnissen oder ein Zähler.Template
# ============================================================
# MODULE 2: STATE
# ============================================================
class AgentState(TypedDict):
# 'messages' is the heartbeat of almost every LangGraph agent.
# Annotated[list, add_messages] means: "this is a list, and when
# a node writes to it, append - don't replace."
messages: Annotated[list[BaseMessage], add_messages]
# Add custom fields below for your specific agent's needs.
# Fields without a reducer are REPLACED each time a node writes to them.
# Example: a simple string field (gets replaced each write)
current_task: str
# Example: a list you want to accumulate (use operator.add as reducer)
# results: Annotated[list[str], operator.add]
# Example: a counter
# iteration_count: int
Über Reducer nachdenken
Reducer verwirren viele Anfänger, daher hier eine Analogie. Stellen Sie sich den Zustand als ein Papierformular vor, das in einem Büro weitergegeben wird. Das Feld messages ist ein Block mit Haftnotizen: Jeder, der das Formular bearbeitet, fügt eine Notiz hinzu, und niemand entfernt die früheren. Das ist add_messages.
Ein Feld ohne Reducer ist ein einzelnes Textfeld. Jeder überschreibt das, was der Vorherige geschrieben hat – das ist LangGraphs Standardverhalten: Das letzte Eingabewert gewinnt.
Es gilt eine einfache Regel: Geben Sie dem messages-Feld immer Annotated[list, add_messages], wählen Sie Annotated[list, operator.add] für jede weitere Liste, die weiter wachsen soll, und lassen Sie alle übrigen Felder unverändert.
Abschnitt 3: Werkzeuge
Zweck
Tools sind die Hände des Agents. Der LLM ist das Gehirn, das vernimmt und entscheidet, kann aber allein nicht in der Welt handeln. Mit Tools kann er im Internet surfen, eine API aufrufen, Code ausführen, Informationen in einer Datenbank abrufen oder jede beliebige von Ihnen definierte Python-Funktion ausführen.
Dieser Abschnitt ist optional. Ein Agent, der nur Text erzeugt und niemals externe Systeme berührt, kann ihn überspringen, obwohl die meisten echten Agenten Tools benötigen.
Wichtige Begriffe
@toolist ein LangChain-Decorator, der eine gewöhnliche Python-Funktion in ein Tool verwandelt, das LangGraph verstehen kann. Die Dokumentation der Funktion wird zur Beschreibung, die der LLM liest, um zu entscheiden, ob er sie verwenden soll.ToolNodeist ein fertig erstellter Knoten, der inlanggraph.prebuiltzu finden ist, und führt die von dem LLM ausgewählten Tools aus, sodass Sie die Ausführungslogik nicht selbst schreiben müssen.
bind_tools ist eine Methode in Ihrer LLM-Instanz, die Tools daran anschließt. Danach weiß das Modell, welche Tools verfügbar sind, und kann statt reinem Text eine spezielle tool_calls-Nachricht senden, um einen Aufruf zu beantragen.Template
# ============================================================
# MODULE 3: TOOLS
# ============================================================
@tool
def search_web(query: str) -> str:
"""Search the web for current information about a topic.
Use this when you need real-time information that is not in
your training data, such as recent news or live prices.
Args:
query: The search query string.
Returns:
A string containing search results.
"""
# Your actual implementation here (e.g., Tavily, SerpAPI, etc.)
# Placeholder for illustration:
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression and return the result.
Use this for any arithmetic, algebra, or numerical computation.
Args:
expression: A valid Python math expression as a string, e.g. '2 + 2 * 10'
Returns:
The computed result as a string.
"""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
# Collect all tools into a list - this is the pattern you always follow
tools = [search_web, calculate]
# Bind tools to the LLM so it knows they exist and can choose to call them
llm_with_tools = llm.bind_tools(tools)
# Create the pre-built ToolNode that will execute tool calls automatically
tool_node = ToolNode(tools)
Die Dokumentation ist wichtiger, als es scheint
Anfänger übersehen das oft: Das Modell verlässt sich auf Ihre Dokumentation, um ein Werkzeug auszuwählen. Es liest den Text innerhalb der dreifachen Anführungszeichen. Wenn die Dokumentation vage ist oder fehlt, weiß das LLM nicht, wann Ihr Werkzeug das Richtige ist. Schreiben Sie daher klare, präzise Beschreibungen, die angeben, was das Werkzeug tut und wann es verwendet werden soll.
Abschnitt 4: Knoten
Zweck
Knoten sind die „Arbeiter“. Jeder ist eine Python-Funktion, die den aktuellen Zustand erhält, etwas wie den Aufruf eines LLMs, die Verarbeitung von Daten oder den Einsatz eines Werkzeugs durchführt und anschließend ein Dictionary mit Aktualisierungen für den Zustand zurückgibt. Das ist die vollständige Beschreibung der Aufgabe.
Ein Knoten entscheidet niemals, was danach geschieht. Er erledigt seine Arbeit und aktualisiert den Zustand; die anschließende Weiterleitung wird im Abschnitt 5 behandelt.
Wichtige Begriffe
stateist der Parameter, den jeder Knoten erhält – ein Dictionary im Format vonAgentState. Darin finden Sie Werte wiestate["messages"]oderstate["current_task"].HumanMessage,AIMessageundSystemMessagesind die drei Nachrichtentypen, die Sie ständig verwenden werden. Der erste enthält das, was der Benutzer gesagt hat, der zweite die Antwort des Modells, und der dritte enthält Ihre Anweisungen an das Modell, also den Systemprompt.invokeist die Methode, mit der Sie das LLM oderllm_with_toolsaufrufen, um eine Antwort zu erhalten. Sie übergeben eine Liste von Nachrichten und erhalten anschließend eineAIMessage.
AgentState entsprechen. Seine Werte sind Updates, nicht der vollständige Zustand, und LangGraph fügt sie mithilfe der von Ihnen definierten Reduzierer zusammen.Template
# ============================================================
# MODULE 4: NODES
# ============================================================
# ── Node: Agent (the reasoning brain) ───────────────────────
def agent_node(state: AgentState) -> dict:
"""The central reasoning node. Calls the LLM and decides
whether to respond or call a tool."""
# Build the message list to send to the LLM.
# Always include a system message to set behavior.
system_prompt = SystemMessage(content=(
"You are a helpful assistant. Use the available tools "
"when you need real-time information or computation. "
"Respond clearly and concisely."
))
# The LLM receives the system prompt + all previous messages in state
messages_to_send = [system_prompt] + state["messages"]
# Call the LLM. Use llm_with_tools if you have tools; plain llm if not.
response = llm_with_tools.invoke(messages_to_send)
# Return the LLM's response as a state update.
# add_messages will APPEND this AIMessage to state["messages"].
return {"messages": [response]}
# ── Node: Summarizer (example of a non-LLM processing node) ─
def summarize_node(state: AgentState) -> dict:
"""Summarizes the conversation so far to keep context short.
This shows that nodes don't have to call an LLM - they can
do any Python processing."""
all_messages = state["messages"]
# Summarize with the LLM (a different prompt, same LLM)
summary_prompt = [
SystemMessage(content="Summarize the following conversation in 2-3 sentences."),
HumanMessage(content=str(all_messages))
]
summary_response = llm.invoke(summary_prompt)
# Replace messages with a fresh start containing just the summary
return {
"messages": [AIMessage(content=f"[Summary] {summary_response.content}")]
}
Bild für Knoten
Stellen Sie sich einen Knoten als Läufer vor, der einen Stab übernimmt – dieser Stab steht für den Zustand. Der Läufer läuft seine Strecke, also erledigt die Arbeit, und gibt den Stab mit aktualisierten Inhalten weiter. Er wählt nicht den nächsten Läufer aus; das ist die Aufgabe dessen, der das Rennen leitet, also der Kanten.
Jede Knotensignatur sieht gleich aus: def my_node(state: AgentState) -> dict. Die Eingabe ist der gesamte Zustand, die Ausgabe ist ein Dictionary, das nur die Felder enthält, die geändert werden sollen.
Abschnitt 5: Kanten und Routing
Zweck
Kanten bilden die Entscheidungslogik des Graphen. Sie verbinden Knoten miteinander und bestimmen, wie die Ausführung abläuft. LangGraph bietet zwei Arten davon.
Statische Kanten verbinden einen Knoten ohne Bedingung mit einem anderen: „Nach dem Zusammenfasser geht es zum Agenten.“ Es handelt sich dabei um keine Entscheidung, sondern nur um eine direkte Verbindung.
Bedingte Kanten betrachten den Zustand und geben einen String zurück, der den nächsten Knoten benennt. Hier liegt die Intelligenz des Agents bezüglich seines eigenen Ablaufs. Soll er ein Tool aufrufen, beenden oder zur Wiederholung zurückkehren?
Wichtige Begriffe
add_edge(A, B)erstellt eine statische Kante, was bedeutet, dass die Steuerung immer von A nach B geht.add_conditional_edges(source, routing_function, mapping)erstellt einen Zweig, der vomsourceausgeht. Die Routenfunktion erhält den aktuellen Zustand und gibt einen String zurück, wobei dasmapping-Wörterbuch diesen String in den tatsächlichen Namen des nächsten Knotens übersetzt.
tools_condition ist eine fertige Routenfunktion aus langgraph.prebuilt. Sie prüft die letzte Nachricht im Zustand auf Tool-Aufrufe. Falls solche vorhanden sind, wird zur "tools"-Kategorie weitergeleitet, andernfalls zur END-Kategorie. Dies ist das Standardmuster für Agenten, die Tools verwenden, und sollte auswendig gelernt werden.START ist die Konstante für den Anfang des Graphen. Das Festlegen eines Eingangspunkts bedeutet das Hinzufügen einer Kante von START zum ersten Knoten.Template
# ============================================================
# MODULE 5: EDGES & ROUTING
# ============================================================
# Import the pre-built tool routing function
from langgraph.prebuilt import tools_condition
# ── Custom Routing Function Example ─────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "summarize", "__end__"]:
"""Custom router for the agent node.
Routing functions always:
1. Receive the current state as input
2. Return a string that maps to the next node (or END)
The return values must match the keys in add_conditional_edges' mapping.
"""
last_message = state["messages"][-1] # Look at what the LLM just said
# Case 1: The LLM decided to call a tool
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# Case 2: The conversation is getting long - summarize before continuing
if len(state["messages"]) > 20:
return "summarize"
# Case 3: The LLM gave a direct answer - we're done
return "__end__" # LangGraph's internal name for END
Ein Bild zur Verkehrsführung
Betrachten Sie eine Verkehrsfunktion als einen Eisenbahnsignalmeister. Ein Zug erreicht einen Bahnhof, wenn ein Knoten seine Aufgabe erledigt hat. Der Meister prüft die Liste, die den aktuellen Zustand darstellt, und setzt den richtigen Schalter um, damit der Zug die richtige Strecke nimmt. Der Meister bewegt den Zug niemals selbst; das übernimmt LangGraph.
Eine Regel, die man im Hinterkopf behalten sollte: Verkehrsfunktionen ändern niemals den Zustand. Sie lesen ihn lediglich aus. Jede Änderung muss im Knoten vorgenommen werden; falls Sie feststellen, dass Sie während der Verkehrsführung den Zustand aktualisieren möchten, bringen Sie diese Logik in einen Knoten.
Abschnitt 6: Aufbau des Graphen
Zweck
Hier kommt alles zusammen. Betrachten Sie diesen Abschnitt als den Bauzeitpunkt. Die Abschnitte 1 bis 5 dienten der Vorbereitung – dabei wurden Materialien definiert sowie Teile zugeschnitten und angeordnet – und nun setzen Sie die Struktur zusammen, indem Sie die Teile miteinander verbinden und kompilieren.
Die Assemblierung folgt stets denselben fünf Schritten, die man sich merken sollte: Initialisieren, Knoten registrieren, Eingangspunkt festlegen, Kanten verbinden, kompilieren.
Wichtige Begriffe
StateGraph(AgentState)erstellt den Builder unter Verwendung Ihrer State-Klasse als Schema, weshalb jeder Knoten Updates liefern muss, die damit kompatibel sind.add_node(name, function)registriert einen Knoten. Der Name ist der String, den Sie in den Kanten verwenden, und die Funktion ist der Knoten aus Abschnitt 4.set_entry_point(name)legt fest, welcher Knoten zuerst ausgeführt wird, wenn der Graph aufgerufen wird.add_edge(A, B)erstellt eine direkte, bedingungslose Verbindung von A nach B.add_conditional_edges(source, router, mapping)fügt einen bedingten Zweig vom Quellknoten hinzu, wobei Ihre Routing-Funktion sowie eine Zuordnung von zurückgegebenen Strings zu Zielnamen verwendet werden.
compile(checkpointer=...) verschließt das Graph in ein ausführbares Objekt (einen CompiledGraph). Danach können weder Knoten noch Kanten hinzugefügt werden. Der optionale checkpointer aktiviert den persistenten Speicher.Template
# ============================================================
# MODULE 6: GRAPH ASSEMBLY
# ============================================================
# ── Step 1: Initialize ──────────────────────────────────────
# Always pass your State class to StateGraph
graph_builder = StateGraph(AgentState)
# ── Step 2: Register All Nodes ──────────────────────────────
# Format: add_node("node_name_as_string", node_function)
# The string name is what you use in ALL edge definitions
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node) # The pre-built ToolNode from Module 3
graph_builder.add_node("summarize", summarize_node)
# ── Step 3: Set Entry Point ─────────────────────────────────
# Which node runs first when we invoke the graph?
graph_builder.set_entry_point("agent")
# ── Step 4: Wire the Edges ──────────────────────────────────
# Conditional edge from agent: check if we need tools, a summary, or we're done
graph_builder.add_conditional_edges(
"agent", # Source node
should_continue, # Routing function from Module 5
{
"tools": "tools", # If router returns "tools" → go to tools node
"summarize": "summarize", # If router returns "summarize" → go to summarize node
"__end__": END, # If router returns "__end__" → stop the graph
}
)
# Static edge: after tools run, always go back to agent (the ReAct loop)
graph_builder.add_edge("tools", "agent")
# Static edge: after summarization, always return to agent
graph_builder.add_edge("summarize", "agent")
# ── Step 5: Compile ─────────────────────────────────────────
# Without checkpointer: no persistent memory (stateless per invocation)
# With checkpointer: memory persists across turns (stateful conversations)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
Bild zur Veranschaulichung
Ein Leiterplatten-Modell ist die beste Vergleichsmöglichkeit. Jeder Aufruf von add_node platziert ein Bauteil. add_edge und add_conditional_edges verbinden die Leitungen zwischen den Bauteilen. compile() führt die abschließende Überprüfung durch und fertigt die Leiterplatte ab. Danach ist sie bereit für die Stromversorgung, die über invoke erfolgt, sowie zum Ausführen ihrer Aufgabe.
Das Template enthält den grundlegenden ReAct-Zyklus: Agent → Werkzeuge → Agent → Werkzeuge und so weiter, bis der Agent entscheidet, dass er fertig ist, und auf END wechselt. Dieser Zyklus aus Überlegung und Handlung bildet die Grundlage für fast jeden LangGraph-Agenten, der Werkzeuge verwendet, und sein Erkennen erklärt den Großteil der Graphen, auf die man stößt.
Abschnitt 7: Eingangspunkt und Aufruf
Zweck
Das ist die Startplattform. Alles zuvor war die Definition, und dieser Abschnitt ist die Ausführung: der Code, der das Graphen ausführt, Eingaben liefert und Ausgaben sammelt.
Es gibt zwei Möglichkeiten, es auszuführen: invoke für einen einzigen Durchlauf und Streaming für schrittweise Ausgabe. Außerdem wird config benötigt, das Dictionary mit den Laufzeitparametern, insbesondere thread_id für den Speicher.
Namen, die man kennen sollte
graph.invoke(input, config)führt das gesamte Graphen aus und gibt den Endzustand zurück. Da es synchron ist, erhält man nichts, bis jeder Schritt abgeschlossen ist.graph.stream(input, config)führt ebenfalls das Graphen aus, gibt aber teilweise aktualisierte Zustände sofort zurück, was sich gut für benutzerorientierte Anwendungen eignet, die einen Live-Progress anzeigen müssen.
config ist das Dictionary, das bei jeder Aufrufung übergeben wird. Sein Schlüsselwert lautet {"configurable": {"thread_id": "some_id"}}. Die Thread-ID funktioniert wie eine Session-ID, die dem Checkpointer mitteilt, welchen Speicher wiederhergestellt und gespeichert werden soll. Zwei Benutzer mit unterschiedlichen Thread-IDs teilen niemals Speicher miteinander.HumanMessage(content=...) dient dazu, Benutzereingaben vor der Übermittlung an das Graph zu verpacken.Template
# ============================================================
# MODULE 7: ENTRYPOINT & INVOCATION
# ============================================================
if __name__ == "__main__":
# ── Config: defines this conversation's memory session ──
# Change thread_id to start a fresh conversation.
# Keep the same thread_id to continue an existing one.
config = {"configurable": {"thread_id": "user-session-001"}}
# ── Single Invocation (synchronous) ─────────────────────
user_input = "What is the current price of Bitcoin?"
result = graph.invoke(
input={"messages": [HumanMessage(content=user_input)]},
config=config
)
# The result is the final state dictionary.
# Access the last message to get the agent's final answer.
final_answer = result["messages"][-1].content
print(f"Agent: {final_answer}")
# ── Streaming Invocation (for real-time output) ──────────
for chunk in graph.stream(
input={"messages": [HumanMessage(content=user_input)]},
config=config,
stream_mode="values" # Yields the full state after each node runs
):
# Each chunk is a state snapshot. The last message shows progress.
latest = chunk["messages"][-1]
if hasattr(latest, "content") and latest.content:
print(f"[Streaming] {latest.content}")
# ── Multi-turn Conversation Loop ─────────────────────────
print("\n--- Starting Interactive Session ---")
while True:
user_text = input("You: ").strip()
if user_text.lower() in ("exit", "quit", "bye"):
break
response = graph.invoke(
input={"messages": [HumanMessage(content=user_text)]},
config=config # Same config = same memory thread
)
print(f"Agent: {response['messages'][-1].content}\n")
Das vollständige Template
Unten finden Sie den vollständigen, minimalen Rahmen, den Sie in ein neues Projekt kopieren können. Es handelt sich um die einzige Datei, die Sie sich merken sollten: Ersetzen Sie die Platzhalterimplementierungen durch Ihre eigene Logik – so erhalten Sie die Grundlage für jeden LangGraph-Agenten.
# ============================================================
# LANGGRAPH CANONICAL AGENT TEMPLATE
# Modules: Imports → State → Tools → Nodes → Edges → Assembly → Entrypoint
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ── MODULE 2: STATE ─────────────────────────────────────────
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# Add your custom fields here
# ── MODULE 3: TOOLS ─────────────────────────────────────────
@tool
def my_tool(input: str) -> str:
"""Describe clearly what this tool does and when the LLM should use it."""
return f"Result for: {input}"
tools = [my_tool]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ─────────────────────────────────────────
def agent_node(state: AgentState) -> dict:
"""The reasoning node. Calls the LLM, optionally triggers tool calls."""
messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# ── MODULE 5: EDGES & ROUTING ───────────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
"""Decide: did the LLM call a tool, or did it give a final answer?"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "__end__": END}
)
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
# ── MODULE 7: ENTRYPOINT ────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
while True:
user_text = input("You: ").strip()
if not user_text or user_text.lower() in ("exit", "quit"):
break
response = graph.invoke(
{"messages": [HumanMessage(content=user_text)]},
config=config
)
print(f"Agent: {response['messages'][-1].content}\n")
Weitergehen: Mehr-Agentensysteme
Sobald das Konzept mit einem einzelnen Agenten selbstverständlich erscheint, verlieren Mehr-Agentensysteme ihren Schrecken. Die zentrale Idee ist, dass jeder Agent einfach ein kompiliertes Graph ist, und ein kompiliertes Graph wiederum ein Knoten innerhalb eines größeren Überwachungsgraphs sein kann.
Begriffe, die man für die Arbeit mit Mehr-Agentensystemen kennen sollte
- Ein Überwachungsknoten ist ein Agent, dessen Aufgabe es nicht ist, dem Benutzer zu antworten, sondern zu entscheiden, welcher Spezialist die Aufgabe übernehmen soll. Er leitet den Datenfluss zwischen den Unteragenten weiter.
- Ein Untergraph ist ein kompiliertes Graph, das als Knoten in einem anderen Graph registriert ist. Er nimmt Zustände entgegen und gibt sie zurück, genauso wie jede gewöhnliche Knotenfunktion.
Command ist ein neueres Muster, das ab LangGraph 0,2 verfügbar ist, bei dem ein Knoten in einem einzigen Wert sowohl eine Zustandsaktualisierung als auch eine Routenanweisung zurückgibt. In einigen mehragentigen Konzepten entfällt dadurch die Notwendigkeit für separate bedingte Kanten.Strukturelles Template
# ── MULTI-AGENT PATTERN ─────────────────────────────────────
# Each specialist is a compiled graph (a subgraph)
# Sub-agent 1: A researcher
researcher_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
researcher = researcher_graph.compile()
# Sub-agent 2: A writer
writer_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
writer = writer_graph.compile()
# ── SUPERVISOR NODE ─────────────────────────────────────────
def supervisor_node(state: AgentState) -> dict:
"""Decides which sub-agent should handle the current task."""
# The supervisor LLM decides: "researcher" or "writer" or "FINISH"
response = supervisor_llm.invoke(state["messages"])
return {"messages": [response], "next_agent": response.content}
def route_to_agent(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
"""Routes to the appropriate sub-agent based on supervisor's decision."""
return state.get("next_agent", "__end__")
# ── SUPERVISOR GRAPH ────────────────────────────────────────
supervisor_builder = StateGraph(AgentState)
supervisor_builder.add_node("supervisor", supervisor_node)
supervisor_builder.add_node("researcher", researcher) # Subgraph as a node!
supervisor_builder.add_node("writer", writer) # Subgraph as a node!
supervisor_builder.set_entry_point("supervisor")
supervisor_builder.add_conditional_edges(
"supervisor",
route_to_agent,
{"researcher": "researcher", "writer": "writer", "__end__": END}
)
supervisor_builder.add_edge("researcher", "supervisor")
supervisor_builder.add_edge("writer", "supervisor")
supervisor_graph = supervisor_builder.compile(checkpointer=MemorySaver())
Die Mehr-Agenten-Version hat dieselbe Struktur wie die Ein-Agenten-Version, und es gelten dieselben sieben Abschnitte. Der einzige Unterschied besteht darin, dass einige der „Knoten“ vollständig kompilierte Graphen sind. Das mentale Modell skaliert ohne Probleme.
Schnellreferenz
Hier finden Sie eine zusammengefasste Karte, die Sie ausdrucken oder anbringen können, um die wichtigsten Ideen an einem Ort zusammenzufassen.
- Zustand:
TypedDictdefiniert die Struktur.Annotated[T, reducer]fügt einem Feld einen Reducer hinzu.add_messagesist der Standard-Reducer für Listen von Nachrichten und fügt hinzu anstatt zu ersetzen.operator.addakkumuliert Listen oder Zahlen. - LLM:
ChatOpenAIoderChatAnthropicist die Klasse, die stets mit dem Namenllmverknüpft ist.bind_tools(tools)teilt dem Modell die verfügbaren Tools mit.invoke(messages)ruft das Modell auf und gibt eine Antwort zurück. - Messages:
HumanMessageenthält die Eingaben des Benutzers,AIMessagedie Ausgaben des Modells undSystemMessageIhre Anweisungen. Sie werden immer als Liste aninvokeübergeben. - Tools:
@toolmacht eine Python-Funktion für das Modell aufrufbar, wobei die Dokumentation als Beschreibung dient.ToolNode(tools)führt Tool-Aufrufe automatisch aus.tools_conditionprüft, ob das Modell nach einem Tool gefragt hat. - Graph:
StateGraph(State)erstellt den Builder,add_node(name, fn)registriert einen Knoten,set_entry_point(name)wählt den ersten aus,add_edge(A, B)verbindet einen bedingungslosen Link,add_conditional_edges(source, router, mapping)verbindet einen Link, der durch den Rückgabewert des Routers ausgewählt wird,compile(checkpointer=...)finalisiert den Graphen undENDmarkiert den Endpunkt. - Aufruf:
graph.invoke(input, config)gibt den endgültigen Zustand zurück,graph.stream(input, config)liefert die Aktualisierungen schrittweise,config = {„configurable“: {"thread_id": "..."}}identifiziert die Speichersitzung, undMemorySaver()ist ein in Speicher liegender Checkpointer.
Es zur Muskelgedächtnis-Funktion machen
Es geht nicht darum, LangGraph von Grund auf zu unterrichten. Es geht vielmehr darum, dem Gehirn eine stabile Struktur zu bieten, die es automatisch nutzen kann – genauso wie ein Musiker auf eine vertraute Akkordfolge zurückgreift oder ein Koch die Zutaten vor dem Anfang zurechtlegt.
Sieben Abschnitte – Importe, Zustand, Werkzeuge, Knoten, Kanten, Zusammenbau und Eingangspunkt – bilden diese Struktur. Die Namen innerhalb jedes Abschnitts sind die „Noten“, und das vollständige Muster ist die „Notenschrift“.
Sowohl ein einfacher Chatbot als auch ein Pipeline mit Dutzenden von Knoten passen in dieselbe Struktur. Die Komplexität nimmt zu, doch die Struktur bleibt unverändert – und genau diese Stabilität macht sie zum Standard.
Eine sinnvolle Herangehensweise ist, mit dem minimalen ReAct-Loop zu beginnen: ein Agentenknoten, ein Werkzeuggrenk und eine bedingte Kante unter Verwendung von tools_condition. Bringen Sie das zunächst zum Funktionieren, fügen Sie anschließend einen weiteren Knoten hinzu und später einen weiteren Agenten. Die Struktur bleibt bei jeder Größe erhalten.
Für Persistenz-Backends, Muster mit menschlicher Beteiligung sowie Deployment-Optionen ist die offizielle LangGraph-Dokumentation der nächste Ansprechpunkt.