fix: handle Gemini function calls without text content

When Gemini returns a function call without text content, the response
was being returned as empty/None, causing 'Invalid response from LLM
call - None or empty' errors.

This fix:
1. Adds _format_function_call_as_action() to format function calls as
   'Action: <name>\nAction Input: <args>' text that CrewAI's parser
   can understand
2. Adds _extract_text_from_parts() to properly extract text from
   response parts instead of relying on response.text
3. Updates _process_response_with_tools() to use the fallback format
   when no text content is available
4. Updates _finalize_streaming_response() with the same fallback logic

Fixes #4093

Co-Authored-By: João <joao@crewai.com>
This commit is contained in:
Devin AI
2025-12-16 08:44:25 +00:00
parent 88d3c0fa97
commit 4938654ede
2 changed files with 366 additions and 22 deletions

View File

@@ -583,6 +583,47 @@ class GeminiCompletion(BaseLLM):
messages_for_event, content, from_agent
)
def _format_function_call_as_action(
self, function_name: str, function_args: dict[str, Any]
) -> str:
"""Format a function call as Action/Action Input text for CrewAI's parser.
When Gemini returns a function call but native tool execution is not available
(i.e., available_functions is empty), this method formats the function call
in the text format that CrewAI's agent parser expects.
Args:
function_name: Name of the function to call
function_args: Arguments for the function call
Returns:
Formatted string in "Action: <name>\nAction Input: <args>" format
"""
import json
try:
args_str = json.dumps(function_args, default=str)
except (TypeError, ValueError):
args_str = str(function_args)
return f"Action: {function_name}\nAction Input: {args_str}"
def _extract_text_from_parts(
self, parts: list[Any]
) -> str:
"""Extract text content from response parts.
Args:
parts: List of response parts from Gemini
Returns:
Concatenated text content from all text parts
"""
text_parts = [
part.text for part in parts if hasattr(part, "text") and part.text
]
return " ".join(text_parts).strip()
def _process_response_with_tools(
self,
response: GenerateContentResponse,
@@ -605,6 +646,8 @@ class GeminiCompletion(BaseLLM):
Returns:
Final response content or function call result
"""
function_call_fallback = ""
if response.candidates and (self.tools or available_functions):
candidate = response.candidates[0]
if candidate.content and candidate.content.parts:
@@ -619,18 +662,35 @@ class GeminiCompletion(BaseLLM):
else {}
)
result = self._handle_tool_execution(
function_name=function_name,
function_args=function_args,
available_functions=available_functions or {},
from_task=from_task,
from_agent=from_agent,
)
if available_functions:
result = self._handle_tool_execution(
function_name=function_name,
function_args=function_args,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
)
if result is not None:
return result
if result is not None:
return result
if not function_call_fallback:
function_call_fallback = self._format_function_call_as_action(
function_name, function_args
)
content = ""
if response.candidates:
candidate = response.candidates[0]
if candidate.content and candidate.content.parts:
content = self._extract_text_from_parts(candidate.content.parts)
if not content:
content = response.text or ""
if not content and function_call_fallback:
content = function_call_fallback
content = response.text or ""
content = self._apply_stop_words(content)
return self._finalize_completion_response(
@@ -718,8 +778,10 @@ class GeminiCompletion(BaseLLM):
"""
self._track_token_usage_internal(usage_data)
function_call_fallback = ""
# Handle completed function calls
if function_calls and available_functions:
if function_calls:
for call_data in function_calls.values():
function_name = call_data["name"]
function_args = call_data["args"]
@@ -732,20 +794,32 @@ class GeminiCompletion(BaseLLM):
if not isinstance(function_args, dict):
function_args = {}
# Execute tool
result = self._handle_tool_execution(
function_name=function_name,
function_args=function_args,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
)
# Try to execute tool if available_functions provided
if available_functions:
result = self._handle_tool_execution(
function_name=function_name,
function_args=function_args,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
)
if result is not None:
return result
if result is not None:
return result
# Store first function call as fallback if no text response
if not function_call_fallback:
function_call_fallback = self._format_function_call_as_action(
function_name, function_args
)
# Use function call fallback if no text response
content = full_response
if not content and function_call_fallback:
content = function_call_fallback
return self._finalize_completion_response(
content=full_response,
content=content,
contents=contents,
response_model=response_model,
from_task=from_task,

View File

@@ -728,3 +728,273 @@ def test_google_streaming_returns_usage_metrics():
assert result.token_usage.prompt_tokens > 0
assert result.token_usage.completion_tokens > 0
assert result.token_usage.successful_requests >= 1
def test_gemini_function_call_without_text_returns_action_format():
"""
Test that when Gemini returns a function call without text content,
the response is formatted as Action/Action Input for CrewAI's parser.
This addresses GitHub issue #4093 where Gemini models would return
"None or empty response" errors when the model returned only a function call.
"""
from unittest.mock import Mock, patch
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
with patch.object(completion.client.models, 'generate_content') as mock_generate:
mock_function_call = Mock()
mock_function_call.name = "search_tool"
mock_function_call.args = {"query": "weather in Tokyo"}
mock_part = Mock()
mock_part.function_call = mock_function_call
mock_part.text = None
mock_content = Mock()
mock_content.parts = [mock_part]
mock_candidate = Mock()
mock_candidate.content = mock_content
mock_response = Mock()
mock_response.candidates = [mock_candidate]
mock_response.text = None
mock_response.usage_metadata = Mock()
mock_response.usage_metadata.prompt_token_count = 100
mock_response.usage_metadata.candidates_token_count = 50
mock_response.usage_metadata.total_token_count = 150
mock_generate.return_value = mock_response
messages = [{"role": "user", "content": "What's the weather in Tokyo?"}]
result = completion.call(messages=messages)
assert result is not None
assert result != ""
assert "Action: search_tool" in result
assert "Action Input:" in result
assert "weather in Tokyo" in result
def test_gemini_function_call_with_text_returns_text():
"""
Test that when Gemini returns both a function call and text content,
the text content is returned (not the Action/Action Input format).
"""
from unittest.mock import Mock, patch
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
with patch.object(completion.client.models, 'generate_content') as mock_generate:
mock_function_call = Mock()
mock_function_call.name = "search_tool"
mock_function_call.args = {"query": "weather"}
mock_function_part = Mock()
mock_function_part.function_call = mock_function_call
mock_function_part.text = None
mock_text_part = Mock()
mock_text_part.function_call = None
mock_text_part.text = "Let me search for the weather information."
mock_content = Mock()
mock_content.parts = [mock_text_part, mock_function_part]
mock_candidate = Mock()
mock_candidate.content = mock_content
mock_response = Mock()
mock_response.candidates = [mock_candidate]
mock_response.text = "Let me search for the weather information."
mock_response.usage_metadata = Mock()
mock_response.usage_metadata.prompt_token_count = 100
mock_response.usage_metadata.candidates_token_count = 50
mock_response.usage_metadata.total_token_count = 150
mock_generate.return_value = mock_response
messages = [{"role": "user", "content": "What's the weather?"}]
result = completion.call(messages=messages)
assert result is not None
assert result != ""
assert "Let me search for the weather information" in result
assert "Action:" not in result
def test_gemini_function_call_with_available_functions_executes_tool():
"""
Test that when available_functions is provided, the function call is executed
instead of being formatted as Action/Action Input.
"""
from unittest.mock import Mock, patch
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
def mock_search_tool(query: str) -> str:
return f"Search results for: {query}"
available_functions = {"search_tool": mock_search_tool}
with patch.object(completion.client.models, 'generate_content') as mock_generate:
mock_function_call = Mock()
mock_function_call.name = "search_tool"
mock_function_call.args = {"query": "weather in Tokyo"}
mock_part = Mock()
mock_part.function_call = mock_function_call
mock_part.text = None
mock_content = Mock()
mock_content.parts = [mock_part]
mock_candidate = Mock()
mock_candidate.content = mock_content
mock_response = Mock()
mock_response.candidates = [mock_candidate]
mock_response.text = None
mock_response.usage_metadata = Mock()
mock_response.usage_metadata.prompt_token_count = 100
mock_response.usage_metadata.candidates_token_count = 50
mock_response.usage_metadata.total_token_count = 150
mock_generate.return_value = mock_response
messages = [{"role": "user", "content": "What's the weather in Tokyo?"}]
result = completion.call(
messages=messages,
available_functions=available_functions
)
assert result == "Search results for: weather in Tokyo"
def test_gemini_streaming_function_call_without_text_returns_action_format():
"""
Test that streaming responses with function calls but no text content
return Action/Action Input format.
"""
from unittest.mock import Mock, patch
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001", stream=True)
with patch.object(completion.client.models, 'generate_content_stream') as mock_stream:
mock_function_call = Mock()
mock_function_call.name = "calculator"
mock_function_call.args = {"expression": "2 + 2"}
mock_part = Mock()
mock_part.function_call = mock_function_call
mock_part.text = None
mock_content = Mock()
mock_content.parts = [mock_part]
mock_candidate = Mock()
mock_candidate.content = mock_content
mock_chunk = Mock()
mock_chunk.candidates = [mock_candidate]
mock_chunk.text = None
mock_chunk.usage_metadata = Mock()
mock_chunk.usage_metadata.prompt_token_count = 50
mock_chunk.usage_metadata.candidates_token_count = 25
mock_chunk.usage_metadata.total_token_count = 75
mock_stream.return_value = [mock_chunk]
messages = [{"role": "user", "content": "Calculate 2 + 2"}]
result = completion.call(messages=messages)
assert result is not None
assert result != ""
assert "Action: calculator" in result
assert "Action Input:" in result
assert "2 + 2" in result
def test_gemini_format_function_call_as_action():
"""
Test the _format_function_call_as_action helper method directly.
"""
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
result = completion._format_function_call_as_action(
function_name="search_tool",
function_args={"query": "test query", "limit": 10}
)
assert "Action: search_tool" in result
assert "Action Input:" in result
assert "test query" in result
assert "10" in result
def test_gemini_format_function_call_as_action_handles_special_types():
"""
Test that _format_function_call_as_action handles non-JSON-serializable types.
"""
from crewai.llms.providers.gemini.completion import GeminiCompletion
from datetime import datetime
completion = GeminiCompletion(model="gemini-2.0-flash-001")
result = completion._format_function_call_as_action(
function_name="date_tool",
function_args={"date": datetime(2024, 1, 1), "value": 42}
)
assert "Action: date_tool" in result
assert "Action Input:" in result
assert "42" in result
def test_gemini_extract_text_from_parts():
"""
Test the _extract_text_from_parts helper method.
"""
from unittest.mock import Mock
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
mock_part1 = Mock()
mock_part1.text = "Hello"
mock_part2 = Mock()
mock_part2.text = "World"
mock_part3 = Mock()
mock_part3.text = None
parts = [mock_part1, mock_part2, mock_part3]
result = completion._extract_text_from_parts(parts)
assert result == "Hello World"
def test_gemini_extract_text_from_parts_empty():
"""
Test _extract_text_from_parts with no text parts.
"""
from unittest.mock import Mock
from crewai.llms.providers.gemini.completion import GeminiCompletion
completion = GeminiCompletion(model="gemini-2.0-flash-001")
mock_part = Mock()
mock_part.text = None
parts = [mock_part]
result = completion._extract_text_from_parts(parts)
assert result == ""