Gemini failed to call tool

[david-andrew]

Issue for tracking general problems when using gemini models (e.g. gemini-2.5-pro) as the ReAct agent

Issues

The main issues I run into are:

1. response times from gemini becoming slower and slower.
2. getting "Response from LLM does not include any content or tool calls"

Slow gemini responses

This one has been pretty hard to diagnose. Basically just tons of calls to the agent are very slow. Seems more correlated with the number of messages rather than the actual length of the messages, but definitely really long chats can also be slow.

I don't really have a good guess for the cause. I was thinking maybe it's related to caching, but I think I do usually see the debug message saying when a cache hit occurs. Perhaps it's related to rate limits and how that's all handled under the hood. Whenever I kill the process if it's taking too long, it usually just seems to be stuck on model.ainvoke(...) so it's when it's talking to the LLM API rather than something in archytas

No tool calls or content

Basically gemini frequently will return empty messages that cause the ReAct loop to throw an uncaught exception. This is the traceback from testing in the terminal:

>>> can you write me a function that will compute the first N digits of pi, where N can be large
Token estimate for query:  273
Actual usage for query:  {'input_tokens': 1064, 'output_tokens': 0, 
'total_tokens': 1064, 'input_token_details': {'cache_read': 0}}
Traceback (most recent call last):
  File "/home/david/dev/AMPT/writing_assistant/backend/test_bug.py", line 138, in simple_test
    response = await agent.react_async(query)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/react.py", line 87, in inner_async
    return await ensure_async(fn(self, *args, **kwargs))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/utils.py", line 60, in ensure_async
    return await fn
           ^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/react.py", line 270, in react_async
    action = await self.handle_message(HumanMessage(content=query))
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/agent.py", line 239, in handle_message
    return await self.execute()
           ^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/agent.py", line 308, in execute
    result = self.model.process_result(raw_result)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/models/gemini.py", line 76, in process_result
    response = super().process_result(response_message)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/david/anaconda3/envs/onr-writing-assistant/lib/python3.12/site-packages/archytas/models/base.py", line 302, in process_result
    raise ValueError("Response from LLM does not include any content or tool calls. This shouldn't happen.")
ValueError: Response from LLM does not include any content or tool calls. This shouldn't happen.

An error occurred: Response from LLM does not include any content or tool calls. This shouldn't happen.

So the error raised at archytas/models/base.py is uncaught, causing the whole ReAct loop to fail when it happens (as opposed to feeding the error message back to the agent so it can self correct). However I had also tried getting the agent to correct by manually telling it of the error, and the issue still persists. It seems largely related to issues with making use of the built-in tool calling api that google implemented for gemini.

Possible causes:

• version of gemini api being used in archytas. There are like 3 different API versions/libraries google has. I updated adhoc-api to use the newest API library, and it has seemed a lot more stable (noting that adhoc doesn't use tools at all, so don't know if this is the root of the tool issue)
• general incompatibilities with how langchain invokes agents, tells them to use tools, etc.

Debug Code

This is the code I was using to try and work through the issue and come up with a possible solution. The big_test case is based on the setup from another project, running it just in the terminal. There's also simple_test which is just a pure archytas agent with only the python tool. Even running the simple test, frequently on the first tool call, it will throw the no tools/content exception.

The last big thing I was trying was to create a fresh BaseArchytasModel child class that directly used the gemini api rather than going through the langchain implementation. Additionally, even in the gemini api, it only makes use of text-based messages rather than using any api support for tool calling. It seems maybe promising, I had a few cases that seemed like they were working, but there was at least one case where I ran into a bug in the archytas implementation:

• at https://github.com/jataware/archytas/blob/main/archytas/agent.py#L317 the task variable is unbound if self.post_execute_task is None came back False

So for continuing to test/diagnose/implement a solution, basically pulling the MyGeminiModel implementation I have below. For convenience, I just pull the agent class from adhoc-api but one could certainly just copy it from adhoc to remove the dependency: https://github.com/jataware/adhoc-api/blob/master/adhoc_api/uaii.py#L199-L362

Note: there are probably other issues with things not aligning with how archytas expects them. The current archytas implementation for agents is pretty unwieldy and hard to match up

from typing import Literal
import requests

from archytas.react import ReActAgent, FailedTaskError
from archytas.models.base import BaseArchytasModel
from archytas.models.gemini import GeminiModel
from archytas.models.openai import OpenAIModel
from archytas.tools import PythonTool
from langchain_core.messages import AIMessage
from easyrepl import REPL
from adhoc_api.uaii import gemini_25_pro, gpt_41, LLMConfig

from .api.map import MapTools
from .api.document import DocumentTools
from .api.schemas import Session, Document
from .prompts import (
    LOAD_BATHYMETRY_PYTHON_PRELUDE,
    INTEL_TOOL_DESCRIPTION,
    RED_FORCE_POSITIONS_TOOL_DESCRIPTION,
    BLUE_FORCE_POSITIONS_TOOL_DESCRIPTION,
    get_unified_system_prompt,
)
from .utils import adhoc_to_tool, csv_to_tool, get_scenario

from pathlib import Path
here = Path(__file__).parent

import os


def get_model_configs(name: Literal['gemini', 'gpt', 'custom']) -> tuple[LLMConfig, BaseArchytasModel]:
    if name == 'gemini':
        model = GeminiModel({'api_key': os.getenv('GEMINI_API_KEY'), 'model_name': 'gemini-2.5-pro'})
        adhoc_model_config = gemini_25_pro
    elif name == 'gpt':
        model = OpenAIModel({'api_key': os.getenv('OPENAI_API_KEY'), 'model_name': 'gpt-4.1'})
        adhoc_model_config = gpt_41
    elif name == 'custom':
        model = MyGeminiModel({'api_key': os.getenv('GEMINI_API_KEY'), 'model_name': 'gemini-2.5-pro'})
        adhoc_model_config = gemini_25_pro
    else:
        raise ValueError(f"Unknown model name: {name}. Use 'gemini' or 'gpt'.")
    
    return adhoc_model_config, model

async def big_test(model_name='gemini'):
    adhoc_model_config, model = get_model_configs(model_name)
    print(f"Big test using model: '{model_name}'")

    # create a dummy session to use for the tools
    session = Session(
        scenario_name='generic',
        session_id='',
        opord_id='test-debug-opord-id',
    )

    scenario = get_scenario('generic')

    
    map_tools = MapTools(session)
    document_tools = DocumentTools(session, session.opord_id)
    python = PythonTool(prelude=LOAD_BATHYMETRY_PYTHON_PRELUDE)
    _, weather = adhoc_to_tool(here/'../docs/openmeteo_marine_weather/api.yaml', adhoc_model_config)
    intel_tool = csv_to_tool(scenario.intel_csv_path, 'intelligence', INTEL_TOOL_DESCRIPTION)    
    red_force_positions_tool = csv_to_tool(scenario.red_force_csv_path, 'red_force_positions', RED_FORCE_POSITIONS_TOOL_DESCRIPTION)
    blue_force_positions_tool = csv_to_tool(scenario.blue_force_csv_path, 'blue_force_positions', BLUE_FORCE_POSITIONS_TOOL_DESCRIPTION)


    tools = [
        map_tools,
        document_tools,
        python,
        weather,
        intel_tool,
        red_force_positions_tool,
        blue_force_positions_tool,
    ]

    agent = ReActAgent(
        model=model,
        tools=tools,
        verbose=True,
    )

    # set up the agent with the system prompt and context
    agent.add_context(get_unified_system_prompt(scenario.platform_capabilities_csv_path))
    agent.add_context(scenario.user_info)
    agent.chat_history.add_message(AIMessage(scenario.hello_message))



    for query in REPL(history_file='.chat'):
        try:
            response = await agent.react_async(query)
            print(response)
        except FailedTaskError as e:
            print(f"Failed task: {e}")
        except Exception as e:
            from traceback import format_exc
            print(format_exc())
            print(f"An error occurred: {e}")
            # break

def big_test_sync(*args, **kwargs):
    import asyncio
    asyncio.run(big_test(*args, **kwargs))


def get_pride_and_prejudice():
    """Long text to stress testing the agent's context window"""
    url = "https://www.gutenberg.org/files/1342/1342-0.txt"
    response = requests.get(url)
    if response.status_code != 200:
        raise Exception(f"Failed to fetch text: {response.status_code}")
    return response.text


async def simple_test(model_name='gemini'):
    _, model = get_model_configs(model_name)
    print(f"Simple test using model: '{model_name}'")
    python = PythonTool()
    tools = [python]

    pride_and_prejudice = get_pride_and_prejudice()

    agent = ReActAgent(
        model=model,
        tools=tools,
        verbose=True,
    )

    # dump a long text into the agent's context
    # agent.add_context(pride_and_prejudice)
    

    for query in REPL(history_file='.chat'):
        try:
            response = await agent.react_async(query)
            print(response)
        except Exception as e:
            from traceback import format_exc
            print(format_exc())
            print(f"An error occurred: {e}")
            # break

def simple_test_sync(*args, **kwargs):
    import asyncio
    asyncio.run(simple_test(*args, **kwargs))





import pdb#pdbpp as pdb
from functools import cache
from adhoc_api.uaii import GeminiAgent, gemini_25_pro, set_gemini_api_key
from langchain_core.messages import SystemMessage, AIMessage, ToolMessage, HumanMessage, BaseMessage
from langchain_core.messages.tool import ToolCall
import asyncio
import rich
from rich import traceback# import install as rich_traceback_install
# set rich to do the traceback formatting
from archytas.tool_utils import get_tool_prompt_description
from archytas.chat_history import AgentResponse
import json
import time
traceback.install(show_locals=True, word_wrap=True)

async def async_wrapper(sync_gen_func, *args, **kwargs):
    for item in sync_gen_func(*args, **kwargs):
        await asyncio.sleep(0)  # Let the event loop breathe
        yield item


class _Model:
    def __init__(self): ...
    def bind_tools(self, tools):
        print(f"Binding tools to model...\n{tools=}")
        self.tools = tools
        pdb.set_trace()
        ...

class MyGeminiModel(BaseArchytasModel):
    # DEFAULT_MODEL: ClassVar[Optional[str]] = None
    # DEFAULT_SUMMARIZATION_RATIO: float = 0.5
    # MODEL_PROMPT_INSTRUCTIONS: str = ""

    # _model: "BaseChatModel"
    # config: ModelConfig
    # lc_tools: "list[StructuredTool] | None"

    def __init__(self, config:dict) -> None:
        self._model: _Model = _Model()
        api_key = config['api_key']
        model_name = config['model_name']
        self._agent = GeminiAgent(model=model_name, system_prompt='', cache_key=None, cache_content='', ttl_seconds=3600)
        self._messages = []
        self.config = config
        self.post_execute_task = None
        # set_gemini_api_key(api_key)
        # super().__init__({model_name: model_name, 'api_key': api_key})

    # def auth(self, api_key: str) -> None:
    #     set_gemini_api_key(api_key)
    
    def initialize_model(self, **kwargs): ...


    @property
    def summarization_threshold(self):
        return float('inf') # make it never try to summarize


    def _system_prompt_setup(self, messages: list[BaseMessage], agent_tools: dict):
        system_message = messages[0]
        assert isinstance(system_message, SystemMessage), f"Expected first message to be a SystemMessage, got {system_message=}"
        
        system_message_str = system_message.content
        # TODO: use archytas to generate the strings for each tool
        # print(f'{agent_tools=}')
        assert len(agent_tools) > 0, f'No tools provided to the agent. Expected at least one tool, got {agent_tools=}'
        system_message_str += "\n\nThe following tools are available:\n"
        for tool_name, tool in agent_tools.items():
            tool_description = get_tool_prompt_description(tool)
            system_message_str += f"{tool_description}\n\n"
        system_message_str += """\
> NOTE: When using a tool, you MUST use the following format:
THOUGHT: <explain why you are using this tool>
TOOL: <tool name>
INPUT: <tool input>


> For example, say you want to use the 'weather' (assuming it was one of the available tools), calling it might look like this (noting that the specific tool inputs required will be specified in the tool's description from the list above):
THOUGHT: I need to know the weather conditions at such and such location.
TOOL: weather
INPUT: {"latitude": 34.0522, "longitude": -118.2437, "hourly": ["temperature_2m", "precipitation"]}

> DO NOT deviate from this output format, and DO NOT include any other text or comments in your response
"""
        print(f"Setting system prompt:\n'''\n{system_message_str}\n'''")

        self._agent.set_system_prompt(system_prompt='', cache_content=system_message_str) #TODO: there might be a bug in adhoc uaii for gemini where system_prompt is not used unless caching was successful...

        # save first message so that ainvoke knows it was accounted for already
        self._messages = [messages[0]]


    def _push_new_messages(self, input: list[BaseMessage]):
        # convert new messages to uaii format
        new_messages = input[len(self._messages):]
        for message in new_messages:
            if isinstance(message, SystemMessage):
                self._agent.push_user_message(f'SYSTEM: {message.content}')
            elif isinstance(message, AIMessage):
                self._agent.push_assistant_message(message.content)
            elif isinstance(message, ToolMessage):
                self._agent.push_user_message(f'TOOL: {message.content}')
            elif isinstance(message, HumanMessage):
                self._agent.push_user_message(message.content)
            else:
                raise ValueError(f"Unexpected message type: {type(message)}.\n{message=}")
        
        self._messages.extend(new_messages)
    
    
    async def ainvoke(self, *, input: list[BaseMessage], temperature: float, agent_tools: dict):
        if len(self._messages) == 0:
            self._system_prompt_setup(input, agent_tools)


        self._push_new_messages(input)
        # gen = self._agent.multishot(self._agent.messages, stream=True)
        parts = []
        async for chunk in async_wrapper(self._agent.multishot, messages=self._agent.messages, stream=True):
            parts.append(chunk)
        response = ''.join(parts)
        print(f'raw response from agent:\n"""\n{response}\n"""')

        num_retries = 0
        max_retries = 3
        while response.strip() == '' and num_retries < max_retries:
            num_retries += 1
            # sometimes the agent returns an empty string, so tell it it made a mistake and to retry
            print("Received empty response, retrying...")
            self._agent.push_user_message("SYSTEM: received empty response. response must be non-empty, and contain a valid tool call or text response. Please retry.")
            async for chunk in async_wrapper(self._agent.multishot, messages=self._agent.messages, stream=True):
                parts.append(chunk)
            response = ''.join(parts)
            print(f'raw response from agent:\n"""\n{response}\n"""')

        # TBD maybe we need to determine if it was a tool call?
        return self.process_message(response) 
    def process_message(self, response: str) -> BaseMessage:
        if 'THOUGHT' not in response or 'TOOL' not in response or 'INPUT' not in response:
            print(f"WARNING: agent didn't format response as a tool call, returning as AIMessage: {response}")
            return AIMessage(content=response)
        
        junk, rest = response.split('THOUGHT', 1)
        thought, rest = rest.split('TOOL', 1)
        tool_name, tool_input = rest.split('INPUT', 1)
        thought = thought.lstrip(':').strip()
        tool_name = tool_name.lstrip(':').strip()
        tool_input = tool_input.lstrip(':').strip()

        args = json.loads(tool_input)
        id = f"{tool_name}:{hash((tool_input, time.time()))}"
        return AIMessage(content=thought, tool_calls=[ToolCall(name=tool_name, args=args, id=id, type='tool_call')])

    def invoke(self, input, *, config=None, stop=None, agent_tools=None, **kwargs):
        raise NotImplementedError("This method is not implemented in MyGeminiModel. Use ainvoke instead.")




if __name__ == '__main__':
    model_name='gemini'#'custom' #'gemini' # 'gpt'
    # big_test_sync(model_name)
    simple_test_sync(model_name)

[david-andrew]

I was able to demonstrate a seemingly stable version of gemini

requirements:

• langgraph
• langchain[all] # not sure if this is actually needed
• langchain-google-genai
• easyrepl

test script:

from langgraph.prebuilt import create_react_agent
from langchain_google_genai import ChatGoogleGenerativeAI
from easyrepl import REPL
import os

async def test3():
    python = PythonTool(debug=True)
    agent = create_react_agent(
        model=ChatGoogleGenerativeAI(model='gemini-2.5-pro', api_key=os.getenv('GEMINI_API_KEY')),
        tools=[python.run],
        prompt="You are a helpful assistant",
        # debug=True,
    )
    messages = []
    for query in REPL(history_file='.chat'):
        messages.append({"role": "user", "content": query})
        gen = agent.astream({"messages": messages}, stream_mode='messages')
        chunks = []
        async for response in gen:
            message, _ = response
            chunk = message.content
            print(chunk, end='', flush=True)
            chunks.append(chunk)
        print()
        messages.append({"role": "assistant", "content": ''.join(chunks)})



from typing import Any
from io import StringIO
import sys

class PythonTool:
    """
    Tool for running python code. If the user asks you to write code, you can run it here.
    """
    def __init__(self, debug: bool = False):
        # create the python env instance
        self.env = Python()
        self.debug = debug

    def run(self, code: str) -> str:
        """
        Runs python code in a python environment.

        The environment is persistent between runs, so any variables created will be available in subsequent runs.
        The only visible effects of this tool are from output to stdout/stderr. If you want to view a result, you MUST print it.

        Args:
            code (str): The code to run

        Returns:
            str: The stdout output of the code
        """
        if self.debug:
            print(f'```python\n{code}\n```')
        out, err = self.env.run_script(code)
        if err:
            raise Exception(err)
        if self.debug:
            print(f'output: {out}')

        return out

class Python:
    def __init__(self):
        self.locals: dict[str, Any] = {}

    def run_script(self, script: str):
        # capture any stdout/stderr from the script
        captured_stdout = StringIO()
        captured_stderr = StringIO()
        sys.stdout = captured_stdout
        sys.stderr = captured_stderr

        # run the script
        try:
            exec(script, self.locals)
        except Exception as e:
            sys.stderr.write(str(e))

        # restore stdout/stderr
        sys.stdout = sys.__stdout__
        sys.stderr = sys.__stderr__

        return captured_stdout.getvalue(), captured_stderr.getvalue()


if __name__ == "__main__":
    import asyncio
    asyncio.run(test3())

Basically it makes use of langgraph's implementation of ReAct rather than relying on the archytas version. It seems this works with basically the version of langchain_google_genai that archytas uses (I tested with 2.1.8, it looks like archytas has 2.1.5 in the poetry.lock file).

So probably being able to fix gemini in archytas would involve a very extensive rewrite to make langgraph the underlying implementation of react rather than the version we have in archytas. A possible alternative might be to provide a separate ReAct implemntation in archytas that just wraps around the langgraph version, and then the user could select that version if they were using gemini (or perhaps archytas could detect if it was gemini and use that version). But there would still be a lot of functionality/machinery in the current archytas ReAct function that needs to be adapted to work with a langgraph wrapper