引言 — 在本篇以实现为重点的教程中，将引导你使用Agno框架构建一个用于物流领域的AI代理。Agno（前身为Phidata）是一个开源框架，旨在创建具有记忆、知识保留、工具集成和复杂推理等能力的高级AI代理。它使开发人员能够无缝集成任何大型语言模型（LLM），管理代理的状态和记忆，甚至协调多个代理协同工作。

Agno的真正强大之处在于其能够构建高性能、特定领域的AI代理，并提供工具来监控和优化生产环境中的这些代理。在本教程中，我们将利用Agno的强大功能，逐步解决物流领域的一个简单问题。到本教程结束时，你将清楚了解如何开始使用Agno，并构建针对你特定领域的实用AI解决方案。

用例：智能配送助手 — 在这个场景中，物流AI代理处理两项关键任务（意图）：配送路线优化和货物跟踪，接下来我将扩展其功能，为配送司机查找附近地点。该代理将理解用户何时想要找到最佳配送路线，以及用户何时需要跟踪包裹，并会提供相应的信息。例如，用户可能会问“从我们的仓库到A、B和C的最佳配送路线是什么？”或“TRK12345货物现在在哪里？”。我们的代理将使用Agno来解释这些请求，调用合适的工具或数据，并返回有用的答案。

现在，让我们使用Python逐步深入实现过程。下面的每个部分都建立在前一个部分的基础上，从设置环境到指导如何部署完成的代理。

设置Agno

在编写代理代码之前，请确保你的开发环境已设置好Agno和任何必要的依赖项：

1. 安装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以使用其他模型提供程序。）

2. 在Python中导入Agno — 安装好包后，你现在可以在Python脚本或笔记本中导入Agno的类。我将主要使用Agent类来创建我们的代理，并使用GPT的OpenAIChat模型包装器。例如：

from agno.agent import Agent 
from agno.models.openai import OpenAIChat # (We will import tool classes or create custom ones in later steps)

这确保了我们的代码可以访问Agno的代理框架和OpenAI模型集成。

创建AI代理

现在，让我们使用Agno创建一个基本的AI代理。至少，一个Agno代理需要一个语言模型来思考和响应。我们将通过OpenAIChat为代理配置GPT-4（或GPT-3.5）（请注意，你可以使用任何开源模型或任何提供商的闭源模型）。我们还将为代理简要描述其角色。

代码 — 初始化一个基本代理：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
# Initialize the agent with an LLM model and a basic role description
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),  # GPT-4 (you can use "gpt-3.5-turbo" or other models as needed)
    description="You are a helpful logistics assistant.",
    markdown=True  # Enable markdown formatting in responses for readability
)

在这段代码中，我创建了一个使用OpenAI模型的代理，并为其添加了一个简单的描述。该描述为代理提供了关于其角色或身份（在本例中为有用的物流助手）的上下文。设置markdown=True意味着代理的答案可以包含Markdown格式，这有助于提高可读性（例如，在回答中使用项目符号列表或表格）。

在这个阶段，代理还没有任何特殊技能——它基本上只是一个带有一点上下文的语言模型。接下来，我们将定义代理应处理的意图（功能），并通过工具和逻辑来实现它们。

定义意图和响应

对于物流助手，我们确定了两个主要意图：

• 路线优化意图：当用户请求优化的配送路线时。
• 货物跟踪意图：当用户通过跟踪ID询问货物的状态或位置时。

在此上下文中定义意图意味着决定代理应如何识别和处理每种类型的请求。我们将使用指令来引导代理，并为其配备工具（函数）以实现每个意图：

1. 意图的代理指令 — 我们为代理提供明确的指令，作为特定领域的指南。这些指令将告诉代理何时使用哪个工具以及如何形成其响应。例如，如果查询包含跟踪号模式，代理应使用跟踪工具；如果查询提到路线规划，则使用路线工具。我们还指示代理给出清晰、简洁的答案。
2. 响应格式 — 我们决定代理应如何响应。对于跟踪查询，响应可能是一个简单的状态消息，而对于路线优化，响应可能列出停靠点的顺序以及总距离或时间。我们将确保代理的答案包含必要的细节（如果合适，还可能包含解释）。

我们可以将这些决策作为代理配置的一部分进行编码。Agno允许我们在创建代理时传递一个instructions参数（作为字符串或字符串列表）。让我们更新代理初始化，添加针对我们两个意图的指令：

代码 — 定义包含意图的代理行为：

# Update agent creation with specific instructions and placeholders for tools
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    description="You are a knowledgeable logistics assistant.",
    instructions=[
        "If the user asks about a shipment status or provides a tracking ID, use the TrackingTool to retrieve the shipment status.",
        "If the user asks about optimizing a delivery route, use the RouteTool to compute the best route.",
        "Provide clear and concise responses. For routes, list the stop order and total distance. For tracking, give the current status and any relevant details."
    ],
    tools=[],  # tools will be added in the next step
    show_tool_calls=True,  # for development, show when the agent calls a tool (useful for debugging)
    markdown=True
)

在这个更新的代码片段中，我们向代理传递了一组指令字符串，这些字符串就像代理的“知识库”，指导它如何处理我们定义的意图。例如，我们明确告诉代理，在收到跟踪查询时使用TrackingTool，在处理与路线相关的查询时使用RouteTool。目前，我们将工具列表留空——在实现实际工具后，我们将在下一节中填充它。我们还设置了show_tool_calls=True，以便在测试期间查看代理的决策过程（即它调用了哪个工具）；这是在开发复杂代理时的最佳实践，以确保它们遵循预期的流程。

定义了意图和响应指南后，下一步是实现工具，这些工具将使我们的代理能够集成实际的物流数据和功能。

将代理与物流数据集成

为了使我们的代理能够完成物流任务，我们需要将其与相关数据和功能集成。在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的代理使用。每个工具将有一个名称（代理用它来引用）和一个执行其功能的方法。Agno的内置工具（如DuckDuckGo搜索工具）遵循类似的模式。我们将为每个工具实现一个简单的run()方法，该方法处理代理的查询并返回结果字符串。

import re
class TrackingTool:
    """Tool to fetch shipment status by tracking ID."""
    def __init__(self):
        self.name = "TrackingTool"
        self.description = "Provides shipment status updates given a tracking ID."
    
    def run(self, query: str) -> str:
        # Extract tracking ID from the query (assume IDs follow format like TRK12345)
        match = re.search(r"\bTRK\d+\b", query.upper())
        if not match:
            return "I couldn't find a tracking number in the query."
        tracking_id = match.group(0)
        # Lookup the tracking ID in our data
        status = tracking_data.get(tracking_id)
        if status:
            return f"Status for **{tracking_id}**: {status}"
        else:
            return f"Sorry, I have no information on tracking ID {tracking_id}."

class RouteTool:
    """Tool to calculate optimal route given an origin and multiple destinations."""
    
    def __init__(self):
        self.name = "RouteTool"
        self.description = "Computes the best delivery route given a start and destinations."
    
    def run(self, query: str) -> str:
        # Expect queries in form "from X to Y, Z, ..." (case-insensitive)
        pattern = re.compile(r"from\s+([\w\s]+)\s+to\s+(.+)", re.IGNORECASE)
        match = pattern.search(query)
        if not match:
            return ("Please specify a route in the format 'from <Origin> to <Dest1>, <Dest2>, ...'.")
        origin = match.group(1).strip()
        dests_part = match.group(2)
        # Split multiple destinations by commas or 'and'
        destinations = [d.strip() for d in re.split(r",| and ", dests_part) if d.strip()]
        # Validate locations exist in our distance matrix
        if origin not in distance_matrix:
            return f"Unknown origin location: {origin}."
        for loc in destinations:
            if loc not in distance_matrix:
                return f"Unknown destination: {loc}."
        # Compute the shortest route visiting all destinations (simple brute-force for demo)
        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
            # No return to origin in this scenario (one-way route finishing at last destination)
            if total < best_distance:
                best_distance = total
                best_route = perm
        route_list = " -> ".join([origin] + list(best_route)) if best_route else origin
        return f"Optimal route: **{route_list}** (Total distance: {best_distance} km)"

让我们来详细解释一下这些工具的功能：

• TrackingTool：在run()方法中，我们使用正则表达式在查询文本中查找类似“TRK12345”的模式。这是我们假设的跟踪ID格式。如果找到该ID，我们会在tracking_data中查找该ID。如果该ID的货物状态可用，工具会返回一条包含状态的消息；如果ID未知，则返回一条致歉消息。这模拟了对跟踪系统的API调用。
• RouteTool：在run()方法中，我们解析查询以获取起点和一系列目的地。我们假设用户会说类似“从仓库到A、B和C”（使用逗号或“和”来分隔多个站点）的内容。我们拆分目的地，然后验证每个位置是否已知。接着，我们通过尝试目的地列表的所有排列组合来计算最优路线（这是一种仅适用于少量站点的暴力求解方法）。我们使用distance_matrix计算每种排列组合的总距离，并找到最短的路线。结果以有序路线（例如，仓库 -> A -> C -> B）和总距离的形式返回。在实际场景中，这个工具可能会调用优化API或使用更高效的算法，但我们的实现说明了这一概念。

现在我们已经有了这些工具，接下来需要将它们集成到代理中。

c. 将工具附加到代理：我们现在可以将这些工具的实例传递给代理的tools参数。这样，代理就可以在决定用户查询需要时（根据我们给它的指令），调用TrackingTool.run()或RouteTool.run()。

# Instantiate the tools
tracking_tool = TrackingTool()
route_tool = RouteTool()

# Attach tools to the agent
agent.tools = [tracking_tool, route_tool]

通过将工具实例添加到agent.tools中，我们为代理配备了使用这些工具的能力。在底层实现中，当我们稍后调用agent.run(...)或agent.print_response(...)时，Agno将允许语言模型对查询进行推理，并决定是否应该使用工具。由于我们提供了指令和工具描述，模型将知道（例如，如果它在查询中看到“TRK”），则调用TrackingTool。Agno管理这个序列：代理将调用工具的run方法，并将结果纳入其最终答案中。

此时，我们的物流AI代理已完全构建完成——它拥有一个模型（大脑）、处理意图的知识（指令），以及执行这些意图的手段（带有数据的工具）。接下来，我们将用一些测试查询来运行代理，以确保一切按预期工作。

运行和测试代理

代理设置完成后，让我们模拟一次对话，看看它的表现如何。我们将测试两个意图：货物跟踪查询和路线优化查询。我们将使用代理的print_response方法（该方法方便地流式传输并打印答案）进行演示，但你也可以使用agent.run()来获取响应对象。

测试货物跟踪：

# Example 1: Tracking query
user_query = "Where is shipment TRK12345 right now?"
agent.print_response(user_query)

预期行为：代理应能在查询中检测到跟踪ID“TRK12345”。根据我们的指令，它将使用TrackingTool。如果设置show_tool_calls=True，你会在控制台中看到一条日志行，指示工具已被调用。然后，代理应输出类似以下内容：

**Status for TRK12345**: In transit at Toronto distribution center

该响应包括跟踪ID和从我们的数据中查找到的状态。答案以清晰的格式呈现给用户。

测试路线优化：

# Example 2: Route optimization query
user_query = "What is the best route from Warehouse to A, B and C?"
agent.print_response(user_query)

预期行为：代理识别出路线规划请求（查询中包含“从……到……”）。它将调用RouteTool来计算最优路径。在内部，工具将计算距离并返回最佳顺序。代理给用户的最终答案可能是：

Optimal route: **Warehouse -> A -> C -> B** (Total distance: 23 km)

这表示从仓库出发前往所有目的地（A、B、C）的最佳路线，并显示总行驶距离。代理通过利用我们的自定义路线逻辑得出该答案——这是我们如何通过特定领域功能扩展 AI 代理的一个有力示例。

请随意测试查询的变化：

• “Track TRK98765”应该返回该 ID 的已交付状态。
• “规划从仓库出发从 A 到 C 和 B 的路线”（不同的措辞）仍应理解为路线优化。
• “TRK00000 在哪里？”（未知 ID）应触发代理响应，表示它没有关于该跟踪 ID 的信息。

每次测试都有助于确保我们的意图得到正确识别，并且工具按预期运行。如果代理的输出不符合预期，你可以优化工具中的指令或解析逻辑。

部署和扩展的最佳实践

构建代理只是第一步。当你准备在现实世界场景（例如，作为物流仪表板的一部分或运营人员的聊天机器人）中部署这个物流AI代理时，请考虑以下最佳实践：

• 使用专用API或服务：将你的代理封装在一个Web服务中，以便轻松集成。例如，你可以使用一个FastAPI应用程序来暴露一个端点，该端点将用户查询转发到agent.run()（运行本地代理API - Agno）。这样，你的物流代理可以通过REST调用（来自Web应用、Slack机器人等）进行访问。

• 容器化：将你的应用程序和Agno代理打包在一个Docker容器中，以实现一致的部署。Agno设计为可以在任何地方运行（你自己的云或本地环境）。容器化确保依赖项（包括Agno和任何数据库库）在开发、预生产和生产环境中保持一致。

• 状态和内存管理：如果你的代理将处理长时间会话或大量查询，请考虑启用Agno的内置内存和状态管理。例如，连接数据库（如带有向量存储的PostgreSQL）可以让代理记住过去的交互或存储频繁访问的路线信息。Agno支持开箱即用地存储代理状态和使用向量数据库进行知识管理。

• 监控和日志记录：使用Agno的监控工具来跟踪代理的性能。记录每个查询和响应，包括任何工具调用和错误。这有助于调试问题（如代理选择了错误的工具）和衡量使用情况。Agno甚至提供了一个代理UI，用于在开发期间进行聊天和评估。

• 水平扩展：对于高吞吐量场景（例如，许多并发用户询问物流查询），水平扩展服务。因为Agno代理实例化是轻量级的（声称非常快且高效），所以生成多个实例或工作进程是可行的。确保你的设计是无状态的，或者通过数据库使用共享内存，以便任何实例都可以处理任何请求。

• 工具和模型优化：随着代理的发展，你可能会添加更多工具（例如，用于实时路线更新的交通API工具，或仓库库存工具）。保持代理的范围集中——如果工具集变得非常大，Agno建议使用一组专门的代理。这可以防止一个代理变得过于笨拙。此外，为任务选择合适的模型——GPT-4功能强大但成本高昂；对于更简单的查询，在生产环境中可能足以使用较小的模型或开源替代方案。

• 安全和验证：如果代理将执行操作（如安排交付或更新数据库），请确保验证来自大型语言模型（LLM）的输出。对工具输入应用检查（我们的工具中的正则表达式解析是基本验证的一个例子）。始终清理任何外部数据，并注意工具函数中的错误处理，以便在数据源不可达时，代理可以优雅地失败。

通过遵循这些实践，你可以确保你的物流AI代理或你构建的任何代理在从原型到生产的过程中都是健壮、高效且可维护的。

结论：物流AI代理的完整实现

在这篇文章中，我们介绍了Agno，并演示了如何使用它一步一步地构建一个以物流为重点的AI代理。我们选择了一个结合路线优化和货物跟踪的使用案例——这是两个常见的物流挑战——并展示了AI代理如何通过自然语言交互处理这两者。在此过程中，我们设置了Agno，定义了代理的意图和指令，通过自定义工具与示例物流数据集成，并测试了代理的能力。我们还涵盖了在真实环境中部署和扩展代理的最佳实践。

最后，以下是我们构建的物流AI代理的完整代码。这汇集了所有讨论过的组件：

import re
from itertools import permutations
from agno.agent import Agent
from agno.models.openai import OpenAIChat

# Sample data for shipments and distances
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}
}
# Define custom tools
class TrackingTool:
    def __init__(self):
        self.name = "TrackingTool"
        self.description = "Provides shipment status updates given a tracking ID."
    def run(self, query: str) -> str:
        match = re.search(r"\bTRK\d+\b", query.upper())
        if not match:
            return "Please provide a valid tracking ID."
        tid = match.group(0)
        status = tracking_data.get(tid)
        return f"Status for {tid}: {status}" if status else f"No information for {tid}."
class RouteTool:
    def __init__(self):
        self.name = "RouteTool"
        self.description = "Computes the best delivery route given a start and destinations."
    def run(self, query: str) -> str:
        m = re.search(r"from\s+([\w\s]+)\s+to\s+(.+)", query, re.IGNORECASE)
        if not m:
            return "Specify route as '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"Unknown origin: {origin}."
        for loc in dests:
            if loc not in distance_matrix:
                return f"Unknown destination: {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"Optimal route: {route_plan} (Total distance: {best_distance} km)"
# Create the agent with model, instructions, and tools
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    description="You are a knowledgeable logistics assistant.",
    instructions=[
        "If the user asks about a shipment or tracking ID, use the TrackingTool.",
        "If the user asks about route optimization or best route, use the RouteTool.",
        "Provide concise and clear answers, including relevant details from the tools."
    ],
    tools=[TrackingTool(), RouteTool()],
    show_tool_calls=False,  # set to False for clean output (True for debugging)
    markdown=True
)
# Example usage
print(agent.run("Where is shipment TRK12345?"))
print(agent.run("Find the best route from Warehouse to A, B and C"))

#output is given below

通过这一实现，你拥有了一个由Agno驱动的物流人工智能代理。在你的环境中插入有效的OpenAI API密钥后运行上述代码，将为示例查询生成响应——实现了我们设定的两个意图的处理。此示例可以进一步扩展和定制：你可以集成真实的API（用于实时跟踪数据或地图服务）、添加更多意图（例如，通过检查设备日志进行预测性维护），甚至创建一个多代理系统，其中一个代理负责路线规划，另一个代理负责库存查询，并根据需要进行协作。Agno的灵活性和性能使得在物流领域试验和扩展此类解决方案成为可能。我所做的只是一个非常简单的扩展。我使用了Oxylabs（7天内5000次免费请求）API来访问谷歌地图，并通过创建FuelStationSearchTool为路线上的送货司机获取附近的加油站。

import os
import requests
from agno.agent import Agent
from agno.models.azure import AzureOpenAI
# Define custom Oxylabs tool
class FuelStationSearchTool:
    def __init__(self):
        self.name = "FuelStationSearchTool"
        self.description = "Fetches nearby fuel stations using Oxylabs Real-Time 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 Error: {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: {rating})")
        if not stations_info:
            return "No nearby fuel stations found."
        return "\n".join(stations_info)
# Initialize Agno agent
agent = Agent(
    model=AzureOpenAI(id="gpt-4o"),
    description="An intelligent logistics assistant helping drivers find nearby services.",
    instructions=[
        "Use FuelStationSearchTool for queries about nearby fuel stations.",
        "List fuel stations clearly with names, addresses, and ratings."
    ],
    markdown=True,
    tools=[FuelStationSearchTool()]
)
# Main execution
if __name__ == "__main__":
    location_query = "NLS, Missisauga, ON"
    agent.print_response(location_query)

仅供参考：NLS是位于安大略省密西沙加的一个仓库。并

且我们从我们的代理那里获取了附近所有的加油站信息。

总结：在本指南中，我们展示了开发人员如何使用Agno构建一个AI代理来应对物流挑战。通过遵循清晰的分步方法——从设置框架、定义代理的范围（意图）、实现集成点，到测试和部署——你可以创建智能代理，实现复杂工作流程的自动化和简化。物流领域只是其中一个例子；同样的模式也可以应用于金融、客户服务等领域。