Brandon/cre 509 hitl multiple rounds of followup (#1702)

* v1 of HITL working

* Drop print statements

* HITL code more robust. Still needs to be refactored.

* refactor and more clear messages

* Fix type issue

* fix tests

* Fix test again

* Drop extra print
This commit is contained in:
Brandon Hancock (bhancock_ai)
2024-12-05 10:14:04 -05:00
committed by GitHub
parent fa373f9660
commit 611526596a
6 changed files with 403 additions and 89 deletions

View File

@@ -3,19 +3,20 @@
import os
from unittest import mock
from unittest.mock import patch
import pytest
from crewai import Agent, Crew, Task
from crewai.agents.cache import CacheHandler
from crewai.agents.crew_agent_executor import CrewAgentExecutor
from crewai.agents.parser import AgentAction, CrewAgentParser, OutputParserException
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
from crewai.llm import LLM
from crewai.tools import tool
from crewai.tools.tool_calling import InstructorToolCalling
from crewai.tools.tool_usage import ToolUsage
from crewai.tools.tool_usage_events import ToolUsageFinished
from crewai.utilities import RPMController
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
from crewai.utilities.events import Emitter
@@ -985,8 +986,7 @@ def test_agent_definition_based_on_dict():
# test for human input
@pytest.mark.vcr(filter_headers=["authorization"])
def test_agent_human_input():
from unittest.mock import patch
# Agent configuration
config = {
"role": "test role",
"goal": "test goal",
@@ -995,6 +995,7 @@ def test_agent_human_input():
agent = Agent(**config)
# Task configuration with human input enabled
task = Task(
agent=agent,
description="Say the word: Hi",
@@ -1002,11 +1003,26 @@ def test_agent_human_input():
human_input=True,
)
with patch.object(CrewAgentExecutor, "_ask_human_input") as mock_human_input:
mock_human_input.return_value = "Don't say hi, say Hello instead!"
# Side effect function for _ask_human_input to simulate multiple feedback iterations
feedback_responses = iter(
[
"Don't say hi, say Hello instead!", # First feedback
"looks good", # Second feedback to exit loop
]
)
def ask_human_input_side_effect(*args, **kwargs):
return next(feedback_responses)
with patch.object(
CrewAgentExecutor, "_ask_human_input", side_effect=ask_human_input_side_effect
) as mock_human_input:
# Execute the task
output = agent.execute_task(task)
mock_human_input.assert_called_once()
assert output == "Hello"
# Assertions to ensure the agent behaves correctly
assert mock_human_input.call_count == 2 # Should have asked for feedback twice
assert output.strip().lower() == "hello" # Final output should be 'Hello'
def test_interpolate_inputs():