Erstellen von Human-In-The-Loop-Agenten-Workflows | Auf dem Weg zur Datenwissenschaft

wie GPT-5.4 von OpenAI und Opus 4.6 von Anthropic haben herausragende Fähigkeiten bei der Ausführung lang andauernder Agentenaufgaben bewiesen.

Infolgedessen sehen wir einen erhöhten Einsatz von LLM-Agenten über Einzel- und Unternehmensumgebungen hinweg, um komplexe Aufgaben zu erledigen, wie z. B. die Durchführung von Finanzanalysen, die Erstellung von Apps und die Durchführung umfangreicher Recherchen.

Diese Agenten können, unabhängig davon, ob sie Teil eines hochautonomen Setups oder eines vordefinierten Workflows sind, mithilfe von Instruments mehrstufige Aufgaben ausführen, um Ziele zu erreichen minimale menschliche Aufsicht.

„Minimal“ bedeutet jedoch nicht, dass keine menschliche Kontrolle erfolgt.

Im Gegenteil bleibt die menschliche Überprüfung aufgrund der inhärenten probabilistischen Natur von LLMs und des Fehlerpotenzials wichtig.

Diese Fehler können sich im Laufe des Arbeitsablaufs ausbreiten und verschlimmern, insbesondere wenn wir zahlreiche Agentenkomponenten aneinanderreihen.

Sie werden die beeindruckenden Fortschritte bemerkt haben, die Agenten im Codierungsbereich gemacht haben. Der Grund dafür ist, dass Code relativ einfach zu überprüfen ist (d. h. er wird entweder ausgeführt oder schlägt fehl und das Suggestions ist sofort sichtbar).

Aber in Bereichen wie der Erstellung von Inhalten, der Recherche oder der Entscheidungsfindung ist die Richtigkeit oft subjektiv und schwieriger automatisch zu bewerten.

Darum Human-in-the-Loop (HITL) Design bleibt entscheidend.

In diesem Artikel erklären wir Ihnen, wie Sie es verwenden LangGraph eine einrichten Human-in-the-Loop Agentischer Workflow für die Erstellung und Veröffentlichung von Inhalten Bluesky.

Inhalt

(1) Grundierung für LangGraph
(2) Beispiel-Workflow
(3) Schlüsselkonzepte
(4) Code-Komplettlösung
(5) Greatest Practices für Interrupts

Die beiliegende finden Sie hier GitHub-Repo hier.

(1) Einführung in LangGraph

LangGraph (Teil der LangChain Ökosystem) ist ein Low-Stage-Agenten-Orchestrierungs-Framework und eine Laufzeitumgebung zum Erstellen von Agenten-Workflows.

Aufgrund seines hohen Maßes an Kontrolle und Anpassbarkeit, das für produktionstaugliche Lösungen von entscheidender Bedeutung ist, ist es mein bevorzugtes Framework.

Während LangChain ein Middleware-Objekt anbietet (HumanInTheLoopMiddleware), um einen einfachen Einstieg in die menschliche Aufsicht bei Agentenanrufen zu ermöglichen, erfolgt dies auf einem hohen Abstraktionsniveau, das die zugrunde liegenden Mechanismen verdeckt.

Im Gegensatz dazu abstrahiert LangGraph weder die Eingabeaufforderungen noch die Architektur und gibt uns so das feinere Maß an Kontrolle, das wir benötigen. Damit können wir explizit Folgendes definieren:

• Wie Daten zwischen Schritten fließen
• Wo Entscheidungen und Codeausführungen stattfinden
• Wo menschliches Eingreifen erforderlich ist

Deshalb werden wir verwenden LangGraph das HITL-Konzept innerhalb eines zu demonstrieren Agenten-Workflow.

Es ist auch hilfreich, zwischen zu unterscheiden Agenten-Workflows Und autonome KI-Agenten.

Agenten-Workflows verfügen über vorgegebene Pfade und sind für die Ausführung in einer definierten Reihenfolge konzipiert, wobei LLMs und/oder Agenten in eine oder mehrere Komponenten integriert sind. Auf der anderen Seite, KI-Agenten selbstständig ein Ziel planen, ausführen und iterieren.

In diesem Artikel konzentrieren wir uns auf Agenten-Workflowsbei dem wir bewusst menschliche Kontrollpunkte in einen vordefinierten Ablauf einfügen.

(2) Beispiel-Workflow

Für unser Beispiel erstellen wir einen Workflow zur Generierung von Social-Media-Inhalten wie folgt:

1. Der Benutzer gibt ein interessantes Thema ein (z. B. „Neueste Nachrichten über Anthropic“).
2. Der Websuchknoten nutzt die Tavily Device zur On-line-Suche nach Artikeln, die zum Prime passen.
3. Das oberste Suchergebnis wird ausgewählt und in ein LLM eingespeist Knoten zur Inhaltserstellung um einen Social-Media-Beitrag zu erstellen.
4. Im Überprüfungsknotengibt es zwei Kontrollpunkte für die menschliche Überprüfung:
   (ich) Präsentieren Sie generierte Inhalte, damit Menschen diese genehmigen, ablehnen oder bearbeiten können.
   (ii) Nach der Genehmigung löst der Workflow Folgendes aus Bluesky API Werkzeug und bittet um eine endgültige Bestätigung, bevor es on-line veröffentlicht wird.

So sieht es aus, wenn es vom Terminal aus ausgeführt wird:

Und hier ist der Reside-Beitrag auf meinem Bluesky-Profil:

Bluesky ist eine soziale Plattform ähnlich wie Twitter (X) und wurde in dieser Demo ausgewählt, weil ihre API viel einfacher zugänglich und zu verwenden ist.

(3) Schlüsselkonzepte

Der Kernmechanismus hinter dem HITL-Setup in LangGraph ist das Konzept von unterbricht.

Unterbricht (mit interrupt() Und Command in LangGraph) ermöglichen es uns Diagrammausführung anhalten an bestimmten Stellen, Anzeige bestimmte Informationen an den Menschen, und erwarten ihre Eingaben, bevor der Arbeitsablauf fortgesetzt wird.

Befehl ist ein vielseitiges Objekt, mit dem wir den Diagrammstatus aktualisieren können (replace), geben Sie den nächsten auszuführenden Knoten an (goto), oder erfassen Sie den Wert, um die Diagrammausführung mit ( fortzusetzen)resume).

So sieht der Ablauf aus:

(1) Beim Erreichen der interrupt() Funktion, die Ausführung wird angehalten und die darin übergebene Nutzlast wird dem Benutzer angezeigt. Die Nutzlast ist eingetroffen interrupt sollte normalerweise im JSON- oder String-Format vorliegen, z. B.

choice = interrupt("Ought to we get KFC for lunch?") # String proven to person

(2) Nachdem der Benutzer geantwortet hat, übergeben wir die Antwortwerte an das Diagramm, um die Ausführung fortzusetzen. Es geht um die Verwendung Command und es ist resume Parameter als Teil des erneuten Aufrufs des Diagramms:

if human_response == "sure":
    return graph.invoke(Command(resume="KFC"))
else:
    return graph.invoke(Command(resume="McDonalds"))

(3) Der Antwortwert in resume wird in zurückgegeben choice Variable, die der Knoten für den Relaxation der Knotenausführung und den nachfolgenden Diagrammfluss verwendet:

if choice == "KFC":
    return Command(goto="kfc_order_node", replace={"lunch_choice": "KFC")
else:
    return Command(goto="mcd_order_node", replace={"lunch_choice": "McDonalds")

Interrupts sind dynamisch und können an einer beliebigen Stelle im Code platziert werden, im Gegensatz zu statischen Haltepunkten, die vor oder nach bestimmten Knoten festgelegt werden.

Allerdings platzieren wir Interrupts normalerweise entweder innerhalb der Knoten oder innerhalb der während der Diagrammausführung aufgerufenen Instruments.

Lassen Sie uns abschließend darüber reden Checkpointer. Wenn ein Workflow bei einer Unterbrechung pausiert, brauchen wir eine Möglichkeit dazu den aktuellen Zustand speichern damit es später weitergehen kann.

Wir brauchen daher einen Kontrollpunkt bestehen den Zustand, damit der Zustand während der Interrupt-Pause nicht verloren geht. Stellen Sie sich einen Kontrollpunkt vor als eine Momentaufnahme des Diagrammstatus zu einem bestimmten Zeitpunkt.

Für die Entwicklung ist es akzeptabel, den Zustand mit dem im Speicher zu speichern InMemorySaver Kontrollpunkt.

Für die Produktion ist es besser, Geschäfte wie zu nutzen Postgres oder Redis. In diesem Sinne verwenden wir die SQLite Kontrollpunkt in diesem Beispiel anstelle eines In-Reminiscence-Speichers.

Um die Grafik sicherzustellen setzt genau an der Stelle fort Wo der Interrupt aufgetreten ist, müssen wir ihn weitergeben und verwenden Thread-ID.

Stellen Sie sich einen Thread als eine einzelne Ausführungssitzung vor (wie eine separate Einzelkonversation), in der jeder Thread eine eindeutige ID hat und seinen eigenen Standing und Verlauf verwaltet.

Die Thread-ID wird übergeben config bei jedem Diagrammaufruf, damit LangGraph weiß, von welchem ​​Zustand aus es nach dem Interrupt fortfahren soll.

Nachdem wir uns nun mit den Konzepten von Interrupts befasst haben, CommandPrüfpunkte und Threads, kommen wir zur Code-Komplettlösung.

Da der Schwerpunkt auf der Human-in-the-Loop-Mechanik liegt, werden wir nicht auf die umfassende Code-Einrichtung eingehen. Besuchen Sie die GitHub-Repo für die vollständige Umsetzung.

(4) Code-Komplettlösung

(4.1) Ersteinrichtung

Wir beginnen mit der Set up der erforderlichen Abhängigkeiten und der Generierung von API-Schlüsseln für Bluesky, OpenAI, LangChain, LangGraph und Tavily.

# necessities.txt
langchain-openai>=1.1.9
langgraph>=1.0.8
langgraph-checkpoint-sqlite>=3.0.3
openai>=2.20.0
tavily-python>=0.7.21

# env.instance
export OPENAI_API_KEY=your_openai_api_key
export TAVILY_API_KEY=your_tavily_api_key
export BLUESKY_HANDLE=yourname.bsky.social
export BLUESKY_APP_PASSWORD=your_bluesky_app_password

(4.2) Zustand definieren

Wir haben das eingerichtet StateDabei handelt es sich um das gemeinsam genutzte, strukturierte Datenobjekt, das als zentraler Speicher des Diagramms dient. Es enthält Felder, die wichtige Informationen erfassen, etwa den Inhalt des Beitrags und den Genehmigungsstatus.

Der post_data Der Schlüssel ist, wo der generierte Beitragsinhalt gespeichert wird.

(4.3) Interrupt auf Knotenebene

Wir haben bereits erwähnt, dass Interrupts auf Knotenebene oder innerhalb von Toolaufrufen auftreten können. Sehen wir uns an, wie Ersteres funktioniert, indem wir den Knoten für die menschliche Überprüfung einrichten.

Der Zweck des Überprüfungsknotens besteht darin, die Ausführung anzuhalten und dem Benutzer den Entwurfsinhalt zur Überprüfung vorzulegen.

Hier sehen wir das interrupt() in Aktion (Zeilen 8 bis 13), wobei die Diagrammausführung im ersten Abschnitt der Knotenfunktion anhält.

Der particulars Schlüssel übergeben interrupt() enthält den generierten Inhalt, während die motion Style löst eine Handlerfunktion aus (handle_content_interrupt()) zur Unterstützung der Rezension:

Der generierte Inhalt wird im Terminal gedruckt, damit der Benutzer ihn anzeigen kann. Er kann ihn unverändert genehmigen, direkt ablehnen oder vor der Genehmigung direkt im Terminal bearbeiten.

Basierend auf der Entscheidung gibt die Handlerfunktion einen von drei Werten zurück:

• True (genehmigt),
• False (abgelehnt), oder
• Zeichenfolgenwert, der dem vom Benutzer bearbeiteten Inhalt entspricht (edited).

Dieser Rückgabewert wird mit an den Überprüfungsknoten zurückgegeben graph.invoke(Command=resume…)wodurch die Ausführung von dort aus fortgesetzt wird interrupt() wurde aufgerufen (Zeile 15) und bestimmt, welcher Knoten als nächstes aufgerufen werden soll: Inhalte genehmigen, ablehnen oder bearbeiten und mit der Genehmigung fortfahren.

(4.4) Interrupt auf Device-Ebene

Interrupts können auch auf Device-Aufrufebene definiert werden. Dies wird im nächsten menschlichen Überprüfungskontrollpunkt im demonstriert Knoten genehmigen bevor der Inhalt on-line auf Bluesky veröffentlicht wird.

Anstatt zu platzieren interrupt() Innerhalb eines Knotens platzieren wir ihn innerhalb des publish_post Device, das Beiträge über die Bluesky API erstellt:

Genau wie das, was wir auf Knotenebene gesehen haben, rufen wir eine Handlerfunktion auf (handle_publish_interrupt), um die menschliche Entscheidung zu erfassen:

Der Rückgabewert dieses Überprüfungsschritts ist entweder:

• {"motion": "verify"}oder
• {"motion": "cancel} ,

Der letzte Teil des Codes (d. h. ab Zeile 19) im publish_post Das Device verwendet diesen Rückgabewert, um zu bestimmen, ob mit der Veröffentlichung auf Bluesky fortgefahren werden soll oder nicht.

(4.5) Diagramm mit Checkpointer einrichten

Als nächstes verbinden wir die Knoten in einem Diagramm zur Kompilierung und führen einen SQLite-Prüfpunkt ein, um Schnappschüsse des Zustands bei jedem Interrupt zu erfassen.

SQLite erlaubt standardmäßig nur die Verwendung durch den Thread, der die Datenbankverbindung erstellt hat. Da LangGraph einen Thread-Pool für Checkpoint-Schreibvorgänge verwendet, müssen wir ihn festlegen check_same_thread=False um diesen Threads auch den Zugriff auf die Verbindung zu ermöglichen.

(4.6) Richten Sie den vollständigen Workflow mit Config ein

Nachdem das Diagramm fertig ist, platzieren wir es nun in einem Workflow, der die Pipeline zur Inhaltsgenerierung in Gang setzt.

Dieser Workflow umfasst die Konfiguration einer Thread-ID, die an jeden Thread übergeben wirdgraph.invoke(). Diese ID ist die Verbindung, die die Aufrufe miteinander verbindet, sodass das Diagramm bei einer Unterbrechung anhält und an der Stelle fortfährt, an der es aufgehört hat.

Vielleicht ist Ihnen das aufgefallen __interrupt__ Geben Sie den Code oben ein. Es ist einfach ein spezieller Schlüssel, den LangGraph dem Ergebnis hinzufügt, wann immer ein interrupt() wird getroffen.

Mit anderen Worten, es ist das Primäres Sign, das angibt, dass die Diagrammausführung angehalten wurde und wartet auf menschliche Eingaben, bevor es fortfährt.

Durch Platzieren __interrupt__ als Teil einer whereas Schleife bedeutet, dass die Schleife ständig prüft, ob noch ein Interrupt läuft. Sobald der Interrupt behoben ist, verschwindet der Schlüssel und die while-Schleife wird beendet.

Wenn der Workflow abgeschlossen ist, können wir ihn wie folgt ausführen:

run_hitl_workflow(question="newest information about Anthropic")

(5) Greatest Practices für Interrupts

Während Interrupts bei der Aktivierung von HITL-Workflows sehr hilfreich sind, können sie bei falscher Verwendung störend sein.

Daher empfehle ich die Lektüre LangGraph-Dokumentation. Hier sind einige praktische Regeln, die Sie beachten sollten:

• Schließen Sie Interrupt-Aufrufe nicht in Strive/Besides-Blöcke ein, da sie sonst die Ausführung nicht ordnungsgemäß anhalten
• Halten Sie Unterbrechungsanrufe jedes Mal in der gleichen Reihenfolge und überspringen Sie sie nicht und ordnen Sie sie nicht neu an
• Übergeben Sie nur JSON-sichere Werte an Interrupts und vermeiden Sie komplexe Objekte
• Stellen Sie sicher, dass jeder Code vor einem Interrupt sicher mehrmals ausgeführt werden kann (dh Idempotenz), oder verschieben Sie ihn nach dem Interrupt

Ich hatte zum Beispiel ein Downside im Internet-Suchknoten, bei dem ich direkt nach der Tavily-Suche eine Unterbrechung platzierte. Die Absicht bestand darin, eine Pause einzulegen und den Benutzern die Möglichkeit zu geben, die Suchergebnisse zur Erstellung von Inhalten zu überprüfen.

Da Interrupts jedoch dadurch funktionieren, dass sie die Knoten, von denen sie aufgerufen wurden, erneut ausführen, hat der Knoten einfach die Websuche erneut ausgeführt und einen anderen Satz von Suchergebnissen weitergeleitet als die, die ich zuvor genehmigt habe.

Daher funktionieren Interrupts am besten als Tor vor einer Aktion, aber wenn wir sie nach einem nicht deterministischen Schritt (wie einer Suche) verwenden, müssen wir das Ergebnis beibehalten, sonst riskieren wir, dass bei der Wiederaufnahme etwas anderes angezeigt wird.

Zum Abschluss

Die menschliche Überprüfung kann bei Agentenaufgaben wie ein Engpass erscheinen, bleibt jedoch kritisch, insbesondere in Bereichen, in denen die Ergebnisse subjektiv oder schwer zu überprüfen sind.

LangGraph erleichtert die Erstellung von HITL-Workflows mit Interrupts und Checkpoints.

Daher besteht die Herausforderung darin, zu entscheiden, wo diese menschlichen Entscheidungspunkte platziert werden sollen, um ein gutes Gleichgewicht zwischen Kontrolle und Effizienz zu finden.