用Agno开发物流AI代理
在这个以实现为重点的教程中,我们将引导你使用 Agno 框架构建一个用于物流的AI代理。Agno,以前称为 Phidata,是一个开源框架,旨在创建具有记忆、知识保留、工具集成和复杂推理能力的高级AI代理。它使开发人员能够无缝地集成任何大型语言模型(LLM),管理代理的状态和记忆,并且甚至可以协调多个协作工作的代理。
Agno 的真正力量在于它能够构建高性能的领域特定AI代理,并提供生产中的监控和优化工具。在这个教程中,我们将利用Agno的强大功能逐步解决物流领域的玩具问题。到本教程结束时,你将对如何开始使用Agno并构建适合您领域的实用AI解决方案有一个清晰的理解。
智能交付助手 — 在此物流AI代理处理两个关键任务(意图):路线优化用于交付和货物跟踪,我将进一步扩展它以查找附近的地点供交付司机使用。该代理将理解用户何时想要找到最佳的交付路线,何时需要跟踪包裹,并将根据适当的信息进行响应。例如,用户可能会问:“从我们的仓库向A、B和C交付包裹的最佳路线是什么?”或者“运输单号TRK12345现在在哪里?”我们的代理将使用Agno来解释这些请求,调用正确的工具或数据,并返回有帮助的答案。
让我们现在使用Python逐步实现。下面的每个部分都建立在前一部分的基础上,从设置环境到部署完成的代理的指导。
1、设置Agno
在编写代理之前,请确保使用Agno和任何必要的依赖项设置您的开发环境:
安装Agno和LLM提供程序 — Agno可以与任何LLM一起工作。在此指南中,我将通过openai
API使用OpenAI的GPT模型。使用pip安装Agno库和OpenAI SDK:
pip install agno openai
Agno轻量级且安装快速。如果您还没有,请获取OpenAI API密钥并将其设置为环境变量,以便代理可以使用它:
export OPENAI_API_KEY=<your_openai_api_key>
(API密钥是必需的,因为我们将通过Agno使用OpenAI的GPT-4模型。或者,您可以根据需要配置Agno与其他模型提供程序。)
在Python中导入Agno — 在安装了包之后,现在可以在Python脚本或笔记本中导入Agno类。我将主要使用Agent
类来创建我们的代理,并使用OpenAIChat
模型包装器来处理GPT。例如:
from agno.agent import Agent
from agno.models.openai import OpenAIChat # (我们将在后续导入工具类或创建自定义类)
这确保了我们的代码可以访问Agno的代理框架和OpenAI模型集成。
如果需要,Agno的文档提供了更多关于安装和设置的详细信息(你的第一个代理 — Agno)。一旦环境准备就绪,我们可以继续构建物流代理。
2、创建AI代理
现在,让我们使用Agno创建一个基本的AI代理。至少,一个Agno代理需要一个语言模型来进行思考和回应。我们将通过OpenAIChat
配置代理使用GPT-4(或GPT-3.5)(请参阅您可以使用任何开源或封闭源模型)。我们还将给代理一个简短的角色描述。
代码 — 初始化一个基本代理:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
# 使用LLM模型和基本角色描述初始化代理
agent = Agent(
model=OpenAIChat(id="gpt-4o"), # GPT-4(您可以根据需要使用"gpt-3.5-turbo"或其他模型)
description="你是一名有用的物流助理。",
markdown=True # 启用响应中的Markdown格式化以提高可读性
)
在这段代码中,我使用OpenAI模型和简单的描述创建了一个Agent
。描述给代理提供了有关其角色或身份的背景信息(在这种情况下,一名有用的物流助理)。设置markdown=True
意味着代理的回答可以包含Markdown格式化,这对于更好的可读性非常有用(例如,在回答中使用项目符号列表或表格)。
在这个阶段,代理没有任何特殊技能——它基本上只是一个带有少量上下文的语言模型。接下来,我们将定义代理应处理的意图(能力)并通过工具和逻辑实现它们。
3、定义意图和响应
对于物流助理,我们确定了两个主要的意图:
- 路线优化意图:当用户要求优化交付路线时。
- 货物跟踪意图:当用户查询某个货物的状态或位置时。
在这种背景下定义意图意味着决定代理应该如何识别和处理每种类型的请求。我们将通过指令引导代理,并为其配备工具(函数)以履行每个意图:
- 代理指令 — 我们为代理提供明确的指令,作为特定领域的指南。这些指令将告诉代理在什么情况下使用哪个工具以及如何形成其响应。例如,如果查询包含跟踪号码模式,则代理应该使用跟踪工具;如果查询提到路线规划,则使用路线工具。我们还指示代理给出清晰简洁的答案。
- 响应格式 — 我们决定代理应该如何响应。对于跟踪查询,响应可能是一个简单的状态消息,而对于路线优化,响应可能是列出停靠顺序和总距离或时间。我们将确保代理的答案包括必要的细节(如果合适的话,还可以包括解释)。
我们可以将这些决策编码为代理配置的一部分。Agno允许我们在创建代理时传递instructions
参数(作为字符串或字符串列表)。让我们更新代理初始化以包含针对我们两个意图的指令:
代码 — 使用意图定义代理的行为:
# 使用特定指令更新代理创建并为工具留出占位符
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
description="你是一名知识丰富的物流助理。",
instructions=[
"如果用户询问货物状态或提供跟踪ID,请使用TrackingTool检索货物状态。",
"如果用户询问优化交付路线,请使用RouteTool计算最佳路线。",
"提供清晰简洁的响应。对于路线,列出停靠顺序和总距离。对于跟踪,提供当前状态和任何相关细节。"
],
tools=[], # 工具将在下一节中添加
show_tool_calls=True, # 在开发过程中显示代理调用工具的时间(调试很有用)
markdown=True
)
在这段更新后的代码中,我们向代理传递了一组指令字符串。这些指令就像代理的“知识库”,告诉它如何处理我们定义的意图。例如,我们明确告诉代理在收到跟踪查询时使用TrackingTool
,在收到与路线相关的查询时使用RouteTool
。我们暂时将tools
列表留空——我们将在下一节中实现实际工具后填充它。我们还设置了show_tool_calls=True
,以便在测试期间查看代理的决策过程(它正在调用哪个工具);这是开发复杂代理的最佳实践,以确保它们遵循预期流程。
定义了意图和响应指南后,下一步是实现工具,将我们的代理与真实的物流数据和功能集成。
4、将代理与物流数据集成
为了使我们的代理能够完成物流任务,我们需要将其与相关数据和功能集成。在 Agno 中,这可以通过创建工具(代理可以在需要时调用的外部函数或 API)来实现。我们将为我们的用例实现两个自定义工具:
ShipmentTrackingTool
— 连接到货运跟踪数据(这里我们将使用一个简单的字典来模拟数据库或 API)。RouteOptimizationTool
— 执行路线计算(这里我们将实现一个简单的路线优化逻辑以进行演示)。
a. 准备示例数据:对于此示例,我们将使用内存数据结构来模拟外部数据源:
Python 字典 tracking_data 将跟踪 ID 映射到其当前状态或位置。
嵌套字典 distance_matrix 将表示不同位置之间的距离(用于路线优化)。这可以替代实际系统中对映射 API 或距离计算器的调用。
# Sample logistics data sources
tracking_data = {
"TRK12345": "In transit at Toronto distribution center",
"TRK98765": "Delivered on 2025-03-09 10:24",
"TRK55555": "Out for delivery - last scanned at Vancouver hub"
}
distance_matrix = {
"Warehouse": {"A": 10, "B": 15, "C": 20},
"A": {"Warehouse": 10, "B": 12, "C": 5},
"B": {"Warehouse": 15, "A": 12, "C": 8},
"C": {"Warehouse": 20, "A": 5, "B": 8}
}
在生产场景中,tracking_data
可以替换为数据库查询或向承运人发出的 API 调用,而 distance_matrix
可以来自像 Google Maps 这样的服务或内部路线规划系统。在我们的教程中,这些硬编码值将让我们测试代理的功能。
b. 实现自定义工具:我们将创建两个工具类,供 Agno 的代理使用。每个工具都有一个 name
(代理用来引用它)和执行其功能的方法。Agno 内置的工具(如 DuckDuckGo 搜索工具)遵循类似的模式。我们将为每个工具实现一个简单的 run()
方法,该方法处理代理的查询并返回结果字符串。
import re
class TrackingTool:
"""通过跟踪 ID 获取货运状态的工具。"""
def __init__(self):
self.name = "TrackingTool"
self.description = "根据跟踪 ID 提供货运状态更新。"
def run(self, query: str) -> str:
# 使用正则表达式从查询中提取跟踪 ID(假设 ID 格式为 TRK12345)
match = re.search(r"\bTRK\d+\b", query.upper())
if not match:
return "我在查询中找不到跟踪号码。"
tracking_id = match.group(0)
# 在我们的数据中查找跟踪 ID
status = tracking_data.get(tracking_id)
if status:
return f"**{tracking_id}**的状态:{status}"
else:
return f"对不起,我没有关于跟踪 ID {tracking_id} 的信息。"
class RouteTool:
"""计算给定起点和多个目的地的最佳路线的工具。"""
def __init__(self):
self.name = "RouteTool"
self.description = "根据起点和目的地计算最佳配送路线。"
def run(self, query: str) -> str:
# 预期查询格式为 "from X to Y, Z, ..."(不区分大小写)
pattern = re.compile(r"from\s+([\w\s]+)\s+to\s+(.+)", re.IGNORECASE)
match = pattern.search(query)
if not match:
return ("请按照 'from <Origin> to <Dest1>, <Dest2>, ...' 的格式指定路线。")
origin = match.group(1).strip()
dests_part = match.group(2)
# 通过逗号或“and”分割多个目的地
destinations = [d.strip() for d in re.split(r",| and ", dests_part) if d.strip()]
# 验证位置是否存在于我们的距离矩阵中
if origin not in distance_matrix:
return f"未知的起始位置:{origin}。"
for loc in destinations:
if loc not in distance_matrix:
return f"未知的目的地:{loc}。"
# 计算访问所有目的地的最佳路线(演示用途采用简单穷举法)
best_route = None
best_distance = float('inf')
from itertools import permutations
for perm in permutations(destinations):
total = 0
current = origin
for nxt in perm:
total += distance_matrix[current][nxt]
current = nxt
# 在此场景中不返回到起始点(单程路线结束于最后一个目的地)
if total < best_distance:
best_distance = total
best_route = perm
route_list = " -> ".join([origin] + list(best_route)) if best_route else origin
return f"最优路线:**{route_list}** (总距离:{best_distance} 公里)"
让我们分解一下这些工具的作用:
- TrackingTool:在
run()
中,我们使用正则表达式查找查询文本中的类似 "TRK12345" 的模式。这是我们假设的跟踪 ID 格式。如果找到,则我们在tracking_data
中查找该 ID。如果存在,工具会返回带有从数据中查找到的状态的消息;否则,会返回一条表示 ID 未知的信息。这模拟了向跟踪系统发出 API 调用的过程。 - RouteTool:在
run()
中,我们解析查询以获取起点和目的地列表。我们假设用户会说类似于“from Warehouse to A, B and C”的话(使用逗号或“and”来分隔多个站点)。我们拆分目的地,并验证每个位置是否存在。然后,我们通过尝试所有目的地列表的排列组合(适合小数量站点的穷举法)来计算最佳路线。我们使用distance_matrix
计算每种排列组合的总距离,并找到最短的一个。结果将以有序路线(例如,Warehouse -> A -> C -> B)的形式返回,并显示总距离。在真实场景中,这个工具可能会调用优化 API 或使用更高效的算法,但我们的实现说明了这一概念。
现在我们已经有了工具,需要将它们与代理集成。
c. 将工具附加到代理:我们现在可以将这些工具的实例传递给代理的 tools
参数。这将允许代理在用户查询时调用 TrackingTool.run()
或 RouteTool.run()
(根据我们给出的指示):
# 实例化工具
tracking_tool = TrackingTool()
route_tool = RouteTool()
# 将工具附加到代理
agent.tools = [tracking_tool, route_tool]
通过将我们的工具实例添加到 agent.tools
,我们已经赋予了代理使用这些工具的能力。在幕后,当我们稍后调用 agent.run(...)
或 agent.print_response(...)
时,Agno 会让语言模型推理出查询,并决定是否需要调用工具。由于我们的指示和工具描述,模型将知道(例如,如果在查询中看到 "TRK"),调用 TrackingTool
。Agno 管理顺序:代理将调用工具的 run
方法,并将其结果纳入最终答案。
至此,我们的物流 AI 代理已完全构建完成——它有一个模型(大脑)、处理意图的知识(指令)以及实现这些意图的方法(带有数据的工具)。接下来,我们将运行代理并进行一些测试查询,以确保一切按预期工作。
5、运行和测试代理
设置好代理后,让我们模拟对话,看看它的表现如何。我们将测试两种意图:货运跟踪查询和路线优化查询。我们将使用代理的 print_response
方法(方便地流式输出和打印答案)进行演示,但你也可以使用 agent.run()
来获取响应对象。
测试货运跟踪:
# 示例 1:跟踪查询
user_query = "TRK12345 目前在哪里?"
agent.print_response(user_query)
预期行为: 代理应该在查询中检测到跟踪 ID“TRK12345”。根据我们的指示,它将使用 TrackingTool
。如果有 show_tool_calls=True
,你会在控制台看到一条日志记录,表明调用了工具。代理随后应输出类似以下内容:
**TRK12345**的状态:正在多伦多分销中心运输
响应包括跟踪 ID 和从数据中查找到的状态。答案清晰地呈现给用户。
测试路线优化:
# 示例 2:路线优化查询
user_query = "从仓库到 A、B 和 C 的最佳路线是什么?"
agent.print_response(user_query)
预期行为: 代理识别出一个路线规划请求(查询包含“from … to …”)。它将调用 RouteTool
来计算最佳路径。工具内部将计算距离并返回最佳序列。代理的最终答案可能是:
最优路线:**Warehouse -> A -> C -> B** (总距离:23 公里)
这表示从仓库出发访问所有目的地(A、B、C)的最佳路线,并显示总行驶距离。代理通过利用我们的自定义路线逻辑得出了这个答案——这是一个强大的例子,展示了我们如何扩展 AI 代理以具备特定领域的功能。
你可以测试各种查询变体:
- “Track TRK98765” 应返回该 ID 的交付状态。
- “Plan a route from A to C and B starting at Warehouse”(不同的措辞)仍应被理解为路线优化。
- “Where is TRK00000?”(一个未知的 ID)应触发代理回应没有关于该跟踪 ID 的信息。
每次测试都能确保我们的意图被正确识别,并且工具按预期工作。如果代理的输出不符合预期,你可以改进指令或工具中的解析逻辑。
6、部署和扩展的最佳实践
构建代理只是第一步。当你准备将这个物流 AI 代理部署到实际场景中(例如作为物流仪表板的一部分或操作人员的聊天机器人)时,请考虑以下最佳实践:
- 使用专用 API 或服务:将你的代理包装在一个 Web 服务中,以便轻松集成。例如,你可以使用 FastAPI 应用程序暴露一个端点,该端点将用户查询转发到
agent.run()
(本地 Docker 指南 - Agno)。这样,你的物流代理可以通过 REST 调用(来自 Web 应用、Slack 机器人等)被访问。 - 容器化:将你的应用程序和 Agno 代理打包成 Docker 容器,以确保一致的部署。Agno 设计用于在任何地方运行(你的云或本地)。容器化确保开发、测试和生产环境中的依赖项(包括 Agno 和任何数据库)保持一致。
- 状态和内存管理:如果你的代理将处理长时间运行的会话或大量查询,请考虑启用 Agno 的内置记忆功能。尝试并使用状态管理。例如,连接数据库(如与向量存储结合的PostgreSQL)可以让代理记住过去的交互或存储频繁访问的路线信息。Agno 支持存储代理状态并使用向量数据库进行知识管理。
- 监控和日志记录: 使用 Agno 的监控工具来跟踪代理的表现。记录每个查询和响应,包括任何工具调用和错误。这有助于调试问题(如代理选择了错误的工具)和衡量使用情况。Agno 甚至提供了代理用户界面,用于开发期间的聊天和评估。
- 扩展部署: 对于高吞吐量场景(例如,许多并发用户询问物流查询),可以水平扩展服务。由于 Agno 代理实例化是轻量级的(声称极其快速和高效),因此实例化多个实例或工作者是可行的。确保您的设计是无状态的,或者通过数据库使用共享内存,以便任何实例都可以处理任何请求。
- 工具和模型优化: 随着代理的增长,您可能会添加更多工具(例如,实时路线更新的交通 API 工具,或仓库库存工具)。保持代理的范围集中——Agno 建议如果工具集变得非常庞大,使用一组专门的代理。这可以防止一个代理变得过于笨重。此外,选择正确的模型——GPT-4 强大但昂贵;对于简单的查询,在生产中可能只需较小的模型或开源替代品就足够了。
- 安全性和验证: 如果代理将执行操作(如调度交付或更新数据库),请确保验证来自 LLM 的输出。对工具输入应用检查(我们的工具中的正则表达式解析是一个基本验证的例子)。始终清理任何外部数据,并注意工具函数中的错误处理,以便代理在数据源无法访问时能够优雅地失败。
遵循这些实践,您可以确保您的物流 AI 代理或任何您构建的代理从原型到生产都是健壮、高效且可维护的。
7、物流 AI 代理的完整实现
在这篇博客中,我们介绍了 Agno 并演示了如何逐步使用它构建一个专注于物流的 AI 代理。我们选择了一个结合路线优化和货物跟踪的用例——两个常见的物流挑战——并展示了 AI 代理如何通过自然语言交互处理两者。在此过程中,我们设置了 Agno,定义了代理的意图和指令,通过自定义工具集成了示例物流数据,并测试了代理的功能。我们还涵盖了在真实环境中部署和扩展代理的最佳实践。
总结一下,这里是我们在构建的物流 AI 代理的 完整代码。这将所有讨论过的组件汇集在一起:
import re
from itertools import permutations
from agno.agent import Agent
from agno.models.openai import OpenAIChat
# 货运和距离的样本数据
tracking_data = {
"TRK12345": "正在多伦多配送中心运输",
"TRK98765": "已于 2025-03-09 10:24 交付",
"TRK55555": "已出发 - 最后一次扫描在温哥华枢纽"
}
distance_matrix = {
"仓库": {"A": 10, "B": 15, "C": 20},
"A": {"仓库": 10, "B": 12, "C": 5},
"B": {"仓库": 15, "A": 12, "C": 8},
"C": {"仓库": 20, "A": 5, "B": 8}
}
# 定义自定义工具
class TrackingTool:
def __init__(self):
self.name = "TrackingTool"
self.description = "提供给定跟踪 ID 的货运状态更新。"
def run(self, query: str) -> str:
match = re.search(r"\bTRK\d+\b", query.upper())
if not match:
return "请提供有效的跟踪 ID。"
tid = match.group(0)
status = tracking_data.get(tid)
return f"{tid} 的状态:{status}" if status else f"没有 {tid} 的信息。"
class RouteTool:
def __init__(self):
self.name = "RouteTool"
self.description = "根据起点和目的地计算最佳送货路线。"
def run(self, query: str) -> str:
m = re.search(r"from\s+([\w\s]+)\s+to\s+(.+)", query, re.IGNORECASE)
if not m:
return "指定路线为 'from <Origin> to <Dest1>, <Dest2>, ...'。"
origin = m.group(1).strip()
dests = [d.strip() for d in re.split(r",| and ", m.group(2)) if d.strip()]
if origin not in distance_matrix:
return f"未知起点:{origin}。"
for loc in dests:
if loc not in distance_matrix:
return f"未知目的地:{loc}。"
best_distance = float('inf')
best_order = None
for perm in permutations(dests):
total = 0
cur = origin
for nxt in perm:
total += distance_matrix[cur][nxt]
cur = nxt
if total < best_distance:
best_distance = total
best_order = perm
route_plan = " -> ".join([origin] + list(best_order)) if best_order else origin
return f"最优路线:{route_plan} (总距离:{best_distance} 公里)"
# 使用模型、指令和工具创建代理
agent = Agent(
model=OpenAIChat(id="gpt-4o"),
description="你是一位知识丰富的物流助理。",
instructions=[
"如果用户询问有关货运或跟踪 ID 的问题,请使用 TrackingTool。",
"如果用户询问关于路线优化或最佳路线的问题,请使用 RouteTool。",
"提供简洁明了的答案,包括工具的相关详细信息。"
],
tools=[TrackingTool(), RouteTool()],
show_tool_calls=False, # 设置为 False 以获得干净的输出(设置为 True 用于调试)
markdown=True
)
# 示例用法
print(agent.run("TRK12345 在哪里?"))
print(agent.run("从仓库到 A、B 和 C 找最佳路线"))
输出如下:
通过此实现,您拥有一个由 Agno 提供支持的物流 AI 代理。运行上述代码(在环境变量中插入有效的 OpenAI API 密钥后)将生成样本查询的响应——满足我们设定的两个意图。
这个示例可以进一步扩展和定制:您可以集成实时 API(用于实时跟踪数据或地图服务),添加更多意图(例如,通过检查设备日志进行预测性维护),甚至创建一个多代理系统,其中一位代理处理路由,另一位代理处理库存查询,根据需要协作。Agno 的灵活性和性能使其成为在物流领域实验和扩展此类解决方案的可行选择。我所做的只是一个简单的扩展。我使用的是 Oxylabs(前 7 天 5000 次免费请求)API 来访问 Google 地图,通过创建 FuelStationSearchTool 来获取路线上的加油站点。
import os
import requests
from agno.agent import Agent
from agno.models.azure import AzureOpenAI
# 定义自定义 Oxylabs 工具
class FuelStationSearchTool:
def __init__(self):
self.name = "FuelStationSearchTool"
self.description = "使用 Oxylabs 实时 API 获取附近的加油站。"
def run(self, location_query, pages=1):
payload = {
'source': 'google_maps',
'domain': 'com',
'query': f'fuel stations in {location_query}',
'pages': pages,
}
response = requests.post(
'https://realtime.oxylabs.io/v1/queries',
auth=(<USERNAME>, <PASSWORD>),
json=payload,
)
if response.status_code != 200:
return f"API 错误:{response.status_code},{response.text}"
data = response.json()
stations_info = []
for result in data.get('results', [])[:5]:
title = result.get('title', 'No Name')
address = result.get('address', 'Address Unavailable')
rating = result.get('rating', 'Rating Unavailable')
stations_info.append(f"{title} - {address} (评分:{rating})")
if not stations_info:
return "附近没有找到加油站。"
return "\n".join(stations_info)
# 初始化 Agno 代理
agent = Agent(
model=AzureOpenAI(id="gpt-4o"),
description="帮助司机找到附近服务的智能物流助手。",
instructions=[
"对于有关附近加油站的问题,请使用 FuelStationSearchTool。",
"清楚列出加油站,包括名称、地址和评分。"
],
markdown=True,
tools=[FuelStationSearchTool()]
)
# 主执行
if __name__ == "__main__":
location_query = "NLS, Missisauga, ON"
agent.print_response(location_query)
注:NLS 是安大略省密西沙加的一个仓库。
这里我们从代理获取所有附近的加油站。
8、结束语
在本指南中,我们展示了开发者如何使用 Agno 构建解决物流挑战的 AI 代理。通过遵循清晰的分步方法——从设置框架、定义代理的范围(意图)、实施集成点到测试和部署——你可以创建能够自动化和简化复杂工作流程的智能代理。
物流领域只是其中一个例子;同样的模式可以应用于金融、客户服务等领域。使用 Agno,创建强大的 AI 代理的门槛比以前更低。我们希望本教程能激发你构建自己的代理来解决你热衷的问题。
原文链接:Building an AI Agent with Agno: A Step-by-Step Guide
汇智网翻译整理,转载请标明出处