第 3 章 · 运行 Agent 与 RunOutput 解析
本章目标：掌握
run()/arun()/print_response()三种运行方式的区别，学会从RunOutput中取出内容、消息与指标，理解多轮对话的正确姿势。
3.1 三种运行方式
Agno 官方文档对运行方式的建议很明确：开发调试用 print_response()，生产代码用 run() 或 arun()。
# run_basics.py
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
load_dotenv()
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
)
# 方式一：开发时直接打印，人类可读，但拿不到返回值
agent.print_response("用一句话解释依赖注入")
# 方式二：同步运行，返回 RunOutput 对象——业务代码用这个
response = agent.run("用一句话解释闭包")
print(response.content) # 真正的回答文本在这里
# 方式三：异步运行（协程），适合高并发服务端
import asyncio
async def main():
resp = await agent.arun("用一句话解释装饰器")
print(resp.content)
asyncio.run(main())print_response 内部其实也调用了 run，只是额外负责格式化打印。写单元测试或 Web 接口时必须用 run/arun，因为你要对返回值做断言或加工。
3.2 RunOutput 核心字段
run() 的返回值是一个 RunOutput 对象，官方定义的核心属性如下：
| 字段 | 含义 |
|---|---|
content | 最终回答内容（字符串；结构化输出时是 Pydantic 对象） |
messages | 本次发送给模型的完整消息列表（含 system/工具消息） |
metrics | token 用量、耗时等指标 |
run_id / session_id | 本次运行的唯一 ID 与所属会话 ID |
agent_id / agent_name | 运行该次请求的 Agent 标识 |
reasoning_content | 思考型模型的推理过程文本 |
# inspect_output.py —— 把 RunOutput 拆开看
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
load_dotenv()
agent = Agent(model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
))
r = agent.run("什么是幂等性？")
print("回答:", r.content[:80], "...")
print("run_id:", r.run_id)
print("消息数:", len(r.messages))
for m in r.messages:
print("-", m.role, ":", str(m.content)[:50])
print("输入 tokens:", r.metrics.input_tokens, "/ 输出 tokens:", r.metrics.output_tokens)metrics 有什么用
metrics 是成本核算的基础：把每次运行的 token 数记录下来，月底就能按用户/功能维度统计开销。第 23 章的可观测性会系统展开。
3.3 多轮对话：为什么第二次运行"失忆"了
先看一个新手必踩的坑：
# memory_trap.py —— 演示默认无记忆
agent = Agent(model=make_model()) # make_model 见 2.3 节写法
agent.run("我叫高小灵，最喜欢的数字是 7")
resp = agent.run("我的名字是什么？") # ❌ 模型答不出来！
print(resp.content)两次 run() 之间没有任何共享状态：每次调用都是独立的全新上下文，模型根本不知道上一次说过什么。解决办法有三种，复杂度递增：
- 手动拼接历史：自己把上一问一答加进下一次输入；
- chat history：让 Agno 自动携带同会话的历史运行（需配合存储，第 10–11 章）;
- Memory：跨会话的用户级记忆（第 12 章）。
本节先用最朴素的手动方式打通概念：
# manual_history.py —— 手动维护对话历史
history = []
def chat(question: str) -> str:
prompt = ""
for q, a in history: # 把历史拼进提示词
prompt += f"用户: {q}\n助手: {a}\n"
prompt += f"用户: {question}"
resp = agent.run(prompt)
history.append((question, resp.content)) # 记录本轮结果
return resp.content
chat("我最喜欢的颜色是蓝色")
print(chat("我喜欢什么颜色？")) # ✅ 能答出蓝色这种写法的问题显而易见：提示词越来越长、token 成本线性上涨、无法持久化。所以它只适合理解原理，生产环境请使用第 10 章的会话机制。
3.4 异步并发运行多个 Agent
arun 的价值在并发场景才能体现：三个独立任务串行要等三次网络往返，异步并发只需最慢一次的时间：
# concurrent_runs.py —— asyncio.gather 并发三个请求
import asyncio
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
load_dotenv()
agent = Agent(model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1"),
)
async def main():
questions = ["解释线程与进程的区别", "解释TCP与UDP的区别", "解释GET与POST的区别"]
# 三个请求同时发出，gather 收集全部结果
results = await asyncio.gather(*(agent.arun(q) for q in questions))
for q, r in zip(questions, results):
print(f"[{q}] -> {r.content[:40]}...")
asyncio.run(main())注意 asyncio.gather 里的生成器表达式：为每个问题创建一个协程。如果某个请求失败会导致整个 gather 抛异常，生产中可用 return_exceptions=True 单独处理失败项。
3.5 本章小结
- 开发调试用
print_response，业务代码用run，并发场景用arun； RunOutput的关键信息：content拿答案、messages看上下文、metrics算成本；- 默认情况下每次
run()相互独立，模型没有跨调用记忆； - 手动拼历史能实现多轮但成本高，正确方案是会话存储 + chat history（第 10–11 章）。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 生产环境的 Web 服务中处理 Agent 请求，推荐使用哪种方式？
2. 连续两次 agent.run() 后，第二次提问涉及第一次的内容，模型却答不上来，原因是？
3. 想统计某次运行消耗了多少 token，应该访问 RunOutput 的哪个字段？
4. 关于 messages 字段，说法正确的是？
🛠️ 动手实践
- 编写脚本分别用
run和arun提同一个问题，打印两者的run_id和metrics，确认它们是相互独立的运行。 - 扩展 3.3 的
manual_history.py：限制历史最多保留 3 轮，超出就丢弃最早的记录，观察回答质量变化。 - 用
asyncio.gather并发询问 5 个互不相关的问题，记录总耗时，再改为串行执行对比时间差。
掌握了同步运行后，进入第 4 章：流式输出与实时响应。