Agno - 构建 Agent

Building Agents - 构建 Agent

原文档：docs.agno.com/agents/buil…

概述

构建有效的 Agent 应该从简单开始：一个模型、一些工具，以及指令。一旦基础功能正常工作，再根据需求逐步添加更多功能。

1. 最简单的 Agent 示例

知识点说明

一个最基本的 Agent 需要三个核心组件：

• model: 语言模型，负责理解和生成文本
• tools: 工具，让 Agent 能够执行特定任务（如搜索、计算等）
• instructions: 指令，告诉 Agent 如何完成任务

代码示例 1: 基础 Agent

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.hackernews import HackerNewsTools

# 创建一个最简单的 Agent
agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[HackerNewsTools()],
    instructions="Write a report on the topic. Output only the report.",
    markdown=True,
)

# 运行 Agent
agent.print_response("Trending startups and products.", stream=True)

代码示例 2: 不带工具的纯文本 Agent

from agno.agent import Agent
from agno.models.deepseek import DeepSeek

# 创建一个纯对话 Agent，不需要工具
agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    instructions="You are a helpful assistant. Answer questions concisely.",
)

# 运行 Agent
agent.print_response("What is artificial intelligence?", stream=True)

代码示例 3: 带多个工具的 Agent

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.calculator import CalculatorTools

# 创建一个有多个工具的 Agent
agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[
        DuckDuckGoTools(),  # 搜索工具
        CalculatorTools(),  # 计算工具
    ],
    instructions="""You are a helpful assistant that can:
1. Search for information using DuckDuckGo
2. Perform calculations using the calculator
Always provide accurate and helpful responses.""",
    markdown=True,
)

# 运行 Agent
agent.print_response("What is the population of Tokyo divided by 1000?", stream=True)

代码示例 4: 带名称和描述的 Agent

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.yfinance import YFinanceTools

# 创建一个带名称和描述的 Agent
agent = Agent(
    name="Finance Assistant",
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[YFinanceTools()],
    instructions="""You are a finance assistant specialized in stock market data.
Provide clear, concise financial information.""",
    description="An agent that helps with financial data and stock information",
    markdown=True,
)

# 运行 Agent
agent.print_response("What is the current price of Apple stock?", stream=True)

代码示例 5: 启用调试模式的 Agent

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

# 创建启用调试的 Agent
agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    debug_mode=True,  # 启用调试模式，查看详细日志
    markdown=True,
)

# 运行 Agent
agent.print_response("Latest news about AI", stream=True)

2. 运行 Agent 的方式

知识点说明

Agno 提供了多种运行 Agent 的方式：

• print_response(): 开发环境使用，美观地打印响应
• run(): 生产环境使用，返回 RunOutput 对象
• arun(): 异步运行，用于异步环境

代码示例 1: 使用 print_response()（推荐开发使用）

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    markdown=True,
)

# 使用 print_response - 自动格式化输出到终端
agent.print_response("What is machine learning?", stream=True)

代码示例 2: 使用 run()（推荐生产使用）

from agno.agent import Agent, RunOutput
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.utils.pprint import pprint_run_response

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    markdown=True,
)

# 使用 run() - 返回 RunOutput 对象
response: RunOutput = agent.run("What is machine learning?")

# 打印响应内容
print(f"Content: {response.content}")
print(f"Run ID: {response.run_id}")
print(f"Session ID: {response.session_id}")

# 或使用 pprint_run_response 美化打印
pprint_run_response(response, markdown=True)

代码示例 3: 使用流式输出

from typing import Iterator
from agno.agent import Agent, RunOutputEvent, RunEvent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    markdown=True,
)

# 使用流式输出
stream: Iterator[RunOutputEvent] = agent.run("What is machine learning?", stream=True)

# 逐块处理响应
for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(chunk.content, end="", flush=True)

代码示例 4: 使用异步运行

import asyncio
from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    markdown=True,
)

async def main():
    # 使用 arun() 异步运行
    response = await agent.arun("What is machine learning?")
    print(f"Response: {response.content}")

# 运行异步函数
asyncio.run(main())

代码示例 5: 使用 input 参数明确指定输入

from agno.agent import Agent, RunOutput
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information and provide summaries.",
    markdown=True,
)

# 使用 input 参数明确指定输入
response: RunOutput = agent.run(input="What is machine learning?")
print(f"Response: {response.content}")

3. 可调用工厂（Callable Factories）

知识点说明

可调用工厂允许你动态决定 Agent 使用哪些工具或知识库。这在以下场景非常有用：

• 不同用户需要不同的工具集
• 根据会话状态动态选择工具
• 根据权限控制工具访问

代码示例 1: 根据角色动态选择工具

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools


def get_tools(run_context: RunContext):
    """
    根据用户角色返回不同的工具集
    """
    role = (run_context.session_state or {}).get("role", "general")
    
    if role == "finance":
        print(f"[DEBUG] Using finance tools for role: {role}")
        return [YFinanceTools()]
    else:
        print(f"[DEBUG] Using general tools for role: {role}")
        return [DuckDuckGoTools()]


agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=get_tools,  # 传入函数而不是列表
)

# 金融角色 - 使用 YFinanceTools
print("=== Finance Role ===")
agent.print_response("AAPL stock price?", session_state={"role": "finance"}, stream=True)

# 普通角色 - 使用 DuckDuckGoTools
print("\n=== General Role ===")
agent.print_response("Latest AI news?", session_state={"role": "general"}, stream=True)

代码示例 2: 根据用户权限选择工具

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.shell import ShellTools


def get_tools_by_permission(run_context: RunContext):
    """
    根据用户权限返回工具集
    """
    user_role = (run_context.session_state or {}).get("user_role", "guest")
    
    tools = [DuckDuckGoTools()]  # 所有用户都有搜索权限
    
    if user_role == "admin":
        tools.append(ShellTools())  # 只有管理员有 shell 权限
        
    return tools


agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=get_tools_by_permission,
    instructions="Help users with their requests.",
)

# 普通用户
print("=== Guest User ===")
agent.print_response("Search for Python tutorials", 
                     session_state={"user_role": "guest"}, 
                     stream=True)

# 管理员
print("\n=== Admin User ===")
agent.print_response("List files in current directory", 
                     session_state={"user_role": "admin"}, 
                     stream=True)

代码示例 3: 根据查询类型选择工具

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.calculator import CalculatorTools
from agno.tools.yfinance import YFinanceTools


def get_tools_by_query(run_context: RunContext):
    """
    根据查询内容智能选择工具
    """
    query = (run_context.session_state or {}).get("query", "").lower()
    
    tools = [DuckDuckGoTools()]  # 默认有搜索工具
    
    # 如果查询包含数学相关词汇，添加计算器
    if any(word in query for word in ["calculate", "math", "+", "-", "*", "/"]):
        tools.append(CalculatorTools())
    
    # 如果查询包含股票相关词汇，添加金融工具
    if any(word in query for word in ["stock", "price", "ticker", "aapl", "goog"]):
        tools.append(YFinanceTools())
        
    return tools


agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=get_tools_by_query,
)

# 数学查询
print("=== Math Query ===")
agent.print_response("Calculate 123 * 456", 
                     session_state={"query": "Calculate 123 * 456"}, 
                     stream=True)

# 股票查询
print("\n=== Stock Query ===")
agent.print_response("What is AAPL stock price?", 
                     session_state={"query": "What is AAPL stock price?"}, 
                     stream=True)

代码示例 4: 带缓存的工具工厂

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools

# 缓存工具实例
_tool_cache = {}


def get_cached_tools(run_context: RunContext):
    """
    使用缓存避免重复创建工具实例
    """
    role = (run_context.session_state or {}).get("role", "general")
    
    cache_key = f"tools_{role}"
    
    if cache_key not in _tool_cache:
        print(f"[DEBUG] Creating new tools for role: {role}")
        if role == "finance":
            _tool_cache[cache_key] = [YFinanceTools()]
        else:
            _tool_cache[cache_key] = [DuckDuckGoTools()]
    else:
        print(f"[DEBUG] Using cached tools for role: {role}")
    
    return _tool_cache[cache_key]


agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=get_cached_tools,
)

# 第一次调用 - 创建工具
print("=== First Call ===")
agent.print_response("AAPL stock price?", 
                     session_state={"role": "finance"}, 
                     stream=True)

# 第二次调用 - 使用缓存
print("\n=== Second Call (Cached) ===")
agent.print_response("GOOGL stock price?", 
                     session_state={"role": "finance"}, 
                     stream=True)

代码示例 5: 异步工具工厂

import asyncio
from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools


async def get_tools_async(run_context: RunContext):
    """
    异步获取工具（例如从数据库或API加载）
    """
    user_id = (run_context.session_state or {}).get("user_id", "default")
    
    # 模拟异步操作（如从数据库查询用户配置）
    await asyncio.sleep(0.1)
    
    print(f"[DEBUG] Async loading tools for user: {user_id}")
    
    # 根据用户配置返回工具
    return [DuckDuckGoTools()]


agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=get_tools_async,
)


async def main():
    response = await agent.arun(
        "Latest AI news?",
        session_state={"user_id": "user_123"}
    )
    print(f"Response: {response.content}")


asyncio.run(main())

4. Agent 配置选项详解

知识点说明

Agent 有很多配置选项可以自定义其行为：

• markdown: 是否使用 markdown 格式输出
• debug_mode: 是否启用调试模式
• show_tool_calls: 是否显示工具调用信息
• system_message: 系统消息
• description: Agent 描述

代码示例 1: 显示工具调用

from agno.agent import Agent
from agno.models.deepseek import DeepSeek
from agno.tools.duckduckgo import DuckDuckGoTools

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    tools=[DuckDuckGoTools()],
    instructions="Search for information.",
    show_tool_calls=True,  # 显示工具调用信息
    markdown=True,
)

agent.print_response("What is Python programming language?", stream=True)

代码示例 2: 自定义系统消息

from agno.agent import Agent
from agno.models.deepseek import DeepSeek

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    system_message="""You are an expert Python programmer.
Always provide code examples in your explanations.
Be concise but thorough.""",
    markdown=True,
)

agent.print_response("Explain list comprehensions in Python", stream=True)

代码示例 3: 禁用流式输出

from agno.agent import Agent
from agno.models.deepseek import DeepSeek

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    instructions="Answer questions concisely.",
)

# 不使用 stream=True，等待完整响应
response = agent.run("What is 2+2?")
print(f"Full response: {response.content}")

代码示例 4: 设置响应格式

from agno.agent import Agent
from agno.models.deepseek import DeepSeek

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    instructions="Provide responses in bullet points.",
    markdown=True,
)

agent.print_response("List 3 benefits of Python", stream=True)

代码示例 5: 设置超时时间

from agno.agent import Agent
from agno.models.deepseek import DeepSeek

agent = Agent(
    model=DeepSeek(id="deepseek-v4-flash"),
    instructions="Answer quickly.",
)

# 注意：实际超时设置取决于模型配置
response = agent.run("Quick answer: What is AI?")
print(f"Response: {response.content}")

5. 完整可运行代码示例

以下是一个综合示例，演示了本章所有核心知识点：

"""
Agno Agent 完整示例 - 构建和运行 Agent

本示例演示：
1. 创建基础 Agent
2. 使用不同方式运行 Agent
3. 使用可调用工厂动态选择工具
4. 配置 Agent 选项

运行前请确保：
- 安装 agno: pip install agno
- 设置 DeepSeek API 密钥: export DEEPSEEK_API_KEY=your_key
"""

import os
import asyncio
from typing import Iterator
from agno.agent import Agent, RunOutput, RunOutputEvent, RunEvent
from agno.models.deepseek import DeepSeek
from agno.run import RunContext
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools
from agno.tools.calculator import CalculatorTools
from agno.utils.pprint import pprint_run_response


# ============================================================
# 第一部分：基础 Agent 创建
# ============================================================

def create_basic_agent():
    """
    创建一个基础 Agent
    演示最简单的 Agent 配置
    """
    print("=" * 60)
    print("示例 1: 基础 Agent")
    print("=" * 60)
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        tools=[DuckDuckGoTools()],
        instructions="Search for information and provide concise summaries.",
        markdown=True,
    )
    
    # 使用 print_response 运行（开发推荐）
    agent.print_response("What is machine learning?", stream=True)
    print("\n")


# ============================================================
# 第二部分：使用 run() 方法
# ============================================================

def run_agent_production():
    """
    使用 run() 方法运行 Agent（生产环境推荐）
    演示如何获取 RunOutput 对象并处理响应
    """
    print("=" * 60)
    print("示例 2: 使用 run() 方法（生产环境）")
    print("=" * 60)
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        tools=[DuckDuckGoTools()],
        instructions="Search for information.",
    )
    
    # 运行 Agent 并获取 RunOutput 对象
    response: RunOutput = agent.run("What is Python?")
    
    # 访问响应的各种属性
    print(f"Content: {response.content}")
    print(f"Run ID: {response.run_id}")
    print(f"Agent ID: {response.agent_id}")
    print(f"Session ID: {response.session_id}")
    print(f"Model: {response.model}")
    print("\n")


# ============================================================
# 第三部分：流式输出
# ============================================================

def run_agent_streaming():
    """
    使用流式输出
    演示如何逐块处理响应
    """
    print("=" * 60)
    print("示例 3: 流式输出")
    print("=" * 60)
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        tools=[DuckDuckGoTools()],
        instructions="Search for information.",
    )
    
    # 使用流式输出
    print("Streaming response: ", end="", flush=True)
    stream: Iterator[RunOutputEvent] = agent.run("What is AI?", stream=True)
    
    for chunk in stream:
        if chunk.event == RunEvent.run_content:
            print(chunk.content, end="", flush=True)
    
    print("\n\n")


# ============================================================
# 第四部分：可调用工厂
# ============================================================

def create_dynamic_tool_agent():
    """
    使用可调用工厂动态选择工具
    根据用户角色返回不同的工具集
    """
    print("=" * 60)
    print("示例 4: 可调用工厂 - 动态工具选择")
    print("=" * 60)
    
    def get_tools(run_context: RunContext):
        """根据角色返回工具"""
        role = (run_context.session_state or {}).get("role", "general")
        
        if role == "finance":
            print(f"[System] Loading finance tools for role: {role}")
            return [YFinanceTools()]
        elif role == "math":
            print(f"[System] Loading calculator tools for role: {role}")
            return [CalculatorTools()]
        else:
            print(f"[System] Loading search tools for role: {role}")
            return [DuckDuckGoTools()]
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        tools=get_tools,  # 传入函数而不是列表
    )
    
    # 测试不同角色
    print("\n--- Finance Role ---")
    agent.print_response("AAPL stock price", 
                         session_state={"role": "finance"}, 
                         stream=True)
    
    print("\n\n--- Math Role ---")
    agent.print_response("Calculate 123 * 456", 
                         session_state={"role": "math"}, 
                         stream=True)
    
    print("\n\n--- General Role ---")
    agent.print_response("What is Python?", 
                         session_state={"role": "general"}, 
                         stream=True)
    print("\n")


# ============================================================
# 第五部分：高级配置
# ============================================================

def create_advanced_agent():
    """
    创建高级配置的 Agent
    演示各种配置选项
    """
    print("=" * 60)
    print("示例 5: 高级配置")
    print("=" * 60)
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        name="Advanced Assistant",
        description="An agent with advanced configuration",
        tools=[DuckDuckGoTools()],
        instructions="Provide detailed and accurate information.",
        system_message="You are an expert assistant. Always cite your sources.",
        show_tool_calls=True,  # 显示工具调用
        markdown=True,
        debug_mode=False,  # 关闭调试模式
    )
    
    agent.print_response("What are the latest developments in AI?", stream=True)
    print("\n")


# ============================================================
# 第六部分：异步运行
# ============================================================

async def run_async_agent():
    """
    异步运行 Agent
    适用于异步环境
    """
    print("=" * 60)
    print("示例 6: 异步运行")
    print("=" * 60)
    
    agent = Agent(
        model=DeepSeek(id="deepseek-v4-flash"),
        tools=[DuckDuckGoTools()],
        instructions="Answer concisely.",
    )
    
    # 使用 arun() 异步运行
    response = await agent.arun("What is async programming?")
    print(f"Response: {response.content}")
    print("\n")


# ============================================================
# 主函数：运行所有示例
# ============================================================

def main():
    """
    主函数：按顺序运行所有示例
    """
    print("\n" + "=" * 60)
    print("Agno Agent 完整示例")
    print("=" * 60 + "\n")
    
    # 检查 API 密钥
    if not os.getenv("DEEPSEEK_API_KEY"):
        print("警告: 未设置 DEEPSEEK_API_KEY 环境变量")
        print("请设置: export DEEPSEEK_API_KEY=your_key\n")
        return
    
    # 运行所有示例
    try:
        create_basic_agent()
        run_agent_production()
        run_agent_streaming()
        create_dynamic_tool_agent()
        create_advanced_agent()
        
        # 异步示例
        asyncio.run(run_async_agent())
        
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
    
    print("\n" + "=" * 60)
    print("所有示例运行完成!")
    print("=" * 60)


# ============================================================
# 入口点
# ============================================================

if __name__ == "__main__":
    main()

运行说明

1. 安装依赖:

   pip install agno
   

2. 设置 API 密钥:

   export DEEPSEEK_API_KEY=your_deepseek_api_key
   

3. 运行示例:

   python building_agents_example.py
   

示例输出结构

运行后会依次展示：

1. 基础 Agent 的流式响应
2. 使用 run() 方法的详细输出信息
3. 流式输出的逐字显示效果
4. 可调用工厂根据角色动态切换工具
5. 高级配置 Agent 的工具调用显示
6. 异步运行的响应