Human-In-The-Loop logic in a cloud-hosted Agents/Tools application

[mrctito]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

N/A

Description

Hello, how do I implement a Human-In-The-Loop logic in a cloud-hosted Agents/Tools application with the front-end being WhatsApp or a WebChat, for example?
Thank you.

System Info

N/A

[dosubot[bot]]

To implement a Human-In-The-Loop logic in a cloud-hosted Agents/Tools application with the front-end being WhatsApp or a WebChat, you can follow these steps:

1. Setup the Environment:

   • Install necessary packages:

     %pip install --upgrade --quiet langchain-community langchain-openai tavily-python
     import dotenv
     dotenv.load_dotenv()

2. Initialize the Tools and Chat Model:

   • Initialize Tavily and an OpenAI chat model:

     from langchain_community.tools.tavily_search import TavilySearchResults
     from langchain_openai import ChatOpenAI
     
     tools = [TavilySearchResults(max_results=1)]
     chat = ChatOpenAI(model="gpt-4o-mini", temperature=0)

3. Create a Prompt Template:

   • Create a prompt template with placeholders for chat history:

     from langchain_core.prompts import ChatPromptTemplate
     
     prompt = ChatPromptTemplate.from_messages(
         [
             ("system", "You are a helpful assistant. You may not need to use tools for every query - the user may just want to chat!"),
             ("placeholder", "{messages}"),
             ("placeholder", "{agent_scratchpad}"),
         ]
     )

4. Assemble the Agent:

   • Create and assemble the agent:

     from langchain.agents import AgentExecutor, create_tool_calling_agent
     
     agent = create_tool_calling_agent(chat, tools, prompt)
     agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

5. Manage Chat History:

   • Optionally, wrap the agent executor to manage chat history:

     from langchain_community.chat_message_histories import ChatMessageHistory
     from langchain_core.runnables.history import RunnableWithMessageHistory
     
     demo_ephemeral_chat_history_for_chain = ChatMessageHistory()
     
     conversational_agent_executor = RunnableWithMessageHistory(
         agent_executor,
         lambda session_id: demo_ephemeral_chat_history_for_chain,
         input_messages_key="messages",
         output_messages_key="output",
     )

6. Invoke the Agent:

   • Interact with the agent:

     from langchain_core.messages import HumanMessage
     
     conversational_agent_executor.invoke(
         {"messages": [HumanMessage("I'm Nemo!")]},
         {"configurable": {"session_id": "unused"}},
     )

7. Integrate with WhatsApp or WebChat:

   • Use appropriate APIs or SDKs to connect your agent with WhatsApp or WebChat. For WhatsApp, you might use Twilio's API, and for WebChat, you might use a web socket or REST API to send and receive messages.

This setup allows you to create a conversational agent that can interact with users via WhatsApp or WebChat, with the ability to involve a human in the loop when necessary ^[1].

To continue talking to Dosu, mention @dosu.

[mrctito]

@dosu, Hello, you didn't answer my question. Human-In-The-Loop means that the agent requests confirmation from a human to process a specific tool. In other words, the human asks something, the agent start the process, and it is interrupted for a specific tool, then agent asks the human for additional information or confirmation, and then continues based on their response. Thank you.

[dosubot[bot]]

To implement a Human-In-The-Loop logic where the agent requests confirmation from a human to process a specific tool in a cloud-hosted Agents/Tools application with the front-end being WhatsApp or a WebChat, you can follow these steps:

1. Set up the environment: Ensure you have the necessary packages installed and environment variables set up.

2. Create the SQL query chain: Use LangChain to create a SQL query chain.

3. Add Human-In-The-Loop logic: Modify the chain to include a step where human confirmation is required before executing the query.

Here is an example implementation:

from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI
from langchain.chains import create_sql_query_chain
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain.callbacks import HumanApprovalCallbackHandler
import json

# Set up the database connection
db = SQLDatabase.from_uri("sqlite:///Chinook.db")

# Initialize the language model
llm = ChatOpenAI()

# Create the initial SQL query chain
chain = create_sql_query_chain(llm, db)

# Define the prompt for query validation
system = """Double check the user's {dialect} query for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins

If there are any of the above mistakes, rewrite the query.
If there are no mistakes, just reproduce the original query with no further commentary.

Output the final SQL query only."""
prompt = ChatPromptTemplate.from_messages(
    [("system", system), ("human", "{query}")]
).partial(dialect=db.dialect)
validation_chain = prompt | llm | StrOutputParser()

# Combine the query chain and validation chain
full_chain = {"query": chain} | validation_chain

# Custom exception for not approved actions
class NotApproved(Exception):
    """Custom exception."""

# Function to request human confirmation
def request_human_confirmation(query):
    # Send the query to the human via WhatsApp or WebChat
    # This is a placeholder for the actual implementation
    tool_strs = json.dumps({"query": query}, indent=2)
    input_msg = (
        f"Do you approve of the following tool invocation\n\n{tool_strs}\n\n"
        "Anything except 'Y'/'Yes' (case-insensitive) will be treated as a no.\n >>>"
    )
    resp = input(input_msg)
    if resp.lower() not in ("yes", "y"):
        raise NotApproved(f"Tool invocation not approved:\n\n{tool_strs}")
    return True

# Adding Human Approval Callback Handler
callbacks = [HumanApprovalCallbackHandler(approve=request_human_confirmation)]

# Example usage
query = full_chain.invoke(
    {
        "question": "What's the average Invoice from an American customer whose Fax is missing since 2003 but before 2010"
    }
)
print("Generated query:", query)

# Request human confirmation before executing the query
try:
    if request_human_confirmation(query):
        result = db.run(query)
        print("Query result:", result)
except NotApproved as e:
    print(e)

In this implementation:

• The request_human_confirmation function simulates sending the query to a human for confirmation via WhatsApp or WebChat.
• The human's response is captured and checked. If the response is not affirmative, a NotApproved exception is raised, stopping the execution.
• If the human confirms, the query is executed; otherwise, it is cancelled.
• The HumanApprovalCallbackHandler is used to enforce manual human approval of inputs going into the tool.

Additionally, you can use the following format instructions to ensure the agent's responses are structured correctly:

PREFIX = """Respond to the human as helpfully and accurately as possible. You have access to the following tools:"""
FORMAT_INSTRUCTIONS = """Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).

Valid "action" values: "Final Answer" or {tool_names}

Provide only ONE action per $JSON_BLOB, as shown:

{{{{
"action": $TOOL_NAME,
"action_input": $INPUT
}}}}

Follow this format:

Question: input question to answer
Thought: consider previous and subsequent steps
Action:

$JSON_BLOB

Observation: action result
... (repeat Thought/Action/Observation N times)
Thought: I know what to respond
Action:

{{{{
"action": "Final Answer",
"action_input": "Final response to human"
}}}}

SUFFIX = """Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:.
Thought:"""

This setup ensures that the agent can request human confirmation before proceeding with specific actions ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[mrctito]

@dosu, this example is not correct. This instruction does NOT work in a WebChat or WhatsApp. 'resp = input(input_msg)'

This example you provided, I already knew and understand how it works. My question is precisely to understand how to do something similar in a cloud environment, where the service running the Agent is stateless and the entry point for the service in the cloud is an HTTP request.

Thank you.

[dosubot[bot]]

To implement a Human-In-The-Loop logic in a cloud-hosted Agents/Tools application with a stateless service and HTTP request entry point, specifically for integration with WebChat or WhatsApp, you can follow these steps:

1. Setup Dependencies and Environment:

   • Install necessary packages and set up environment variables.

   %pip install --upgrade --quiet langchain langchain-community langchainhub langchain-chroma beautifulsoup4
   import getpass
   import os
   
   if not os.environ.get("OPENAI_API_KEY"):
       os.environ["OPENAI_API_KEY"] = getpass.getpass()

2. Load and Index Content:

   • Load, chunk, and index the content to create a retriever.

   import bs4
   from langchain_chroma import Chroma
   from langchain_community.document_loaders import WebBaseLoader
   from langchain_openai import OpenAIEmbeddings
   from langchain_text_splitters import RecursiveCharacterTextSplitter
   
   loader = WebBaseLoader(
       web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
       bs_kwargs=dict(
           parse_only=bs4.SoupStrainer(
               class_=("post-content", "post-title", "post-header")
           )
       ),
   )
   docs = loader.load()
   
   text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
   splits = text_splitter.split_documents(docs)
   vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
   retriever = vectorstore.as_retriever()

3. Create Question-Answering Chain:

   • Incorporate the retriever into a question-answering chain.

   from langchain_core.prompts import ChatPromptTemplate
   from langchain.chains.combine_documents import create_stuff_documents_chain
   from langchain.chains import create_retrieval_chain
   from langchain_openai import ChatOpenAI
   
   llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
   
   system_prompt = (
       "You are an assistant for question-answering tasks. "
       "Use the following pieces of retrieved context to answer "
       "the question. If you don't know the answer, say that you "
       "don't know. Use three sentences maximum and keep the "
       "answer concise."
       "\n\n"
       "{context}"
   )
   
   prompt = ChatPromptTemplate.from_messages(
       [
           ("system", system_prompt),
           ("human", "{input}"),
       ]
   )
   
   question_answer_chain = create_stuff_documents_chain(llm, prompt)
   rag_chain = create_retrieval_chain(retriever, question_answer_chain)

4. Add Chat History:

   • Update the prompt to support historical messages and create a history-aware retriever.

   from langchain_core.prompts import MessagesPlaceholder
   from langchain.chains import create_history_aware_retriever
   from langchain_core.runnables.history import RunnableWithMessageHistory
   from langchain_community.chat_message_histories import ChatMessageHistory
   from langchain_core.chat_history import BaseChatMessageHistory
   
   contextualize_q_system_prompt = (
       "Given a chat history and the latest user question "
       "which might reference context in the chat history, "
       "formulate a standalone question which can be understood "
       "without the chat history. Do NOT answer the question, "
       "just reformulate it if needed and otherwise return it as is."
   )
   contextualize_q_prompt = ChatPromptTemplate.from_messages(
       [
           ("system", contextualize_q_system_prompt),
           MessagesPlaceholder("chat_history"),
           ("human", "{input}"),
       ]
   )
   history_aware_retriever = create_history_aware_retriever(
       llm, retriever, contextualize_q_prompt
   )
   
   store = {}
   
   def get_session_history(session_id: str) -> BaseChatMessageHistory:
       if session_id not in store:
           store[session_id] = ChatMessageHistory()
       return store[session_id]
   
   conversational_rag_chain = RunnableWithMessageHistory(
       rag_chain,
       get_session_history,
       input_messages_key="input",
       history_messages_key="chat_history",
       output_messages_key="answer",
   )

5. Invoke the Chain:

   • Use the chain to handle user queries, maintaining session history.

   conversational_rag_chain.invoke(
       {"input": "What is Task Decomposition?"},
       config={
           "configurable": {"session_id": "abc123"}
       }
   )["answer"]
   
   conversational_rag_chain.invoke(
       {"input": "What are common ways of doing it?"},
       config={"configurable": {"session_id": "abc123"}},
   )["answer"]

This setup allows you to integrate with WebChat or WhatsApp by handling HTTP requests at the entry point, maintaining a stateless service while managing session history for each user interaction ^[1].

To continue talking to Dosu, mention @dosu.

[mrctito]

@dosu, you changed the scenario. It's not at all what I asked. Let me explain again:

I want to write a LangChain application that runs inside a cloud container. This application implements Agent/Tools.

My client is WhatsApp. The user will write a command, which will be sent to my application via HTTP, and within the entry point, I will execute the Agent/Tool. The agent will run, and when it selects a tool, it should ask the user for confirmation before executing.

So, what is the pattern for this situation?

[dosubot[bot]]

To write a LangChain application that runs inside a cloud container, where the client is WhatsApp, and the agent asks the user for confirmation before executing a selected tool, you can follow these steps:

1. Set up Azure Container Apps dynamic sessions:

   • Authenticate with Azure using DefaultAzureCredential.
   • Create a session pool and set the pool management endpoint.
   • Install the necessary packages.

2. Create a LangChain agent:

   • Use SessionsPythonREPLTool to interact with the Azure Container Apps.
   • Implement a mechanism to ask for user confirmation before executing a tool.

3. Integrate with WhatsApp:

   • Use a service like Twilio to handle WhatsApp messages.

Here is a simplified example:

import getpass
from langchain_azure_dynamic_sessions import SessionsPythonREPLTool
from langchain import hub
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
from twilio.rest import Client

# Set up Azure Container Apps dynamic sessions
POOL_MANAGEMENT_ENDPOINT = getpass.getpass()
tool = SessionsPythonREPLTool(pool_management_endpoint=POOL_MANAGEMENT_ENDPOINT)

# Set up LangChain agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_tool_calling_agent(llm, [tool], prompt)

agent_executor = AgentExecutor(
    agent=agent, tools=[tool], verbose=True, handle_parsing_errors=True
)

# Twilio setup for WhatsApp
account_sid = 'your_account_sid'
auth_token = 'your_auth_token'
client = Client(account_sid, auth_token)
from_whatsapp_number = 'whatsapp:+14155238886'
to_whatsapp_number = 'whatsapp:+your_number'

def send_whatsapp_message(body):
    message = client.messages.create(
        body=body,
        from_=from_whatsapp_number,
        to=to_whatsapp_number
    )
    return message.sid

def get_user_confirmation():
    send_whatsapp_message("Do you want to execute the tool? Reply with 'yes' or 'no'.")
    # Implement a way to receive and process the user's response
    # This is a placeholder for the actual implementation
    user_response = receive_whatsapp_message()
    return user_response.lower() == 'yes'

# Example usage
user_input = "what's sin of pi . if it's negative generate a random number between 0 and 5. if it's positive between 5 and 10."
send_whatsapp_message(f"Received input: {user_input}")

if get_user_confirmation():
    response = agent_executor.invoke({"input": user_input})
    send_whatsapp_message(f"Execution result: {response}")
else:
    send_whatsapp_message("Execution cancelled by the user.")

This example demonstrates how to set up the LangChain agent, integrate with Azure Container Apps, and use Twilio to handle WhatsApp messages. The get_user_confirmation function sends a confirmation message to the user and waits for their response before executing the tool.

Additionally, you can use the following format for the agent's prompt to ensure proper tool usage:

SYSTEM_MESSAGE_PREFIX = """Answer the following questions as best you can. You have access to the following tools:"""
FORMAT_INSTRUCTIONS = """The way you use the tools is by specifying a json blob.
Specifically, this json should have a `action` key (with the name of the tool to use) and a `action_input` key (with the input to the tool going here).

The only values that should be in the "action" field are: {tool_names}

The $JSON_BLOB should only contain a SINGLE action, do NOT return a list of multiple actions. Here is an example of a valid $JSON_BLOB:

{{{{
"action": $TOOL_NAME,
"action_input": $INPUT
}}}}

ALWAYS use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action:

$JSON_BLOB

Observation: the result of the action
... (this Thought/Action/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question"""
SYSTEM_MESSAGE_SUFFIX = """Begin! Reminder to always use the exact characters `Final Answer` when responding."""
HUMAN_MESSAGE = "{input}\n\n{agent_scratchpad}"

This setup ensures that the agent can request human confirmation before proceeding with specific actions ^[1][2].

To continue talking to Dosu, mention @dosu.

[habberrih]

Did you find any solutions to this? @mrctito
I'm using Rest API with NestJS and LangGraphJS

[mrctito]

No, I did not Em dom., 25 de mai. de 2025, 11:33, Abdullah Habberrih < ***@***.***> escreveu:

…

Did you find any solutions to this? @mrctito <https://github.com/mrctito> I'm using Rest API with NestJS and LangGraphJS — Reply to this email directly, view it on GitHub <#26823 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAGFVVS35Q53HMLCU5RDN3D3AHIDRAVCNFSM6AAAAABOYZSIFCVHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTGMRWGQYDGMY> . You are receiving this because you were mentioned.Message ID: ***@***.*** com>