how to intercept human_input callback

[obaid]

In my example below, the custom_ask_human_input method never gets triggered the when the agent goes into human_tool. Any ideas why?

import time
import requests
from langchain_openai import ChatOpenAI
from crewai import Agent, Task, Crew, Process
from crewai.agents import CrewAgentExecutor
from langchain_community.agent_toolkits.load_tools import load_tools

# Custom function to handle human input
def custom_ask_human_input(self, final_answer: dict) -> str:
    prompt = self._i18n.slice("getting_input").format(final_answer=final_answer)
    
    # Do some custom logic here....

    # For now, just return a placeholder response
    return "Human input handled by backend"

# Override the method in CrewAgentExecutor
CrewAgentExecutor._ask_human_input = custom_ask_human_input



# Define agents and tasks
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
human_tools = load_tools(["human"])

interviewer = Agent(
    role='Interviewer',
    goal='making questions about business to human',
    backstory="You are an interviewer of businessmen.",
    llm=llm,
    verbose=True,
    allow_delegation=False,
    tools=human_tools
)

writer = Agent(
    role='Writer',
    goal='Summarizing interview',
    backstory="You are a writer and your expertise lies in summarizing interviews.",
    llm=llm,
    verbose=True,
    allow_delegation=False
)

interviewer_task = Task(
    description="""
Your task is to interview successful entrepreneurs. 
Ask three business questions. Ask one question at a time.
When three questions have been answered by the interviewee, conclude by responding with a list of questions and answers, followed by "finish" on a new line.
To ask a question, use the "human" tool and provide the question to be asked as the "Action Input".
""",
    max_inter=3,
    expected_output="a coherent question",
    agent=interviewer,
    tools=human_tools,
    human_input=True  # Enable human input
)

writer_task = Task(
    description="Summarize the content of the provided interview",
    max_inter=3,
    agent=writer,
    expected_output="a coherent summary",
    context=[interviewer_task]
)

items_crew = Crew(
    agents=[interviewer, writer],
    tasks=[interviewer_task, writer_task],
    process=Process.sequential,
    memory=True,
    verbose=True,
    full_output=True
)

# Function to start the crew
def start_crew():
    response = items_crew.kickoff()
    print("## Final Result\n" + response)

# Start the crew
start_crew()

[theCyberTech]

_ask_human_input is a method in base_agent_executor_mixin.py not CrewAgentExecutor

Also I think you might be calling CrewAgentExecutor wrong. It lives in executor.py in agents folder

[BokisAngelov]

Hi,

A bit late on this, but since I found a workaround, someone else may come to this post.

First of all, I'm not using any tools, instead, I'm overriding the core function _ask_human_iput.

First, you need to import correctly CrewAgentExecutorMixin, where the function is implemented and then just make the adjustment below.

Change this:
CrewAgentExecutor._ask_human_input = custom_ask_human_input

to this:
CrewAgentExecutorMixin._ask_human_input = custom_ask_human_input

[lucasgomide]

What's your use case? Would it be resolved by emitting events about HIL?

[obaid]

@lucasgomide what would be the correct way to implement this behaviour?