How to Use the agent.run() Method for a Full Agentic Loop in Needle

The agent.run() method automates the complete agentic loop by iteratively prompting the Needle model, executing registered Python tool functions, and feeding structured results back until a final answer is generated or the maximum step limit is reached.

The Needle library provides a streamlined interface for building autonomous AI agents capable of external tool interaction. At the heart of this capability lies the agent.run() method implemented in needle/__init__.py (lines 283-315), which orchestrates a self-contained reasoning cycle that handles multi-step tool invocation without manual loop management.

How agent.run() Implements the Agentic Loop

The agent.run() method drives the entire execution cycle through a structured nine-step process defined in the Needle source code. It manages the conversation flow between the language model and registered functions, handling everything from audio input normalization to result serialization.

Input Preparation and Initial Completion

The method begins by normalizing optional audio inputs using the internal _prepare_audio helper function. It then generates an initial response by calling _complete, which transmits the user query to the underlying Needle engine along with the prepared payload and max_new_tokens budget (defaulting to 256 tokens).

Iterative Tool Execution

The core logic operates within a for loop bounded by the max_steps parameter (defaulting to 8 iterations). During each iteration, the method extracts function_calls from the model's response dictionary. If no calls are detected or the response type is not "call", the loop terminates immediately and returns the accumulated results.

When function calls are present, the method performs temporal grounding checks using _ungrounded_paths to identify arguments containing dates without licensed year contexts. In strict mode (enabled via the strict parameter), these ungrounded calls trigger an immediate error via _annotate_ungrounded, while non-strict mode allows execution to proceed with warnings.

Function Resolution and Result Feedback

For each valid call, the method retrieves the corresponding Python callable from the self._functions registry. Unknown tools generate error entries in the results list, while valid functions execute and append their return values to the executed accumulator. The method then serializes these results using json.dumps and feeds them back into the model via _complete with ground=False, enabling the next reasoning iteration to incorporate the tool outputs.

Customizing Loop Behavior

The agent.run() method exposes several parameters to control execution boundaries and safety constraints.

max_steps limits the total number of tool-call iterations. The default value of 8 prevents infinite loops in complex reasoning scenarios, though you may increase this for workflows requiring extensive tool chaining.

strict controls grounding validation for temporal arguments. When set to True, the method aborts execution upon detecting ungrounded date paths through _ungrounded_paths, enforcing licensing constraints. Setting this to False permits the model to proceed with potentially unverified temporal references.

max_new_tokens defines the token budget for each individual completion call. The default of 256 tokens suffices for most queries, but increasing this accommodates elaborate tool arguments or detailed reasoning chains.

Practical Implementation Examples

The following patterns demonstrate typical usage scenarios for the agent.run() method against the Needle source implementation.

Basic Weather Query

This example registers a simple weather tool and executes a single-step lookup:

from needle import Needle

def get_weather(location):
    return f"Sunny and 22°C in {location}"

agent = Needle(
    model="needle-3b",
    tools={"weather": get_weather}
)

response = agent.run(
    query="What's the weather in Paris tomorrow?",
    max_steps=5,
    strict=True
)

print(response["results"])  # List containing: ["Sunny and 22°C in Paris"]

print(response["content"])   # Final answer from the model

Handling Ungrounded Dates

When working with temporal data that may fall outside licensed date ranges, adjust the strictness parameter:

response = agent.run(
    query="Schedule a meeting for March 15, 2040",
    max_steps=3,
    strict=False  # Bypass grounding checks for future dates

)

Manual Loop Comparison

Without agent.run(), developers must manually orchestrate the interaction cycle:

# Manual implementation of the agentic loop

payload = agent._prepare_audio(None)
response = agent._complete("Query", 256, payload)

for _ in range(8):  # Manual max_steps enforcement

    calls = response.get("function_calls")
    if not calls:
        break
    results = []
    for call in calls:
        fn = agent._functions.get(call.get("name"))
        results.append(fn(**call.get("arguments", {})))
    response = agent._complete(
        json.dumps(results), 
        256, 
        None, 
        ground=False
    )

The agent.run() method encapsulates this complexity, guaranteeing consistent error handling and automatic result accumulation under response["results"].

Summary

• The agent.run() method in needle/__init__.py (lines 283-315) automates the complete agentic loop from initial prompt to final answer.
• It supports up to 8 iterations by default, customizable via max_steps, with each iteration capped at 256 tokens unless overridden by max_new_tokens.
• Strict mode enforces temporal grounding by validating date arguments through _ungrounded_paths, aborting ungrounded calls when enabled.
• The method handles tool resolution via the self._functions registry and automatically serializes results back to the model using JSON formatting.
• Internal helpers _prepare_audio, _complete, and _annotate_ungrounded manage audio normalization, model inference, and grounding checks respectively.

Frequently Asked Questions

How does agent.run() handle unknown tool functions?

When the model requests a function not registered in self._functions, the method appends an error dictionary to the results list indicating the unknown tool and continues processing other calls in the same iteration. These errors appear in the final response["results"] array, allowing the model to adjust its strategy in subsequent reasoning steps.

What happens if the model exceeds the max_steps limit?

If the loop reaches the max_steps threshold before the model produces a final answer without tool calls, the method returns the accumulated executed results and the last response state. The "results" key contains all successful tool executions from the limited iterations, while "content" reflects the model's final output at the step limit.

Can I use agent.run() without any registered tools?

Yes, though the method is optimized for tool-augmented workflows. Without registered functions in self._functions, any model response containing function_calls will generate error entries for unknown tools. If the model returns a direct answer without calling tools, the method returns immediately after the initial _complete call with an empty results list.

How does strict grounding validation work internally?

The method invokes _ungrounded_paths to scan response arguments for date references lacking licensed year contexts. In strict mode (enabled by strict=True), detected ungrounded paths cause the method to abort those specific calls and return licensing errors through _annotate_ungrounded. This enforcement happens before function execution via self._functions, preventing invalid temporal queries from reaching registered Python callables.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}