effidevFlutter · Cloudflare-Edge · Cloud-Kostenoptimierung
Deutsch

Multi-Agenten-Systeme für die Produktion mit LangGraph: Zustandsverwaltung & Zyklische Workflows

LangGraph Multi-Agenten-Architektur

Komplexe Unternehmensanforderungen erfordern Multi-Agenten-Systeme, bei denen spezialisierte Agenten (Researcher, Coder, Reviewer) zusammenarbeiten.

LangGraph ermöglicht durch eine graphbasierte Zustandsmaschine zyklische Schleifen (Loops), bedingte Verzweigungen und Persistenz via Checkpointer.

Wichtige Erkenntnisse

  • Zentraler StateGraph: Alle Knoten greifen auf ein gemeinsames TypedDict-Zustandsschema zu.
  • Zyklische Workflows: add_conditional_edges ermöglicht automatische Korrekturschleifen.
  • Human-in-the-Loop: Unterbrechung für menschliche Freigaben vor kritischen Schritten.

Python-Beispiel

from typing import Annotated, TypedDict, Literal
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_step: str

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

def researcher_node(state: AgentState):
    response = llm.invoke([HumanMessage(content="Recherchiere das Thema.")] + state["messages"])
    return {"messages": [response], "next_step": "writer"}

def writer_node(state: AgentState):
    response = llm.invoke([HumanMessage(content="Schreibe den Bericht.")] + state["messages"])
    return {"messages": [response], "next_step": "end"}

def router(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
    step = state.get("next_step", "researcher")
    if step == "writer": return "writer"
    elif step == "end": return END
    return "researcher"

workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", router)
workflow.add_conditional_edges("writer", router)

app = workflow.compile(checkpointer=MemorySaver())

Fazit

LangGraph ist das ideale Framework, um experimentelle KI-Skripte in hochverfügbare Multi-Agenten-Systeme für Unternehmen zu verwandeln.