第 24 章 · 性能优化与生产实践
本章目标:从模型选型、并发吞吐、成本控制到评估回归与容器化部署,掌握把 Agno 应用推向生产的完整清单。
24.1 模型分级路由:让合适的模型干合适的活
生产成本的大头是"用大模型干了小模型的活"。按任务难度分级:
python
import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
def make_model(model_id: str) -> OpenAIChat:
"""统一的三方模型工厂:所有档位走同一个 OpenAI 兼容端点"""
return OpenAIChat(
id=model_id,
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
)
# 分类/抽取/格式化:小模型,快且便宜
classifier = Agent(
name="Classifier",
model=make_model("deepseek-chat"),
output_schema=Category, # 结构化输出天然适合小模型
instructions=["把用户请求分类为 question/complaint/other"],
)
# 复杂推理/创作:大模型或思考模型
analyst = Agent(
name="Analyst",
model=make_model("deepseek-reasoner"),
instructions=["对用户投诉做根因分析并给出处理建议"],
)配合第 21 章的 Router/Condition,就能实现"小模型先分类、大模型只接难活"的漏斗结构。Team 中还可以只给 leader 配大模型,成员继承或显式用小模型。
24.2 并发吞吐:asyncio.gather + arun
Agent 级并发用 arun() 配合标准 asyncio:
python
import asyncio
async def analyze(symbol: str) -> str:
result = await analyst.arun(f"分析 {symbol} 最近一周的走势风险")
return f"{symbol}: {result.content}"
async def main():
symbols = ["AAPL", "NVDA", "TSLA", "AMZN"]
# 并发跑 4 个分析,总耗时≈最慢一个,而非四者之和
results = await asyncio.gather(*[analyze(s) for s in symbols])
print("\n".join(results))
asyncio.run(main())Workflow 层面则用 Parallel 步骤自动并发(见第 21 章)。注意并发上限要与服务商限流匹配——用 asyncio.Semaphore 控制同时在飞的请求数。
24.3 成本控制四板斧
- 分级路由(见 24.1)——通常能砍掉 50% 以上成本;
- 控制上下文:
add_history_to_context=True时用num_history_runs限制带入的历史轮数,避免会话越长越贵; - 限制工具循环:给 Agent 设
max_tool_calls类上限、给 Team 的 Tasks 模式设max_iterations,防止单次运行失控; - 用 metrics 做预算门禁:每次运行后检查
metrics.total_tokens,超预算时降级到小模型或直接熔断。
python
result = agent.run("长任务...")
if result.metrics and result.metrics.total_tokens > 50_000:
# 记录并告警:单次运行 token 超过预算
print("WARN: token 超预算", result.metrics.total_tokens)24.4 评估(Evals):把质量变成可回归的测试
改提示词、换模型、动工具之后,你怎么知道没把产品改坏?Agno 官方的 Evals 把预期行为变成可重复执行的检查,可以在开发和 CI 里跑:
python
from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval
evaluation = AccuracyEval(
name="calculator-accuracy",
model=make_model("deepseek-chat"),
agent=calc_agent,
input="10 的阶乘是多少?",
expected_output="3628800",
)
result = evaluation.run(print_results=True)
assert result is not None and result.avg_score >= 8四类评估按需选用:
| 问题 | 评估类型 |
|---|---|
| 回复是否匹配预期答案 | Accuracy |
| 是否满足自定义质量标准 | Agent as Judge |
| 是否调用了预期的工具与参数 | Reliability |
| 代码耗时与内存表现 | Performance |
多个用例组合成 Eval Suite,接入 CI 后,每次提交都自动跑一遍智能体回归。
24.5 Docker 化部署 AgentOS
AgentOS 就是一个 FastAPI 应用,容器化非常直接:
dockerfile
FROM python:3.12-slim AS base
WORKDIR /app
# 先装依赖,利用层缓存
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 7777
# 生产不用 reload;worker 数按 CPU 调整
CMD ["uvicorn", "agent_os:app", "--host", "0.0.0.0", "--port", "7777", "--workers", "2"]yaml
# docker-compose.yml 关键片段
services:
agentos:
build: .
environment:
- DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY} # 密钥走环境变量,绝不进镜像
volumes:
- agent-data:/app/tmp # SqliteDb 文件持久化
db:
image: postgres:16 # 生产建议 PostgresDb 替换 SQLite
volumes:
agent-data:生产要点:
- 密钥管理:API Key 只通过环境变量或密钥管理服务注入,不写死在代码或镜像里;
- 数据库升级:SQLite 单文件适合开发,生产换
PostgresDb(会话、记忆、追踪同库或分库); - 开启鉴权与追踪:
authorization=True+tracing=True(第 22、23 章); - 健康检查:对
/info端点配置存活探测。
24.6 上线 Checklist
上线前逐项打勾:
- [ ] 所有 Agent/Team/Workflow 都设置了显式
id - [ ] 三方模型的 Key 走环境变量,仓库中无明文密钥
- [ ] AgentOS 已开启
authorization=True,公网入口有 HTTPS 与限流 - [ ] 会话/记忆使用 PostgresDb 等生产级存储,并有备份策略
- [ ]
tracing=True且追踪写入独立数据库,保留策略明确 - [ ] 核心链路有 Eval Suite 并接入 CI,改动可回归
- [ ] 每个运行路径都有
max_iterations/token 预算等失控保护 - [ ] 监控覆盖成本(tokens)、延迟(P95)、质量(工具失败率)三张看板
- [ ] 用小流量灰度验证后再放量
24.7 本章小结
- 模型分级路由是性价比最高的优化:分类抽取用小模型,推理创作用大/思考模型;
- Agent 并发用
arun()+asyncio.gather(配 Semaphore 限流),Workflow 用Parallel; - 成本控制 = 分级 + 控历史轮数 + 限迭代 + metrics 预算门禁;
- Evals(Accuracy/Judge/Reliability/Performance)把质量变成 CI 里可回归的断言;
- 容器化部署 AgentOS 与部署 FastAPI 无异:密钥环境变量化、SQLite 换 Postgres、开鉴权与追踪。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 「先让小模型把请求分类,再决定是否交给大模型处理」,最适合的实现组合是?
2. 想验证「换新提示词后工具调用行为没有退化」,应该用哪类评估?
3. 关于生产环境中的 API Key 管理,正确做法是?
4. 开发期用 SQLite 存会话,上生产时官方建议?
🛠️ 动手实践
- 为第 20 章的写作流水线接入
AccuracyEval:准备 5 个主题与预期要点,跑通一次评估并接入 GitHub Actions。 - 用
asyncio.gather+Semaphore(3)并发分析 10 个主题,对比串行与并行的总耗时(用metrics.duration记录)。 - 按本章 checklist 逐项检查你在前面章节写的项目,把未达标项整理成一份改造任务清单。
🎉 恭喜完成 Agno 教程全部 24 章!接下来可以进入 CrewAI 教程,对比两大多智能体框架的设计哲学。