func_tool

Tool is LLM’s extended capability which is one of the core design pattern of Agent. All tools can be wrapped in a FunctionTool class. This helps to standardize the tool interface and metadata to communicate with the Agent.

Functions

find_instance_name_from_self(instance)	Attempt to find the variable name of the instance in the calling context.
is_running_in_event_loop()

Classes

FunctionTool(fn[, component, definition])

Describing and Parsing(to LLM) and executing a function.

is_running_in_event_loop() → bool

find_instance_name_from_self(instance)

Attempt to find the variable name of the instance in the calling context.

Parameters:

instance – The instance to find the name for.

Returns:

The variable name of the instance, if found; otherwise, None.

class FunctionTool(fn: Callable[[...], Any] | Awaitable[Callable[[...], Any]], component: Component | None = None, definition: FunctionDefinition | None = None)

Bases: Component

Describing and Parsing(to LLM) and executing a function.

Supports both normal callable functions and methods(__call__) of a component. When component is used, we support both the training and eval mode.

Note

When the eval mode, it outputs FunctionOutput, and when the training mode, it outputs Parameter with data as FunctionOutput.

Parameters:

• fn (Callable) – The function to be executed.

• component (Component, optional) – The component that owns the function. Defaults to None.

• definition (FunctionDefinition, optional) – The definition of the function. Defaults to None.

Function be used by LLM as a tool to achieve a specific task.

What function can you pass as a tool? 1. Any unbound function you wrote outside of a class. 2. Any class method you wrote in your component. It can call self and other methods inside of your component. 3. When the function is using a trainable component, and you can directly use the component’s method as a tool or wrap it in a function. But you need to make sure to pass the component to the tool.

Here are some examples:

from adalflow.core.func_tool import FunctionTool
class AgenticRAG(Component):
    def __init__(self, ...):
        super().__init__()
        self.retriever = Retriever()
        self.llm = Generator()

        def retriever_as_tool(input: str) -> str:
            r"Used as a retriever tool."
            return self.retriever(input)

        tools = [FunctionTool(retriever_as_tool, component=self.retriever),
                    FunctionTool(self.llm.__call__, component=self.llm)]
        # if you have trainable component, this will ensure it can be trained together with your whole task pipeline
        # if you dont want to train them and simply treating them as a tool, you can call like this
        # tools = [FunctionTool(retriever_as_tool), FunctionTool(self.llm.__call__, component=self.llm)]

Features:

• Supports both synchronous and asynchronous functions via call and acall.

• Creates a FunctionDefinition from the function using get_fun_schema.

• Executes the function with arguments.

  • Parses the function call expression (FunctionExpression) into Function (name, args, kwargs).

  • Executes the function using one of the following methods:

    • Via call with args and kwargs.

    • Via eval, without any context or sandboxing.

    • Via sandboxed execution directly using sandbox_exec.

A FunctionTool allows other GradComponent(as a tool) to pass through correctly.

property is_async: bool

forward(*args, **kwargs) → Parameter

Forward the function tool.

call(*args: Any, **kwargs: Any) → FunctionOutput

Execute the function synchronously.

Example:

import time
def sync_function_1():
    time.sleep(1)
    return "Function 1 completed"

tool_1 = FunctionTool(sync_function_1)
output = tool_1.call()

bicall(*args: Any, **kwargs: Any) → FunctionOutput | Parameter

Execute the function synchronously.

Example:

import time
def sync_function_1():
    time.sleep(1)
    return "Function 1 completed"

tool_1 = FunctionTool(sync_function_1)
output = tool_1.call()

async acall(*args: Any, **kwargs: Any) → FunctionOutput

Execute the function asynchronously.

Need to be called in an async function or using asyncio.run.

Example:

import asyncio
async def async_function_1():
    await asyncio.sleep(1)  # Simulate async work
    return "Function 1 completed"

async def call_async_function():
    tool_1 = FunctionTool(async_function_1)
    output = await tool_1.acall()

asyncio.run(call_async_function())

execute(*args, **kwargs) → FunctionOutput

Execute the function synchronously or asynchronously based on the function type.

No matter of the function type, you can run the function using both asyncio and without asyncio.

Use it with caution as it might block the event loop.

Example:

import asyncio
import time

async def async_function_1():
    await asyncio.sleep(1)
    return "Function 1 completed"

def sync_function_1():
    time.sleep(1)
    return "Function 1 completed"

async def async_function_2():
    await asyncio.sleep(2)
    return "Function 2 completed"

def sync_function_2():
    time.sleep(2)
    return "Function 2 completed"

async_tool_1 = FunctionTool(async_function_1)
sync_tool_1 = FunctionTool(sync_function_2)
async_tool_2 = FunctionTool(async_function_2)
sync_tool_2 = FunctionTool(sync_function_2)

def run_sync_and_async_mix_without_wait():
    # both sync and async tool can use execute
    # sync tool can also use call
    # takes 5 seconds (1+1+2) + overhead
    start_time = time.time()
    results = [
        async_tool_1.execute(),
        sync_tool_1.execute(),
        sync_tool_2.call(),
    ]
    end_time = time.time()
    print(f"run_sync_and_async_mix_without_wait time: {end_time - start_time}")
    return results

async def run_sync_and_async_mix():
    # both sync and async tool can use execute&to_thread
    # async tool can also use acall without to_thread
    # takes a bit over 2 seconds max(2)
    start_time = time.time()
    results = await asyncio.gather(
        async_tool_1.execute(),
        sync_tool_1.execute(),
        async_tool_2.acall(),
    )
    end_time = time.time()
    print(f"run_sync_and_async_mix time: {end_time - start_time}")
    return results

run_sync_and_async_mix_without_wait()
asyncio.run(run_sync_and_async_mix())