langgraph/how-tos/human_in_the_loop/edit-graph-state/

[giscus[bot]]

langgraph/how-tos/human_in_the_loop/edit-graph-state/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/edit-graph-state/

[dangchinh25]

Is it possible to update a value of a state? Per my understanding, using .updateState feels like we are simulating a node call ourselves, not modifying a specific state. If I have a tool run which returns a result, can I modify that result before continuing?

Edit: This is what I'm prototyping right now, is this the intended pattern?

import "dotenv/config";

import {
  BaseMessage,
  HumanMessage,
  ToolMessage,
} from "@langchain/core/messages";
import {
  MemorySaver,
  START,
  StateGraph,
  StateGraphArgs,
} from "@langchain/langgraph";
import { executeWithElapsedTime, invokeGraph, prettyPrint } from "../utils";
import { IState } from "./common/types";
import { callModelNode, routeMessage, toolNode } from "./common/nodes";
import { StateSnapshot } from "@langchain/langgraph/dist/pregel/types";
import { RunnableConfig } from "@langchain/core/runnables";

const graphState: StateGraphArgs<IState>["channels"] = {
  messages: {
    reducer: (x: BaseMessage[], y: BaseMessage[]) => x.concat(y),
    default: () => [],
  },
};

const main = async () => {
  const workflow = new StateGraph<IState>({
    channels: graphState,
  })
    .addNode("agent", callModelNode)
    .addNode("tools", toolNode)
    .addEdge(START, "agent")
    .addConditionalEdges("agent", routeMessage)
    .addEdge("tools", "agent");

  // **Persistence**
  // Human-in-the-loop workflows require a checkpointer to ensure
  // nothing is lost between interactions
  const checkpointer = new MemorySaver();

  // **Interrupt**
  // To always interrupt before a particular node, pass the name of the node to `interruptBefore` when compiling.
  const graph = workflow.compile({
    checkpointer,
    interruptAfter: ["tools"],
  });
  let config: RunnableConfig = {
    configurable: { thread_id: "example-thread-1" },
  };

  // 1st interaction
  console.log("---First iteraction---");
  let inputState: IState = {
    messages: [new HumanMessage("What is the current weather like in SF")],
  };
  await invokeGraph<IState, IState>(graph, inputState, config);

  // Simulating user view the output of the tool call and then manually updating the output
  // We are replacing the last state with the new content from the user
  const stateHistory = graph.getStateHistory(config);

  let currentState: StateSnapshot | undefined;
  let previousState: StateSnapshot | undefined;
  let stateIndex = 0;

  for await (const state of stateHistory) {
    if (stateIndex === 0) {
      currentState = state;
    }
    if (stateIndex === 1) {
      previousState = state;
      break;
    }
    stateIndex++;
  }

  if (previousState && currentState) {
    const currentStateValues = currentState.values as IState;
    const lastMessage =
      currentStateValues.messages[currentStateValues.messages.length - 1];

    if (
      lastMessage._getType() === "tool" &&
      (lastMessage as ToolMessage).tool_call_id
    ) {
      const newMessage = new ToolMessage(
        "It's cold outside, with a high of 15C",
        (lastMessage as ToolMessage).tool_call_id
      );

      await graph.updateState(
        previousState.config,
        {
          messages: [newMessage],
        },
        "tools"
      );

      console.log("---Updated state---");
      prettyPrint(newMessage);
    }
  }

  // 2nd interaction, resuming the graph
  // Running an interrupted graph with null as the input means to "proceed as if the interruption didn't occur."
  console.log("---Second iteraction---");
  await invokeGraph<IState, null>(graph, null, config);
};

executeWithElapsedTime(main);

[hwchase17]

using .updateState feels like we are simulating a node call ourselves, not modifying a specific state

This is correct - calling this method will append a new state checkpoint, not modify an existing checkpoint

If I have a tool run which returns a result, can I modify that result before continuing?

In order to do this, you need to make sure the reducer that you define for the state object support this. Read more about reducers here: https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers

The MessagesState (https://langchain-ai.github.io/langgraph/concepts/low_level/#messagestate) actually supports exactly what you are asking for. It does this by using the IDs of the messages to overwrite messages. So for your example, you would pass in a ToolMessage with the same ID as the one you want to overwrite, and then the new state would have that updated message

[dangchinh25]

Do we currently have any guides for this? If not, can you provide a small snippet for it here?

[luisramirez7]

+1

[edsonbersan]

Is it possible to override reducers? If a reducer only support list appending, would list overriting be possible by a HITL?

[edsonbersan]

*overriding

[dangchinh25]

Yes, if you use regular StateGraph instead of MessageGraph you can declare your own reducer

[JianXiao2021]

Hello! I would like to ask if there is an asynchronous version of the function update_state ? I want to clear some temporary state data after streaming the output, but when I run the code I got this error when executing update_state:

The AsyncSqliteSaver does not support synchronous methods. Consider using the SqliteSaver instead.

A simplified version of my code:

from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver
from utils.helper import load_config
from langchain_core.messages import HumanMessage
import asyncio
from graph.graph import create_graph

def clean_tmp_state(app, config):
    app.update_state(
        config,
        {
            "tmp_result": "",
            "tmp_history": []
        }
    )

async def main():
    load_config("config.yaml")
    graph = create_graph(model_name)
    memory = AsyncSqliteSaver.from_conn_string(":memory:")
    app = graph.compile(checkpointer=memory)
    config = {"configurable": {"thread_id": "1"}}

    while True:
        user_message = input("User:")
        print("AI:")
        async for event in app.astream_events({"messages": HumanMessage(content=user_message)}, config=config, version="v2"):
		    // code for streaming output

        clean_tmp_state(app, config)

if __name__ == "__main__":
    asyncio.run(main())

[pcliupc]

there is an aupdate_state function.

[saiadityavzure]

How to delete an entire state with a certain thread_id ? Is there a way?
How to delete a thread id state from the MongoDB?