mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Updates to Tools (#1835)
* updated tool samples * mypy and readme fixes * updated call logic * added function invocation config * added include detailed error * added tests * updated FRC exception handling * updated tests * fix oai test * fix name in sample * imporoved tests coverage and removed some dead code paths
This commit is contained in:
Eduard van Valkenburg
committed by
GitHub
Unverified
GitHub
parent
d81b579111
commit
51b32ed1ac
@@ -19,7 +19,7 @@ from ._middleware import (
|
||||
)
|
||||
from ._serialization import SerializationMixin
|
||||
from ._threads import ChatMessageStoreProtocol
|
||||
from ._tools import ToolProtocol
|
||||
from ._tools import FUNCTION_INVOKING_CHAT_CLIENT_MARKER, FunctionInvocationConfiguration, ToolProtocol
|
||||
from ._types import ChatMessage, ChatOptions, ChatResponse, ChatResponseUpdate, ToolMode, prepare_messages
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -357,6 +357,10 @@ class BaseChatClient(SerializationMixin, ABC):
|
||||
|
||||
self.middleware = middleware
|
||||
|
||||
self.function_invocation_configuration = (
|
||||
FunctionInvocationConfiguration() if hasattr(self.__class__, FUNCTION_INVOKING_CHAT_CLIENT_MARKER) else None
|
||||
)
|
||||
|
||||
def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
|
||||
"""Convert the instance to a dictionary.
|
||||
|
||||
|
||||
@@ -71,6 +71,7 @@ logger = get_logger()
|
||||
__all__ = [
|
||||
"FUNCTION_INVOKING_CHAT_CLIENT_MARKER",
|
||||
"AIFunction",
|
||||
"FunctionInvocationConfiguration",
|
||||
"HostedCodeInterpreterTool",
|
||||
"HostedFileSearchTool",
|
||||
"HostedMCPSpecificApproval",
|
||||
@@ -84,7 +85,8 @@ __all__ = [
|
||||
|
||||
logger = get_logger()
|
||||
FUNCTION_INVOKING_CHAT_CLIENT_MARKER: Final[str] = "__function_invoking_chat_client__"
|
||||
DEFAULT_MAX_ITERATIONS: Final[int] = 10
|
||||
DEFAULT_MAX_ITERATIONS: Final[int] = 40
|
||||
DEFAULT_MAX_CONSECUTIVE_ERRORS_PER_REQUEST: Final[int] = 3
|
||||
TChatClient = TypeVar("TChatClient", bound="ChatClientProtocol")
|
||||
# region Helpers
|
||||
|
||||
@@ -156,34 +158,19 @@ def _parse_inputs(
|
||||
# region Tools
|
||||
@runtime_checkable
|
||||
class ToolProtocol(Protocol):
|
||||
"""Represents a generic tool that can be specified to an AI service.
|
||||
"""Represents a generic tool.
|
||||
|
||||
This protocol defines the interface that all tools must implement to be compatible
|
||||
with the agent framework.
|
||||
with the agent framework. It is implemented by various tool classes such as HostedMCPTool,
|
||||
HostedWebSearchTool, and AIFunction's. A AIFunction is usually created by the `ai_function` decorator.
|
||||
|
||||
Since each connector needs to parse tools differently, users can pass a dict to
|
||||
specify a service-specific tool when no abstraction is available.
|
||||
|
||||
Attributes:
|
||||
name: The name of the tool.
|
||||
description: A description of the tool, suitable for use in describing the purpose to a model.
|
||||
additional_properties: Additional properties associated with the tool.
|
||||
|
||||
Examples:
|
||||
.. code-block:: python
|
||||
|
||||
from agent_framework import ToolProtocol
|
||||
|
||||
|
||||
class CustomTool:
|
||||
def __init__(self, name: str, description: str) -> None:
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.additional_properties = None
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"CustomTool(name={self.name})"
|
||||
|
||||
|
||||
# Tool now implements ToolProtocol
|
||||
tool: ToolProtocol = CustomTool("my_tool", "Does something useful")
|
||||
"""
|
||||
|
||||
name: str
|
||||
@@ -201,22 +188,11 @@ class ToolProtocol(Protocol):
|
||||
class BaseTool(SerializationMixin):
|
||||
"""Base class for AI tools, providing common attributes and methods.
|
||||
|
||||
This class provides the foundation for creating custom tools with serialization support.
|
||||
Used as the base class for the various tools in the agent framework, such as HostedMCPTool,
|
||||
HostedWebSearchTool, and AIFunction.
|
||||
|
||||
Examples:
|
||||
.. code-block:: python
|
||||
|
||||
from agent_framework import BaseTool
|
||||
|
||||
|
||||
class MyCustomTool(BaseTool):
|
||||
def __init__(self, name: str, custom_param: str) -> None:
|
||||
super().__init__(name=name, description="My custom tool")
|
||||
self.custom_param = custom_param
|
||||
|
||||
|
||||
tool = MyCustomTool(name="custom", custom_param="value")
|
||||
print(tool) # MyCustomTool(name=custom, description=My custom tool)
|
||||
Since each connector needs to parse tools differently, this class is not exposed directly to end users.
|
||||
In most cases, users can pass a dict to specify a service-specific tool when no abstraction is available.
|
||||
"""
|
||||
|
||||
DEFAULT_EXCLUDE: ClassVar[set[str]] = {"additional_properties"}
|
||||
@@ -551,6 +527,10 @@ def _default_histogram() -> Histogram:
|
||||
TClass = TypeVar("TClass", bound="SerializationMixin")
|
||||
|
||||
|
||||
class EmptyInputModel(BaseModel):
|
||||
"""An empty input model for functions with no parameters."""
|
||||
|
||||
|
||||
class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
|
||||
"""A tool that wraps a Python function to make it callable by AI models.
|
||||
|
||||
@@ -602,8 +582,10 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
|
||||
name: str,
|
||||
description: str = "",
|
||||
approval_mode: Literal["always_require", "never_require"] | None = None,
|
||||
max_invocations: int | None = None,
|
||||
max_invocation_exceptions: int | None = None,
|
||||
additional_properties: dict[str, Any] | None = None,
|
||||
func: Callable[..., Awaitable[ReturnT] | ReturnT],
|
||||
func: Callable[..., Awaitable[ReturnT] | ReturnT] | None = None,
|
||||
input_model: type[ArgsT] | Mapping[str, Any] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
@@ -614,6 +596,10 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
|
||||
description: A description of the function.
|
||||
approval_mode: Whether or not approval is required to run this tool.
|
||||
Default is that approval is not needed.
|
||||
max_invocations: The maximum number of times this function can be invoked.
|
||||
If None, there is no limit. Should be at least 1.
|
||||
max_invocation_exceptions: The maximum number of exceptions allowed during invocations.
|
||||
If None, there is no limit. Should be at least 1.
|
||||
additional_properties: Additional properties to set on the function.
|
||||
func: The function to wrap.
|
||||
input_model: The Pydantic model that defines the input parameters for the function.
|
||||
@@ -630,21 +616,56 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
|
||||
self.func = func
|
||||
self.input_model = self._resolve_input_model(input_model)
|
||||
self.approval_mode = approval_mode or "never_require"
|
||||
if max_invocations is not None and max_invocations < 1:
|
||||
raise ValueError("max_invocations must be at least 1 or None.")
|
||||
if max_invocation_exceptions is not None and max_invocation_exceptions < 1:
|
||||
raise ValueError("max_invocation_exceptions must be at least 1 or None.")
|
||||
self.max_invocations = max_invocations
|
||||
self.invocation_count = 0
|
||||
self.max_invocation_exceptions = max_invocation_exceptions
|
||||
self.invocation_exception_count = 0
|
||||
self._invocation_duration_histogram = _default_histogram()
|
||||
self.type: Literal["ai_function"] = "ai_function"
|
||||
|
||||
@property
|
||||
def declaration_only(self) -> bool:
|
||||
"""Indicate whether the function is declaration only (i.e., has no implementation)."""
|
||||
return self.func is None
|
||||
|
||||
def _resolve_input_model(self, input_model: type[ArgsT] | Mapping[str, Any] | None) -> type[ArgsT]:
|
||||
if input_model:
|
||||
if inspect.isclass(input_model) and issubclass(input_model, BaseModel):
|
||||
return input_model
|
||||
if isinstance(input_model, Mapping):
|
||||
return cast(type[ArgsT], _create_model_from_json_schema(self.name, input_model))
|
||||
raise TypeError("input_model must be a Pydantic BaseModel subclass or a JSON schema dict.")
|
||||
return cast(type[ArgsT], _create_input_model_from_func(self.func, self.name))
|
||||
"""Resolve the input model for the function."""
|
||||
if input_model is None:
|
||||
if self.func is None:
|
||||
return cast(type[ArgsT], EmptyInputModel)
|
||||
return cast(type[ArgsT], _create_input_model_from_func(func=self.func, name=self.name))
|
||||
if inspect.isclass(input_model) and issubclass(input_model, BaseModel):
|
||||
return input_model
|
||||
if isinstance(input_model, Mapping):
|
||||
return cast(type[ArgsT], _create_model_from_json_schema(self.name, input_model))
|
||||
raise TypeError("input_model must be a Pydantic BaseModel subclass or a JSON schema dict.")
|
||||
|
||||
def __call__(self, *args: Any, **kwargs: Any) -> ReturnT | Awaitable[ReturnT]:
|
||||
"""Call the wrapped function with the provided arguments."""
|
||||
return self.func(*args, **kwargs)
|
||||
if self.func is None:
|
||||
raise ToolException(f"Function '{self.name}' is declaration only and cannot be invoked.")
|
||||
if self.max_invocations is not None and self.invocation_count >= self.max_invocations:
|
||||
raise ToolException(
|
||||
f"Function '{self.name}' has reached its maximum invocation limit, you can no longer use this tool."
|
||||
)
|
||||
if (
|
||||
self.max_invocation_exceptions is not None
|
||||
and self.invocation_exception_count >= self.max_invocation_exceptions
|
||||
):
|
||||
raise ToolException(
|
||||
f"Function '{self.name}' has reached its maximum exception limit, "
|
||||
f"you tried to use this tool too many times and it kept failing."
|
||||
)
|
||||
self.invocation_count += 1
|
||||
try:
|
||||
return self.func(*args, **kwargs)
|
||||
except Exception:
|
||||
self.invocation_exception_count += 1
|
||||
raise
|
||||
|
||||
async def invoke(
|
||||
self,
|
||||
@@ -664,6 +685,8 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
|
||||
Raises:
|
||||
TypeError: If arguments is not an instance of the expected input model.
|
||||
"""
|
||||
if self.declaration_only:
|
||||
raise ToolException(f"Function '{self.name}' is declaration only and cannot be invoked.")
|
||||
global OBSERVABILITY_SETTINGS
|
||||
from .observability import OBSERVABILITY_SETTINGS
|
||||
|
||||
@@ -833,7 +856,7 @@ def _parse_annotation(annotation: Any) -> Any:
|
||||
return annotation
|
||||
|
||||
|
||||
def _create_input_model_from_func(func: Callable[..., Any], tool_name: str) -> type[BaseModel]:
|
||||
def _create_input_model_from_func(func: Callable[..., Any], name: str) -> type[BaseModel]:
|
||||
"""Create a Pydantic model from a function's signature."""
|
||||
sig = inspect.signature(func)
|
||||
fields = {
|
||||
@@ -844,7 +867,7 @@ def _create_input_model_from_func(func: Callable[..., Any], tool_name: str) -> t
|
||||
for pname, param in sig.parameters.items()
|
||||
if pname not in {"self", "cls"}
|
||||
}
|
||||
return create_model(f"{tool_name}_input", **fields) # type: ignore[call-overload, no-any-return]
|
||||
return create_model(f"{name}_input", **fields) # type: ignore[call-overload, no-any-return]
|
||||
|
||||
|
||||
# Map JSON Schema types to Pydantic types
|
||||
@@ -907,6 +930,8 @@ def ai_function(
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
approval_mode: Literal["always_require", "never_require"] | None = None,
|
||||
max_invocations: int | None = None,
|
||||
max_invocation_exceptions: int | None = None,
|
||||
additional_properties: dict[str, Any] | None = None,
|
||||
) -> AIFunction[Any, ReturnT]: ...
|
||||
|
||||
@@ -918,6 +943,8 @@ def ai_function(
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
approval_mode: Literal["always_require", "never_require"] | None = None,
|
||||
max_invocations: int | None = None,
|
||||
max_invocation_exceptions: int | None = None,
|
||||
additional_properties: dict[str, Any] | None = None,
|
||||
) -> Callable[[Callable[..., ReturnT | Awaitable[ReturnT]]], AIFunction[Any, ReturnT]]: ...
|
||||
|
||||
@@ -928,6 +955,8 @@ def ai_function(
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
approval_mode: Literal["always_require", "never_require"] | None = None,
|
||||
max_invocations: int | None = None,
|
||||
max_invocation_exceptions: int | None = None,
|
||||
additional_properties: dict[str, Any] | None = None,
|
||||
) -> AIFunction[Any, ReturnT] | Callable[[Callable[..., ReturnT | Awaitable[ReturnT]]], AIFunction[Any, ReturnT]]:
|
||||
"""Decorate a function to turn it into a AIFunction that can be passed to models and executed automatically.
|
||||
@@ -940,6 +969,22 @@ def ai_function(
|
||||
with a string description as the second argument. You can also use Pydantic's
|
||||
``Field`` class for more advanced configuration.
|
||||
|
||||
Args:
|
||||
func: The function to decorate.
|
||||
|
||||
Keyword Args:
|
||||
name: The name of the function. If not provided, the function's ``__name__``
|
||||
attribute will be used.
|
||||
description: A description of the function. If not provided, the function's
|
||||
docstring will be used.
|
||||
approval_mode: Whether or not approval is required to run this tool.
|
||||
Default is that approval is not needed.
|
||||
max_invocations: The maximum number of times this function can be invoked.
|
||||
If None, there is no limit, should be at least 1.
|
||||
max_invocation_exceptions: The maximum number of exceptions allowed during invocations.
|
||||
If None, there is no limit, should be at least 1.
|
||||
additional_properties: Additional properties to set on the function.
|
||||
|
||||
Note:
|
||||
When approval_mode is set to "always_require", the function will not be executed
|
||||
until explicit approval is given, this only applies to the auto-invocation flow.
|
||||
@@ -997,6 +1042,8 @@ def ai_function(
|
||||
name=tool_name,
|
||||
description=tool_desc,
|
||||
approval_mode=approval_mode,
|
||||
max_invocations=max_invocations,
|
||||
max_invocation_exceptions=max_invocation_exceptions,
|
||||
additional_properties=additional_properties or {},
|
||||
func=f,
|
||||
)
|
||||
@@ -1009,10 +1056,123 @@ def ai_function(
|
||||
# region Function Invoking Chat Client
|
||||
|
||||
|
||||
class FunctionInvocationConfiguration(SerializationMixin):
|
||||
"""Configuration for function invocation in chat clients.
|
||||
|
||||
This class is created automatically on every chat client that supports function invocation.
|
||||
This means that for most cases you can just alter the attributes on the instance, rather then creating a new one.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
# Create an OpenAI chat client
|
||||
client = OpenAIChatClient(api_key="your_api_key")
|
||||
|
||||
# Disable function invocation
|
||||
client.function_invocation_config.enabled = False
|
||||
|
||||
# Set maximum iterations to 10
|
||||
client.function_invocation_config.max_iterations = 10
|
||||
|
||||
# Enable termination on unknown function calls
|
||||
client.function_invocation_config.terminate_on_unknown_calls = True
|
||||
|
||||
# Add additional tools for function execution
|
||||
client.function_invocation_config.additional_tools = [my_custom_tool]
|
||||
|
||||
# Enable detailed error information in function results
|
||||
client.function_invocation_config.include_detailed_errors = True
|
||||
|
||||
# You can also create a new configuration instance if needed
|
||||
new_config = FunctionInvocationConfiguration(
|
||||
enabled=True,
|
||||
max_iterations=20,
|
||||
terminate_on_unknown_calls=False,
|
||||
additional_tools=[another_tool],
|
||||
include_detailed_errors=False,
|
||||
)
|
||||
|
||||
# and then assign it to the client
|
||||
client.function_invocation_config = new_config
|
||||
|
||||
|
||||
Attributes:
|
||||
enabled: Whether function invocation is enabled.
|
||||
When this is set to False, the client will not attempt to invoke any functions,
|
||||
because the tool mode will be set to None.
|
||||
max_iterations: Maximum number of function invocation iterations.
|
||||
Each request to this client might end up making multiple requests to the model. Each time the model responds
|
||||
with a function call request, this client might perform that invocation and send the results back to the
|
||||
model in a new request. This property limits the number of times such a roundtrip is performed. The value
|
||||
must be at least one, as it includes the initial request.
|
||||
If you want to fully disable function invocation, use the ``enabled`` property.
|
||||
The default is 40.
|
||||
max_consecutive_errors_per_request: Maximum consecutive errors allowed per request.
|
||||
The maximum number of consecutive function call errors allowed before stopping
|
||||
further function calls for the request.
|
||||
The default is 3.
|
||||
terminate_on_unknown_calls: Whether to terminate on unknown function calls.
|
||||
When False, call requests to any tools that aren't available to the client
|
||||
will result in a response message automatically being created and returned to the inner client stating that
|
||||
the tool couldn't be found. This behavior can help in cases where a model hallucinates a function, but it's
|
||||
problematic if the model has been made aware of the existence of tools outside of the normal mechanisms, and
|
||||
requests one of those. ``additional_tools`` can be used to help with that. But if instead the consumer wants
|
||||
to know about all function call requests that the client can't handle, this can be set to True. Upon
|
||||
receiving a request to call a function that the client doesn't know about, it will terminate the function
|
||||
calling loop and return the response, leaving the handling of the function call requests to the consumer of
|
||||
the client.
|
||||
additional_tools: Additional tools to include for function execution.
|
||||
These will not impact the requests sent by the client, which will pass through the
|
||||
``tools`` unmodified. However, if the inner client requests the invocation of a tool
|
||||
that was not in ``ChatOptions.tools``, this ``additional_tools`` collection will also be consulted to look
|
||||
for a corresponding tool. This is useful when the service might have been pre-configured to be aware of
|
||||
certain tools that aren't also sent on each individual request. These tools are treated the same as
|
||||
``declaration_only`` tools and will be returned to the user.
|
||||
include_detailed_errors: Whether to include detailed error information in function results.
|
||||
When set to True, detailed error information such as exception type and message
|
||||
will be included in the function result content when a function invocation fails.
|
||||
When False, only a generic error message will be included.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
enabled: bool = True,
|
||||
max_iterations: int = DEFAULT_MAX_ITERATIONS,
|
||||
max_consecutive_errors_per_request: int = DEFAULT_MAX_CONSECUTIVE_ERRORS_PER_REQUEST,
|
||||
terminate_on_unknown_calls: bool = False,
|
||||
additional_tools: Sequence[ToolProtocol] | None = None,
|
||||
include_detailed_errors: bool = False,
|
||||
) -> None:
|
||||
"""Initialize FunctionInvocationConfiguration.
|
||||
|
||||
Args:
|
||||
enabled: Whether function invocation is enabled.
|
||||
max_iterations: Maximum number of function invocation iterations.
|
||||
max_consecutive_errors_per_request: Maximum consecutive errors allowed per request.
|
||||
terminate_on_unknown_calls: Whether to terminate on unknown function calls.
|
||||
additional_tools: Additional tools to include for function execution.
|
||||
include_detailed_errors: Whether to include detailed error information in function results.
|
||||
"""
|
||||
self.enabled = enabled
|
||||
if max_iterations < 1:
|
||||
raise ValueError("max_iterations must be at least 1.")
|
||||
self.max_iterations = max_iterations
|
||||
if max_consecutive_errors_per_request < 0:
|
||||
raise ValueError("max_consecutive_errors_per_request must be 0 or more.")
|
||||
self.max_consecutive_errors_per_request = max_consecutive_errors_per_request
|
||||
self.terminate_on_unknown_calls = terminate_on_unknown_calls
|
||||
self.additional_tools = additional_tools or []
|
||||
self.include_detailed_errors = include_detailed_errors
|
||||
|
||||
|
||||
async def _auto_invoke_function(
|
||||
function_call_content: "FunctionCallContent | FunctionApprovalResponseContent",
|
||||
custom_args: dict[str, Any] | None = None,
|
||||
*,
|
||||
config: FunctionInvocationConfiguration,
|
||||
tool_map: dict[str, AIFunction[BaseModel, Any]],
|
||||
sequence_index: int | None = None,
|
||||
request_index: int | None = None,
|
||||
@@ -1025,6 +1185,7 @@ async def _auto_invoke_function(
|
||||
custom_args: Additional custom arguments to merge with parsed arguments.
|
||||
|
||||
Keyword Args:
|
||||
config: The function invocation configuration.
|
||||
tool_map: A mapping of tool names to AIFunction instances.
|
||||
sequence_index: The index of the function call in the sequence.
|
||||
request_index: The index of the request iteration.
|
||||
@@ -1037,29 +1198,33 @@ async def _auto_invoke_function(
|
||||
KeyError: If the requested function is not found in the tool map.
|
||||
"""
|
||||
from ._types import (
|
||||
FunctionApprovalRequestContent,
|
||||
FunctionApprovalResponseContent,
|
||||
FunctionCallContent,
|
||||
FunctionResultContent,
|
||||
)
|
||||
|
||||
# Note: The scenarios for approval_mode="always_require", declaration_only, and
|
||||
# terminate_on_unknown_calls are all handled in _try_execute_function_calls before
|
||||
# this function is called. This function only handles the actual execution of approved,
|
||||
# non-declaration-only functions.
|
||||
|
||||
tool: AIFunction[BaseModel, Any] | None = None
|
||||
if isinstance(function_call_content, FunctionCallContent):
|
||||
if function_call_content.type == "function_call":
|
||||
tool = tool_map.get(function_call_content.name)
|
||||
# Tool should exist because _try_execute_function_calls validates this
|
||||
if tool is None:
|
||||
raise KeyError(f"No tool or function named '{function_call_content.name}'")
|
||||
if tool.approval_mode == "always_require":
|
||||
return FunctionApprovalRequestContent(id=function_call_content.call_id, function_call=function_call_content)
|
||||
exc = KeyError(f'Function "{function_call_content.name}" not found.')
|
||||
return FunctionResultContent(
|
||||
call_id=function_call_content.call_id,
|
||||
result=f'Error: Requested function "{function_call_content.name}" not found.',
|
||||
exception=exc,
|
||||
)
|
||||
else:
|
||||
if isinstance(function_call_content, FunctionApprovalResponseContent):
|
||||
if function_call_content.approved:
|
||||
tool = tool_map.get(function_call_content.function_call.name)
|
||||
if tool is None:
|
||||
# we assume it is a hosted tool
|
||||
return function_call_content
|
||||
function_call_content = function_call_content.function_call
|
||||
else:
|
||||
raise ToolException("Unapproved tool cannot be executed.")
|
||||
# Note: Unapproved tools (approved=False) are handled in _replace_approval_contents_with_results
|
||||
# and never reach this function, so we only handle approved=True cases here.
|
||||
tool = tool_map.get(function_call_content.function_call.name)
|
||||
if tool is None:
|
||||
# we assume it is a hosted tool
|
||||
return function_call_content
|
||||
function_call_content = function_call_content.function_call
|
||||
|
||||
parsed_args: dict[str, Any] = dict(function_call_content.parse_arguments() or {})
|
||||
|
||||
@@ -1068,10 +1233,10 @@ async def _auto_invoke_function(
|
||||
try:
|
||||
args = tool.input_model.model_validate(merged_args)
|
||||
except ValidationError as exc:
|
||||
return FunctionResultContent(
|
||||
call_id=function_call_content.call_id,
|
||||
exception=exc,
|
||||
)
|
||||
message = "Error: Argument parsing failed."
|
||||
if config.include_detailed_errors:
|
||||
message = f"{message} Exception: {exc}"
|
||||
return FunctionResultContent(call_id=function_call_content.call_id, result=message, exception=exc)
|
||||
if not middleware_pipeline or (
|
||||
not hasattr(middleware_pipeline, "has_middlewares") and not middleware_pipeline.has_middlewares
|
||||
):
|
||||
@@ -1086,10 +1251,10 @@ async def _auto_invoke_function(
|
||||
result=function_result,
|
||||
)
|
||||
except Exception as exc:
|
||||
return FunctionResultContent(
|
||||
call_id=function_call_content.call_id,
|
||||
exception=exc,
|
||||
)
|
||||
message = "Error: Function failed."
|
||||
if config.include_detailed_errors:
|
||||
message = f"{message} Exception: {exc}"
|
||||
return FunctionResultContent(call_id=function_call_content.call_id, result=message, exception=exc)
|
||||
# Execute through middleware pipeline if available
|
||||
from ._middleware import FunctionInvocationContext
|
||||
|
||||
@@ -1117,10 +1282,10 @@ async def _auto_invoke_function(
|
||||
result=function_result,
|
||||
)
|
||||
except Exception as exc:
|
||||
return FunctionResultContent(
|
||||
call_id=function_call_content.call_id,
|
||||
exception=exc,
|
||||
)
|
||||
message = "Error: Function failed."
|
||||
if config.include_detailed_errors:
|
||||
message = f"{message} Exception: {exc}"
|
||||
return FunctionResultContent(call_id=function_call_content.call_id, result=message, exception=exc)
|
||||
|
||||
|
||||
def _get_tool_map(
|
||||
@@ -1141,7 +1306,7 @@ def _get_tool_map(
|
||||
return ai_function_list
|
||||
|
||||
|
||||
async def _execute_function_calls(
|
||||
async def _try_execute_function_calls(
|
||||
custom_args: dict[str, Any],
|
||||
attempt_idx: int,
|
||||
function_calls: Sequence["FunctionCallContent"] | Sequence["FunctionApprovalResponseContent"],
|
||||
@@ -1149,6 +1314,7 @@ async def _execute_function_calls(
|
||||
| Callable[..., Any] \
|
||||
| MutableMapping[str, Any] \
|
||||
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]",
|
||||
config: FunctionInvocationConfiguration,
|
||||
middleware_pipeline: Any = None, # Optional MiddlewarePipeline to avoid circular imports
|
||||
) -> Sequence["Contents"]:
|
||||
"""Execute multiple function calls concurrently.
|
||||
@@ -1158,22 +1324,33 @@ async def _execute_function_calls(
|
||||
attempt_idx: The index of the current attempt iteration.
|
||||
function_calls: A sequence of FunctionCallContent to execute.
|
||||
tools: The tools available for execution.
|
||||
config: Configuration for function invocation.
|
||||
middleware_pipeline: Optional middleware pipeline to apply during execution.
|
||||
|
||||
Returns:
|
||||
A list of Contents containing the results of each function call.
|
||||
A list of Contents containing the results of each function call,
|
||||
or the approval requests if any function requires approval,
|
||||
or the original function calls if any are declaration only.
|
||||
"""
|
||||
from ._types import FunctionApprovalRequestContent, FunctionCallContent
|
||||
|
||||
tool_map = _get_tool_map(tools)
|
||||
approval_tools = [tool_name for tool_name, tool in tool_map.items() if tool.approval_mode == "always_require"]
|
||||
declaration_only = [tool_name for tool_name, tool in tool_map.items() if tool.declaration_only]
|
||||
additional_tool_names = [tool.name for tool in config.additional_tools] if config.additional_tools else []
|
||||
# check if any are calling functions that need approval
|
||||
# if so, we return approval request for all
|
||||
approval_needed = False
|
||||
declaration_only_flag = False
|
||||
for fcc in function_calls:
|
||||
if isinstance(fcc, FunctionCallContent) and fcc.name in approval_tools:
|
||||
approval_needed = True
|
||||
break
|
||||
if isinstance(fcc, FunctionCallContent) and (fcc.name in declaration_only or fcc.name in additional_tool_names):
|
||||
declaration_only_flag = True
|
||||
break
|
||||
if config.terminate_on_unknown_calls and isinstance(fcc, FunctionCallContent) and fcc.name not in tool_map:
|
||||
raise KeyError(f'Error: Requested function "{fcc.name}" not found.')
|
||||
if approval_needed:
|
||||
# approval can only be needed for Function Call Contents, not Approval Responses.
|
||||
return [
|
||||
@@ -1181,6 +1358,9 @@ async def _execute_function_calls(
|
||||
for fcc in function_calls
|
||||
if isinstance(fcc, FunctionCallContent)
|
||||
]
|
||||
if declaration_only_flag:
|
||||
# return the declaration only tools to the user, since we cannot execute them.
|
||||
return [fcc for fcc in function_calls if isinstance(fcc, FunctionCallContent)]
|
||||
|
||||
# Run all function calls concurrently
|
||||
return await asyncio.gather(*[
|
||||
@@ -1191,6 +1371,7 @@ async def _execute_function_calls(
|
||||
sequence_index=seq_idx,
|
||||
request_index=attempt_idx,
|
||||
middleware_pipeline=middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
for seq_idx, function_call in enumerate(function_calls)
|
||||
])
|
||||
@@ -1334,17 +1515,17 @@ def _handle_function_calls_response(
|
||||
# because the underlying function may not preserve it in kwargs
|
||||
stored_middleware_pipeline = kwargs.get("_function_middleware_pipeline")
|
||||
|
||||
# Get max_iterations from instance additional_properties or class attribute
|
||||
instance_max_iterations: int = DEFAULT_MAX_ITERATIONS
|
||||
if hasattr(self, "additional_properties") and self.additional_properties:
|
||||
instance_max_iterations = self.additional_properties.get("max_iterations", DEFAULT_MAX_ITERATIONS)
|
||||
elif hasattr(self.__class__, "MAX_ITERATIONS"):
|
||||
instance_max_iterations = getattr(self.__class__, "MAX_ITERATIONS", DEFAULT_MAX_ITERATIONS)
|
||||
# Get the config for function invocation (not part of ChatClientProtocol, hence getattr)
|
||||
config: FunctionInvocationConfiguration | None = getattr(self, "function_invocation_configuration", None)
|
||||
if not config:
|
||||
# Default config if not set
|
||||
config = FunctionInvocationConfiguration()
|
||||
|
||||
errors_in_a_row: int = 0
|
||||
prepped_messages = prepare_messages(messages)
|
||||
response: "ChatResponse | None" = None
|
||||
fcc_messages: "list[ChatMessage]" = []
|
||||
for attempt_idx in range(instance_max_iterations):
|
||||
for attempt_idx in range(config.max_iterations if config.enabled else 0):
|
||||
fcc_todo = _collect_approval_responses(prepped_messages)
|
||||
if fcc_todo:
|
||||
tools = _extract_tools(kwargs)
|
||||
@@ -1352,13 +1533,29 @@ def _handle_function_calls_response(
|
||||
approved_responses = [resp for resp in fcc_todo.values() if resp.approved]
|
||||
approved_function_results: list[Contents] = []
|
||||
if approved_responses:
|
||||
approved_function_results = await _execute_function_calls(
|
||||
approved_function_results = await _try_execute_function_calls(
|
||||
custom_args=kwargs,
|
||||
attempt_idx=attempt_idx,
|
||||
function_calls=approved_responses,
|
||||
tools=tools, # type: ignore
|
||||
middleware_pipeline=stored_middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
if any(
|
||||
fcr.exception is not None
|
||||
for fcr in approved_function_results
|
||||
if isinstance(fcr, FunctionResultContent)
|
||||
):
|
||||
errors_in_a_row += 1
|
||||
# no need to reset the counter here, since this is the start of a new attempt.
|
||||
if errors_in_a_row >= config.max_consecutive_errors_per_request:
|
||||
logger.warning(
|
||||
"Maximum consecutive function call errors reached (%d). "
|
||||
"Stopping further function calls for this request.",
|
||||
config.max_consecutive_errors_per_request,
|
||||
)
|
||||
# break out of the loop and do the fallback response
|
||||
break
|
||||
_replace_approval_contents_with_results(prepped_messages, fcc_todo, approved_function_results)
|
||||
|
||||
response = await func(self, messages=prepped_messages, **kwargs)
|
||||
@@ -1381,15 +1578,15 @@ def _handle_function_calls_response(
|
||||
if function_calls and tools:
|
||||
# Use the stored middleware pipeline instead of extracting from kwargs
|
||||
# because kwargs may have been modified by the underlying function
|
||||
function_call_results: list[Contents] = await _execute_function_calls(
|
||||
function_call_results: list[Contents] = await _try_execute_function_calls(
|
||||
custom_args=kwargs,
|
||||
attempt_idx=attempt_idx,
|
||||
function_calls=function_calls,
|
||||
tools=tools, # type: ignore
|
||||
middleware_pipeline=stored_middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
|
||||
# Check if we have approval requests in the results
|
||||
# Check if we have approval requests or function calls (not results) in the results
|
||||
if any(isinstance(fccr, FunctionApprovalRequestContent) for fccr in function_call_results):
|
||||
# Add approval requests to the existing assistant message (with tool_calls)
|
||||
# instead of creating a separate tool message
|
||||
@@ -1402,6 +1599,26 @@ def _handle_function_calls_response(
|
||||
result_message = ChatMessage(role="assistant", contents=function_call_results)
|
||||
response.messages.append(result_message)
|
||||
return response
|
||||
if any(isinstance(fccr, FunctionCallContent) for fccr in function_call_results):
|
||||
# the function calls are already in the response, so we just continue
|
||||
return response
|
||||
|
||||
if any(
|
||||
fcr.exception is not None
|
||||
for fcr in function_call_results
|
||||
if isinstance(fcr, FunctionResultContent)
|
||||
):
|
||||
errors_in_a_row += 1
|
||||
if errors_in_a_row >= config.max_consecutive_errors_per_request:
|
||||
logger.warning(
|
||||
"Maximum consecutive function call errors reached (%d). "
|
||||
"Stopping further function calls for this request.",
|
||||
config.max_consecutive_errors_per_request,
|
||||
)
|
||||
# break out of the loop and do the fallback response
|
||||
break
|
||||
else:
|
||||
errors_in_a_row = 0
|
||||
|
||||
# add a single ChatMessage to the response with the results
|
||||
result_message = ChatMessage(role="tool", contents=function_call_results)
|
||||
@@ -1482,16 +1699,16 @@ def _handle_function_calls_streaming_response(
|
||||
# because the underlying function may not preserve it in kwargs
|
||||
stored_middleware_pipeline = kwargs.get("_function_middleware_pipeline")
|
||||
|
||||
# Get max_iterations from instance additional_properties or class attribute
|
||||
instance_max_iterations: int = DEFAULT_MAX_ITERATIONS
|
||||
if hasattr(self, "additional_properties") and self.additional_properties:
|
||||
instance_max_iterations = self.additional_properties.get("max_iterations", DEFAULT_MAX_ITERATIONS)
|
||||
elif hasattr(self.__class__, "MAX_ITERATIONS"):
|
||||
instance_max_iterations = getattr(self.__class__, "MAX_ITERATIONS", DEFAULT_MAX_ITERATIONS)
|
||||
# Get the config for function invocation (not part of ChatClientProtocol, hence getattr)
|
||||
config: FunctionInvocationConfiguration | None = getattr(self, "function_invocation_configuration", None)
|
||||
if not config:
|
||||
# Default config if not set
|
||||
config = FunctionInvocationConfiguration()
|
||||
|
||||
errors_in_a_row: int = 0
|
||||
prepped_messages = prepare_messages(messages)
|
||||
fcc_messages: "list[ChatMessage]" = []
|
||||
for attempt_idx in range(instance_max_iterations):
|
||||
for attempt_idx in range(config.max_iterations if config.enabled else 0):
|
||||
fcc_todo = _collect_approval_responses(prepped_messages)
|
||||
if fcc_todo:
|
||||
tools = _extract_tools(kwargs)
|
||||
@@ -1499,13 +1716,21 @@ def _handle_function_calls_streaming_response(
|
||||
approved_responses = [resp for resp in fcc_todo.values() if resp.approved]
|
||||
approved_function_results: list[Contents] = []
|
||||
if approved_responses:
|
||||
approved_function_results = await _execute_function_calls(
|
||||
approved_function_results = await _try_execute_function_calls(
|
||||
custom_args=kwargs,
|
||||
attempt_idx=attempt_idx,
|
||||
function_calls=approved_responses,
|
||||
tools=tools, # type: ignore
|
||||
middleware_pipeline=stored_middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
if any(
|
||||
fcr.exception is not None
|
||||
for fcr in approved_function_results
|
||||
if isinstance(fcr, FunctionResultContent)
|
||||
):
|
||||
errors_in_a_row += 1
|
||||
# no need to reset the counter here, since this is the start of a new attempt.
|
||||
_replace_approval_contents_with_results(prepped_messages, fcc_todo, approved_function_results)
|
||||
|
||||
all_updates: list["ChatResponseUpdate"] = []
|
||||
@@ -1551,15 +1776,16 @@ def _handle_function_calls_streaming_response(
|
||||
if function_calls and tools:
|
||||
# Use the stored middleware pipeline instead of extracting from kwargs
|
||||
# because kwargs may have been modified by the underlying function
|
||||
function_call_results: list[Contents] = await _execute_function_calls(
|
||||
function_call_results: list[Contents] = await _try_execute_function_calls(
|
||||
custom_args=kwargs,
|
||||
attempt_idx=attempt_idx,
|
||||
function_calls=function_calls,
|
||||
tools=tools, # type: ignore
|
||||
middleware_pipeline=stored_middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
|
||||
# Check if we have approval requests in the results
|
||||
# Check if we have approval requests or function calls (not results) in the results
|
||||
if any(isinstance(fccr, FunctionApprovalRequestContent) for fccr in function_call_results):
|
||||
# Add approval requests to the existing assistant message (with tool_calls)
|
||||
# instead of creating a separate tool message
|
||||
@@ -1575,6 +1801,26 @@ def _handle_function_calls_streaming_response(
|
||||
yield ChatResponseUpdate(contents=function_call_results, role="assistant")
|
||||
response.messages.append(result_message)
|
||||
return
|
||||
if any(isinstance(fccr, FunctionCallContent) for fccr in function_call_results):
|
||||
# the function calls were already yielded.
|
||||
return
|
||||
|
||||
if any(
|
||||
fcr.exception is not None
|
||||
for fcr in function_call_results
|
||||
if isinstance(fcr, FunctionResultContent)
|
||||
):
|
||||
errors_in_a_row += 1
|
||||
if errors_in_a_row >= config.max_consecutive_errors_per_request:
|
||||
logger.warning(
|
||||
"Maximum consecutive function call errors reached (%d). "
|
||||
"Stopping further function calls for this request.",
|
||||
config.max_consecutive_errors_per_request,
|
||||
)
|
||||
# break out of the loop and do the fallback response
|
||||
break
|
||||
else:
|
||||
errors_in_a_row = 0
|
||||
|
||||
# add a single ChatMessage to the response with the results
|
||||
result_message = ChatMessage(role="tool", contents=function_call_results)
|
||||
@@ -1648,10 +1894,6 @@ def use_function_invocation(
|
||||
if getattr(chat_client, FUNCTION_INVOKING_CHAT_CLIENT_MARKER, False):
|
||||
return chat_client
|
||||
|
||||
# Set MAX_ITERATIONS as a class variable if not already set
|
||||
if not hasattr(chat_client, "MAX_ITERATIONS"):
|
||||
chat_client.MAX_ITERATIONS = DEFAULT_MAX_ITERATIONS # type: ignore
|
||||
|
||||
try:
|
||||
chat_client.get_response = _handle_function_calls_response( # type: ignore
|
||||
func=chat_client.get_response, # type: ignore
|
||||
|
||||
@@ -502,8 +502,6 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
|
||||
tool_outputs = []
|
||||
if function_result_content.result:
|
||||
output = prepare_function_call_results(function_result_content.result)
|
||||
elif function_result_content.exception:
|
||||
output = "Error: " + str(function_result_content.exception)
|
||||
else:
|
||||
output = "No output received."
|
||||
tool_outputs.append(ToolOutput(tool_call_id=call_id, output=output))
|
||||
|
||||
@@ -380,11 +380,6 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
|
||||
args["tool_call_id"] = content.call_id
|
||||
if content.result is not None:
|
||||
args["content"] = prepare_function_call_results(content.result)
|
||||
elif content.exception is not None:
|
||||
# Send the exception message to the model
|
||||
# Otherwise we won't have any channels to talk to OpenAI
|
||||
# TODO(yuge): This should ideally be customizable
|
||||
args["content"] = "Error: " + str(content.exception)
|
||||
case _:
|
||||
if "content" not in args:
|
||||
args["content"] = []
|
||||
|
||||
@@ -501,8 +501,6 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
|
||||
}
|
||||
if content.result:
|
||||
args["output"] = prepare_function_call_results(content.result)
|
||||
if content.exception:
|
||||
args["output"] = "Error: " + str(content.exception)
|
||||
return args
|
||||
case FunctionApprovalRequestContent():
|
||||
return {
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -63,6 +63,26 @@ def test_ai_function_decorator_without_args():
|
||||
assert test_tool(1, 2) == 3
|
||||
|
||||
|
||||
def test_ai_function_without_args():
|
||||
"""Test the ai_function decorator."""
|
||||
|
||||
@ai_function
|
||||
def test_tool() -> int:
|
||||
"""A simple function that adds two numbers."""
|
||||
return 1 + 2
|
||||
|
||||
assert isinstance(test_tool, ToolProtocol)
|
||||
assert isinstance(test_tool, AIFunction)
|
||||
assert test_tool.name == "test_tool"
|
||||
assert test_tool.description == "A simple function that adds two numbers."
|
||||
assert test_tool.parameters() == {
|
||||
"properties": {},
|
||||
"title": "test_tool_input",
|
||||
"type": "object",
|
||||
}
|
||||
assert test_tool() == 3
|
||||
|
||||
|
||||
async def test_ai_function_decorator_with_async():
|
||||
"""Test the ai_function decorator with an async function."""
|
||||
|
||||
|
||||
@@ -689,12 +689,15 @@ def test_function_result_exception_handling(openai_unit_test_env: dict[str, str]
|
||||
# Test with exception (no result)
|
||||
test_exception = ValueError("Test error message")
|
||||
message_with_exception = ChatMessage(
|
||||
role="tool", contents=[FunctionResultContent(call_id="call-123", exception=test_exception)]
|
||||
role="tool",
|
||||
contents=[
|
||||
FunctionResultContent(call_id="call-123", result="Error: Function failed.", exception=test_exception)
|
||||
],
|
||||
)
|
||||
|
||||
openai_messages = client._openai_chat_message_parser(message_with_exception)
|
||||
assert len(openai_messages) == 1
|
||||
assert openai_messages[0]["content"] == "Error: Test error message"
|
||||
assert openai_messages[0]["content"] == "Error: Function failed."
|
||||
assert openai_messages[0]["tool_call_id"] == "call-123"
|
||||
|
||||
|
||||
|
||||
@@ -218,9 +218,14 @@ This directory contains samples demonstrating the capabilities of Microsoft Agen
|
||||
|
||||
| File | Description |
|
||||
|------|-------------|
|
||||
| [`getting_started/tools/ai_tool_with_approval.py`](./getting_started/tools/ai_tool_with_approval.py) | Demonstration of a tool with approvals |
|
||||
| [`getting_started/tools/ai_tool_with_approval_and_threads.py`](./getting_started/tools/ai_tool_with_approval_and_threads.py) | Tool Approvals with Threads |
|
||||
| [`getting_started/tools/failing_tools.py`](./getting_started/tools/failing_tools.py) | Tool exceptions handled by returning the error for the agent to recover from |
|
||||
| [`getting_started/tools/ai_function_declaration_only.py`](./getting_started/tools/ai_function_declaration_only.py) | Function declarations without implementations for testing agent reasoning |
|
||||
| [`getting_started/tools/ai_function_from_dict_with_dependency_injection.py`](./getting_started/tools/ai_function_from_dict_with_dependency_injection.py) | Creating AI functions from dictionary definitions using dependency injection |
|
||||
| [`getting_started/tools/ai_function_recover_from_failures.py`](./getting_started/tools/ai_function_recover_from_failures.py) | Graceful error handling when tools raise exceptions |
|
||||
| [`getting_started/tools/ai_function_with_approval.py`](./getting_started/tools/ai_function_with_approval.py) | User approval workflows for function calls without threads |
|
||||
| [`getting_started/tools/ai_function_with_approval_and_threads.py`](./getting_started/tools/ai_function_with_approval_and_threads.py) | Tool approval workflows using threads for conversation history management |
|
||||
| [`getting_started/tools/ai_function_with_max_exceptions.py`](./getting_started/tools/ai_function_with_max_exceptions.py) | Limiting tool failure exceptions using max_invocation_exceptions |
|
||||
| [`getting_started/tools/ai_function_with_max_invocations.py`](./getting_started/tools/ai_function_with_max_invocations.py) | Limiting total tool invocations using max_invocations |
|
||||
| [`getting_started/tools/ai_functions_in_class.py`](./getting_started/tools/ai_functions_in_class.py) | Using ai_function decorator with class methods for stateful tools |
|
||||
|
||||
## Workflows
|
||||
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
# Tools Examples
|
||||
|
||||
This folder contains examples demonstrating how to use AI functions (tools) with the Agent Framework. AI functions allow agents to interact with external systems, perform computations, and execute custom logic.
|
||||
|
||||
## Examples
|
||||
|
||||
| File | Description |
|
||||
|------|-------------|
|
||||
| [`ai_function_declaration_only.py`](ai_function_declaration_only.py) | Demonstrates how to create function declarations without implementations. Useful for testing agent reasoning about tool usage or when tools are defined elsewhere. Shows how agents request tool calls even when the tool won't be executed. |
|
||||
| [`ai_function_from_dict_with_dependency_injection.py`](ai_function_from_dict_with_dependency_injection.py) | Shows how to create AI functions from dictionary definitions using dependency injection. The function implementation is injected at runtime during deserialization, enabling dynamic tool creation and configuration. Note: This serialization/deserialization feature is in active development. |
|
||||
| [`ai_function_recover_from_failures.py`](ai_function_recover_from_failures.py) | Demonstrates graceful error handling when tools raise exceptions. Shows how agents receive error information and can recover from failures, deciding whether to retry or respond differently based on the exception. |
|
||||
| [`ai_function_with_approval.py`](ai_function_with_approval.py) | Shows how to implement user approval workflows for function calls without using threads. Demonstrates both streaming and non-streaming approval patterns where users can approve or reject function executions before they run. |
|
||||
| [`ai_function_with_approval_and_threads.py`](ai_function_with_approval_and_threads.py) | Demonstrates tool approval workflows using threads for automatic conversation history management. Shows how threads simplify approval workflows by automatically storing and retrieving conversation context. Includes both approval and rejection examples. |
|
||||
| [`ai_function_with_max_exceptions.py`](ai_function_with_max_exceptions.py) | Shows how to limit the number of times a tool can fail with exceptions using `max_invocation_exceptions`. Useful for preventing expensive tools from being called repeatedly when they keep failing. |
|
||||
| [`ai_function_with_max_invocations.py`](ai_function_with_max_invocations.py) | Demonstrates limiting the total number of times a tool can be invoked using `max_invocations`. Useful for rate-limiting expensive operations or ensuring tools are only called a specific number of times per conversation. |
|
||||
| [`ai_functions_in_class.py`](ai_functions_in_class.py) | Shows how to use `ai_function` decorator with class methods to create stateful tools. Demonstrates how class state can control tool behavior dynamically, allowing you to adjust tool functionality at runtime by modifying class properties. |
|
||||
|
||||
## Key Concepts
|
||||
|
||||
### AI Function Features
|
||||
|
||||
- **Function Declarations**: Define tool schemas without implementations for testing or external tools
|
||||
- **Dependency Injection**: Create tools from configurations with runtime-injected implementations
|
||||
- **Error Handling**: Gracefully handle and recover from tool execution failures
|
||||
- **Approval Workflows**: Require user approval before executing sensitive or important operations
|
||||
- **Invocation Limits**: Control how many times tools can be called or fail
|
||||
- **Stateful Tools**: Use class methods as tools to maintain state and dynamically control behavior
|
||||
|
||||
### Common Patterns
|
||||
|
||||
#### Basic Tool Definition
|
||||
|
||||
```python
|
||||
from agent_framework import ai_function
|
||||
from typing import Annotated
|
||||
|
||||
@ai_function
|
||||
def my_tool(param: Annotated[str, "Description"]) -> str:
|
||||
"""Tool description for the AI."""
|
||||
return f"Result: {param}"
|
||||
```
|
||||
|
||||
#### Tool with Approval
|
||||
|
||||
```python
|
||||
@ai_function(approval_mode="always_require")
|
||||
def sensitive_operation(data: Annotated[str, "Data to process"]) -> str:
|
||||
"""This requires user approval before execution."""
|
||||
return f"Processed: {data}"
|
||||
```
|
||||
|
||||
#### Tool with Invocation Limits
|
||||
|
||||
```python
|
||||
@ai_function(max_invocations=3)
|
||||
def limited_tool() -> str:
|
||||
"""Can only be called 3 times total."""
|
||||
return "Result"
|
||||
|
||||
@ai_function(max_invocation_exceptions=2)
|
||||
def fragile_tool() -> str:
|
||||
"""Can only fail 2 times before being disabled."""
|
||||
return "Result"
|
||||
```
|
||||
|
||||
#### Stateful Tools with Classes
|
||||
|
||||
```python
|
||||
class MyTools:
|
||||
def __init__(self, mode: str = "normal"):
|
||||
self.mode = mode
|
||||
|
||||
def process(self, data: Annotated[str, "Data to process"]) -> str:
|
||||
"""Process data based on current mode."""
|
||||
if self.mode == "safe":
|
||||
return f"Safely processed: {data}"
|
||||
return f"Processed: {data}"
|
||||
|
||||
# Create instance and use methods as tools
|
||||
tools = MyTools(mode="safe")
|
||||
agent = client.create_agent(tools=tools.process)
|
||||
|
||||
# Change behavior dynamically
|
||||
tools.mode = "normal"
|
||||
```
|
||||
|
||||
### Error Handling
|
||||
|
||||
When tools raise exceptions:
|
||||
1. The exception is captured and sent to the agent as a function result
|
||||
2. The agent receives the error message and can reason about what went wrong
|
||||
3. The agent can retry with different parameters, use alternative tools, or explain the issue to the user
|
||||
4. With invocation limits, tools can be disabled after repeated failures
|
||||
|
||||
### Approval Workflows
|
||||
|
||||
Two approaches for handling approvals:
|
||||
|
||||
1. **Without Threads**: Manually manage conversation context, including the query, approval request, and response in each iteration
|
||||
2. **With Threads**: Thread automatically manages conversation history, simplifying the approval workflow
|
||||
|
||||
## Usage Tips
|
||||
|
||||
- Use **declaration-only** functions when you want to test agent reasoning without execution
|
||||
- Use **dependency injection** for dynamic tool configuration and plugin architectures
|
||||
- Implement **approval workflows** for operations that modify data, spend money, or require human oversight
|
||||
- Set **invocation limits** to prevent runaway costs or infinite loops with expensive tools
|
||||
- Handle **exceptions gracefully** to create robust agents that can recover from failures
|
||||
- Use **class-based tools** when you need to maintain state or dynamically adjust tool behavior at runtime
|
||||
|
||||
## Running the Examples
|
||||
|
||||
Each example is a standalone Python script that can be run directly:
|
||||
|
||||
```bash
|
||||
uv run python ai_function_with_approval.py
|
||||
```
|
||||
|
||||
Make sure you have the necessary environment variables configured (like `OPENAI_API_KEY` or Azure credentials) before running the examples.
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
from agent_framework import AIFunction
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
|
||||
"""
|
||||
Example of how to create a function that only consists of a declaration without an implementation.
|
||||
This is useful when you want the agent to use tools that are defined elsewhere or when you want
|
||||
to test the agent's ability to reason about tool usage without executing them.
|
||||
|
||||
The only difference is that you provide an AIFunction without a function.
|
||||
If you need a input_model, you can still provide that as well.
|
||||
"""
|
||||
|
||||
|
||||
async def main():
|
||||
function_declaration = AIFunction[None, None](
|
||||
name="get_current_time",
|
||||
description="Get the current time in ISO 8601 format.",
|
||||
)
|
||||
|
||||
agent = OpenAIResponsesClient().create_agent(
|
||||
name="DeclarationOnlyToolAgent",
|
||||
instructions="You are a helpful agent that uses tools.",
|
||||
tools=function_declaration,
|
||||
)
|
||||
query = "What is the current time?"
|
||||
print(f"User: {query}")
|
||||
result = await agent.run(query)
|
||||
print(f"Result: {result.to_json(indent=2)}\n")
|
||||
|
||||
|
||||
"""
|
||||
Expected result:
|
||||
User: What is the current time?
|
||||
Result: {
|
||||
"type": "agent_run_response",
|
||||
"messages": [
|
||||
{
|
||||
"type": "chat_message",
|
||||
"role": {
|
||||
"type": "role",
|
||||
"value": "assistant"
|
||||
},
|
||||
"contents": [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_0flN9rfGLK8LhORy4uMDiRSC",
|
||||
"name": "get_current_time",
|
||||
"arguments": "{}",
|
||||
"fc_id": "fc_0fd5f269955c589f016904c46584348195b84a8736e61248de"
|
||||
}
|
||||
],
|
||||
"author_name": "DeclarationOnlyToolAgent",
|
||||
"additional_properties": {}
|
||||
}
|
||||
],
|
||||
"response_id": "resp_0fd5f269955c589f016904c462d5cc819599d28384ba067edc",
|
||||
"created_at": "2025-10-31T15:14:58.000000Z",
|
||||
"usage_details": {
|
||||
"type": "usage_details",
|
||||
"input_token_count": 63,
|
||||
"output_token_count": 145,
|
||||
"total_token_count": 208,
|
||||
"openai.reasoning_tokens": 128
|
||||
},
|
||||
"additional_properties": {}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,188 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import FunctionCallContent, FunctionResultContent, ai_function
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
|
||||
"""
|
||||
Some tools are very expensive to run, so you may want to limit the number of times
|
||||
it tries to call them and fails. This sample shows a tool that can only raise exceptions a
|
||||
limited number of times.
|
||||
"""
|
||||
|
||||
|
||||
# we trick the AI into calling this function with 0 as denominator to trigger the exception
|
||||
@ai_function(max_invocation_exceptions=1)
|
||||
def safe_divide(
|
||||
a: Annotated[int, "Numerator"],
|
||||
b: Annotated[int, "Denominator"],
|
||||
) -> str:
|
||||
"""Divide two numbers can be used with 0 as denominator."""
|
||||
try:
|
||||
result = a / b # Will raise ZeroDivisionError
|
||||
except ZeroDivisionError as exc:
|
||||
print(f" Tool failed with error: {exc}")
|
||||
raise
|
||||
|
||||
return f"{a} / {b} = {result}"
|
||||
|
||||
|
||||
async def main():
|
||||
# tools = Tools()
|
||||
agent = OpenAIResponsesClient().create_agent(
|
||||
name="ToolAgent",
|
||||
instructions="Use the provided tools.",
|
||||
tools=[safe_divide],
|
||||
)
|
||||
thread = agent.get_new_thread()
|
||||
print("=" * 60)
|
||||
print("Step 1: Call divide(10, 0) - tool raises exception")
|
||||
response = await agent.run("Divide 10 by 0", thread=thread)
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
print("Step 2: Call divide(100, 0) - will refuse to execute due to max_invocation_exceptions")
|
||||
response = await agent.run("Divide 100 by 0", thread=thread)
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
print(f"Number of tool calls attempted: {safe_divide.invocation_count}")
|
||||
print(f"Number of tool calls failed: {safe_divide.invocation_exception_count}")
|
||||
print("Replay the conversation:")
|
||||
assert thread.message_store
|
||||
assert thread.message_store.list_messages
|
||||
for idx, msg in enumerate(await thread.message_store.list_messages()):
|
||||
if msg.text:
|
||||
print(f"{idx + 1} {msg.author_name or msg.role}: {msg.text} ")
|
||||
for content in msg.contents:
|
||||
if isinstance(content, FunctionCallContent):
|
||||
print(
|
||||
f"{idx + 1} {msg.author_name}: calling function: {content.name} with arguments: {content.arguments}"
|
||||
)
|
||||
if isinstance(content, FunctionResultContent):
|
||||
print(f"{idx + 1} {msg.role}: {content.result if content.result else content.exception}")
|
||||
|
||||
|
||||
"""
|
||||
Expected Output:
|
||||
============================================================
|
||||
Step 1: Call divide(10, 0) - tool raises exception
|
||||
Tool failed with error: division by zero
|
||||
[2025-10-31 15:39:53 - /Users/edvan/Work/agent-framework/python/packages/core/agent_framework/_tools.py:718 - ERROR]
|
||||
Function failed. Error: division by zero
|
||||
Response: Division by zero is undefined in standard arithmetic. There is no finite value for 10 ÷ 0.
|
||||
|
||||
If you want alternatives:
|
||||
- A valid example: 10 ÷ 2 = 5.
|
||||
- To handle safely in code, you can check the denominator first (e.g., in Python: if b == 0:
|
||||
handle error else: compute a/b).
|
||||
- If you’re curious about limits: as x → 0+, 10/x → +∞; as x → 0−, 10/x → −∞; there is no finite limit.
|
||||
|
||||
Would you like me to show a safe division snippet in a specific language, or compute something else?
|
||||
============================================================
|
||||
Step 2: Call divide(100, 0) - will refuse to execute due to max_invocations
|
||||
[2025-10-31 15:40:09 - /Users/edvan/Work/agent-framework/python/packages/core/agent_framework/_tools.py:718 - ERROR]
|
||||
Function failed. Error: Function 'safe_divide' has reached its maximum exception limit, you tried to use this
|
||||
tool too many times and it kept failing.
|
||||
Response: Division by zero is undefined in standard arithmetic, so 100 ÷ 0 has no finite value.
|
||||
|
||||
If you’re coding and want safe handling, here are quick patterns in a few languages:
|
||||
|
||||
- Python
|
||||
def safe_divide(a, b):
|
||||
if b == 0:
|
||||
return None # or raise an exception
|
||||
return a / b
|
||||
|
||||
safe_divide(100, 0) # -> None
|
||||
|
||||
- JavaScript
|
||||
function safeDivide(a, b) {
|
||||
if (b === 0) return undefined; // or throw
|
||||
return a / b;
|
||||
}
|
||||
|
||||
safeDivide(100, 0) // -> undefined
|
||||
|
||||
- Java
|
||||
public static Double safeDivide(double a, double b) {
|
||||
if (b == 0.0) throw new ArithmeticException("Divide by zero");
|
||||
return a / b;
|
||||
}
|
||||
|
||||
safeDivide(100, 0) // -> exception
|
||||
|
||||
- C/C++
|
||||
double safeDivide(double a, double b) {
|
||||
if (b == 0.0) return std::numeric_limits<double>::infinity(); // or handle error
|
||||
return a / b;
|
||||
}
|
||||
|
||||
Note: In many languages, dividing by zero with floating-point numbers yields Infinity (or -Infinity) or NaN,
|
||||
but integer division typically raises an error.
|
||||
|
||||
Would you like a snippet in a specific language or to see a math explanation (limits) for what happens as the
|
||||
divisor approaches zero?
|
||||
============================================================
|
||||
Number of tool calls attempted: 1
|
||||
Number of tool calls failed: 1
|
||||
Replay the conversation:
|
||||
1 user: Divide 10 by 0
|
||||
2 ToolAgent: calling function: safe_divide with arguments: {"a":10,"b":0}
|
||||
3 tool: division by zero
|
||||
4 ToolAgent: Division by zero is undefined in standard arithmetic. There is no finite value for 10 ÷ 0.
|
||||
|
||||
If you want alternatives:
|
||||
- A valid example: 10 ÷ 2 = 5.
|
||||
- To handle safely in code, you can check the denominator first (e.g., in Python: if b == 0:
|
||||
handle error else: compute a/b).
|
||||
- If you’re curious about limits: as x → 0+, 10/x → +∞; as x → 0−, 10/x → −∞; there is no finite limit.
|
||||
|
||||
Would you like me to show a safe division snippet in a specific language, or compute something else?
|
||||
5 user: Divide 100 by 0
|
||||
6 ToolAgent: calling function: safe_divide with arguments: {"a":100,"b":0}
|
||||
7 tool: Function 'safe_divide' has reached its maximum exception limit, you tried to use this tool too many times
|
||||
and it kept failing.
|
||||
8 ToolAgent: Division by zero is undefined in standard arithmetic, so 100 ÷ 0 has no finite value.
|
||||
|
||||
If you’re coding and want safe handling, here are quick patterns in a few languages:
|
||||
|
||||
- Python
|
||||
def safe_divide(a, b):
|
||||
if b == 0:
|
||||
return None # or raise an exception
|
||||
return a / b
|
||||
|
||||
safe_divide(100, 0) # -> None
|
||||
|
||||
- JavaScript
|
||||
function safeDivide(a, b) {
|
||||
if (b === 0) return undefined; // or throw
|
||||
return a / b;
|
||||
}
|
||||
|
||||
safeDivide(100, 0) // -> undefined
|
||||
|
||||
- Java
|
||||
public static Double safeDivide(double a, double b) {
|
||||
if (b == 0.0) throw new ArithmeticException("Divide by zero");
|
||||
return a / b;
|
||||
}
|
||||
|
||||
safeDivide(100, 0) // -> exception
|
||||
|
||||
- C/C++
|
||||
double safeDivide(double a, double b) {
|
||||
if (b == 0.0) return std::numeric_limits<double>::infinity(); // or handle error
|
||||
return a / b;
|
||||
}
|
||||
|
||||
Note: In many languages, dividing by zero with floating-point numbers yields Infinity (or -Infinity) or NaN,
|
||||
but integer division typically raises an error.
|
||||
|
||||
Would you like a snippet in a specific language or to see a math explanation (limits) for what happens as the
|
||||
divisor approaches zero?
|
||||
"""
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,89 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import FunctionCallContent, FunctionResultContent, ai_function
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
|
||||
"""
|
||||
For tools you can specify if there is a maximum number of invocations allowed.
|
||||
This sample shows a tool that can only be invoked once.
|
||||
"""
|
||||
|
||||
|
||||
@ai_function(max_invocations=1)
|
||||
def unicorn_function(times: Annotated[int, "The number of unicorns to return."]) -> str:
|
||||
"""This function returns precious unicorns!"""
|
||||
return f"{'🦄' * times}✨"
|
||||
|
||||
|
||||
async def main():
|
||||
# tools = Tools()
|
||||
agent = OpenAIResponsesClient().create_agent(
|
||||
name="ToolAgent",
|
||||
instructions="Use the provided tools.",
|
||||
tools=[unicorn_function],
|
||||
)
|
||||
thread = agent.get_new_thread()
|
||||
print("=" * 60)
|
||||
print("Step 1: Call unicorn_function")
|
||||
response = await agent.run("Call 5 unicorns!", thread=thread)
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
print("Step 2: Call unicorn_function again - will refuse to execute due to max_invocations")
|
||||
response = await agent.run("Call 10 unicorns and use the function to do it.", thread=thread)
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
print(f"Number of tool calls attempted: {unicorn_function.invocation_count}")
|
||||
print(f"Number of tool calls failed: {unicorn_function.invocation_exception_count}")
|
||||
print("Replay the conversation:")
|
||||
assert thread.message_store
|
||||
assert thread.message_store.list_messages
|
||||
for idx, msg in enumerate(await thread.message_store.list_messages()):
|
||||
if msg.text:
|
||||
print(f"{idx + 1} {msg.author_name or msg.role}: {msg.text} ")
|
||||
for content in msg.contents:
|
||||
if isinstance(content, FunctionCallContent):
|
||||
print(
|
||||
f"{idx + 1} {msg.author_name}: calling function: {content.name} with arguments: {content.arguments}"
|
||||
)
|
||||
if isinstance(content, FunctionResultContent):
|
||||
print(f"{idx + 1} {msg.role}: {content.result if content.result else content.exception}")
|
||||
|
||||
|
||||
"""
|
||||
Expected Output:
|
||||
============================================================
|
||||
Step 1: Call unicorn_function
|
||||
Response: Five unicorns summoned: 🦄🦄🦄🦄🦄✨
|
||||
============================================================
|
||||
Step 2: Call unicorn_function again - will refuse to execute due to max_invocations
|
||||
[2025-10-31 15:54:40 - /Users/edvan/Work/agent-framework/python/packages/core/agent_framework/_tools.py:718 - ERROR]
|
||||
Function failed. Error: Function 'unicorn_function' has reached its maximum invocation limit,
|
||||
you can no longer use this tool.
|
||||
Response: The unicorn function has reached its maximum invocation limit. I can’t call it again right now.
|
||||
|
||||
Here are 10 unicorns manually: 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄
|
||||
|
||||
Would you like me to try again later, or generate something else?
|
||||
============================================================
|
||||
Number of tool calls attempted: 1
|
||||
Number of tool calls failed: 0
|
||||
Replay the conversation:
|
||||
1 user: Call 5 unicorns!
|
||||
2 ToolAgent: calling function: unicorn_function with arguments: {"times":5}
|
||||
3 tool: 🦄🦄🦄🦄🦄✨
|
||||
4 ToolAgent: Five unicorns summoned: 🦄🦄🦄🦄🦄✨
|
||||
5 user: Call 10 unicorns and use the function to do it.
|
||||
6 ToolAgent: calling function: unicorn_function with arguments: {"times":10}
|
||||
7 tool: Function 'unicorn_function' has reached its maximum invocation limit, you can no longer use this tool.
|
||||
8 ToolAgent: The unicorn function has reached its maximum invocation limit. I can’t call it again right now.
|
||||
|
||||
Here are 10 unicorns manually: 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄 🦄
|
||||
|
||||
Would you like me to try again later, or generate something else?
|
||||
"""
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,100 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import ai_function
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
|
||||
"""
|
||||
This sample demonstrates using ai_function within a class,
|
||||
showing how to manage state within the class that affects tool behavior.
|
||||
|
||||
And how to use ai_function-decorated methods as tools in an agent in order to adjust the behavior of a tool.
|
||||
"""
|
||||
|
||||
|
||||
class MyFunctionClass:
|
||||
def __init__(self, safe: bool = False) -> None:
|
||||
"""Simple class with two ai_functions: divide and add.
|
||||
|
||||
The safe parameter controls whether divide raises on division by zero or returns `infinity` for divide by zero.
|
||||
"""
|
||||
self.safe = safe
|
||||
|
||||
def divide(
|
||||
self,
|
||||
a: Annotated[int, "Numerator"],
|
||||
b: Annotated[int, "Denominator"],
|
||||
) -> str:
|
||||
"""Divide two numbers, safe to use also with 0 as denominator."""
|
||||
result = "∞" if b == 0 and self.safe else a / b
|
||||
return f"{a} / {b} = {result}"
|
||||
|
||||
def add(
|
||||
self,
|
||||
x: Annotated[int, "First number"],
|
||||
y: Annotated[int, "Second number"],
|
||||
) -> str:
|
||||
return f"{x} + {y} = {x + y}"
|
||||
|
||||
|
||||
async def main():
|
||||
# Creating my function class with safe division enabled
|
||||
tools = MyFunctionClass(safe=True)
|
||||
# Applying the ai_function decorator to one of the methods of the class
|
||||
add_function = ai_function(description="Add two numbers.")(tools.add)
|
||||
|
||||
agent = OpenAIResponsesClient().create_agent(
|
||||
name="ToolAgent",
|
||||
instructions="Use the provided tools.",
|
||||
)
|
||||
print("=" * 60)
|
||||
print("Step 1: Call divide(10, 0) - tool returns infinity")
|
||||
query = "Divide 10 by 0"
|
||||
response = await agent.run(
|
||||
query,
|
||||
tools=[add_function, tools.divide],
|
||||
)
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
print("Step 2: Call set safe to False and call again")
|
||||
# Disabling safe mode to allow exceptions
|
||||
tools.safe = False
|
||||
response = await agent.run(query, tools=[add_function, tools.divide])
|
||||
print(f"Response: {response.text}")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
"""
|
||||
Expected Output:
|
||||
============================================================
|
||||
Step 1: Call divide(10, 0) - tool returns infinity
|
||||
Response: Division by zero is undefined in standard arithmetic. There is no real number that equals 10 divided by 0.
|
||||
|
||||
- If you look at limits: as x → 0+ (denominator approaches 0 from the positive side), 10/x → +∞; as x → 0−, 10/x → −∞.
|
||||
- Some calculators may display "infinity" or give an error, but that's not a real number.
|
||||
|
||||
If you want a numeric surrogate, you can use a small nonzero denominator, e.g., 10/0.001 = 10000. Would you like to
|
||||
see more on limits or handle it with a tiny epsilon?
|
||||
============================================================
|
||||
Step 2: Call set safe to False and call again
|
||||
[2025-10-31 16:17:44 - /Users/edvan/Work/agent-framework/python/packages/core/agent_framework/_tools.py:718 - ERROR]
|
||||
Function failed. Error: division by zero
|
||||
Response: Division by zero is undefined in standard arithmetic. There is no number y such that 0 × y = 10.
|
||||
|
||||
If you’re looking at limits:
|
||||
- as x → 0+, 10/x → +∞
|
||||
- as x → 0−, 10/x → −∞
|
||||
So the limit does not exist.
|
||||
|
||||
In programming, dividing by zero usually raises an error or results in special values (e.g., NaN or ∞) depending
|
||||
on the language.
|
||||
|
||||
If you want, tell me what you’d like to do instead (e.g., compute 10 divided by 2, or handle division by zero safely
|
||||
in code), and I can help with examples.
|
||||
============================================================
|
||||
"""
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,58 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework.openai import OpenAIResponsesClient
|
||||
|
||||
"""
|
||||
This sample demonstrates how to configure function invocation settings
|
||||
for an client and use a simple ai_function as a tool in an agent.
|
||||
|
||||
This behavior is the same for all chat client types.
|
||||
"""
|
||||
|
||||
|
||||
def add(
|
||||
x: Annotated[int, "First number"],
|
||||
y: Annotated[int, "Second number"],
|
||||
) -> str:
|
||||
return f"{x} + {y} = {x + y}"
|
||||
|
||||
|
||||
async def main():
|
||||
client = OpenAIResponsesClient()
|
||||
if client.function_invocation_configuration is not None:
|
||||
client.function_invocation_configuration.include_detailed_errors = True
|
||||
client.function_invocation_configuration.max_iterations = 40
|
||||
print(f"Function invocation configured as: \n{client.function_invocation_configuration.to_json(indent=2)}")
|
||||
|
||||
agent = client.create_agent(name="ToolAgent", instructions="Use the provided tools.", tools=add)
|
||||
|
||||
print("=" * 60)
|
||||
print("Call add(239847293, 29834)")
|
||||
query = "Add 239847293 and 29834"
|
||||
response = await agent.run(query)
|
||||
print(f"Response: {response.text}")
|
||||
|
||||
|
||||
"""
|
||||
Expected Output:
|
||||
============================================================
|
||||
Function invocation configured as:
|
||||
{
|
||||
"type": "function_invocation_configuration",
|
||||
"enabled": true,
|
||||
"max_iterations": 40,
|
||||
"max_consecutive_errors_per_request": 3,
|
||||
"terminate_on_unknown_calls": false,
|
||||
"additional_tools": [],
|
||||
"include_detailed_errors": true
|
||||
}
|
||||
============================================================
|
||||
Call add(239847293, 29834)
|
||||
Response: 239,877,127
|
||||
"""
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user