Skip to content

第 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 成本控制四板斧

  1. 分级路由(见 24.1)——通常能砍掉 50% 以上成本;
  2. 控制上下文add_history_to_context=True 时用 num_history_runs 限制带入的历史轮数,避免会话越长越贵;
  3. 限制工具循环:给 Agent 设 max_tool_calls 类上限、给 Team 的 Tasks 模式设 max_iterations,防止单次运行失控;
  4. 用 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 存会话,上生产时官方建议?

🛠️ 动手实践

  1. 为第 20 章的写作流水线接入 AccuracyEval:准备 5 个主题与预期要点,跑通一次评估并接入 GitHub Actions。
  2. asyncio.gather + Semaphore(3) 并发分析 10 个主题,对比串行与并行的总耗时(用 metrics.duration 记录)。
  3. 按本章 checklist 逐项检查你在前面章节写的项目,把未达标项整理成一份改造任务清单。

🎉 恭喜完成 Agno 教程全部 24 章!接下来可以进入 CrewAI 教程,对比两大多智能体框架的设计哲学。