In diesem Artikel erfahren Sie, wie Sie zustandsgesteuerte Unterbrechungen in LangGraph implementieren, damit ein Agenten-Workflow zur menschlichen Genehmigung angehalten werden kann, bevor er mit der Ausführung fortfährt.
Zu den Themen, die wir behandeln werden, gehören:
- Was staatlich verwaltete Unterbrechungen sind und warum sie in Agenten-KI-Systemen wichtig sind.
- So definieren Sie einen einfachen LangGraph-Workflow mit einem gemeinsamen Agentenstatus und ausführbaren Knoten.
- So unterbrechen Sie die Ausführung, aktualisieren den gespeicherten Standing mit menschlicher Genehmigung und setzen den Workflow fort.
Lesen Sie weiter für alle Informationen.
Aufbau eines „Human-in-the-Loop“-Genehmigungsportals für autonome Agenten
Bild vom Herausgeber
Einführung
In Agentische KI-SystemeWenn die Ausführungspipeline eines Agenten absichtlich angehalten wird, liegt ein sogenannter a vor staatlich verwaltete Unterbrechung. Genau wie bei einem gespeicherten Videospiel wird der „Zustand“ eines pausierten Agenten – seine aktiven Variablen, Kontext, Speicher und geplanten Aktionen – dauerhaft gespeichert, wobei der Agent in einen Ruhe- oder Wartezustand versetzt wird, bis ein externer Auslöser seine Ausführung fortsetzt.
Die Bedeutung staatlich verwalteter Unterbrechungen hat aus mehreren Gründen mit dem Fortschritt bei hochautonomen, agentenbasierten KI-Anwendungen zugenommen. Sie fungieren nicht nur als wirksame Sicherheitsleitplanken für die Wiederherstellung nach andernfalls irreversiblen Aktionen in Situationen mit hohem Risiko, sondern sie ermöglichen es auch Human-in-the-Loop-Genehmigung und Korrektur. Ein menschlicher Vorgesetzter kann den Standing eines pausierten Agenten neu konfigurieren und unerwünschte Konsequenzen verhindern, bevor aufgrund einer falschen Reaktion Aktionen ausgeführt werden.
LangGrapheine Open-Supply-Bibliothek zum Erstellen von Stateful Massive Language Mannequin (LLM)-Anwendungen, unterstützt agentenbasierte Arbeitsabläufe mit Human-in-the-Loop-Mechanismen und zustandsverwalteten Unterbrechungen und verbessert so die Robustheit gegenüber Fehlern.
Dieser Artikel führt alle diese Elemente zusammen und zeigt Schritt für Schritt, wie zustandsgesteuerte Unterbrechungen mithilfe von LangGraph in Python im Rahmen eines Human-in-the-Loop-Ansatzes implementiert werden. Während die meisten der unten definierten Beispielprozesse dazu gedacht sind, von einem Agenten automatisiert zu werden, zeigen wir auch, wie man den Workflow an einem wichtigen Punkt stoppt, an dem eine menschliche Überprüfung erforderlich ist, bevor die Ausführung wieder aufgenommen wird.
Schritt-für-Schritt-Anleitung
Zuerst wir pip set up langgraph und nehmen Sie die notwendigen Importe für dieses Praxisbeispiel vor:
|
aus Tippen Import TypedDict aus Langgraph.Graph Import StateGraph, ENDE aus Langgraph.Kontrollpunkt.Erinnerung Import MemorySaver |
Beachten Sie, dass eine der importierten Klassen benannt ist StateGraph. LangGraph verwendet Zustandsgraphen um zyklische, komplexe Arbeitsabläufe zu modellieren, an denen Agenten beteiligt sind. Es gibt Staaten stellt den gemeinsam genutzten Speicher des Methods dar (auch bekannt als Datennutzlast) und Knoten Stellt Aktionen dar, die die Ausführungslogik definieren, die zum Aktualisieren dieses Standing verwendet wird. Sowohl Zustände als auch Knoten müssen explizit definiert und mit Prüfpunkten versehen werden. Machen wir das jetzt.
|
Klasse AgentState(TypedDict): Entwurf: str genehmigt: bool gesendet: bool |
Der Agentenstatus ist ähnlich wie ein Python-Wörterbuch strukturiert, da er von erbt TypedDict. Der Standing verhält sich wie unsere „Speicherdatei“, wenn er zwischen Knoten übertragen wird.
Was die Knoten betrifft, werden wir zwei davon definieren, von denen jeder eine Aktion darstellt: das Verfassen einer E-Mail und das Versenden dieser.
|
def draft_node(Zustand: AgentState): drucken(„(Agent): Verfassen der E-Mail …“) # Der Agent erstellt einen Entwurf und aktualisiert den Standing zurückkehren {„Entwurf“: „Hallo! Ihr Server-Replace ist bereit zur Bereitstellung.“, „genehmigt“: FALSCH, „gesendet“: FALSCH} def send_node(Zustand: AgentState): drucken(F„(Agent): Wache wieder auf! Genehmigungsstatus wird überprüft …“) Wenn Zustand.erhalten(„genehmigt“): drucken(„(System): E-MAIL SENDEN ->“, Zustand(„Entwurf“)) zurückkehren {„gesendet“: WAHR} anders: drucken(„(System): Entwurf wurde abgelehnt. E-Mail abgebrochen.“) zurückkehren {„gesendet“: FALSCH} |
Der draft_node() Die Funktion simuliert eine Agentenaktion, die eine E-Mail verfasst. Um den Agenten dazu zu bringen, eine echte Aktion auszuführen, würden Sie die ersetzen print() Anweisungen, die das Verhalten simulieren, mit tatsächlichen Anweisungen, die es ausführen. Das wichtigste hier zu beachtende Element ist das von der Funktion zurückgegebene Objekt: ein Wörterbuch, dessen Felder mit denen in der zuvor definierten Agentenstatusklasse übereinstimmen.
Mittlerweile ist die send_node() Funktion simuliert den Vorgang des Sendens der E-Mail. Aber es gibt einen Haken: Die Kernlogik für den Human-in-the-Loop-Mechanismus liegt hier, insbesondere in der Überprüfung des authorised Standing. Nur wenn die authorised Feld wurde auf eingestellt True – durch einen Menschen, wie wir sehen werden, oder durch einen simulierten menschlichen Eingriff – wird die E-Mail tatsächlich gesendet. Auch hier werden die Aktionen durch einfache Simulationen simuliert print() Der Einfachheit halber werden die folgenden Anweisungen verwendet, wobei der Schwerpunkt auf dem staatlich verwalteten Unterbrechungsmechanismus liegt.
Was brauchen wir noch? Ein Agenten-Workflow wird durch ein Diagramm mit mehreren verbundenen Zuständen beschrieben. Definieren wir eine einfache, lineare Abfolge von Aktionen wie folgt:
|
Arbeitsablauf = StateGraph(AgentState) # Aktionsknoten hinzufügen Arbeitsablauf.add_node(„draft_message“, draft_node) Arbeitsablauf.add_node(„Nachricht senden“, send_node) # Knoten durch Kanten verbinden: Begin -> Entwurf -> Senden -> Ende Arbeitsablauf.set_entry_point(„draft_message“) Arbeitsablauf.add_edge(„draft_message“, „Nachricht senden“) Arbeitsablauf.add_edge(„Nachricht senden“, ENDE) |
Um den datenbankähnlichen Mechanismus zu implementieren, der den Agentenstatus speichert, und um die zustandsgesteuerte Unterbrechung einzuführen, wenn der Agent im Begriff ist, eine Nachricht zu senden, verwenden wir diesen Code:
|
# MemorySaver ist wie unsere „Datenbank“ zum Speichern von Zuständen Erinnerung = MemorySaver() # DAS IST EIN WICHTIGER TEIL UNSERES PROGRAMMS: Dem Agenten sagen, er solle vor dem Senden eine Pause einlegen App = Arbeitsablauf.kompilieren( Kontrollpunkt=Erinnerung, interrupt_before=(„Nachricht senden“) ) |
Jetzt kommt die eigentliche Motion. Das werden wir Führen Sie den Aktionsgraphen aus vor ein paar Augenblicken definiert. Beachten Sie unten, dass eine Thread-ID verwendet wird, damit der Speicher den Workflow-Standing über alle Ausführungen hinweg verfolgen kann.
|
config = {„konfigurierbar“: {„thread_id“: „Demo-Thread-1“}} initial_state = {„Entwurf“: „“, „genehmigt“: FALSCH, „gesendet“: FALSCH} drucken(„n— ERSTES GRAPH WIRD AUSGEFÜHRT —“) # Das Diagramm führt „draft_node“ aus, erreicht dann den Haltepunkt und hält an. für Ereignis In App.Strom(initial_state, config): passieren |
Als nächstes kommt der „Human-in-the-Loop“-Second, bei dem der Fluss angehalten wird und die Zustimmung des Menschen durch die Einstellung simuliert wird authorised Zu True:
|
drucken(„n— GRAPH PAUSE —“) aktueller_Zustand = App.get_state(config) drucken(F„Nächster auszuführender Knoten: {current_state.subsequent}“) # Sollte „send_message“ anzeigen drucken(F„Aktueller Entwurf: ‚{current_state.values(‘draft‘)}‘“) # Simulation eines Menschen, der den E-Mail-Entwurf überprüft und genehmigt drucken(„n (Mensch): Entwurf wird überprüft… Sieht intestine aus. Genehmigend!“) # WICHTIG: Der Zustand wird mit der Entscheidung des Menschen aktualisiert App.update_state(config, {„genehmigt“: WAHR}) |
Dadurch wird das Diagramm fortgesetzt und die Ausführung abgeschlossen.
|
drucken(„n— GRAFIK FORTSETZEN —“) # Wir übergeben „None“, da die Eingabe dem Diagramm mitteilt, dass es einfach dort fortfahren soll, wo es aufgehört hat für Ereignis In App.Strom(Keiner, config): passieren drucken(„n— ENDGÜLTIGER STATUS —“) drucken(App.get_state(config).Werte) |
Das Ganze Ausgabe Das von diesem simulierten Workflow gedruckte Dokument sollte wie folgt aussehen:
|
—– LÄUFT ANFÄNGLICH GRAPH —– (Agent): Abfassung Die E-Mail... —– GRAPH PAUSE —– Nächste Knoten Zu ausführen: (‚Nachricht senden‘,) Aktuell Entwurf: ‚Hallo! Ihr Server-Replace ist bereit zur Bereitstellung.‘ (Menschlich): Überprüfung Entwurf... Aussehen Intestine. Zustimmend! —– WIEDERAUFNAHME GRAPH —– (Agent): Aufwachen zurück hoch! Überprüfung Genehmigung Standing... (System): SENDEN E-MAIL -> Hallo! Dein Server aktualisieren Ist bereit Zu Sei eingesetzt. —– FINALE ZUSTAND —– {‚Entwurf‘: ‚Hallo! Ihr Server-Replace ist bereit zur Bereitstellung.‘, ‚genehmigt‘: WAHR, ‚gesendet‘: WAHR} |
Zusammenfassung
In diesem Artikel wurde veranschaulicht, wie zustandsgesteuerte Unterbrechungen in agentenbasierten Arbeitsabläufen durch die Einführung von Human-in-the-Loop-Mechanismen implementiert werden können – eine wichtige Funktion in kritischen Szenarien mit hohem Risiko, in denen vollständige Autonomie möglicherweise nicht wünschenswert ist. Wir haben LangGraph, eine leistungsstarke Bibliothek zum Erstellen agentengesteuerter LLM-Anwendungen, verwendet, um einen von diesen Regeln gesteuerten Workflow zu simulieren.