Langgraph Human In The Loop dengan soket

Melalui fungsi gangguan langgraph, saya mengetahui bahawa manusia boleh campur tangan di tengah-tengah pelaksanaan Agen.

Tetapi jika anda melihat contoh, semuanya mengabaikan interaksi manusia. Apakah yang perlu saya lakukan untuk mendapatkan pengesahan daripada pengguna? Saya rasa ada tiga cara utama.

Menggunakan Pelayan API Langgraph

Anda boleh menjalankan pelayan API langgraph dengan docker menggunakan langgraph cli, kemudian jalankan graf, tukar keadaan dan mulakan semula dengan langgraph SDK.

Item yang disediakan oleh langgraph mesti digunakan mengikut cara yang disediakan. Terdapat banyak tetapan, dan nampaknya sukar untuk menyepadukannya dengan kod saya

Pengurusan graf pada pelayan

Ini ialah kaedah untuk melaksanakan hanya bahagian yang diperlukan pelayan API Langgraph di atas pada pelayan tersuai saya. Sebagai contoh, apabila menjalankan graf, klien yang melaksanakan graf dan pusat pemeriksaan graf mesti disimpan, dan selepas pengesahan pengguna, graf mesti dimuatkan semula dan keadaan berubah mengikut respons pengguna untuk dijalankan semula

Mungkin banyak perkara yang perlu difikirkan.

sambungan soket

Apabila melaksanakan Ejen, ia menyambungkan soket dan berinteraksi dengan pengguna melalui soket. Ia berfungsi dengan hanya menambah langkah menerima pengesahan pengguna melalui sambungan soket dan komunikasi soket dalam kod contoh sedia ada

Sebaliknya, mungkin sukar untuk melaksanakan penstriman yang ditaip seperti menaip.

Dilaksanakan dengan sambungan soket

Pertama sekali, saya ingin melaksanakannya dengan cara yang tidak meningkatkan kerumitan sebanyak mungkin, jadi saya melaksanakannya dengan sambungan soket

Pelayan menggunakan NestJs dan pelanggan menggunakan NextJs.

pelayan

Pertama, buat get laluan untuk sambungan Websocket. Sambungan telah dibuat pada ejen/mula dan ejen telah dilaksanakan serta-merta.

Kuncinya mudah. Apabila soket disambungkan, ejen segera dibuat dan dilaksanakan, dan apabila ia terganggu, mesej permintaan pengesahan dihantar kepada pelanggan dan menunggu. Setelah pengesahan diselesaikan, graf diteruskan.

@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, (response: 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 });
  }
}

Ejen yang digunakan dalam kod di atas ialah ejen yang secara berurutan menggunakan langkah 1 2 3 di bawah dalam dokumen langgraph.

pelanggan

  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;

Pelanggan mencipta cangkuk untuk menguruskan permulaan ejen dan statusnya.

Mewujudkan sambungan dan mengemas kini status Permintaan pengesahan apabila permintaan pengesahan tiba. Lihat sahaja status Permintaan pengesahan dalam komponen UI dan timbulkan tetingkap kepada pengguna.

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,
  };
};

Atas ialah kandungan terperinci Langgraph Human In The Loop dengan soket. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!