Agno Agents 可以并发执行多个工具,让您能够高效地处理模型进行的函数调用。当函数涉及耗时操作时,这一点尤其有价值。它能提高响应速度并缩短整体执行时间。
下面是一个示例:
import asyncio
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.log import logger
async def atask1(delay: int):
"""Simulate a task that takes a random amount of time to complete
Args:
delay (int): The amount of time to delay the task
"""
logger.info("Task 1 has started")
for _ in range(delay):
await asyncio.sleep(1)
logger.info("Task 1 has slept for 1s")
logger.info("Task 1 has completed")
return f"Task 1 completed in {delay:.2f}s"
async def atask2(delay: int):
"""Simulate a task that takes a random amount of time to complete
Args:
delay (int): The amount of time to delay the task
"""
logger.info("Task 2 has started")
for _ in range(delay):
await asyncio.sleep(1)
logger.info("Task 2 has slept for 1s")
logger.info("Task 2 has completed")
return f"Task 2 completed in {delay:.2f}s"
async def atask3(delay: int):
"""Simulate a task that takes a random amount of time to complete
Args:
delay (int): The amount of time to delay the task
"""
logger.info("Task 3 has started")
for _ in range(delay):
await asyncio.sleep(1)
logger.info("Task 3 has slept for 1s")
logger.info("Task 3 has completed")
return f"Task 3 completed in {delay:.2f}s"
async_agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[atask2, atask1, atask3],
show_tool_calls=True,
markdown=True,
)
asyncio.run(
async_agent.aprint_response("Please run all tasks with a delay of 3s", stream=True)
)
运行 Agent:
pip install -U agno openai
export OPENAI_API_KEY=***
python async_tools.py
如何使用:
- 为您的 Agent 提供一个工具列表,最好是异步工具以获得最佳性能。不过,同步函数也可以使用,因为它们将在单独的线程上并发执行。
- 使用
arun
或 aprint_response
方法运行 Agent,以启用工具调用的并发执行。
工具的并发执行需要模型支持并行函数调用。例如,OpenAI 模型有一个
parallel_tool_calls
参数(默认启用),允许同时请求和执行多个工具调用。
在此示例中,gpt-4o
调用 atask1
、atask2
和 atask3
三个工具。通常情况下,这些工具调用会顺序执行,但通过使用 aprint_response
函数,它们会并发运行,从而缩短执行时间。

Responses are generated using AI and may contain mistakes.