从零到一：用Agno框架构建你的第一个多智能体协作系统

1. 初识Agno框架：为什么选择它来构建智能体？

在当今AI技术快速发展的时代，多智能体系统正成为解决复杂问题的有力工具。Agno作为一个新兴的高性能Python框架，凭借其轻量级设计、完整的功能生态和出色的性能表现，迅速获得了开发者的青睐。

Agno最显著的特点是它提供了五级智能体架构，从基础工具代理到复杂的团队协作代理，开发者可以根据需求选择合适的抽象级别。与同类框架相比，Agno在以下方面表现突出：

• 性能优势：官方测试显示，在实例化1000次带工具的智能体任务中，Agno在时间和内存使用上均优于LangGraph等流行框架
• 开发体验：清晰的API设计和详尽的文档大大降低了学习曲线
• 功能完备：内置100+实用工具，支持知识管理、记忆模块和多种数据库集成
• 灵活扩展：通过MCP协议支持自定义工具和第三方服务集成

# 基础安装命令
pip install agno openai duckduckgo-search

对于初学者而言，Agno提供了平滑的学习路径。你可以从一个简单的问答代理开始，逐步扩展到具有记忆、知识检索能力的复杂代理，最终构建多智能体协作系统。这种渐进式的设计理念让学习过程更加自然。

2. 搭建开发环境与第一个智能体

2.1 环境准备

开始之前，我们需要设置Python虚拟环境并安装必要依赖：

# 创建并激活虚拟环境
python3 -m venv agno_env
source agno_env/bin/activate  # Linux/Mac
# agno_env\Scripts\activate  # Windows

# 安装核心包
pip install agno openai

如果你计划使用向量数据库或需要多模态支持，可以安装额外组件：

# 可选：向量数据库支持
pip install pgvector lancedb

# 可选：多模态扩展
pip install opencv-python elevenlabs

2.2 构建基础智能体

让我们从最简单的例子开始 - 创建一个能够回答问题的AI助手：

from agno.agent import Agent
from agno.models.openai import OpenAIChat

# 初始化模型和智能体
model = OpenAIChat(id="gpt-4")  # 使用OpenAI兼容接口
agent = Agent(
    model=model,
    description="你是一个乐于助人的AI助手",
    markdown=True  # 让输出格式化为Markdown
)

# 与智能体交互
response = agent.run("请用中文简单介绍一下你自己")
print(response.content)

这段代码展示了Agno的核心工作流程：

1. 选择或创建一个大语言模型实例
2. 定义智能体的基本属性和能力
3. 通过run()方法与智能体交互

关键参数说明：

参数	类型	说明	示例
model	BaseModel	语言模型实例	OpenAIChat
description	str	智能体的角色描述	"专业的技术支持专家"
instructions	List[str]	具体行为指令	["回答要简洁专业"]
markdown	bool	是否使用Markdown格式化输出	True

3. 增强智能体能力：记忆、知识与工具

基础智能体虽然能回答问题，但缺乏记忆和实际能力。让我们通过三个关键扩展来提升它的实用性。

3.1 添加对话记忆

使用Sqlite数据库存储对话历史：

from agno.db.sqlite import SqliteDb
from agno.session import AgentSession

# 配置数据库
db = SqliteDb(db_file="agno_session.db", session_table="conversations")

# 创建带记忆的智能体
agent = Agent(
    model=model,
    db=db,
    add_history_to_context=True,  # 将历史对话加入上下文
    num_history_runs=3  # 保留最近3轮对话
)

# 带会话ID的交互
session_id = "user_123"
agent.run("我上次问过什么问题？", session_id=session_id)

3.2 集成知识库

Agno支持多种知识源，包括PDF、网页和数据库：

from agno.knowledge.pdf import PDFKnowledgeBase

# 加载PDF知识库
knowledge = PDFKnowledgeBase(path="data/产品手册.pdf")

# 创建知识增强型智能体
agent = Agent(
    model=model,
    knowledge=knowledge,
    description="产品技术支持专家"
)

response = agent.run("我们产品的保修政策是什么？")

3.3 使用工具扩展能力

Agno内置100+实用工具，也可以轻松自定义：

from agno.tools.duckduckgo import DuckDuckGoTools

# 内置工具示例
agent = Agent(
    model=model,
    tools=[DuckDuckGoTools()],  # 添加网页搜索工具
    description="实时信息查询助手"
)

# 查询实时信息
agent.run("今天硅谷有什么重要科技新闻？")

# 自定义工具示例
def calculate_age(birth_year: int) -> int:
    """计算当前年龄"""
    return 2025 - birth_year

agent = Agent(tools=[calculate_age])
agent.run("1990年出生的人现在多少岁？")

提示：使用debug_mode=True参数可以查看详细的提示词和工具调用日志，这对调试复杂交互非常有用

4. 构建多智能体协作系统

当单个智能体无法满足复杂需求时，Agno的Team功能允许创建协作式多智能体系统。

4.1 基础团队配置

让我们创建一个包含新闻代理和天气代理的协作团队：

from agno.team import Team

# 定义新闻代理
news_agent = Agent(
    model=model,
    role="新闻记者",
    tools=[DuckDuckGoTools()],
    description="负责获取最新新闻资讯"
)

# 定义天气代理
def get_weather(city: str) -> str:
    # 实际项目中这里会调用天气API
    return f"{city}今天晴，气温22-28°C"

weather_agent = Agent(
    model=model,
    role="气象专家",
    tools=[get_weather],
    description="提供全球城市天气预报"
)

# 创建团队
team = Team(
    model=model,
    name="资讯服务团队",
    members=[news_agent, weather_agent],
    description="综合提供新闻和天气信息服务"
)

# 使用团队
response = team.run("杭州的天气怎么样？上海有什么科技新闻？")
print(response.content)

团队工作原理：

1. Team Leader接收用户请求
2. 分析任务类型并路由到合适的成员
3. 收集成员响应并整合最终结果

4.2 高级团队协作

对于更复杂的场景，可以构建层级式团队结构：

# 创建子团队 - 技术支持组
support_agent1 = Agent(
    model=model,
    role="安装专家",
    description="解决软件安装问题"
)

support_agent2 = Agent(
    model=model,
    role="故障排查",
    description="诊断系统错误"
)

support_team = Team(
    model=model,
    name="技术支持组",
    members=[support_agent1, support_agent2]
)

# 创建子团队 - 销售咨询组
sales_agent = Agent(
    model=model,
    role="销售顾问",
    description="处理产品购买咨询"
)

# 创建顶级团队
company_team = Team(
    model=model,
    name="客户服务部",
    members=[support_team, sales_agent],
    description="处理所有客户咨询和问题"
)

# 使用层级团队
response = company_team.run("我的软件安装失败了，错误代码500")

4.3 团队协作优化技巧

1. 明确角色定义：每个成员应有清晰的职责范围
2. 工具专业化：避免工具重复，确保每个工具由最合适的代理掌握
3. 上下文共享：使用team_shared_memory实现成员间数据共享
4. 性能监控：利用AgentOS面板观察团队协作效率

# 共享内存示例
team = Team(
    model=model,
    members=[...],
    shared_memory={"customer_level": "VIP"}  # 共享数据
)

5. 实战：构建智能客服系统

让我们综合运用所学知识，构建一个完整的电商客服系统。

5.1 系统架构设计

我们的系统将包含以下智能体：

1. 接待代理：处理初始问候和问题分类
2. 订单代理：查询和处理订单相关咨询
3. 产品代理：回答产品详情和技术问题
4. 投诉代理：处理客户投诉和退款请求

5.2 代码实现

from agno.agent import Agent
from agno.team import Team
from agno.db.postgres import PostgresDb

# 数据库连接
db = PostgresDb(
    dbname="ecommerce",
    user="admin",
    password="securepass",
    host="localhost"
)

# 1. 接待代理
reception_agent = Agent(
    model=model,
    role="接待员",
    description="欢迎客户并分类问题类型",
    db=db,
    instructions=[
        "保持友好专业的语气",
        "准确识别问题类型：订单、产品、投诉"
    ]
)

# 2. 订单代理
def query_order(order_id: str) -> dict:
    """模拟订单查询"""
    return {
        "status": "已发货",
        "items": ["智能音箱×1"],
        "tracking": "SF123456789"
    }

order_agent = Agent(
    model=model,
    role="订单专员",
    tools=[query_order],
    description="处理所有订单相关查询",
    db=db
)

# 3. 产品代理
product_knowledge = PDFKnowledgeBase("data/product_catalog.pdf")
product_agent = Agent(
    model=model,
    role="产品专家",
    knowledge=product_knowledge,
    description="回答产品规格、功能等问题",
    markdown=True
)

# 4. 投诉代理
complaint_agent = Agent(
    model=model,
    role="客户关系",
    description="处理投诉和退款请求",
    instructions=[
        "保持同理心",
        "根据政策提供解决方案"
    ],
    db=db
)

# 构建客服团队
customer_service = Team(
    model=model,
    name="电商客服中心",
    members=[reception_agent, order_agent, product_agent, complaint_agent],
    description="提供全方位的客户服务支持",
    shared_memory={"company_policy": "data/policy.md"}
)

# 使用示例
questions = [
    "我上周买的音箱怎么还没到？",
    "你们旗舰手机的防水等级是多少？",
    "我要投诉！收到的商品有损坏"
]

for question in questions:
    response = customer_service.run(question)
    print(f"Q: {question}\nA: {response.content}\n")

5.3 系统优化建议

1. 性能优化：

   • 为高频工具添加缓存机制
   • 使用异步处理耗时操作
   • 定期清理无用会话数据

2. 用户体验提升：

   • 添加个性化记忆（用户偏好记录）
   • 实现多轮对话上下文保持
   • 支持多模态输入（图片、语音）

3. 运维监控：

   • 集成AgentOS控制面板
   • 设置关键指标报警
   • 定期审查对话日志优化提示词

# 异步处理示例
async def async_query():
    agent = Agent(...)
    response = await agent.arun("异步查询示例")

在实际项目中，你可以根据业务需求继续扩展这个系统，比如添加库存查询代理、支付问题代理等。Agno的模块化设计使得这种扩展变得非常简单。