Endless Tool Loop (Vertex Gemini)

[jtbaker]

Hi, thanks for the great framework - it's been very productive for me so far. I've been hitting an issue implementing an agentic pattern using some tools that I was hoping to get some direction on. I'm not sure if it can be handled through the type system, prompting, or the agent/model arguments. I've tried variations on all of them with seemingly little effect.

High level, I have one top level Agent that has tools that call out to sub-agents, - a data agent and a news agent. These agents have tools that get data and news and summarize what they find as they relate to the query.

I'm hitting an issue with the news agent get_articles tool where it seems to get placed into an endless loop, calling the news API over and over (getting the same results) and eventually leading to getting 429 (Resource Exhausted) responses from the Gemini API. Using the debugger, I can see that news_agent.run(...) is only getting invoked once, so the issue seems to be with how news_agent is converting its tool return type into its agent return type (list[WPArticle] -> list[NewsSummary]). I think understanding how this works and being able to manage the control flow of agent tools on some level will be pretty important for developers of agentic applications like this.

I've tried various configurations on NewsAgent, including
end_strategy="early", defer_model_check=False, which don't seem to have any effect on the behavior. Every tool has the retries=1 parameter set too, and explicit system prompting to only invoke said tool one time.

from pydantic_ai import Agent, RunContext
import yaml
from pydantic import BaseModel, Field
from pydantic_ai.models.vertexai import VertexAIModel
from datetime import datetime
from typing import Optional
from io import StringIO


class WPArticle(BaseModel):
    # omitted for brevity
    pass


class NewsRequest(BaseModel):
    query: str = Field(
        ...,
        description="(individual components of the user's query with locations, topics)",
    )
    start_date: Optional[datetime] = Field(
        None,
        description="The Start Date for the user's query. Only include this if it is explicitly requested.",
    )
    end_date: Optional[datetime] = Field(
        None,
        description="The End Date for the user's query. Only include this if it is explicitly requested.",
    )


class NewsSummary(BaseModel):
    date: datetime = Field(..., description="The date the article was published on")
    summary: str = Field(
        ...,
        description="A brief summary of the article, and how it may relate to the user's query",
    )
    title: str = Field(..., description="The title of the article")
    link: str = Field(..., description="The link to the article")

    def to_yaml(self) -> str:
        json_body = self.model_dump(mode="json")
        return yaml.dump(json_body)

news_agent = Agent(
    "google-vertex:gemini-1.5-flash",
    system_prompt=[
        "You are an expert at searching ___ news articles, using sematic search terms are included or are related to their query, and summarizing their content as they relate to user queries. You call your tool exactly one time to find relevant results - the responses to your tool calls will not change.",
    ],
    model_settings={"temperature": 0, "timeout": 15},
    result_type=list[NewsSummary],
    # end_strategy="early",
    defer_model_check=False,
)


@news_agent.tool_plain(retries=1)
async def get_articles(search: NewsRequest) -> list[WPArticle]:
    """
    Fetch articles from the ___ API
    * Use this tool ONE TIME PER RUN to retrieve relevant articles for a search string.

    Args:
        search: NewsRequest.
            object to search news articles with, including:
                 query: str (individual components of the user's query with locations, topics)
                 start_date: date (optional)
                 end_date: date (optional)

    Returns:
        A list of WPArticle objects representing the article contents, date, and title. This list may be empty if no records were found.
    """
    logging.info("Running get_articles...")
    # return []
    query = search.query.replace('""', "")
    async with NewsApiClient() as client:
        res = await client.get_articles(
            search=query, start_date=search.start_date, end_date=search.end_date
        )
        return res  # I can see that this returns the top 5 articles as WPArticles (pydantic models) related to the user query when debugging - but the tool evaluates over and over, having the effect of spamming the API I'm interacting with. It's the same search and response every time.


response_agent = Agent("google-vertex:gemini-1.5-pro")


@response_agent.tool(retries=1)
async def get_data(ctx: RunContext, query: str):
    # ... longer running op than news
    pass


@response_agent.tool(retries=1)
async def get_news(ctx: RunContext, query: str) -> str:
    """
    Tool to page our org's news API and summarize articles relevant to a user's query.
    """
    response = await news_agent.run(query)
    with StringIO() as buffer:
        for item in response.data:
            buffer.write(item.to_yaml())
        return buffer.getvalue()

Is there some way that I can better indicate to Gemini through the Agent/tool that the results of a tool call are idempotent, given the same input, and that it should only invoke that tool once while attempting to converge?

[izzyacademy]

@jtbaker for scenarios like this, you might want to leverage caching with a K-V store so that if a specific request has already been made within a specified time frame, you could simply return the cached results and not reach out to the remote service for a while. It could help you avoid the 429 errors. You could use the SHA1 or MD5 hash of the query as a look up key with your Key-Value store (Redis or similar).

[jtbaker]

@jtbaker for scenarios like this, you might want to leverage caching with a K-V store so that if a specific request has already been made within a specified time frame, you could simply return the cached results and not reach out to the remote service for a while. It could help you avoid the 429 errors. You could use the SHA1 or MD5 hash of the query as a look up key with your Key-Value store (Redis or similar).

Thanks for the response @izzyacademy. If I understand correctly, that would help with the requests to the WP API, but not with the downstream requests to Gemini, which is attempting to summarize the article content, right? Because the tool call is finishing, but Gemini is deciding to invoke it multiple times.

[Kludex]

@jtbaker Sorry for the delay here.

Can you please share a code I can copy/paste and it works?

[atomic-235]

I have similar situation when using qwen-235b with venice.ai api. I'm using MCP for tools. Using rate limit for the query with same parameters wouldn't work since it should be defined per graph call and not globally. Is there way to limit number of calls for specific tool for agent? Is it possible to use something like ctx.state for that?

[atomic-235]

"""
Agent Tool Call Deduplication

Uses agent.iter() to manually iterate through agent execution and inject error messages for duplicate tool calls.
"""

from typing import Any, Set
import json
import hashlib
from dataclasses import dataclass, field
from loguru import logger
from pydantic_ai import Agent
from pydantic_graph import End
from pydantic_ai.messages import ModelRequest, ToolReturnPart


@dataclass
class ToolCallTracker:
    """Tracks tool calls within a pipeline execution to prevent duplicates."""
    
    # Track calls within current execution
    seen_calls: Set[str] = field(default_factory=set)
    
    def _get_call_hash(self, tool_name: str, args: dict) -> str:
        """Generate hash for a tool call."""
        call_data = {"tool": tool_name, "args": args}
        call_str = json.dumps(call_data, sort_keys=True)
        return hashlib.sha256(call_str.encode()).hexdigest()[:16]
    
    def is_duplicate_call(self, tool_name: str, args: dict) -> bool:
        """Check if this tool call is a duplicate."""
        call_hash = self._get_call_hash(tool_name, args)
        if call_hash in self.seen_calls:
            logger.warning(f"🚫 Duplicate tool call detected: {tool_name} with args {args}")
            return True
        
        self.seen_calls.add(call_hash)
        return False
    
    def reset(self):
        """Reset the tracker for a new message."""
        self.seen_calls.clear()
        logger.debug("🔄 Tool call tracker reset")


async def run_agent_with_deduplication(
    agent: Agent, 
    message: str, 
    tracker: ToolCallTracker,
    agent_name: str = "Agent",
    deps: Any = None,
    **kwargs
) -> Any:
    """Run an agent with tool call deduplication using agent.iter()."""
    
    logger.info(f"🔄 Running {agent_name} with deduplication")
    
    async with agent.iter(message, deps=deps, **kwargs) as agent_run:
        node = agent_run.next_node
        
        while not isinstance(node, End):
            if agent.is_call_tools_node(node):
                # Check each tool call in the response for duplicates
                tool_return_parts = []
                
                for part in node.model_response.parts:
                    if hasattr(part, 'tool_name') and hasattr(part, 'args'):
                        tool_name = part.tool_name
                        
                        # Parse args if they're a string
                        if isinstance(part.args, str):
                            try:
                                args = json.loads(part.args)
                            except json.JSONDecodeError:
                                args = {"raw_args": part.args}
                        else:
                            args = part.args
                        
                        tool_call_id = getattr(part, 'tool_call_id', 'unknown')
                        
                        # Check if this is a duplicate call
                        if tracker.is_duplicate_call(tool_name, args):
                            # Create error message for duplicate call
                            error_content = (
                                f"🚫 DUPLICATE TOOL CALL DETECTED: You already called '{tool_name}' "
                                f"with identical parameters {args}. Please try a different approach "
                                f"or use different parameters."
                            )
                            
                            # Create ToolReturnPart with error message
                            tool_return_part = ToolReturnPart(
                                tool_name=tool_name,
                                content=error_content,
                                tool_call_id=tool_call_id,
                                part_kind='tool-return'
                            )
                            tool_return_parts.append(tool_return_part)
                            
                            logger.warning(f"🚫 Injected error message for duplicate call: {tool_name}")
                
                # If we found duplicates, inject error messages and advance agent execution
                if tool_return_parts:
                    # Create a model request with the error tool returns
                    error_request = ModelRequest(parts=tool_return_parts)
                    
                    # Add this to the message history and continue
                    agent_run.ctx.state.message_history.append(error_request)
                    
                    # Continue to next step - the agent should see the error messages
                    # and hopefully avoid making the same calls again
                    logger.info(f"📨 Injected {len(tool_return_parts)} error messages into message history")
            
            # Continue to next node
            node = await agent_run.next(node)
        
        # Return the final result
        return agent_run.result 

This approach works somehow.

[atomic-235]

I'm using it this way in the graph node

if ctx.state.tool_call_tracker is None:
    ctx.state.tool_call_tracker = ToolCallTracker()

# Run asset identification with duplicate prevention
asset_run_output = await run_agent_with_deduplication(
    asset_identifier_agent,
    enhanced_message,
    ctx.state.tool_call_tracker,
    "AssetIdentifier",
    deps=asset_identifier_deps
)

[strawgate]

I am experiencing this now as well -- gemini just repeating tool calls

[tai-bo]

Same here. 936 final_result tool call: