How `run()` Implements the Agentic Loop in Needle: A Deep Dive into Tool Orchestration

The run() method implements a closed-loop agentic cycle that prompts the underlying model, executes discovered function calls, and feeds results back iteratively until the model completes its reasoning or reaches the step limit.

The agentic loop is the core mechanism that enables Needle to function as an autonomous agent capable of tool use. In the cactus-compute/needle repository, this behavior is encapsulated in the run() method within needle/__init__.py, which orchestrates a continuous cycle of generation, execution, and feedback between the language model and Python functions.

Architecture of the Agentic Loop

The run() method establishes a closed-loop interaction between the model and the execution environment. According to the source code in needle/__init__.py, the process follows a strict six-step workflow that repeats until the model no longer requests tool calls or the iteration limit is reached.

Step 1: Initial Completion

The loop begins when run() calls self.complete(query, max_new_tokens) at lines 27-29. This forwards the user query to the underlying C-engine via needle_complete, returning a response envelope containing the model's initial reply and any discovered function calls.

Step 2: Iteration Control

A for loop iterates up to max_steps (default 8) as shown at lines 30-34. On each iteration, the method checks whether the latest response is a "call" type and whether function_calls are present. If neither condition is met, the loop terminates early.

Step 3: Function Resolution and Execution

For every function call in the envelope (lines 35-44), the method looks up the corresponding Python callable in self._functions. If the function is missing, it produces an error object. Otherwise, it invokes the function with the arguments supplied by the model, catching exceptions and converting them to error objects to prevent loop breakage.

Step 4: Result Accumulation

Results or error objects from the batch of calls are appended to the executed list (lines 45-46), preserving the precise order of execution for the final response.

Step 5: Context Feedback

The accumulated results are JSON-encoded via _jsonable and fed back into the model through another self.complete() call (lines 46-47). This asks the model to continue the conversation with the new execution context, effectively closing the feedback loop.

Step 6: Final Response Assembly

After the loop concludes—either because the model stopped issuing calls or max_steps was hit—the method attaches the full history of executed tool results under the "results" key (lines 47-48) and returns the final response dictionary.

Practical Implementation

To use the agentic loop in practice, instantiate the Needle class with your tools and invoke run():

from needle import Needle

def get_weather(location: str) -> str:
    return f"Sunny in {location}"

agent = Needle(tools=[get_weather])
response = agent.run("What is the weather in Paris?", max_steps=5)

print(response["text"])     # Final model output

print(response["results"])  # List of tool execution results

The internal loop logic resembles this simplified implementation:

def _agentic_loop(agent, query):
    response = agent.complete(query, max_new_tokens=256)
    executed = []
    
    for _ in range(8):  # max_steps default

        calls = response.get("function_calls") or []
        if response.get("type") != "call" or not calls:
            break
            
        results = [
            agent._functions[c["name"]](**c.get("arguments", {}))
            for c in calls
        ]
        executed.extend(results)
        response = agent.complete(
            json.dumps(results), 
            max_new_tokens=256
        )
    
    response["results"] = executed
    return response

Key Source Files

Understanding the agentic loop requires familiarity with these core components:

• needle/__init__.py – Contains the Needle class and the run() method implementation that drives the agentic loop.

• needle/agent/tools.py – Provides schema generation via build_schema and the tool decorator used by run() to discover and register Python functions as model-accessible tools.

• needle/model/run.py – Supplies the underlying language model generation utilities (generate, batch_generate) that power the complete() method used within the loop.

Summary

• The run() method in needle/__init__.py implements a closed-loop agentic cycle that enables autonomous tool use.
• The loop executes up to max_steps (default 8) iterations, checking for function calls in each model response.
• Function resolution occurs against self._functions, with robust error handling to ensure the loop continues even if individual tools fail.
• Execution results are JSON-encoded via _jsonable and fed back into the model via subsequent complete() calls, maintaining conversation context.
• The final response includes both the model's text output and a complete history of tool executions under the "results" key.

Frequently Asked Questions

What happens if a function called by the agent does not exist?

If the model requests a function not registered in self._functions, the run() method generates an error object for that specific call and continues processing remaining calls. This error is then fed back to the model in the next iteration, allowing the agent to potentially recover or adjust its strategy.

How does Needle prevent infinite loops in the agentic cycle?

The loop enforces a hard limit via the max_steps parameter, which defaults to 8 iterations. Once this limit is reached, the method returns the current response regardless of whether additional function calls were requested by the model.

Can the agentic loop handle multiple function calls in a single iteration?

Yes. The implementation processes function_calls as a batch within each iteration. It resolves and executes every call in the envelope simultaneously (lines 35-44), accumulates all results, and feeds them back to the model in a single context update.

What format does the model receive for tool execution results?

Results are encoded as JSON using the internal _jsonable helper before being passed to subsequent complete() calls. This ensures the model receives structured, serializable data representing the outcome of each function execution or any errors that occurred.

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