LangGraph: Agenten mit Tool Calling und begrenzter Ausführung für Code-Verifikation erstellen

Autonome Code-Modifikationen erfordern mehr als eine klassische lineare Prompt-Kette. Wenn ein Agent ein Repository bearbeitet, muss er Dateien einlesen, Änderungen anwenden, Test-Suites ausführen, Ausgaben analysieren und sich bei auftretenden Regressionen selbst korrigieren. Allerdings bergen zyklische Agentenmuster – die häufig auf dem ReAct-Paradigma (Reasoning and Acting) basieren – operationelle Risiken, wenn sie unkontrolliert ablaufen. Trifft ein Agent auf fehlerhafte Abhängigkeiten oder dauerhaft fehlschlagende Tests, kann er in kostspielige, endlose Ausführungsschleifen geraten.

Auf Basis der von Google AI dokumentierten ReAct-Architektur für LangGraph überträgt diese Anleitung grundlegende State-Graph-Mechanismen auf einen spezialisierten, in der Ausführung begrenzten Coding-Agenten. Die Implementierung führt Code-Änderungen durch, verifiziert diese anhand automatisierter Test-Suites und setzt ein striktes Iterationsbudget durch.

---

Kernarchitektur: State, Nodes und Conditional Edges

LangGraph modelliert Agenten-Workflows als gerichtete Graphen. Anders als statische gerichtete azyklische Graphen (DAGs) unterstützt LangGraph von Haus aus Zyklen, die für das iterative Schlussfolgern von Agenten unerlässlich sind.

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

Der Graph stützt sich auf drei architektonische Kernelemente:

1. State: Eine zentralisierte, typisierte Datenstruktur, die den Nachrichtenverlauf, Ausführungsmetadaten und explizite Schrittzähler speichert.
2. Nodes: Reguläre Python-Funktionen, die den aktuellen Zustand empfangen, isolierte Berechnungen ausführen (Modellaufruf oder Tool-Ausführung) und aktualisierte Zustandsattribute zurückgeben.
3. Conditional Edges: Entscheidungsfunktionen, die den Zustand prüfen (beispielsweise das Vorhandensein von Tool Calls oder das Überschreiten eines Iterationslimits), um zu bestimmen, ob der Ablauf zum Tool-Runner verzweigt oder bei END beendet wird.

---

Voraussetzungen und Abhängigkeiten

Stellen Sie sicher, dass Ihre Entwicklungsumgebung die erforderlichen Pakete für die Graph-Orchestrierung und Modelltreiber enthält:

pip install langgraph langchain-core langchain-openai

Um den Agenten über ein OpenAI-kompatibles Gateway mit Inferenz-Providern zu verbinden, konfigurieren Sie Ihre Umgebungsvariablen. Die Nutzung von OneMux vereinfacht das zentrale Key-Management und die Ausgabenkontrolle über mehrere Modelle hinweg:

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(Hinweis: Prüfen Sie stets, ob das gewählte Zielmodell strukturierte Tool-Calling-Parameter über OpenAI-kompatible Endpunkte unterstützt).

---

Schritt 1: Begrenzten Agenten-State definieren

In standardmäßigen Chat-Workflows übernimmt MessagesState das Anhängen des Gesprächsverlaufs. Für Programmieraufgaben definieren wir AgentState mithilfe von Pythons TypedDict explizit, um Nachrichtensequenzen zusammen mit einem Zähler (step_count) und einem Höchstwert (max_steps) zu verwalten.

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

Der add_messages-Reducer stellt sicher, dass neue Nachrichtenlisten sauber an den bisherigen Konversationsverlauf angehängt werden, anstatt vorherige Beobachtungen zu überschreiben.

---

Schritt 2: Dateisystem- und Verifikations-Tools implementieren

Der Agent benötigt Werkzeuge, um Quelldateien einzusehen, Änderungen vorzunehmen und Repository-Tests auszuführen. Diese Operationen kapseln wir mit dem @tool-Dekorator von LangChain.

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

Schritt 3: Modell-Setup über die Unified API

Wir initialisieren das Modell mit ChatOpenAI und verweisen dabei auf den Proxy-Endpunkt des Unified Providers. Konsultieren Sie die OneMux Docs und den Quickstart Guide, um die Parameterkompatibilität für Function Calling sicherzustellen.

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

Schritt 4: Graph-Nodes und Schleifen-Schutzmechanismen aufbauen

Nodes führen die Programmlogik aus und inkrementieren die Zähler. Verweilt ein Agent in einer wiederholten Schleife, ohne einen fehlgeschlagenen Test zu beheben, erkennt die Kantenlogik die Bedingung step_count >= max_steps und leitet den Ablauf direkt zum Abschluss des Graphen weiter.

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

Schritt 5: Graph-Kompilierung und Zusammenbau

Fügen Sie die Komponenten mittels StateGraph zusammen und kompilieren Sie den Workflow:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

Schritt 6: Code-Änderungen mit Test-Verifikation ausführen

Um den Workflow zu starten, definieren Sie eine Programmieraufgabe zusammen mit strikten Ausführungsgrenzen:

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

Grenzen und Absicherung für den Produktivbetrieb

Beachten Sie beim Einsatz von Code-modifizierenden Agenten in Produktionsumgebungen folgende Schutzmaßnahmen:

• Sandbox-Ausführung: Das lokale Ausführen beliebiger Shell-Befehle wie pytest birgt das Risiko von Umgebungsbeschädigungen oder unsicherer Skriptausführung. Isolieren Sie Testläufe daher in kurzlebigen Docker-Containern oder sicheren Sandboxes.
• Token-Limits im State: Umfangreiche Tool-Ausgaben – wie mehrseitige Stacktraces – können den Kontextrahmen schnell erschöpfen. Kürzen Sie Testausgaben sinnvoll, bevor sie in die State-Historie geschrieben werden.
• Modellkosten steuern: Iterative ReAct-Schleifen vervielfachen den Token-Verbrauch. Überwachen Sie die Nutzung über das OneMux Pricing-Dashboard und evaluieren Sie kostengünstigere Reasoning-Modelle unter OneMux Models, um die Kosten pro Aufgabe zu optimieren.

---

Quellen

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Aktualisiert am 2. September 2026).