ReAct in LangGraph: Vom selbstgebauten Loop zu ToolNode und StateGraph

Eine einfache Schleife zur Aufrufung von Tools erfolgt in einem Durchlauf: Das Modell bittet um Tools, die Ergebnisse kommen zurück und anschließend wird eine Antwort gesendet. Aufgaben, die aufgeschaltete Aufrufe erfordern, bei denen der zweite Aufruf von dem abhängt, was der erste zurückgegeben hat, benötigen einen Agenten, der denken, handeln, das Ergebnis prüfen und erneut handeln kann. Dieser Zyklus ist das ReAct-Muster. Unten finden Sie eine Veranschaulichung, eine handgeschriebene ReAct-Schleife sowie denselben Agenten, der mit LangGraph neu aufgebaut wurde – inklusive Tool-Node, Tool-Nachrichten, bedingtem Kante und Zustandsgraphen.

Die Komponenten, die LangGraph dafür bereitstellt:

• Tool-Node: Analysiert Tool-Aufrufe, führt die Tools aus und gibt die Ergebnisse zurück.
• Tool-Nachricht: Übermittelt die Ausgabe eines Tools an das LLM.
• tools_condition: Eine Routing-Funktion, die nach Tool-Aufrufen sucht und den bedingten Kante steuert.
• StateGraph: Steuert den Ablauf des Workflows.

Zustand: ein eingegebenes Dictionary, das die sich ändernden Daten des Agents enthält.

Knoten: Python-Funktionen, die den Zustand aktualisieren, wobei ein Tool-Knoten einer davon ist.

Kanten: Übergänge zwischen Knoten, wobei bedingte Kanten Zweige und Schleifen ermöglichen.

Einrichtung

Halten Sie Geheimnisse in einer .env-Datei im Projektverzeichnis:

OPENAI_API_KEY="<Your OpenAI API Key>"
TAVILY_API_KEY=<Your TAVILY API Key>

Laden Sie sie über eine Pydantic BaseSettings-Klasse:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Die Projektstruktur:

project/
├── react_loop/
│   └── weather_assistant.py # implement a vanilla react loop
│
├── react_langgraph     # implement ReAct by LangGraph workflow
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|
├── tools
│   ├── tool_call.py  # implement simple tool-calling loop
|   └── weather.py    # get_weather LangChain tool using open source APIs
|
|──    # implement tool-calling loop using LangChain tool
├── config.py         # pydantic BaseConfig
├── utils.py          # utility method to print messages
├── .env              # environment variable definition
├── .gitignore
└── requirements.txt  # package requirements

Was ReAct bedeutet

ReAct steht für Reasoning plus Acting. Ein Agent, der eine mehrstufige Aufgabe erledigt, denkt nach, handelt, beobachtet und denkt erneut nach, bis die Aufgabe abgeschlossen ist. Der grundlegende Zyklus erlaubt nur eine Runde Tool-Ausführung, und genau diese Lücke schließt ReAct.

Ein Lauf umfasst sechs Schritte:

1. Der Benutzer sendet eine Nachricht.
2. Das LLM liest die Anfrage, ermittelt, welche Informationen benötigt werden, und entscheidet, ob und welche Tools aufgerufen werden sollen.
3. Der Agent führt ein oder mehrere Tools aus, wie zum Beispiel das Suchen im Internet, das Abfragen einer API oder das Senden einer E-Mail.
4. Die Ergebnisse kommen zurück und werden dem aktuellen Zustand des Agents hinzugefügt.
5. Das LLM bewertet die neuen Informationen und antwortet entweder oder bittet um weitere Tool-Aufrufe.
6. Sobald keine weiteren Aktionen mehr notwendig sind, wird die endgültige Antwort gesendet.

Der Loop ist mächtig, weil der Agent seinen Plan entsprechend dem, was er gerade gelernt hat, überarbeiten kann, sodass jede Entscheidung von früheren Ergebnissen abhängen kann.

Ein einfacher ReAct-Loop

Diese Version befindet sich in react_loop/weather_assistant.py und ähnelt dem Ein-Pass-Loop, mit zwei Unterschieden:

async def chat(self, message: str):
    # Add user message
    self.messages.append(HumanMessage(content=message))

    # loop to handle serial tool calls
    while True:

        # Get AI response with potential tool calls
        response = await self.llm_with_tools.ainvoke(self.messages)
        self.messages.append(response)

        # Check if there is any tools to call
        if not response.tool_calls:
            # no tools to call
            break
            # Execute each tool call

        # process tool calls asynchronously
        results = await asyncio.gather(*[self.tools[tc["name"]].ainvoke(tc) for tc in response.tool_calls])
        self.messages.extend(results)

Zunächst laufen die Tool-Aufrufe asynchron mit asyncio.gather, sodass mehrere gleichzeitig ausgeführt werden können. Zweitens befindet sich der Schritt zur Ausführung des Tools innerhalb einer while-Schleife, die den Zyklus „Begründung dann Handlung“ wiederholt, bis die Antwort des Modells keine Tool-Aufrufe mehr enthält.

Derselbe Agent in LangGraph

Die manuelle Schleife funktioniert, zwingt einen jedoch dazu, den Zyklus „Begründung-Handlung“, die Nachrichtenliste, das Tool-Wörterbuch sowie die Ergebnissammlung selbst zu verwalten. LangGraph übernimmt all das: ToolNode parsiert die Aufrufe und führt die Tools aus, StateGraph speichert die Nachrichten im Graphenzustand, und die tools_condition-Kante ersetzt die manuell geschriebene while-Schleife sowie deren Überprüfung der Tool-Aufrufe.

Zustand. Ein typisierter Wörterbuch enthält die Nachrichtenliste (react_langgraph/state.py):

from typing import TypedDict, Annotated
from langgraph.graph.message import AnyMessage, add_messages

class ReactState(TypedDict):
    """
    Define the state schema for our hotel assistant.

    Attributes:
        messages: list of messages
    """

    messages: Annotated[list[AnyMessage], add_messages]

Die Knoten lesen und aktualisieren Nachrichten darüber, und der add_messages Reducer fügt hinzu, was auch immer ein Knoten zurückgibt.

Knoten. In react_langgraph/nodes.py erstellt die Einrichtung eine Liste von LangChain-Tools, einen Systemprompt, der dem Modell mitteilt, wie es diese Tools verwenden soll, sowie ein OpenAI-Modell, das an diese Tool-Liste gebunden ist:

tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY), get_weather]

SYSTEM_PROMPT = f"""You are an assistant for question-answering tasks with access to multiple tools.
    use these tools to retrieve context to answer the question. If you don't know the answer,
    just say that you don't know. Use three sentences maximum and keep the answer concise.

    Tool Use Instructions:

    - Use tools thoughtfully—only when they add value to the response
    - Chain tools when needed (e.g., search for info, then use that to search
      for relevant hotels)
      - Be conversational and helpful
    """

system_message = SystemMessage(content=SYSTEM_PROMPT)

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)


async def run_agent_reasoning(state: ReactState) -> dict:
    """
    run agent reasoning node
    :param state: ReactState, manage graph state
    :return: dict containing new message
    """

    print("----AGENT REASONING------")
    messages = [system_message] + state["messages"]

    if isinstance(messages[-1], ToolMessage):
        print_message(messages[-1])

    response = await llm_with_tools.ainvoke(messages)
    print_message(response)

    return {"messages": [response]}


tool_node = ToolNode(tools)

Zwei Knoten entstehen. Der Reasoning-Knoten erstellt die Nachrichtenliste, ruft das LLM auf und gibt dessen Antwort zurück. Der Act-Knoten ist ein ToolNode, der Tool-Aufrufe parsiert, die Tools ausführt und Tool-Nachrichten zurückgibt.

Graph. In react_langgraph/graph.py registriert ein StateGraph die Knoten und verbindet sie mithilfe von Kanten:

workflow = StateGraph(ReactState)

workflow.add_node(AGENT_REASON, run_agent_reasoning)
workflow.add_node(ACT, tool_node)

workflow.add_edge(START, AGENT_REASON)
workflow.add_edge(ACT, AGENT_REASON)

workflow.add_conditional_edges(
    AGENT_REASON,
    tools_condition,
    {"tools": ACT, END: END}
)tion is invoked first

Die Ausführung folgt vier Regeln. Der Graph beginnt am Reasoning-Node. Wenn die Antwort des LLM Toolaufrufe enthält, leitet tools_condition zum Act-Node weiter. Nachdem die Tools ausgeführt wurden, kehrt der Kontrollfluss stets zum Reasoning zurück. Wenn die Antwort keine Toolaufrufe enthält, endet der Graph und gibt die endgültige Antwort zurück. Das spiegelt den ReAct-Zyklus ohne manuellen Loop oder manuell verwaltete Nachrichten wider.

Kompilieren und ausführen. Durch das Kompilieren entsteht der Agent, und ein Diagramm des Graphen kann als PNG exportiert werden:

react_agent = workflow.compile()

react_agent.get_graph().draw_mermaid_png(output_file_path="graphs.png")

question = """
           tell me the first hotel you find in Boston if
            its temperature is lower than 30 degree."""


async def main():
    print("hello ReAct agent by Langraph")
    message = HumanMessage(content=question)
    res = await react_agent.ainvoke({"messages": [message]})


if __name__ == "__main__":
    asyncio.run(main())

Rufen Sie es mit einer Benachrichtigung für den Benutzer auf, und die resultierende Nachrichtenliste zeigt die Abwechslung der Schritte zwischen dem Reasoning-Node und dem Tool-Node, was einen funktionsfähigen mehrstufigen Ablauf darstellt.

Stand der Dinge

Drei LangGraph-Funktionen übernehmen die Arbeit: ToolNode zur Suche und Ausführung von Tools, tools_condition für die Schleifensteuerung sowie add_messages für die automatische Nachrichtenverarbeitung. Nachdem ReAct verstanden und implementiert wurde, geht es als nächstes darum, zu einem LangGraph RAG-Pipeline zurückzukehren und es vollständig agierend zu gestalten.