一、运行代理

1. 运行代理

以下是运行代理的方法。响应在response对象中捕获。

from typing import Iterator
from agno.agent import Agent, RunResponse
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

# Run agent and return the response as a variable
# 运行agent并且返回 RunResponse对象
response: RunResponse = agent.run("Tell me a 5 second short story about a robot")

# Print the response in markdown format
# 使用markdown格式打印response
pprint_run_response(response, markdown=True)

2. 流式响应

要启用流式处理，请在调用 run（） 时设置 stream=True。这将返回 RunResponseEvent 对象的迭代器，而不是单个响应。

from typing import Iterator
from agno.agent import Agent, RunResponseEvent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4-mini"))

# Run agent and return the response as a stream
# 运行agent时设置 stream=True ，代表的是流式响应
response_stream: Iterator[RunResponseEvent] = agent.run(
    "Tell me a 5 second short story about a lion",
    stream=True
)

# Print the response stream in markdown format
pprint_run_response(response_stream, markdown=True)

3. 流式处理中间步骤

要获得更详细的流式处理，您可以通过设置 stream_intermediate_steps=True 来启用中间步骤。这将提供有关代理内部流程的实时更新。

# Stream with intermediate steps
# 流式处理中间步骤
response_stream: Iterator[RunResponseEvent] = agent.run(
    "Tell me a 5 second short story about a lion",
    stream=True,
    stream_intermediate_steps=True
)

4. 处理事件

您可以通过迭代响应流来在事件到达时对其进行处理

response_stream = agent.run("Your prompt", stream=True, stream_intermediate_steps=True)

for event in response_stream:
    if event.event == "RunResponseContent":
        print(f"Content: {event.content}")
    elif event.event == "ToolCallStarted":
        print(f"Tool call started: {event.tool}")
    elif event.event == "ReasoningStep":
        print(f"Reasoning step: {event.content}")

5. 存储事件

您可以存储在 RunResponse 对象上运行期间发生的所有事件。

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), store_events=True)

response = agent.run("Tell me a 5 second short story about a lion", stream=True, stream_intermediate_steps=True)
pprint_run_response(response)

for event in agent.run_response.events:
    print(event.event)

默认情况下，不存储 RunResponseContentEvent 事件。您可以通过设置 events_to_skip 参数来修改跳过的事件。

例如：

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"), store_events=True, events_to_skip=[RunEvent.run_started.value])

6. 事件类型

根据代理的配置，Agent.run（） 和 Agent.arun（） 函数会生成以下事件：

核心活动

RunStarted： 指示运行的开始
RunResponseContent： 将模型的响应文本作为单独的块
RunCompleted： 表示运行成功完成
RunError： 指示运行期间发生错误
RunCancelled： 表示运行已取消

控制流事件

RunPaused： 指示运行已暂停
RunContinued： 表示已继续暂停的运行

工具事件

ToolCallStarted： 指示工具调用的开始
ToolCallCompleted： 表示工具调用完成，包括工具调用结果

推理事件

ReasoningStarted： 指示代理的推理过程的开始
ReasoningStep： 包含推理过程中的单个步骤
ReasoningCompleted： 表示推理过程完成

内存事件

MemoryUpdateStarted： 指示代理正在更新其内存
MemoryUpdateCompleted ： 表示内存更新完成

二、计量指标

了解 Agno 中的代理运行和会话指标
当您在 Agno 中运行代理时，您获得的响应 （RunResponse） 包括有关运行的详细指标。这些指标可帮助您了解资源使用情况（如令牌使用情况和时间 ）、性能以及模型和工具调用的其他方面。

指标在多个级别可用：

• 每条消息 ：每条消息（助手、工具等）都有自己的量度。
• 按工具调用 ：每个工具执行都有自己的指标。
• Aggregated：RunResponse 聚合运行中所有消息的指标。

1. 示例用法

假设您有一个执行某些任务的代理，并且您希望在运行该代理后分析指标。以下是访问和打印指标的方法：
您运行以下代码来创建代理，并使用以下配置运行它：

from typing import Iterator

from agno.agent import Agent, RunResponse
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from rich.pretty import pprint

agent = Agent(
    model=Gemini(id="gemini-2.0-flash-001"),
    tools=[YFinanceTools(stock_price=True)],
    markdown=True,
    show_tool_calls=True,
)

agent.print_response(
    "What is the stock price of NVDA", stream=True
)

# Print metrics per message
# 打印每条消息的 度量指标
if agent.run_response.messages:
    for message in agent.run_response.messages:
        if message.role == "assistant":
            if message.content:
                print(f"Message: {message.content}")
            elif message.tool_calls:
                print(f"Tool calls: {message.tool_calls}")
            print("---" * 5, "Metrics", "---" * 5)
            pprint(message.metrics)
            print("---" * 20)

# Print the aggregated metrics for the whole run
# 为整个运行的所有请求打印 聚合指标
print("---" * 5, "Collected Metrics", "---" * 5)
pprint(agent.run_response.metrics)
# Print the aggregated metrics for the whole session
# 为整个会话打印指标
print("---" * 5, "Session Metrics", "---" * 5)
pprint(agent.session_metrics)

2. 工具执行指标

本节提供每个工具执行的指标。它包括有关各个工具调用的资源使用情况和性能的详细信息。

3. 消息指标

在这里，您可以查看来自代理的每条消息响应的指标。所有 “assistant” 响应都将具有这样的指标，帮助您了解消息级别的性能和资源使用情况。

4. 聚合运行指标

聚合指标提供了整个运行的全面视图。这包括所有消息和工具调用的摘要，让您全面了解代理的性能和资源使用情况。

同样，对于会话指标，您可以查看会话中所有运行的聚合指标，从而深入了解代理在多次运行中的整体性能和资源使用情况。

指标是如何聚合的

每条消息 ：每条消息（助手、工具等）都有自己的 metrics 对象。
Run-level：RunResponse.metrics 是一个字典，其中每个键（例如 input_tokens）映射到运行中所有助理消息的值列表。
会话级别 ：SessionMetrics（请参阅 agent.session_metrics）聚合会话中所有运行的指标。

消息度量参数

• input_tokens 模型的提示/输入中的token数。
• output_tokens 模型作为输出生成的token数。
• total_tokens: 使用的令牌总数（输入 + 输出）。
• prompt_tokens: 提示中的 Tokens（与 OpenAI 中的 input_tokens 相同）。
• completion_tokens: 完成中的 Tokens （与 OpenAI 中的 output_tokens 相同）。
• audio_tokens: 音频令牌总数（如果使用音频输入/输出）。
• input_audio_tokens: 输入中的 Audio Tokens。
• output_audio_tokens 音频令牌。
  …

三、会话

1. 基础概念

当我们调用 Agent.run（） 时，它会创建一个无状态的单一 Agent 运行。
但是，如果我们想继续这个运行，即进行多轮对话怎么办？这就是 sessions 的用武之地。会话是连续运行的集合。
在实践中，会话是用户和 Agent 之间的多轮对话。使用 session_id，我们可以连接多次运行的对话历史记录和状态。

以下是一些相关的关键组件

• Session : 会话是连续运行的集合，就像用户和 Agent 之间的多轮对话一样。会话由 session_id 标识，每个轮次都是一次运行 。
• Run: 与 Agent 的每次交互（即聊天或轮次）都称为 run。运行由 run_id 标识，Agent.run（） 在调用时创建新 run_id。
• Messages: 是在模型和 Agent 之间发送的单个消息。消息是 Agent 和模型之间的通信协议。

让我们从一个使用 Agent 创建单个运行的示例开始。系统会自动生成一个 run_id 和一个 session_id （因为我们没有提供一个来继续对话）。此运行尚未与用户关联。

from typing import Iterator
from agno.agent import Agent, RunResponse
from agno.models.openai import OpenAIChat
from agno.utils.pprint import pprint_run_response

agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))

# Run agent and return the response as a variable
agent.print_response("Tell me a 5 second short story about a robot")

2. 多用户、多会话代理

与 Agent 交互的每个用户都会获得一组唯一的会话，您可以让多个用户同时与同一个 Agent 交互。
设置 user_id 以将用户连接到其与 Agent 的会话。
在下面的示例中，我们设置了一个 session_id 来演示如何同时与多个用户进行多轮次对话。在生产环境中，session_id 是自动生成的。

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.memory.v2 import Memory

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # Multi-user, multi-session only work with Memory.v2
    memory=Memory(),
    add_history_to_messages=True,
    num_history_runs=3,
)

user_1_id = "user_101"
user_2_id = "user_102"

user_1_session_id = "session_101"
user_2_session_id = "session_102"

# user1 开始会话
agent.print_response(
    "Tell me a 5 second short story about a robot.",
    user_id=user_1_id,
    session_id=user_1_session_id,
)
# user1继续会话
agent.print_response("Now tell me a joke.", user_id=user_1_id, session_id=user_1_session_id)

# user2开始会话
agent.print_response("Tell me about quantum physics.", user_id=user_2_id, session_id=user_2_session_id)
# user2继续会话
agent.print_response("What is the speed of light?", user_id=user_2_id, session_id=user_2_session_id)

# 总结user1的session的历史会话
agent.print_response(
    "Give me a summary of our conversation.",
    user_id=user_1_id,
    session_id=user_1_session_id,
)

3. 从最近 N 个会话中获取消息

在某些情况下，您可能希望从最近 N 个会话中获取消息，以便在对话中提供上下文或连续性。
以下是如何实现此目的的示例：

import os

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.sqlite import SqliteStorage

os.remove("tmp/data.db")

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    user_id="user_1",
    storage=SqliteStorage(table_name="agent_sessions_new", db_file="tmp/data.db"),
    search_previous_sessions_history=True,  # 允许搜索历史会话
    num_history_sessions=2,  # 仅包含会话中最近的两条会话
    show_tool_calls=True,
)

session_1_id = "session_1_id"
session_2_id = "session_2_id"
session_3_id = "session_3_id"
session_4_id = "session_4_id"
session_5_id = "session_5_id"

agent.print_response("What is the capital of South Africa?", session_id=session_1_id)
agent.print_response("What is the capital of China?", session_id=session_2_id)
agent.print_response("What is the capital of France?", session_id=session_3_id)
agent.print_response("What is the capital of Japan?", session_id=session_4_id)
agent.print_response(
    "What did I discuss in my previous conversations?", session_id=session_5_id
)  # 仅包含会话中的最近两条