agno 接入 Nebius 模型实战环境配置、流式输出、结构化返回与工具调用的完整指南【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoNebius 是 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_factorylambda: 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中的任意一个。启动前确保已安装 agno 核心库agno包位于 libs/agno/agno已设置NEBIUS_API_KEY环境变量使用知识库与会话存储示例时需安装ddgs、sqlalchemy、pgvector、pypdf等依赖见下文对应章节。基础用法同步、异步与流式对话Nebius Basic 示例 展示了最核心的四种调用方式。先创建一个最简 Agentfrom agno.agent import Agent from agno.models.nebius import Nebius agent Agent( modelNebius(), markdownTrue, )markdownTrue会让 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, streamTrue) # --- 异步调用 --- asyncio.run(agent.aprint_response(write a two sentence horror story)) # --- 异步 流式 --- asyncio.run(agent.aprint_response(write a two sentence horror story, streamTrue))四种模式对照模式方法适用场景同步print_response(...)简单脚本、命令行演示同步 流式print_response(..., streamTrue)终端逐步输出、实时反馈异步asyncio.run(agent.aprint_response(...))Web 后端、并发任务异步 流式aprint_response(..., streamTrue)高吞吐服务、SSE 推送由于 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( modelNebius(idQwen/Qwen3-30B-A3B), tools[WebSearchTools()], markdownTrue, )运行部分与 basic 一致覆盖同步、同步流式、异步、异步流式四种模式if __name__ __main__: # --- Sync --- agent.print_response(Whats happening in France?) # --- Sync Streaming --- agent.print_response(Whats happening in France?, streamTrue) # --- Async --- asyncio.run(agent.aprint_response(Whats happening in France?)) # --- Async Streaming --- asyncio.run(agent.aprint_response(Whats happening in France?, streamTrue))要点WebSearchTools()依赖ddgs包示例文件头部注释明确要求uv pip install ddgs ...使用前请先安装模型需要具备函数调用tool calling能力才能驱动工具Nebius 提供的 Qwen 系列模型已支持该能力除WebSearchTools外你也可以传入 agno 内置的其他工具集见 cookbook/91_tools例如文件工具、计算器、日历等用法完全一致。结构化输出用 Pydantic 约束模型返回 JSONStructured 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( ..., descriptionProvide a nice setting for a blockbuster movie. ) ending: str Field( ..., descriptionEnding of the movie. If not available, provide a happy ending., ) genre: str Field( ..., descriptionGenre of the movie. If not available, select action, thriller or romantic comedy., ) name: str Field(..., descriptionGive a name to this movie) characters: List[str] Field(..., descriptionName of characters for this movie.) storyline: str Field( ..., description3 sentence storyline for the movie. Make it exciting! )然后将output_schema传给 Agentstructured_output_agent Agent( modelNebius(idQwen/Qwen3-30B-A3B), descriptionYou are a helpful assistant. Summarize the movie script based on the location in a JSON object., output_schemaMovieScript, ) 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便于调试时美化打印若模型返回内容不符合 schemaagno 会引导模型自我修正保证下游拿到的是合法 JSON。知识库检索Nebius pgvector 实现 RAGKnowledge 示例 将 Nebius 模型与KnowledgePgVector组合实现检索增强生成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 postgresqlpsycopg://ai:ailocalhost:5532/ai knowledge Knowledge( vector_dbPgVector(table_namerecipes, db_urldb_url), ) # Add content to the knowledge knowledge.insert(urlhttps://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf) agent Agent(modelNebius(idQwen/Qwen3-30B-A3B), knowledgeknowledge) agent.print_response(How to make Thai curry?, markdownTrue)工作流程拆解PgVector(table_namerecipes, db_urldb_url)在 PostgreSQL启用 pgvector 扩展中创建向量表recipesknowledge.insert(url...)拉取远程 PDF 并切分、向量化后写入向量库此处pypdf负责解析 PDFAgent 回答 How to make Thai curry? 时会先从知识库检索相关片段再交给 Nebius 模型生成基于检索结果的答案db_url中postgresqlpsycopg://ai:ailocalhost: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 postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url) agent Agent( modelNebius(), dbdb, tools[WebSearchTools()], add_history_to_contextTrue, ) agent.print_response(How many people live in Canada?) agent.print_response(What is their national anthem called?)这里体现了一个典型的多轮对话设计dbdb指定PostgresDb作为会话与消息存储对话历史会被落库add_history_to_contextTrue将历史消息注入到每轮请求的上下文中使第二问 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( modelNebius( idwrong_model_id, retries3, # 请求重试次数 delay_between_retries1, # 每次重试间的延迟秒 exponential_backoffTrue, # 为 True 时延迟随重试次数指数翻倍 ), ) agent.print_response(What is the capital of France?)三个重试参数的行为可在 agno 模型基类 libs/agno/agno/models/base.py 中找到权威定义参数默认值含义retries0请求失败后的最大重试次数delay_between_retries1相邻两次重试之间的基础延迟秒exponential_backoffFalse若为True第attempt次重试的延迟为delay_between_retries * (2 ** attempt)即每次翻倍从源码看指数退避的延迟计算逻辑为delay_between_retries * (2**attempt)第 1 次重试延迟 1 秒、第 2 次 2 秒、第 3 次 4 秒……这能有效避免在服务限流或瞬时故障时对端点造成请求风暴。生产环境中建议开启exponential_backoffTrue并依据你的服务可用性预算设置合理的retries次数示例中的retries3是常用实践。小结与进阶路径本文围绕 cookbook/90_models/nebius 的完整示例覆盖了 Nebius 模型的六个实战维度基础对话同步/异步/流式、工具调用、结构化输出、知识库 RAG、会话持久化与重试策略并下钻到 nebius.py 源码验证了NEBIUS_API_KEY注入、默认端点https://api.tokenfactory.nebius.com/v1/、OpenAILike兼容协议等实现细节。你可以按以下路径继续深入若想接入其他 OpenAI 兼容模型参考同目录下的其他模型 Cookbookcookbook/90_models切换模型只需替换modelNebius(...)一处若想扩展工具能力浏览 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
agno 接入 Nebius 模型实战：环境配置、流式输出、结构化返回与工具调用的完整指南
agno 接入 Nebius 模型实战环境配置、流式输出、结构化返回与工具调用的完整指南【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoNebius 是 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_factorylambda: 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中的任意一个。启动前确保已安装 agno 核心库agno包位于 libs/agno/agno已设置NEBIUS_API_KEY环境变量使用知识库与会话存储示例时需安装ddgs、sqlalchemy、pgvector、pypdf等依赖见下文对应章节。基础用法同步、异步与流式对话Nebius Basic 示例 展示了最核心的四种调用方式。先创建一个最简 Agentfrom agno.agent import Agent from agno.models.nebius import Nebius agent Agent( modelNebius(), markdownTrue, )markdownTrue会让 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, streamTrue) # --- 异步调用 --- asyncio.run(agent.aprint_response(write a two sentence horror story)) # --- 异步 流式 --- asyncio.run(agent.aprint_response(write a two sentence horror story, streamTrue))四种模式对照模式方法适用场景同步print_response(...)简单脚本、命令行演示同步 流式print_response(..., streamTrue)终端逐步输出、实时反馈异步asyncio.run(agent.aprint_response(...))Web 后端、并发任务异步 流式aprint_response(..., streamTrue)高吞吐服务、SSE 推送由于 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( modelNebius(idQwen/Qwen3-30B-A3B), tools[WebSearchTools()], markdownTrue, )运行部分与 basic 一致覆盖同步、同步流式、异步、异步流式四种模式if __name__ __main__: # --- Sync --- agent.print_response(Whats happening in France?) # --- Sync Streaming --- agent.print_response(Whats happening in France?, streamTrue) # --- Async --- asyncio.run(agent.aprint_response(Whats happening in France?)) # --- Async Streaming --- asyncio.run(agent.aprint_response(Whats happening in France?, streamTrue))要点WebSearchTools()依赖ddgs包示例文件头部注释明确要求uv pip install ddgs ...使用前请先安装模型需要具备函数调用tool calling能力才能驱动工具Nebius 提供的 Qwen 系列模型已支持该能力除WebSearchTools外你也可以传入 agno 内置的其他工具集见 cookbook/91_tools例如文件工具、计算器、日历等用法完全一致。结构化输出用 Pydantic 约束模型返回 JSONStructured 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( ..., descriptionProvide a nice setting for a blockbuster movie. ) ending: str Field( ..., descriptionEnding of the movie. If not available, provide a happy ending., ) genre: str Field( ..., descriptionGenre of the movie. If not available, select action, thriller or romantic comedy., ) name: str Field(..., descriptionGive a name to this movie) characters: List[str] Field(..., descriptionName of characters for this movie.) storyline: str Field( ..., description3 sentence storyline for the movie. Make it exciting! )然后将output_schema传给 Agentstructured_output_agent Agent( modelNebius(idQwen/Qwen3-30B-A3B), descriptionYou are a helpful assistant. Summarize the movie script based on the location in a JSON object., output_schemaMovieScript, ) 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便于调试时美化打印若模型返回内容不符合 schemaagno 会引导模型自我修正保证下游拿到的是合法 JSON。知识库检索Nebius pgvector 实现 RAGKnowledge 示例 将 Nebius 模型与KnowledgePgVector组合实现检索增强生成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 postgresqlpsycopg://ai:ailocalhost:5532/ai knowledge Knowledge( vector_dbPgVector(table_namerecipes, db_urldb_url), ) # Add content to the knowledge knowledge.insert(urlhttps://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf) agent Agent(modelNebius(idQwen/Qwen3-30B-A3B), knowledgeknowledge) agent.print_response(How to make Thai curry?, markdownTrue)工作流程拆解PgVector(table_namerecipes, db_urldb_url)在 PostgreSQL启用 pgvector 扩展中创建向量表recipesknowledge.insert(url...)拉取远程 PDF 并切分、向量化后写入向量库此处pypdf负责解析 PDFAgent 回答 How to make Thai curry? 时会先从知识库检索相关片段再交给 Nebius 模型生成基于检索结果的答案db_url中postgresqlpsycopg://ai:ailocalhost: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 postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url) agent Agent( modelNebius(), dbdb, tools[WebSearchTools()], add_history_to_contextTrue, ) agent.print_response(How many people live in Canada?) agent.print_response(What is their national anthem called?)这里体现了一个典型的多轮对话设计dbdb指定PostgresDb作为会话与消息存储对话历史会被落库add_history_to_contextTrue将历史消息注入到每轮请求的上下文中使第二问 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( modelNebius( idwrong_model_id, retries3, # 请求重试次数 delay_between_retries1, # 每次重试间的延迟秒 exponential_backoffTrue, # 为 True 时延迟随重试次数指数翻倍 ), ) agent.print_response(What is the capital of France?)三个重试参数的行为可在 agno 模型基类 libs/agno/agno/models/base.py 中找到权威定义参数默认值含义retries0请求失败后的最大重试次数delay_between_retries1相邻两次重试之间的基础延迟秒exponential_backoffFalse若为True第attempt次重试的延迟为delay_between_retries * (2 ** attempt)即每次翻倍从源码看指数退避的延迟计算逻辑为delay_between_retries * (2**attempt)第 1 次重试延迟 1 秒、第 2 次 2 秒、第 3 次 4 秒……这能有效避免在服务限流或瞬时故障时对端点造成请求风暴。生产环境中建议开启exponential_backoffTrue并依据你的服务可用性预算设置合理的retries次数示例中的retries3是常用实践。小结与进阶路径本文围绕 cookbook/90_models/nebius 的完整示例覆盖了 Nebius 模型的六个实战维度基础对话同步/异步/流式、工具调用、结构化输出、知识库 RAG、会话持久化与重试策略并下钻到 nebius.py 源码验证了NEBIUS_API_KEY注入、默认端点https://api.tokenfactory.nebius.com/v1/、OpenAILike兼容协议等实现细节。你可以按以下路径继续深入若想接入其他 OpenAI 兼容模型参考同目录下的其他模型 Cookbookcookbook/90_models切换模型只需替换modelNebius(...)一处若想扩展工具能力浏览 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考