Home
Blog
如何利用 Agno 和 Milvus 构建生产就绪的多代理系统

如何利用 Agno 和 Milvus 构建生产就绪的多代理系统

Tutorials

February 10, 2026

Min Yin

如果您一直在构建人工智能 Agents，您可能会遇到这样的问题：您的演示效果很好，但要将其投入生产却完全是另一回事。

我们在前面的文章中介绍了代理内存管理和 Rerankers。现在，让我们来应对更大的挑战--建立能在生产中真正站稳脚跟的 Agents。

现实情况是：生产环境非常混乱。单个 Agents 很少能胜任，这就是多 Agents 系统随处可见的原因。但目前可用的框架往往分为两类：一类是轻量级的，演示效果很好，但在实际负载下就会崩溃；另一类是功能强大的，但学习和构建都要花很长时间。

我最近一直在尝试使用Agno，它似乎找到了一个合理的中间点--既注重生产准备，又不过分复杂。几个月来，该项目在 GitHub 上获得了 37,000 多颗星，这表明其他开发者也觉得它很有用。

在这篇文章中，我将分享在使用 Agno 和Milvus作为内存层构建多代理系统时学到的知识。我们将探讨 Agno 与 LangGraph 等替代方案的比较，并介绍一个完整的实现方案，您可以亲自尝试。

Agno 是什么？

Agno是专为生产应用而构建的多代理框架。它有两个不同的层：

Agno 框架层：您定义代理逻辑的地方
AgentOS 运行层：将逻辑转化为可实际部署的 HTTP 服务

可以这样理解：框架层定义 Agents 应该做什么，而 AgentOS 处理如何执行和提供这些工作。

框架层

这是你直接使用的部分。它引入了三个核心概念：

Agent：处理特定类型的任务
团队：协调多个代理解决复杂问题
工作流：定义执行顺序和结构

我最欣赏的一点是：你不需要学习新的 DSL 或绘制流程图。Agents 的行为是通过标准 Python 函数调用定义的。该框架处理 LLM 调用、工具执行和内存管理。

AgentOS 运行时层

AgentOS 专为通过异步执行处理大量请求而设计，其无状态架构使扩展变得简单易行。

主要功能包括

内置 FastAPI 集成，可将 Agents 作为 HTTP 端点公开
会话管理和流式响应
监控端点
支持横向扩展

在实践中，AgentOS 处理大部分基础架构工作，这让您可以专注于代理逻辑本身。

Agno 架构的高层视图如下所示。

Agno 与 LangGraph 的对比

要了解 Agno 的优势所在，让我们将其与 LangGraph 进行比较--LangGraph 是使用最广泛的多代理框架之一。

LangGraph使用基于图的状态机。您可以将整个代理工作流程模型化为一张图：步骤是节点，执行路径是边。在流程固定且严格有序的情况下，这种方法效果很好。但对于开放式或对话式场景，这可能会让人感觉受到限制。随着交互变得越来越动态，维持一个清晰的图变得越来越难。

Agno采用了不同的方法。它不是一个纯粹的协调层，而是一个端到端的系统。您只需定义代理行为，AgentOS 就会自动将其作为生产就绪的 HTTP 服务--内置监控、可扩展性和多轮对话支持。无需单独的 API 网关，无需自定义会话管理，无需额外的操作符。

下面是一个快速比较：

维度	LangGraph	Agno
协调模型	使用节点和边明确定义图	用 Python 定义的声明式工作流
状态管理	由开发人员定义和管理自定义状态类	内置内存系统
调试和可观察性	LangSmith（付费）	AgentOS UI（开源）
运行时模型	集成到现有运行时中	基于 FastAPI 的独立服务
部署复杂性	需要通过 LangServe 进行额外设置	开箱即用

LangGraph 可为您提供更多灵活性和细粒度控制。Agno 进行了优化，可加快产品上市时间。正确的选择取决于您的项目阶段、现有基础设施以及所需的定制化程度。如果您不确定，使用这两种系统进行小型概念验证可能是最可靠的决定方式。

为 Agents 内存层选择 Milvus

一旦选择了框架，下一个决定就是如何存储内存和知识。为此，我们使用了 Milvus。Milvus是专为人工智能工作负载构建的最受欢迎的开源向量数据库，在GitHub 上拥有超过42,000+ 个星级。

Agno 原生支持 Milvus。 agno.vectordb.milvus 模块封装了连接管理、自动重试、批量写入和 Embeddings 生成等生产功能。你不需要自己建立连接池或处理网络故障--几行 Python 代码就能为你提供一个工作向量内存层。

Milvus 可根据你的需求进行扩展。它支持三种部署模式：

Milvus Lite：轻量级、基于文件--非常适合本地开发和测试
独立部署：生产工作负载的单服务器部署
分布式：用于大规模场景的全集群

你可以从 Milvus Lite 开始，在本地验证你的代理内存，然后随着流量的增长转到独立或分布式--而无需改变你的应用代码。当您在早期阶段快速迭代，但需要明确的扩展路径时，这种灵活性尤其有用。

逐步实现：使用 Milvus 构建生产就绪的 Agno Agent

让我们从头开始构建一个可投入生产的 Agents。

我们将从一个简单的单个 Agents 示例开始，展示完整的工作流程。然后，我们将把它扩展到一个多 Agents 系统。AgentOS 会自动将所有内容打包成可调用的 HTTP 服务。

1.使用 Docker 部署 Milvus 单机版

(1) 下载部署文件

**wget** **
<https://github.com/Milvus-io/Milvus/releases/download/v2.****5****.****12****/Milvus-standalone-docker-compose.yml> -O docker-compose.yml**

(2) 启动 Milvus 服务

docker-compose up -d

docker-compose ps -a

2.核心实施

import os
from pathlib import Path
from agno.os import AgentOS
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.milvus import Milvus
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.db.sqlite import SqliteDb
os.environ\["OPENAI_API_KEY"\] = "you-key-here"
data_dir = Path("./data")
data_dir.mkdir(exist_ok=True)
knowledge_base = Knowledge(
    contents_db=SqliteDb(
        db_file=str(data_dir / "knowledge_contents.db"),
        knowledge_table="knowledge_contents",
    ),
    vector_db=Milvus(
        collection="agno_knowledge",
        uri="http://192.168.x.x:19530",
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)
*# Create Agent*
agent = Agent(
    name="Knowledge Assistant",
    model=OpenAIChat(id="gpt-4o"),
    instructions=\[
        "You are a knowledge base assistant that helps users query and manage knowledge base content.",
        "Answer questions in English.",
        "Always search the knowledge base before answering questions.",
        "If the knowledge base is empty, kindly prompt the user to upload documents."
    \],
    knowledge=knowledge_base,
    search_knowledge=True,
    db=SqliteDb(
        db_file=str(data_dir / "agent.db"),
        session_table="agent_sessions",
    ),
    add_history_to_context=True,
    markdown=True,
)
agent_os = AgentOS(agents=\[agent\])
app = agent_os.get_app()
if __name__ == "__main__":
    print("\n🚀 Starting service...")
    print("📍 http://localhost:7777")
    print("💡 Please upload documents to the knowledge base in the UI\n")
    agent_os.serve(app="knowledge_agent:app", port=7777, reload=False)

(1) 运行 Agents

**python** **knowledge_agent.py**

3.连接到 AgentOS 控制台

https://os.agno.com/

(1) 创建账户并登录

(2) 将代理连接到 AgentOS

(3) 配置公开端口和 Agents 名称

(4) 添加文件并在 Milvus 中建立索引

(5) 端对端测试 Agents

在此设置中，Milvus 处理高性能语义检索。当知识库助手收到一个技术问题时，它会调用search_knowledge 工具嵌入查询，从 Milvus 中检索最相关的文档块，并将这些结果作为其响应的基础。

Milvus 提供三种部署选项，您可以选择适合您操作要求的架构，同时在所有部署模式中保持应用级 API 的一致性。

上面的演示展示了核心检索和生成流程。不过，要将这一设计应用到生产环境中，还需要对几个架构方面进行更详细的讨论。

如何在 Agents 之间共享检索结果

Agno 的团队模式有一个share_member_interactions=True 选项，允许后来的代理继承先前代理的全部交互历史。在实践中，这意味着当第一个 Agents 从 Milvus 检索信息时，后面的 Agents 可以重复使用这些结果，而不用再次运行相同的搜索。

好处检索成本可在整个团队中分摊。一个向量搜索支持多个 Agents，减少了冗余查询。
缺点检索质量会被放大。如果初始搜索返回的结果不完整或不准确，那么这个错误就会传播给依赖它的每个 Agents。

这就是为什么检索准确性在多 Agents 系统中更为重要的原因。糟糕的检索不仅会降低一个 Agents 的响应，还会影响整个团队。

下面是一个团队设置示例：

from agno.team import Team
analyst = Agent(
    name="Data Analyst",
    model=OpenAIChat(id="gpt-4o"),
    knowledge=knowledge_base,
    search_knowledge=True,
    instructions=\["Analyze data and extract key metrics"\]
)
writer = Agent(
    name="Report Writer",
    model=OpenAIChat(id="gpt-4o"),
    knowledge=knowledge_base,
    search_knowledge=True,
    instructions=\["Write reports based on the analysis results"\]
)
team = Team(
    agents=\[analyst, writer\],
    share_member_interactions=True,  *# Share knowledge retrieval results*
)

为什么 Agno 和 Milvus 要分开分层？

在此架构中，Agno位于对话和协调层。它负责管理对话流、协调 Agents 和维护对话状态，并将会话历史记录保存在关系数据库中。系统的实际领域知识（如产品文档和技术报告）则单独处理，并以向量嵌入的形式存储在Milvus 中。这种明确的划分使会话逻辑和知识存储完全分离。

操作符为何如此重要？

独立扩展：随着 Agno 需求的增长，增加更多的 Agno 实例。随着查询量的增长，通过增加查询节点来扩展 Milvus。每一层都是独立扩展的。
不同的硬件需求：Agno 需要占用 CPU 和内存（LLM 推断、工作流执行）。Milvus 针对高吞吐量向量检索（磁盘 I/O，有时 GPU 加速）进行了优化。将它们分开可防止资源争用。
成本优化：您可以为每一层独立调整和分配资源。

这种分层方法为您提供了一个更高效、更有弹性、更适合生产的架构。

将 Agno 与 Milvus 结合使用时的监控内容

Agno 具有内置的评估功能，但添加 Milvus 后，您需要监控的内容会更多。根据我们的经验，应重点关注以下三个方面：

检索质量：Milvus 返回的文档与查询是否真正相关，还是只是在向量层面上表面相似？
答案忠实性：最终的响应是以检索内容为基础，还是 LLM 在生成无据可循的主张？
端到端延迟细分：不要只跟踪总响应时间。将其按阶段细分--Embeddings 生成、向量搜索、上下文组装、LLM 推断--这样你就能确定哪里出现了延迟。

举个实际例子：当你的 Milvus Collections 从 100 万个向量增加到 1000 万个向量时，你可能会发现检索延迟在逐渐增加。这通常是一个信号，表明需要调整索引参数（如nlist 和nprobe ）或考虑从独立部署转为分布式部署。