Human in Loop and FastAPI Integration

[BharahthyKannan]

Trying to implement the human in-loop feature with Fast API. I have two endpoints.
The first endpoint, issues Interrupt
second endpoint with the same Checkpoint and thread ID I need to use ainvoke with "command" to resume.

I can get interrupt in the first endpoint. The second endpoint does not have a reference to the previous graph object to issue command. Is there any example I can refer to?

[Alsouidan]

@BharahthyKannan Would you like to update us with what you have reached in this issue?

[Alsouidan]

@Alsouidan are you using any external storage or in memory ?

In memory for now

[KiranCNayak]

@BharahthyKannan, if possible, can you give a gist of how you have implemented Fast API on top of LangGraph's HITL? I have a similar usecase that I need to implement.

[YassinNouh21]

U could use command

[udaylunawat]

Would love to know the same!

[YassinNouh21]

quick update u can use invoke with command and the problem will be solved and just add new attibute for human_input with true and false from this u will wait for the graph to compile

[JuGit-pk]

See what i have been implemented, its the mock for listening for the user input,

def process_chunks(chunks):
for chunk in chunks:
for node_id, value in chunk.items():
if node_id == "interrupt":
print(value[0].value)
user_input = input(value[0].value)
new_chunks = graph.stream(Command(resume=user_input), config=config)
process_chunks(new_chunks)
else:
print("node_id", node_id)
# print("value", value)
if value.get("messages"):
for message in value["messages"]:
message.pretty_print()

chunks = graph.stream({"messages": HumanMessage(content="First Message to Initiate the State")}, config=config)

process_chunks(chunks)

[CarryCKW]

1. when you use interrupt() func somewhere in your langgraph node, the raised GraphInterruptexception will break the http stream response;
2. then you can catch the event_data from code async for agent, _, event_data in graph.astream() and check whether the isinstance(event_data, dict) and event_data.get("__interrupt__", "") , then you can yield to http resopnse;
3. to resume the langgraph runs, just send http request again with the same chat_id which equals to the thread_id of runtime config from last conversation.

[CarryCKW]

1. interrupt_input = interrupt({
    "agent": "script_input",
   

   })

2.     async with get_maintenance_graph(agent_name="maintenance", chat_id=request.chat_id) as graph:
        async for agent, _, event_data in graph.astream(
                input=input_,
                config=config_,
                stream_mode=["messages", "updates", "custom"],
                subgraphs=True
        ):
            if isinstance(event_data, dict):
                if event_data.get("__interrupt__", ""):
                    ive = event_data.get("__interrupt__")[0].value
                    yield _make_event("interrupt", {"somekey": "somevalue"})
                continue
   

3. input_ = Command(resume={"input": request.file_content})
   config_ = {
   "configurable": {
   "thread_id": request.chat_id, # the same chat_id which equals to the thread_id of runtime config from last conversation.
   },
   "recursion_limit": 100,
   }
   async for agent, , event_data in graph.astream(
   input=input,
   config=config_,
   stream_mode=["messages", "updates", "custom"],
   subgraphs=True
   ):