andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-08 23:58:34 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

imporve HITL (#2169)

Browse Source 
* imporve HITL

* fix failing test

* fix failing test part 2

* Drop extra logs that were causing confusion

---------

Co-authored-by: Lorenze Jay <63378463+lorenzejay@users.noreply.github.com>
This commit is contained in:
Brandon Hancock (bhancock_ai)
2025-02-20 12:01:49 -05:00
committed by GitHub
parent 00c2f5043e
commit 14503bc43b
5 changed files with 31 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
tests/agent_test.py
View File

@@ -1,7 +1,6 @@
"""Test Agent creation and execution basic functionality."""
import os
from datetime import UTC, datetime, timezone
from unittest import mock
from unittest.mock import patch
@@ -9,7 +8,7 @@ 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.crew_agent_executor import AgentFinish, CrewAgentExecutor
from crewai.agents.parser import AgentAction, CrewAgentParser, OutputParserException
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
@@ -999,23 +998,35 @@ def test_agent_human_input():
# 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
"Don't say hi, say Hello instead!", # First feedback: instruct change
"", # Second feedback: empty string signals acceptance
]
)
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:
# Patch both _ask_human_input and _invoke_loop to avoid real API/network calls.
with (
patch.object(
CrewAgentExecutor,
"_ask_human_input",
side_effect=ask_human_input_side_effect,
) as mock_human_input,
patch.object(
CrewAgentExecutor,
"_invoke_loop",
return_value=AgentFinish(output="Hello", thought="", text=""),
) as mock_invoke_loop,
):
# Execute the task
output = agent.execute_task(task)
# 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'
# Assertions to ensure the agent behaves correctly.
# It should have requested feedback twice.
assert mock_human_input.call_count == 2
# The final result should be processed to "Hello"
assert output.strip().lower() == "hello"
def test_interpolate_inputs():