企业官网建设流程全解析

agno 接入 Nebius 模型实战：环境配置、流式输出、结构化返回与工具调用的完整指南

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

Nebius 是 agno 官方支持的第三方模型提供商之一，可通过 OpenAI 兼容协议（OpenAILike）直接接入。本文以 cookbook/90_models/nebius 目录下的完整示例为核心，系统讲解在 agno 中配置 Nebius 模型、运行同步/异步与流式对话、调用联网工具、输出结构化 JSON、挂载知识库与会话存储，以及配置重试策略的完整方案；同时结合 Nebius 模型源码 揭示其底层实现原理。读完本文，你将能够在自己的 agno 项目中以 Nebius 作为后端模型，快速搭建从"最简 Agent"到"带知识库与持久化存储的生产级 Agent"的完整链路。

Nebius 模型在 agno 中的定位与源码实现

在开始写代码之前，先理解 Nebius 模型在 agno 中的实现方式，这有助于你正确使用它的默认行为与扩展能力。

从 libs/agno/agno/models/nebius/nebius.py 可以看到，Nebius是一个继承自OpenAILike的数据类（dataclass），其关键默认值如下：

@dataclass class Nebius(OpenAILike): id: str = "openai/gpt-oss-20b" # 默认聊天模型 ID name: str = "Nebius" provider: str = "Nebius" api_key: Optional[str] = field(default_factory=lambda: getenv("NEBIUS_API_KEY")) base_url: str = "https://api.tokenfactory.nebius.com/v1/"

要点说明：

• API Key 通过环境变量注入：api_key默认从环境变量NEBIUS_API_KEY读取。若未设置该环境变量，调用_get_client_params()时会抛出ModelAuthenticationError（提示 "NEBIUS_API_KEY not set..."），因此运行示例前务必先export NEBIUS_API_KEY=你的密钥。
• OpenAI 兼容端点：base_url指向 Nebius Token Factory 的/v1/端点，由于继承自OpenAILike，底层使用 OpenAI SDK 的客户端参数体系（api_key、organization、base_url、timeout、max_retries、default_headers、default_query等），并支持通过client_params追加自定义客户端参数。
• 模型 ID 可覆盖：源码中的默认id是openai/gpt-oss-20b；而目录内的示例实际多采用Qwen/Qwen3-30B-A3B（见 knowledge.py、structured_output.py、tool_use.py）。这说明创建Nebius(id="...")时可以自由指定你订阅的任意模型 ID。

环境准备与示例运行方式

官方 README 给出了运行该目录下任意示例的统一命令（需要先按 scripts/dev_setup.sh 等脚本准备好虚拟环境）：

.venvs/demo/bin/python cookbook/90_models/nebius/<example>.py

其中<example>.py可替换为basic.py、tool_use.py、structured_output.py、knowledge.py、db.py、retry.py中的任意一个。启动前确保：

1. 已安装 agno 核心库（agno包位于 libs/agno/agno）；
2. 已设置NEBIUS_API_KEY环境变量；
3. 使用知识库与会话存储示例时，需安装ddgs、sqlalchemy、pgvector、pypdf等依赖（见下文对应章节）。

基础用法：同步、异步与流式对话

Nebius Basic 示例 展示了最核心的四种调用方式。先创建一个最简 Agent：

from agno.agent import Agent from agno.models.nebius import Nebius agent = Agent( model=Nebius(), markdown=True, )

markdown=True会让 Agent 以 Markdown 格式输出回答，便于终端阅读。随后在一个if __name__ == "__main__":块中依次演示四种执行模式：

if __name__ == "__main__": # --- 同步调用 --- agent.print_response("write a two sentence horror story") # --- 同步 + 流式 --- agent.print_response("write a two sentence horror story", stream=True) # --- 异步调用 --- asyncio.run(agent.aprint_response("write a two sentence horror story")) # --- 异步 + 流式 --- asyncio.run(agent.aprint_response("write a two sentence horror story", stream=True))

四种模式对照：

由于 Nebius 底层走 OpenAI 兼容协议，print_response/aprint_response的参数（如stream、markdown）与 agno 其他模型一致，切换到 Nebius 时无需改动调用代码。

工具调用：为 Nebius Agent 挂载联网搜索

Tool Use 示例 演示了让 Nebius 驱动的 Agent 调用WebSearchTools完成实时信息检索（例如查询法国正在发生的事件）：

import asyncio from agno.agent import Agent from agno.models.nebius import Nebius from agno.tools.websearch import WebSearchTools agent = Agent( model=Nebius(id="Qwen/Qwen3-30B-A3B"), tools=[WebSearchTools()], markdown=True, )

运行部分与 basic 一致，覆盖同步、同步流式、异步、异步流式四种模式：

if __name__ == "__main__": # --- Sync --- agent.print_response("Whats happening in France?") # --- Sync + Streaming --- agent.print_response("Whats happening in France?", stream=True) # --- Async --- asyncio.run(agent.aprint_response("Whats happening in France?")) # --- Async + Streaming --- asyncio.run(agent.aprint_response("Whats happening in France?", stream=True))

要点：

• WebSearchTools()依赖ddgs包（示例文件头部注释明确要求uv pip install ddgs ...），使用前请先安装；
• 模型需要具备函数调用（tool calling）能力才能驱动工具；Nebius 提供的 Qwen 系列模型已支持该能力；
• 除WebSearchTools外，你也可以传入 agno 内置的其他工具集（见 cookbook/91_tools），例如文件工具、计算器、日历等，用法完全一致。

结构化输出：用 Pydantic 约束模型返回 JSON

Structured Output 示例 展示如何让 Nebius 按预定义的 Pydantic 模型返回结构化 JSON。首先定义输出结构——一个电影剧本大纲：

from typing import List from agno.agent import Agent from agno.models.nebius import Nebius from pydantic import BaseModel, Field class MovieScript(BaseModel): setting: str = Field( ..., description="Provide a nice setting for a blockbuster movie." ) ending: str = Field( ..., description="Ending of the movie. If not available, provide a happy ending.", ) genre: str = Field( ..., description="Genre of the movie. If not available, select action, thriller or romantic comedy.", ) name: str = Field(..., description="Give a name to this movie") characters: List[str] = Field(..., description="Name of characters for this movie.") storyline: str = Field( ..., description="3 sentence storyline for the movie. Make it exciting!" )

然后将output_schema传给 Agent：

structured_output_agent = Agent( model=Nebius(id="Qwen/Qwen3-30B-A3B"), description="You are a helpful assistant. Summarize the movie script based on the location in a JSON object.", output_schema=MovieScript, ) structured_output_agent.print_response("New York")

使用要点：

• Field(..., description=...)的作用：Pydantic 字段描述会作为模型生成 JSON 的提示词约束，描述写得越明确，字段内容越贴合预期；
• 字段类型即约束：List[str]会强制模型输出字符串数组，setting/ending/genre等必填字段必须全部返回，模型输出会被 agno 校验并反序列化为MovieScript实例；
• output_schema与response_model：此处使用output_schema声明输出结构，agno 同时支持RunOutput/response_model等结构化返回机制（示例头部以# noqa导入RunOutput和rich.pretty.pprint，便于调试时美化打印）；若模型返回内容不符合 schema，agno 会引导模型自我修正，保证下游拿到的是合法 JSON。

知识库检索：Nebius + pgvector 实现 RAG

Knowledge 示例 将 Nebius 模型与Knowledge+PgVector组合，实现检索增强生成（RAG）。前置依赖见文件头部注释：

uv pip install ddgs sqlalchemy pgvector pypdf

核心代码如下：

from agno.agent import Agent from agno.knowledge.knowledge import Knowledge from agno.models.nebius import Nebius from agno.vectordb.pgvector import PgVector db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" knowledge = Knowledge( vector_db=PgVector(table_name="recipes", db_url=db_url), ) # Add content to the knowledge knowledge.insert(url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf") agent = Agent(model=Nebius(id="Qwen/Qwen3-30B-A3B"), knowledge=knowledge) agent.print_response("How to make Thai curry?", markdown=True)

工作流程拆解：

1. PgVector(table_name="recipes", db_url=db_url)在 PostgreSQL（启用 pgvector 扩展）中创建向量表recipes；
2. knowledge.insert(url=...)拉取远程 PDF 并切分、向量化后写入向量库（此处pypdf负责解析 PDF）；
3. Agent 回答 "How to make Thai curry?" 时，会先从知识库检索相关片段，再交给 Nebius 模型生成基于检索结果的答案；
4. db_url中postgresql+psycopg://ai:ai@localhost:5532/ai表示使用 psycopg 驱动连接本地 5532 端口（该端口与仓库 scripts/run_pgvector.sh 的默认映射一致）的ai数据库。

若你本地尚未启动 pgvector，可参考 scripts/run_pgvector.sh 启动配套容器，再运行本示例。

会话持久化：Nebius Agent 的历史记忆与联网回答

DB 示例 演示如何把多轮对话历史持久化到 PostgreSQL，让 Agent 在后续提问中"记住"上一轮的上下文：

from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.nebius import Nebius from agno.tools.websearch import WebSearchTools db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" db = PostgresDb(db_url=db_url) agent = Agent( model=Nebius(), db=db, tools=[WebSearchTools()], add_history_to_context=True, ) agent.print_response("How many people live in Canada?") agent.print_response("What is their national anthem called?")

这里体现了一个典型的多轮对话设计：

• db=db：指定PostgresDb作为会话与消息存储，对话历史会被落库；
• add_history_to_context=True：将历史消息注入到每轮请求的上下文中，使第二问 "What is their national anthem called?" 能正确关联到第一问的主题"加拿大"；
• tools=[WebSearchTools()]：第一问的人口数据属实时信息，Agent 会调用联网搜索获取事实，再组织答案。

重试策略：应对瞬时故障与限流

Retry 示例 演示了 Nebius 请求失败时的自动重试配置。示例特意使用一个故意写错的模型 ID来触发重试：

from agno.agent import Agent from agno.models.nebius import Nebius wrong_model_id = "nebius-wrong-id" agent = Agent( model=Nebius( id=wrong_model_id, retries=3, # 请求重试次数 delay_between_retries=1, # 每次重试间的延迟（秒） exponential_backoff=True, # 为 True 时，延迟随重试次数指数翻倍 ), ) agent.print_response("What is the capital of France?")

三个重试参数的行为可在 agno 模型基类 libs/agno/agno/models/base.py 中找到权威定义：

从源码看，指数退避的延迟计算逻辑为delay_between_retries * (2**attempt)：第 1 次重试延迟 1 秒、第 2 次 2 秒、第 3 次 4 秒……这能有效避免在服务限流或瞬时故障时对端点造成请求风暴。生产环境中建议开启exponential_backoff=True，并依据你的服务可用性预算设置合理的retries次数（示例中的retries=3是常用实践）。

小结与进阶路径

本文围绕 cookbook/90_models/nebius 的完整示例，覆盖了 Nebius 模型的六个实战维度：基础对话（同步/异步/流式）、工具调用、结构化输出、知识库 RAG、会话持久化与重试策略，并下钻到 nebius.py 源码验证了NEBIUS_API_KEY注入、默认端点https://api.tokenfactory.nebius.com/v1/、OpenAILike兼容协议等实现细节。

你可以按以下路径继续深入：

• 若想接入其他 OpenAI 兼容模型，参考同目录下的其他模型 Cookbook（cookbook/90_models），切换模型只需替换model=Nebius(...)一处；
• 若想扩展工具能力，浏览 cookbook/91_tools 中数十种现成工具（如WebSearchTools、文件、数据库、日历等）并直接挂载到 Agent；
• 若想了解Knowledge、PgVector、PostgresDb的更多参数（如会话表名、向量维度、检索 top-k 等），可阅读 cookbook/07_knowledge 与 cookbook/06_storage 的进阶章节。

所有示例均可通过.venvs/demo/bin/python cookbook/90_models/nebius/<example>.py一键运行，快用你的NEBIUS_API_KEY跑通第一条 Nebius Agent 对话吧。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno