Langgraph Human In The Loop mit Steckdose

Durch die Unterbrechungsfunktion von Langgraph habe ich gelernt, dass Menschen mitten in der Agentenausführung eingreifen können.

Aber wenn man sich die Beispiele ansieht, ignorieren sie alle einfach die menschliche Interaktion. Was muss ich tun, um tatsächlich eine Bestätigung vom Benutzer zu erhalten? Ich denke, es gibt drei Hauptwege.

Verwendung des Langgraph API-Servers

Sie können den Langgraph-API-Server mit Docker über die Langgraph-CLI ausführen, dann das Diagramm ausführen, den Status ändern und es mit dem Langgraph-SDK neu starten.

Die von langgraph bereitgestellten Artikel müssen in der vorgesehenen Weise verwendet werden. Es gibt viele Einstellungen und es scheint schwierig zu sein, sie in meinen Code zu integrieren

Diagrammverwaltung auf dem Server

Dies ist eine Methode, um nur die notwendigen Teile des oben genannten Langgraph-API-Servers auf meinem benutzerdefinierten Server zu implementieren. Wenn Sie beispielsweise ein Diagramm ausführen, müssen der Client, der das Diagramm ausgeführt hat, und der Diagrammprüfpunkt gespeichert werden. Nach der Bestätigung des Benutzers muss das Diagramm erneut geladen und der Status entsprechend der Antwort des Benutzers geändert werden, um erneut ausgeführt zu werden

Vielleicht gibt es viel zu bedenken.

Steckdosenanschluss

Beim Ausführen des Agenten verbindet er einen Socket und interagiert über den Socket mit dem Benutzer. Dies funktioniert durch einfaches Hinzufügen des Schritts zum Empfangen einer Benutzerbestätigung über eine Socket-Verbindung und eine Socket-Kommunikation im vorhandenen Beispielcode

Stattdessen kann es schwierig sein, Streaming zu implementieren, das wie das Tippen funktioniert.

Implementiert mit Socket-Verbindung

Zuerst wollte ich es so implementieren, dass die Komplexität möglichst nicht zunimmt, also habe ich es mit einer Socket-Verbindung implementiert

Der Server verwendet NestJs und der Client verwendet NextJs.

Server

Erstellen Sie zunächst ein Gateway für die Websocket-Verbindung. Beim Agenten/Start wurde eine Verbindung hergestellt und der Agent wurde sofort ausgeführt.

@WebSocketGateway({
  namespace: "/",
  transport: ["websocket", "polling"],
  path: "/agent/start",
  cors: {
    origin: "*",
    methods: ["GET", "POST"],
    credentials: true,
  },
})
export class AgentGateway implements OnGatewayConnection, OnGatewayDisconnect {
  @WebSocketServer()
  server: Server;
  protected readonly logger = new Logger(this.constructor.name);

  constructor(
    private readonly agentFactory: AgentFactory
  ) {}

  private pendingConfirmations = new Map<string boolean> void>();

  // Handle new connections
  handleConnection(client: Socket) {
    console.log(`Client connected: ${client.id}`);

    // Option 1: Get actionData from query parameters
    const actionData: { agent: AgentName } = client.handshake.query.actionData
      ? JSON.parse(client.handshake.query.actionData as string)
      : null;

    if (actionData) {
      this.startAgentProcess(client, actionData);
    } else {
      // If no actionData is provided, you can wait for an event
      client.emit("error", "No action data provided");
      client.disconnect();
    }
  }

  // Handle disconnections
  handleDisconnect(client: Socket) {
    console.log(`Client disconnected: ${client.id}`);
    this.pendingConfirmations.delete(client.id);
  }

  // Send confirmation request
  async sendConfirmationRequest(clientId: string, data: any): Promise<boolean> {
    return new Promise((resolve) => {
      this.pendingConfirmations.set(clientId, resolve);
      this.server.to(clientId).emit("confirmation_request", data);

      // Optional timeout for response
      setTimeout(() => {
        if (this.pendingConfirmations.has(clientId)) {
          this.pendingConfirmations.delete(clientId);
          resolve(false); // Default to 'false' if timeout occurs
        }
      }, 3000000); // 3000 seconds timeout
    });
  }

  // Handle client's confirmation response
  @SubscribeMessage("confirmation_response")
  handleClientResponse(
    @MessageBody() data: { confirmed: boolean },
    @ConnectedSocket() client: Socket
  ) {
    const resolve = this.pendingConfirmations.get(client.id);
    if (resolve) {
      resolve(data.confirmed);
      this.pendingConfirmations.delete(client.id);
    }
  }

  // Start the agent process
  private async startAgentProcess(
    client: Socket,
    actionData: { agent: AgentName }
  ) {
    const graph = await this.agentFactory.create({
      agentName: actionData.agent,
    });

    const initialInput = { input: "hello world" };

    // Thread
    const graphStateConfig = {
      configurable: { thread_id: "1" },
      streamMode: "values" as const,
    };

    // Run the graph until the first interruption
    for await (const event of await graph.stream(
      initialInput,
      graphStateConfig
    )) {
      this.logAndEmit(client, `--- ${event.input} ---`);
    }

    // Will log when the graph is interrupted, after step 2.
    this.logAndEmit(client, "---GRAPH INTERRUPTED---");

    const userConfirmed = await this.sendConfirmationRequest(client.id, {
      message: "Do you want to proceed with this action?",
      actionData,
    });

    if (userConfirmed) {
      // If approved, continue the graph execution. We must pass `null` as
      // the input here, or the graph will
      for await (const event of await graph.stream(null, graphStateConfig)) {
        this.logAndEmit(client, `--- ${event.input} ---`);
      }
      this.logAndEmit(client, "---ACTION EXECUTED---");
    } else {
      this.logAndEmit(client, "---ACTION CANCELLED---");
    }

    // Optionally disconnect the client
    client.disconnect();
  }

  private logAndEmit(client: Socket, message: string) {
    console.log(message);
    client.emit("message", { message });
  }
}
</boolean></string>

Der Schlüssel ist einfach. Wenn ein Socket verbunden ist, wird sofort ein Agent erstellt und ausgeführt. Wenn er unterbrochen wird, wird eine Bestätigungsanforderungsnachricht an den Client gesendet und gewartet. Sobald die Bestätigung gelöst ist, wird die Grafik fortgesetzt.

Der im obigen Code verwendete Agent ist ein Agent, der die Schritte 1 2 3 unten im Langgraph-Dokument nacheinander ausführt.

  const GraphState = Annotation.Root({
    input: Annotation<string>,
  });

  const step1 = (state: typeof GraphState.State) => {
    console.log("---Step 1---");
    return state;
  };

  const step2 = (state: typeof GraphState.State) => {
    console.log("---Step 2---");
    return state;
  };

  const step3 = (state: typeof GraphState.State) => {
    console.log("---Step 3---");
    return state;
  };

  const builder = new StateGraph(GraphState)
    .addNode("step1", step1)
    .addNode("step2", step2)
    .addNode("step3", step3)
    .addEdge(START, "step1")
    .addEdge("step1", "step2")
    .addEdge("step2", "step3")
    .addEdge("step3", END);

  // Set up memory
  const graphStateMemory = new MemorySaver();

  const graph = builder.compile({
    checkpointer: graphStateMemory,
    interruptBefore: ["step3"],
  });
  return graph;
</string>

Kunde

Der Client erstellt einen Hook, um den Agentenstart und seinen Status zu verwalten.

import { useRef, useState } from "react";
import io, { Socket } from "socket.io-client";

export const useAgentSocket = () => {
  const socketRef = useRef<socket null>(null);
  const [confirmationRequest, setConfirmationRequest] = useState<any>(null);
  const [messages, setMessages] = useState<string>([]);

  const connectAndRun = (actionData: any) => {
    return new Promise((resolve, reject) => {
      socketRef.current = io("http://localhost:8000", {
        path: "/agent/start",
        transports: ["websocket", "polling"],
        query: {
          actionData: JSON.stringify(actionData),
        },
      });

      socketRef.current.on("connect", () => {
        console.log("Connected:", socketRef.current?.id);
        resolve(void 0);
      });

      socketRef.current.on("connect_error", (error) => {
        console.error("Connection error:", error);
        reject(error);
      });

      // Listen for confirmation requests
      socketRef.current.on("confirmation_request", (data) => {
        setConfirmationRequest(data);
      });

      // Listen for messages
      socketRef.current.on("message", (data) => {
        console.log("Received message:", data);
        setMessages((prevMessages) => [...prevMessages, data.message]);
      });

      socketRef.current.on("disconnect", () => {
        console.log("Disconnected from server");
      });
    });
  };

  const sendConfirmationResponse = (confirmed: boolean) => {
    if (socketRef.current) {
      socketRef.current.emit("confirmation_response", { confirmed });
      setConfirmationRequest(null);
    }
  };

  const disconnectSocket = () => {
    if (socketRef.current) {
      socketRef.current.disconnect();
    }
  };

  const clearMessages = () => {
    setMessages([]);
  };

  return {
    confirmationRequest,
    sendConfirmationResponse,
    connectAndRun,
    disconnectSocket,
    messages,
    clearMessages,
  };
};
</string></any></socket>

Stellt eine Verbindung her und aktualisiert den Bestätigungsanfragestatus, wenn eine Bestätigungsanfrage eintrifft. Schauen Sie sich einfach den Bestätigungsanfragestatus in der UI-Komponente an und öffnen Sie ein Fenster für den Benutzer.

Das obige ist der detaillierte Inhalt vonLanggraph Human In The Loop mit Steckdose. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!