From 2d40cec5c7c3ecfa556093ef36c2c4fe28855bfb Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 26 Oct 2026 01:00:45 -0500 Subject: [PATCH 1/2] Add an OpenAI-compatible model provider sample for Baseten, Fireworks AI, and Together AI One Worker script covers all three providers: a provider table holds each endpoint's base URL, API key environment variable, and a default open-weight model with tool calling (gpt-oss-120b everywhere), and an OpenAICompatibleModelProvider turns an entry into a Chat Completions model with client retries off, so Temporal owns the retries. The Workflow names no model, so it runs unchanged on whichever provider the Worker was started with (--provider baseten|fireworks|together). Each provider entry carries its own snipsync marker so the Temporal docs page for that provider can quote it. Offline tests cover the provider table, the client configuration, and the Workflow with a scripted model. --- openai_agents/README.md | 2 +- openai_agents/model_providers/README.md | 23 +++ .../openai_compatible_providers.py | 88 +++++++++++ .../run_openai_compatible_worker.py | 66 +++++++++ .../run_openai_compatible_workflow.py | 29 ++++ .../workflows/openai_compatible_workflow.py | 35 +++++ tests/openai_agents/model_providers_test.py | 138 ++++++++++++++++++ 7 files changed, 380 insertions(+), 1 deletion(-) create mode 100644 openai_agents/model_providers/openai_compatible_providers.py create mode 100644 openai_agents/model_providers/run_openai_compatible_worker.py create mode 100644 openai_agents/model_providers/run_openai_compatible_workflow.py create mode 100644 openai_agents/model_providers/workflows/openai_compatible_workflow.py create mode 100644 tests/openai_agents/model_providers_test.py diff --git a/openai_agents/README.md b/openai_agents/README.md index c0b79225..707e3445 100644 --- a/openai_agents/README.md +++ b/openai_agents/README.md @@ -36,7 +36,7 @@ Each directory contains a complete example with its own README for detailed inst - **[Hosted MCP](./hosted_mcp/README.md)** - Using the MCP client functionality of the OpenAI Responses API. - **[MCP v2](./mcp_v2/README.md)** - MCP SDK v2 using named server factories and durable workflow proxies. - **[MCP](./mcp/README.md)** - Legacy MCP provider examples retained for compatibility. -- **[Model Providers](./model_providers/README.md)** - Using custom LLM providers (e.g., Anthropic via LiteLLM). +- **[Model Providers](./model_providers/README.md)** - Using custom LLM providers: Anthropic via LiteLLM, OpenRouter, a local gpt-oss with Ollama, and open-weight models on Baseten, Fireworks AI, and Together AI. - **[Research Bot](./research_bot/README.md)** - Multi-agent research system with specialized roles: a planner agent, search agent, and writer agent working together to conduct comprehensive research. - **[Customer Service](./customer_service/README.md)** - Interactive customer service agent with escalation capabilities, demonstrating conversational workflows. - **[Reasoning Content](./reasoning_content/README.md)** - Example of how to retrieve the thought process of reasoning models. diff --git a/openai_agents/model_providers/README.md b/openai_agents/model_providers/README.md index c9b93986..d36d2570 100644 --- a/openai_agents/model_providers/README.md +++ b/openai_agents/model_providers/README.md @@ -47,6 +47,29 @@ uv run openai_agents/model_providers/run_openrouter_workflow.py The workflow uses `openrouter/auto`, so OpenRouter picks a model per request; change `OPENROUTER_MODEL` in [workflows/openrouter_workflow.py](workflows/openrouter_workflow.py) to pin any OpenRouter model slug. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. See the [openrouter](../../openrouter) sample for calling OpenRouter directly from Activities with cost tracking and budgets. +#### Open-weight models on Baseten, Fireworks AI, and Together AI +[Baseten](https://www.baseten.co/), [Fireworks AI](https://fireworks.ai/), and [Together AI](https://www.together.ai/) serve open-weight models behind the OpenAI Chat Completions API, so one Worker script covers all three. It points the OpenAI client at the provider's base URL and supplies a default model; the Workflow names no model, so the same Workflow runs on whichever provider the Worker was started with. + +| Provider | API key | Default model | +|---|---|---| +| Baseten | `BASETEN_API_KEY` | `openai/gpt-oss-120b` | +| Fireworks AI | `FIREWORKS_API_KEY` | `accounts/fireworks/models/gpt-oss-120b` | +| Together AI | `TOGETHER_API_KEY` | `openai/gpt-oss-120b` | + +Start the worker for one provider (`baseten`, `fireworks`, or `together`): +```bash +export TOGETHER_API_KEY="your_together_api_key" + +uv run openai_agents/model_providers/run_openai_compatible_worker.py --provider together +``` + +Then run the example in a separate terminal: +```bash +uv run openai_agents/model_providers/run_openai_compatible_workflow.py +``` + +The providers are defined in [openai_compatible_providers.py](openai_compatible_providers.py). Add another OpenAI-compatible endpoint by appending to `PROVIDERS`, or run an agent on a different model by setting `model=` on the `Agent`. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. The Temporal docs have a page per provider: [Baseten](https://docs.temporal.io/develop/python/integrations/baseten), [Fireworks AI](https://docs.temporal.io/develop/python/integrations/fireworks-ai), and [Together AI](https://docs.temporal.io/develop/python/integrations/together-ai). + ### Extra #### GPT-OSS with Ollama diff --git a/openai_agents/model_providers/openai_compatible_providers.py b/openai_agents/model_providers/openai_compatible_providers.py new file mode 100644 index 00000000..1f6a5fd7 --- /dev/null +++ b/openai_agents/model_providers/openai_compatible_providers.py @@ -0,0 +1,88 @@ +"""Open-weight model providers with OpenAI-compatible APIs. + +Baseten, Fireworks AI, and Together AI all serve open-weight models behind the +OpenAI Chat Completions API, so the stock ``openai`` client reaches them once +it is pointed at the provider's base URL. The Workflow is the same for all of +them; the Worker chooses the provider and the default model. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + +from agents import Model, ModelProvider, OpenAIChatCompletionsModel +from openai import AsyncOpenAI + + +@dataclass(frozen=True) +class OpenAICompatibleProvider: + """An OpenAI-compatible inference endpoint and a default model on it.""" + + name: str + base_url: str + api_key_env: str + # A model with tool-calling support, in the provider's naming scheme. + model: str + + +# @@@SNIPSTART python-openai-agents-baseten-provider +BASETEN = OpenAICompatibleProvider( + name="baseten", + base_url="https://inference.baseten.co/v1", + api_key_env="BASETEN_API_KEY", + model="openai/gpt-oss-120b", +) +# @@@SNIPEND + +# @@@SNIPSTART python-openai-agents-fireworks-provider +FIREWORKS = OpenAICompatibleProvider( + name="fireworks", + base_url="https://api.fireworks.ai/inference/v1", + api_key_env="FIREWORKS_API_KEY", + model="accounts/fireworks/models/gpt-oss-120b", +) +# @@@SNIPEND + +# @@@SNIPSTART python-openai-agents-together-provider +TOGETHER = OpenAICompatibleProvider( + name="together", + base_url="https://api.together.ai/v1", + api_key_env="TOGETHER_API_KEY", + model="openai/gpt-oss-120b", +) +# @@@SNIPEND + +PROVIDERS: dict[str, OpenAICompatibleProvider] = { + provider.name: provider for provider in (BASETEN, FIREWORKS, TOGETHER) +} + + +# @@@SNIPSTART python-openai-agents-openai-compatible-model-provider +class OpenAICompatibleModelProvider(ModelProvider): + """OpenAI Agents SDK model provider for an OpenAI-compatible endpoint. + + Client retries are off: the Temporal plugin runs each model call as an + Activity, and the Activity's Retry Policy owns the retries. An agent that + names no model gets the provider's default; an agent can name any other + model the provider serves. + """ + + def __init__(self, provider: OpenAICompatibleProvider) -> None: + self._provider = provider + self._client = AsyncOpenAI( + base_url=provider.base_url, + api_key=os.environ[provider.api_key_env], + max_retries=0, + ) + + def get_model(self, model_name: str | None) -> Model: + # Chat Completions is the API all of these providers serve. The Agents + # SDK defaults to the Responses API, which not every provider offers. + return OpenAIChatCompletionsModel( + model=model_name or self._provider.model, + openai_client=self._client, + ) + + +# @@@SNIPEND diff --git a/openai_agents/model_providers/run_openai_compatible_worker.py b/openai_agents/model_providers/run_openai_compatible_worker.py new file mode 100644 index 00000000..aa6a05bd --- /dev/null +++ b/openai_agents/model_providers/run_openai_compatible_worker.py @@ -0,0 +1,66 @@ +"""Worker that runs the agent on Baseten, Fireworks AI, or Together AI.""" + +import argparse +import asyncio +import logging +from datetime import timedelta + +from agents import set_tracing_disabled +from temporalio.client import Client +from temporalio.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin +from temporalio.worker import Worker + +from openai_agents.model_providers.openai_compatible_providers import ( + PROVIDERS, + OpenAICompatibleModelProvider, +) +from openai_agents.model_providers.workflows.openai_compatible_workflow import ( + OpenAICompatibleAgentWorkflow, +) + + +async def main(provider_name: str) -> None: + # Disable Agents SDK tracing: the default exporter sends traces to OpenAI's + # backend, which needs an OpenAI API key that this sample does not have. + set_tracing_disabled(disabled=True) + + logging.basicConfig(level=logging.WARNING) + logging.getLogger("temporalio.workflow").setLevel(logging.DEBUG) + + # @@@SNIPSTART python-openai-agents-openai-compatible-worker + client = await Client.connect( + "localhost:7233", + plugins=[ + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + # A custom model provider must set a timeout. Raise it for + # slow models or deployments that scale up from zero. + start_to_close_timeout=timedelta(seconds=60), + # The plugin heartbeats, so a Worker that dies mid-call is + # noticed in seconds rather than after the full timeout. + heartbeat_timeout=timedelta(seconds=10), + ), + model_provider=OpenAICompatibleModelProvider(PROVIDERS[provider_name]), + ), + ], + ) + + worker = Worker( + client, + task_queue="openai-agents-model-providers-task-queue", + workflows=[OpenAICompatibleAgentWorkflow], + ) + await worker.run() + # @@@SNIPEND + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--provider", + choices=sorted(PROVIDERS), + required=True, + help="which provider serves the model", + ) + args = parser.parse_args() + asyncio.run(main(args.provider)) diff --git a/openai_agents/model_providers/run_openai_compatible_workflow.py b/openai_agents/model_providers/run_openai_compatible_workflow.py new file mode 100644 index 00000000..93720efd --- /dev/null +++ b/openai_agents/model_providers/run_openai_compatible_workflow.py @@ -0,0 +1,29 @@ +import asyncio + +from temporalio.client import Client +from temporalio.openai_agents import OpenAIAgentsPlugin + +from openai_agents.model_providers.workflows.openai_compatible_workflow import ( + OpenAICompatibleAgentWorkflow, +) + + +async def main() -> None: + client = await Client.connect( + "localhost:7233", + plugins=[ + OpenAIAgentsPlugin(), + ], + ) + + result = await client.execute_workflow( + OpenAICompatibleAgentWorkflow.run, + "What's the weather in Tokyo?", + id="openai-agents-openai-compatible-workflow-id", + task_queue="openai-agents-model-providers-task-queue", + ) + print(f"Result: {result}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/openai_agents/model_providers/workflows/openai_compatible_workflow.py b/openai_agents/model_providers/workflows/openai_compatible_workflow.py new file mode 100644 index 00000000..7eeba9f4 --- /dev/null +++ b/openai_agents/model_providers/workflows/openai_compatible_workflow.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from agents import Agent, ModelSettings, Runner, function_tool +from temporalio import workflow + + +# @@@SNIPSTART python-openai-agents-openai-compatible-workflow +@workflow.defn +class OpenAICompatibleAgentWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + # Tools that run inside the Workflow must be async: the Agents SDK runs + # sync tools in a thread, which the Workflow sandbox does not allow. + @function_tool + async def get_weather(city: str) -> str: + workflow.logger.debug(f"Getting weather for {city}") + return f"The weather in {city} is sunny." + + # No model is named here, so the Worker's model provider supplies its + # default. Set model= to any model the provider serves to override it. + # Start from empty model settings: an agent that names no model gets + # the Agents SDK's defaults for OpenAI's default model (a reasoning + # effort and `verbosity`), which other providers reject. + agent = Agent( + name="Assistant", + instructions="You only respond in haikus. When asked about the weather always use the tool to get the current weather.", + model_settings=ModelSettings(), + tools=[get_weather], + ) + + result = await Runner.run(agent, prompt) + return result.final_output + + +# @@@SNIPEND diff --git a/tests/openai_agents/model_providers_test.py b/tests/openai_agents/model_providers_test.py new file mode 100644 index 00000000..9f327f02 --- /dev/null +++ b/tests/openai_agents/model_providers_test.py @@ -0,0 +1,138 @@ +"""Offline tests for the OpenAI-compatible model provider sample.""" + +import uuid +from collections.abc import AsyncIterator +from datetime import timedelta +from typing import Any + +import pytest +from agents import ( + AgentOutputSchemaBase, + Handoff, + Model, + ModelResponse, + ModelSettings, + ModelTracing, + OpenAIChatCompletionsModel, + Tool, + TResponseInputItem, +) +from temporalio.client import Client +from temporalio.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin +from temporalio.openai_agents.testing import ResponseBuilders, TestModelProvider +from temporalio.worker import Worker + +from openai_agents.model_providers.openai_compatible_providers import ( + PROVIDERS, + OpenAICompatibleModelProvider, + OpenAICompatibleProvider, +) +from openai_agents.model_providers.workflows.openai_compatible_workflow import ( + OpenAICompatibleAgentWorkflow, +) + + +def test_provider_table() -> None: + assert set(PROVIDERS) == {"baseten", "fireworks", "together"} + for name, provider in PROVIDERS.items(): + assert provider.name == name + assert provider.base_url.startswith("https://") + assert provider.api_key_env.endswith("_API_KEY") + assert provider.model + + +@pytest.mark.parametrize("provider", PROVIDERS.values(), ids=lambda p: p.name) +def test_model_provider_targets_the_endpoint( + monkeypatch: pytest.MonkeyPatch, provider: OpenAICompatibleProvider +) -> None: + monkeypatch.setenv(provider.api_key_env, "test-key") + model_provider = OpenAICompatibleModelProvider(provider) + + default = model_provider.get_model(None) + assert isinstance(default, OpenAIChatCompletionsModel) + assert default.model == provider.model + client = default._client + assert str(client.base_url).rstrip("/") == provider.base_url + assert client.api_key == "test-key" + # Temporal owns retries: one HTTP call per Activity attempt. + assert client.max_retries == 0 + + override = model_provider.get_model("some/other-model") + assert isinstance(override, OpenAIChatCompletionsModel) + assert override.model == "some/other-model" + + +def test_model_provider_requires_the_api_key(monkeypatch: pytest.MonkeyPatch) -> None: + provider = PROVIDERS["together"] + monkeypatch.delenv(provider.api_key_env, raising=False) + with pytest.raises(KeyError, match=provider.api_key_env): + OpenAICompatibleModelProvider(provider) + + +class RecordingModel(Model): + """Scripted model that records the settings each call was made with.""" + + def __init__(self, responses: list[ModelResponse]) -> None: + self._responses = iter(responses) + self.settings: list[ModelSettings] = [] + + async def get_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + **kwargs: Any, + ) -> ModelResponse: + self.settings.append(model_settings) + return next(self._responses) + + async def stream_response(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]: + raise NotImplementedError + yield + + +async def test_workflow_uses_the_tool_and_the_default_model(client: Client) -> None: + # The Workflow names no model, so the provider's default is used; a + # scripted model stands in for the provider here. + model = RecordingModel( + [ + ResponseBuilders.tool_call('{"city": "Tokyo"}', "get_weather"), + ResponseBuilders.output_message("Sunny over Tokyo"), + ] + ) + config = client.config() + config["plugins"] = [ + *config["plugins"], + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=30) + ), + model_provider=TestModelProvider(model), + ), + ] + client = Client(**config) + task_queue = f"openai-agents-openai-compatible-{uuid.uuid4()}" + + async with Worker( + client, + task_queue=task_queue, + workflows=[OpenAICompatibleAgentWorkflow], + ): + result = await client.execute_workflow( + OpenAICompatibleAgentWorkflow.run, + "What's the weather in Tokyo?", + id=f"openai-compatible-{uuid.uuid4()}", + task_queue=task_queue, + ) + + assert result == "Sunny over Tokyo" + # The Agents SDK's defaults for OpenAI's default model (a reasoning effort + # and verbosity) must not reach a provider that rejects them. + assert len(model.settings) == 2 + for settings in model.settings: + assert settings.reasoning is None + assert settings.verbosity is None From 53b94564a7278c468b180f3e28a80f17dab4d476 Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 26 Oct 2026 12:01:19 -0500 Subject: [PATCH 2/2] Route model calls by provider prefix so one Worker serves every provider Review feedback: restarting the Worker to switch providers is not how a real deployment works. The model provider now holds every provider whose API key is set and routes `/` names, such as `together/openai/gpt-oss-120b`, to the matching endpoint. The model name travels in the Workflow input, so Workflows on different providers run side by side on the same Worker, and the starter runs one Workflow per `--model` concurrently. Naming the model on the agent also lets the Agents SDK drop the default settings it keeps for OpenAI's default model, so the explicit `ModelSettings()` workaround is gone. A model name for a provider without a key, or with an unknown prefix, fails its Activity without retrying; the starter prints that root cause. Verified live on Fireworks AI alongside a keyless Together AI Workflow on the same Worker: the first completed through two model Activities, the second failed on its first attempt with the "set TOGETHER_API_KEY" error. --- openai_agents/model_providers/README.md | 20 +- .../openai_compatible_providers.py | 73 ++++--- .../run_openai_compatible_worker.py | 26 +-- .../run_openai_compatible_workflow.py | 60 +++++- .../workflows/openai_compatible_workflow.py | 23 +- tests/openai_agents/model_providers_test.py | 197 +++++++++++++----- 6 files changed, 282 insertions(+), 117 deletions(-) diff --git a/openai_agents/model_providers/README.md b/openai_agents/model_providers/README.md index d36d2570..998cd640 100644 --- a/openai_agents/model_providers/README.md +++ b/openai_agents/model_providers/README.md @@ -48,27 +48,27 @@ uv run openai_agents/model_providers/run_openrouter_workflow.py The workflow uses `openrouter/auto`, so OpenRouter picks a model per request; change `OPENROUTER_MODEL` in [workflows/openrouter_workflow.py](workflows/openrouter_workflow.py) to pin any OpenRouter model slug. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. See the [openrouter](../../openrouter) sample for calling OpenRouter directly from Activities with cost tracking and budgets. #### Open-weight models on Baseten, Fireworks AI, and Together AI -[Baseten](https://www.baseten.co/), [Fireworks AI](https://fireworks.ai/), and [Together AI](https://www.together.ai/) serve open-weight models behind the OpenAI Chat Completions API, so one Worker script covers all three. It points the OpenAI client at the provider's base URL and supplies a default model; the Workflow names no model, so the same Workflow runs on whichever provider the Worker was started with. +[Baseten](https://www.baseten.co/), [Fireworks AI](https://fireworks.ai/), and [Together AI](https://www.together.ai/) serve open-weight models behind the OpenAI Chat Completions API. One Worker serves all three: it builds a client for each provider whose API key is set, and each Workflow picks a provider through the model name it gives its agent, `/`. Workflows on different providers run side by side on the same Worker. -| Provider | API key | Default model | +| Provider | API key | Example model name | |---|---|---| -| Baseten | `BASETEN_API_KEY` | `openai/gpt-oss-120b` | -| Fireworks AI | `FIREWORKS_API_KEY` | `accounts/fireworks/models/gpt-oss-120b` | -| Together AI | `TOGETHER_API_KEY` | `openai/gpt-oss-120b` | +| Baseten | `BASETEN_API_KEY` | `baseten/openai/gpt-oss-120b` | +| Fireworks AI | `FIREWORKS_API_KEY` | `fireworks/accounts/fireworks/models/gpt-oss-120b` | +| Together AI | `TOGETHER_API_KEY` | `together/openai/gpt-oss-120b` | -Start the worker for one provider (`baseten`, `fireworks`, or `together`): +Export the keys you have and start the worker: ```bash export TOGETHER_API_KEY="your_together_api_key" -uv run openai_agents/model_providers/run_openai_compatible_worker.py --provider together +uv run openai_agents/model_providers/run_openai_compatible_worker.py ``` -Then run the example in a separate terminal: +Then run the example in a separate terminal. Pass `--model` once per Workflow to start; with no `--model`, it starts one Workflow per provider whose key is set in that shell: ```bash -uv run openai_agents/model_providers/run_openai_compatible_workflow.py +uv run openai_agents/model_providers/run_openai_compatible_workflow.py --model together/openai/gpt-oss-120b ``` -The providers are defined in [openai_compatible_providers.py](openai_compatible_providers.py). Add another OpenAI-compatible endpoint by appending to `PROVIDERS`, or run an agent on a different model by setting `model=` on the `Agent`. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. The Temporal docs have a page per provider: [Baseten](https://docs.temporal.io/develop/python/integrations/baseten), [Fireworks AI](https://docs.temporal.io/develop/python/integrations/fireworks-ai), and [Together AI](https://docs.temporal.io/develop/python/integrations/together-ai). +Any model the provider serves works with its prefix. A model name for a provider without a key, or with an unknown prefix, fails its Workflow without retrying. The providers are defined in [openai_compatible_providers.py](openai_compatible_providers.py); add another OpenAI-compatible endpoint by appending to `PROVIDERS`. Tools that run inside the Workflow must be `async`: the Agents SDK runs sync tools in a thread, which the Workflow sandbox does not allow. The Temporal docs have a page per provider: [Baseten](https://docs.temporal.io/develop/python/integrations/baseten), [Fireworks AI](https://docs.temporal.io/develop/python/integrations/fireworks-ai), and [Together AI](https://docs.temporal.io/develop/python/integrations/together-ai). ### Extra diff --git a/openai_agents/model_providers/openai_compatible_providers.py b/openai_agents/model_providers/openai_compatible_providers.py index 1f6a5fd7..43b4faea 100644 --- a/openai_agents/model_providers/openai_compatible_providers.py +++ b/openai_agents/model_providers/openai_compatible_providers.py @@ -2,28 +2,34 @@ Baseten, Fireworks AI, and Together AI all serve open-weight models behind the OpenAI Chat Completions API, so the stock ``openai`` client reaches them once -it is pointed at the provider's base URL. The Workflow is the same for all of -them; the Worker chooses the provider and the default model. +it is pointed at the provider's base URL. One Worker serves every provider +whose API key is set, and each Workflow picks a provider through the model +name it gives its agent: ``/``, such as +``together/openai/gpt-oss-120b``. """ from __future__ import annotations import os +from collections.abc import Iterable from dataclasses import dataclass from agents import Model, ModelProvider, OpenAIChatCompletionsModel from openai import AsyncOpenAI +from temporalio.exceptions import ApplicationError @dataclass(frozen=True) class OpenAICompatibleProvider: - """An OpenAI-compatible inference endpoint and a default model on it.""" + """An OpenAI-compatible inference endpoint and an example model on it.""" + # The prefix that selects this provider in a model name. name: str base_url: str api_key_env: str - # A model with tool-calling support, in the provider's naming scheme. - model: str + # A model with tool-calling support, in the provider's naming scheme. The + # starter runs it when no model is given. + default_model: str # @@@SNIPSTART python-openai-agents-baseten-provider @@ -31,7 +37,7 @@ class OpenAICompatibleProvider: name="baseten", base_url="https://inference.baseten.co/v1", api_key_env="BASETEN_API_KEY", - model="openai/gpt-oss-120b", + default_model="openai/gpt-oss-120b", ) # @@@SNIPEND @@ -40,7 +46,7 @@ class OpenAICompatibleProvider: name="fireworks", base_url="https://api.fireworks.ai/inference/v1", api_key_env="FIREWORKS_API_KEY", - model="accounts/fireworks/models/gpt-oss-120b", + default_model="accounts/fireworks/models/gpt-oss-120b", ) # @@@SNIPEND @@ -49,7 +55,7 @@ class OpenAICompatibleProvider: name="together", base_url="https://api.together.ai/v1", api_key_env="TOGETHER_API_KEY", - model="openai/gpt-oss-120b", + default_model="openai/gpt-oss-120b", ) # @@@SNIPEND @@ -60,29 +66,48 @@ class OpenAICompatibleProvider: # @@@SNIPSTART python-openai-agents-openai-compatible-model-provider class OpenAICompatibleModelProvider(ModelProvider): - """OpenAI Agents SDK model provider for an OpenAI-compatible endpoint. + """Routes ``/`` names to the matching OpenAI-compatible endpoint. - Client retries are off: the Temporal plugin runs each model call as an - Activity, and the Activity's Retry Policy owns the retries. An agent that - names no model gets the provider's default; an agent can name any other - model the provider serves. + One Worker serves every provider whose API key is set, so Workflows on + different providers run side by side on the same Task Queue. Client + retries are off: the plugin runs each model call as a Temporal Activity, + and the Activity's Retry Policy owns the retries. """ - def __init__(self, provider: OpenAICompatibleProvider) -> None: - self._provider = provider - self._client = AsyncOpenAI( - base_url=provider.base_url, - api_key=os.environ[provider.api_key_env], - max_retries=0, - ) + def __init__(self, providers: Iterable[OpenAICompatibleProvider]) -> None: + self._providers = {provider.name: provider for provider in providers} + self._clients: dict[str, AsyncOpenAI] = {} + for provider in self._providers.values(): + api_key = os.environ.get(provider.api_key_env) + if api_key: + self._clients[provider.name] = AsyncOpenAI( + base_url=provider.base_url, api_key=api_key, max_retries=0 + ) + + @property + def configured(self) -> list[str]: + """Names of the providers whose API key is set.""" + return sorted(self._clients) def get_model(self, model_name: str | None) -> Model: + provider_name, _, model = (model_name or "").partition("/") + provider = self._providers.get(provider_name) + if provider is None or not model: + # A mistake in the Workflow's input: fail the Activity without retrying. + raise ApplicationError( + f"Model name {model_name!r} must be / with a " + f"provider in {sorted(self._providers)}", + non_retryable=True, + ) + client = self._clients.get(provider_name) + if client is None: + raise ApplicationError( + f"Set {provider.api_key_env} and restart the Worker to use {provider_name}", + non_retryable=True, + ) # Chat Completions is the API all of these providers serve. The Agents # SDK defaults to the Responses API, which not every provider offers. - return OpenAIChatCompletionsModel( - model=model_name or self._provider.model, - openai_client=self._client, - ) + return OpenAIChatCompletionsModel(model=model, openai_client=client) # @@@SNIPEND diff --git a/openai_agents/model_providers/run_openai_compatible_worker.py b/openai_agents/model_providers/run_openai_compatible_worker.py index aa6a05bd..9f54e84c 100644 --- a/openai_agents/model_providers/run_openai_compatible_worker.py +++ b/openai_agents/model_providers/run_openai_compatible_worker.py @@ -1,8 +1,8 @@ -"""Worker that runs the agent on Baseten, Fireworks AI, or Together AI.""" +"""Worker that serves Baseten, Fireworks AI, and Together AI from one process.""" -import argparse import asyncio import logging +import sys from datetime import timedelta from agents import set_tracing_disabled @@ -19,7 +19,7 @@ ) -async def main(provider_name: str) -> None: +async def main() -> None: # Disable Agents SDK tracing: the default exporter sends traces to OpenAI's # backend, which needs an OpenAI API key that this sample does not have. set_tracing_disabled(disabled=True) @@ -28,6 +28,14 @@ async def main(provider_name: str) -> None: logging.getLogger("temporalio.workflow").setLevel(logging.DEBUG) # @@@SNIPSTART python-openai-agents-openai-compatible-worker + # One provider for every endpoint whose API key is set. The model name on + # each Workflow's agent selects the endpoint. + model_provider = OpenAICompatibleModelProvider(PROVIDERS.values()) + if not model_provider.configured: + env_vars = ", ".join(provider.api_key_env for provider in PROVIDERS.values()) + sys.exit(f"No provider API key is set; export at least one of {env_vars}.") + print(f"Serving providers: {', '.join(model_provider.configured)}", flush=True) + client = await Client.connect( "localhost:7233", plugins=[ @@ -40,7 +48,7 @@ async def main(provider_name: str) -> None: # noticed in seconds rather than after the full timeout. heartbeat_timeout=timedelta(seconds=10), ), - model_provider=OpenAICompatibleModelProvider(PROVIDERS[provider_name]), + model_provider=model_provider, ), ], ) @@ -55,12 +63,4 @@ async def main(provider_name: str) -> None: if __name__ == "__main__": - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument( - "--provider", - choices=sorted(PROVIDERS), - required=True, - help="which provider serves the model", - ) - args = parser.parse_args() - asyncio.run(main(args.provider)) + asyncio.run(main()) diff --git a/openai_agents/model_providers/run_openai_compatible_workflow.py b/openai_agents/model_providers/run_openai_compatible_workflow.py index 93720efd..25ba7387 100644 --- a/openai_agents/model_providers/run_openai_compatible_workflow.py +++ b/openai_agents/model_providers/run_openai_compatible_workflow.py @@ -1,29 +1,67 @@ +"""Start one Workflow per model name; they run concurrently on the same Worker.""" + +import argparse import asyncio +import os -from temporalio.client import Client +from temporalio.client import Client, WorkflowFailureError +from temporalio.exceptions import FailureError from temporalio.openai_agents import OpenAIAgentsPlugin +from openai_agents.model_providers.openai_compatible_providers import PROVIDERS from openai_agents.model_providers.workflows.openai_compatible_workflow import ( + OpenAICompatibleAgentInput, OpenAICompatibleAgentWorkflow, ) -async def main() -> None: +async def run_one(client: Client, model: str) -> None: + try: + result = await client.execute_workflow( + OpenAICompatibleAgentWorkflow.run, + OpenAICompatibleAgentInput( + prompt="What's the weather in Tokyo?", model=model + ), + id=f"openai-agents-openai-compatible-{model.replace('/', '-')}", + task_queue="openai-agents-model-providers-task-queue", + ) + print(f"{model}:\n{result}\n") + except WorkflowFailureError as e: + # Report the root cause, such as the provider's "Set FIREWORKS_API_KEY". + cause = e.cause + while isinstance(cause, FailureError) and cause.cause is not None: + cause = cause.cause + print(f"{model}: failed: {cause}\n") + + +async def main(models: list[str]) -> None: client = await Client.connect( "localhost:7233", plugins=[ OpenAIAgentsPlugin(), ], ) - - result = await client.execute_workflow( - OpenAICompatibleAgentWorkflow.run, - "What's the weather in Tokyo?", - id="openai-agents-openai-compatible-workflow-id", - task_queue="openai-agents-model-providers-task-queue", - ) - print(f"Result: {result}") + # Each Workflow names its own model, so these run side by side on one + # Worker, each on its own provider. + await asyncio.gather(*(run_one(client, model) for model in models)) if __name__ == "__main__": - asyncio.run(main()) + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--model", + action="append", + dest="models", + metavar="PROVIDER/MODEL", + help="model to run, such as together/openai/gpt-oss-120b; repeat for " + "several. Defaults to each provider whose API key is set in this shell.", + ) + args = parser.parse_args() + models = args.models or [ + f"{provider.name}/{provider.default_model}" + for provider in PROVIDERS.values() + if os.environ.get(provider.api_key_env) + ] + if not models: + parser.error("pass --model or export a provider API key") + asyncio.run(main(models)) diff --git a/openai_agents/model_providers/workflows/openai_compatible_workflow.py b/openai_agents/model_providers/workflows/openai_compatible_workflow.py index 7eeba9f4..5884d6dd 100644 --- a/openai_agents/model_providers/workflows/openai_compatible_workflow.py +++ b/openai_agents/model_providers/workflows/openai_compatible_workflow.py @@ -1,14 +1,24 @@ from __future__ import annotations -from agents import Agent, ModelSettings, Runner, function_tool +from dataclasses import dataclass + +from agents import Agent, Runner, function_tool from temporalio import workflow # @@@SNIPSTART python-openai-agents-openai-compatible-workflow +@dataclass +class OpenAICompatibleAgentInput: + prompt: str + # "/", such as "together/openai/gpt-oss-120b". The Worker's + # model provider routes the agent's model calls to that provider. + model: str + + @workflow.defn class OpenAICompatibleAgentWorkflow: @workflow.run - async def run(self, prompt: str) -> str: + async def run(self, input: OpenAICompatibleAgentInput) -> str: # Tools that run inside the Workflow must be async: the Agents SDK runs # sync tools in a thread, which the Workflow sandbox does not allow. @function_tool @@ -16,19 +26,14 @@ async def get_weather(city: str) -> str: workflow.logger.debug(f"Getting weather for {city}") return f"The weather in {city} is sunny." - # No model is named here, so the Worker's model provider supplies its - # default. Set model= to any model the provider serves to override it. - # Start from empty model settings: an agent that names no model gets - # the Agents SDK's defaults for OpenAI's default model (a reasoning - # effort and `verbosity`), which other providers reject. agent = Agent( name="Assistant", instructions="You only respond in haikus. When asked about the weather always use the tool to get the current weather.", - model_settings=ModelSettings(), + model=input.model, tools=[get_weather], ) - result = await Runner.run(agent, prompt) + result = await Runner.run(agent, input.prompt) return result.final_output diff --git a/tests/openai_agents/model_providers_test.py b/tests/openai_agents/model_providers_test.py index 9f327f02..f9453cf9 100644 --- a/tests/openai_agents/model_providers_test.py +++ b/tests/openai_agents/model_providers_test.py @@ -1,5 +1,6 @@ """Offline tests for the OpenAI-compatible model provider sample.""" +import asyncio import uuid from collections.abc import AsyncIterator from datetime import timedelta @@ -10,6 +11,7 @@ AgentOutputSchemaBase, Handoff, Model, + ModelProvider, ModelResponse, ModelSettings, ModelTracing, @@ -17,56 +19,88 @@ Tool, TResponseInputItem, ) -from temporalio.client import Client +from agents.items import TResponseStreamEvent +from temporalio.client import Client, WorkflowFailureError +from temporalio.exceptions import ActivityError, ApplicationError, RetryState from temporalio.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin -from temporalio.openai_agents.testing import ResponseBuilders, TestModelProvider +from temporalio.openai_agents.testing import ResponseBuilders from temporalio.worker import Worker from openai_agents.model_providers.openai_compatible_providers import ( PROVIDERS, OpenAICompatibleModelProvider, - OpenAICompatibleProvider, ) from openai_agents.model_providers.workflows.openai_compatible_workflow import ( + OpenAICompatibleAgentInput, OpenAICompatibleAgentWorkflow, ) +TOGETHER_MODEL = "together/openai/gpt-oss-120b" +FIREWORKS_MODEL = "fireworks/accounts/fireworks/models/gpt-oss-120b" + def test_provider_table() -> None: assert set(PROVIDERS) == {"baseten", "fireworks", "together"} for name, provider in PROVIDERS.items(): assert provider.name == name + assert "/" not in name assert provider.base_url.startswith("https://") assert provider.api_key_env.endswith("_API_KEY") - assert provider.model + assert provider.default_model + +def _provider_with_keys( + monkeypatch: pytest.MonkeyPatch, *names: str +) -> OpenAICompatibleModelProvider: + for provider in PROVIDERS.values(): + monkeypatch.delenv(provider.api_key_env, raising=False) + for name in names: + monkeypatch.setenv(PROVIDERS[name].api_key_env, f"{name}-test-key") + return OpenAICompatibleModelProvider(PROVIDERS.values()) -@pytest.mark.parametrize("provider", PROVIDERS.values(), ids=lambda p: p.name) -def test_model_provider_targets_the_endpoint( - monkeypatch: pytest.MonkeyPatch, provider: OpenAICompatibleProvider + +def test_routes_a_prefixed_model_name_to_its_endpoint( + monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.setenv(provider.api_key_env, "test-key") - model_provider = OpenAICompatibleModelProvider(provider) - - default = model_provider.get_model(None) - assert isinstance(default, OpenAIChatCompletionsModel) - assert default.model == provider.model - client = default._client - assert str(client.base_url).rstrip("/") == provider.base_url - assert client.api_key == "test-key" + model_provider = _provider_with_keys(monkeypatch, "together", "fireworks") + assert model_provider.configured == ["fireworks", "together"] + + model = model_provider.get_model(TOGETHER_MODEL) + assert isinstance(model, OpenAIChatCompletionsModel) + assert model.model == "openai/gpt-oss-120b" + client = model._client + assert str(client.base_url).rstrip("/") == PROVIDERS["together"].base_url + assert client.api_key == "together-test-key" # Temporal owns retries: one HTTP call per Activity attempt. assert client.max_retries == 0 - override = model_provider.get_model("some/other-model") - assert isinstance(override, OpenAIChatCompletionsModel) - assert override.model == "some/other-model" + fireworks = model_provider.get_model(FIREWORKS_MODEL) + assert isinstance(fireworks, OpenAIChatCompletionsModel) + assert fireworks.model == "accounts/fireworks/models/gpt-oss-120b" + assert ( + str(fireworks._client.base_url).rstrip("/") == PROVIDERS["fireworks"].base_url + ) + + +@pytest.mark.parametrize( + "model_name", [None, "openai/gpt-oss-120b", "nope/model", "together", "together/"] +) +def test_rejects_a_name_without_a_known_provider_without_retrying( + monkeypatch: pytest.MonkeyPatch, model_name: str | None +) -> None: + model_provider = _provider_with_keys(monkeypatch, "together") + with pytest.raises(ApplicationError) as err: + model_provider.get_model(model_name) + assert err.value.non_retryable -def test_model_provider_requires_the_api_key(monkeypatch: pytest.MonkeyPatch) -> None: - provider = PROVIDERS["together"] - monkeypatch.delenv(provider.api_key_env, raising=False) - with pytest.raises(KeyError, match=provider.api_key_env): - OpenAICompatibleModelProvider(provider) +def test_rejects_a_provider_without_a_key_without_retrying( + monkeypatch: pytest.MonkeyPatch, +) -> None: + model_provider = _provider_with_keys(monkeypatch, "together") + with pytest.raises(ApplicationError, match="FIREWORKS_API_KEY") as err: + model_provider.get_model(FIREWORKS_MODEL) + assert err.value.non_retryable class RecordingModel(Model): @@ -90,20 +124,32 @@ async def get_response( self.settings.append(model_settings) return next(self._responses) - async def stream_response(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]: - raise NotImplementedError - yield + def stream_response( + self, *args: Any, **kwargs: Any + ) -> AsyncIterator[TResponseStreamEvent]: + raise NotImplementedError() -async def test_workflow_uses_the_tool_and_the_default_model(client: Client) -> None: - # The Workflow names no model, so the provider's default is used; a - # scripted model stands in for the provider here. - model = RecordingModel( - [ - ResponseBuilders.tool_call('{"city": "Tokyo"}', "get_weather"), - ResponseBuilders.output_message("Sunny over Tokyo"), - ] - ) +class RecordingProvider(ModelProvider): + """Stands in for the endpoints: one scripted model per model name.""" + + def __init__(self, scripts: dict[str, list[ModelResponse]]) -> None: + self.models = {name: RecordingModel(script) for name, script in scripts.items()} + self.requested: list[str | None] = [] + + def get_model(self, model_name: str | None) -> Model: + self.requested.append(model_name) + return self.models[model_name or ""] + + +def _script(answer: str) -> list[ModelResponse]: + return [ + ResponseBuilders.tool_call('{"city": "Tokyo"}', "get_weather"), + ResponseBuilders.output_message(answer), + ] + + +def _with_provider(client: Client, provider: ModelProvider) -> Client: config = client.config() config["plugins"] = [ *config["plugins"], @@ -111,10 +157,22 @@ async def test_workflow_uses_the_tool_and_the_default_model(client: Client) -> N model_params=ModelActivityParameters( start_to_close_timeout=timedelta(seconds=30) ), - model_provider=TestModelProvider(model), + model_provider=provider, ), ] - client = Client(**config) + return Client(**config) + + +async def test_workflows_on_different_providers_share_one_worker( + client: Client, +) -> None: + provider = RecordingProvider( + { + TOGETHER_MODEL: _script("Sunny over Tokyo"), + FIREWORKS_MODEL: _script("Clear Tokyo sky"), + } + ) + client = _with_provider(client, provider) task_queue = f"openai-agents-openai-compatible-{uuid.uuid4()}" async with Worker( @@ -122,17 +180,56 @@ async def test_workflow_uses_the_tool_and_the_default_model(client: Client) -> N task_queue=task_queue, workflows=[OpenAICompatibleAgentWorkflow], ): - result = await client.execute_workflow( - OpenAICompatibleAgentWorkflow.run, - "What's the weather in Tokyo?", - id=f"openai-compatible-{uuid.uuid4()}", - task_queue=task_queue, + results = await asyncio.gather( + *( + client.execute_workflow( + OpenAICompatibleAgentWorkflow.run, + OpenAICompatibleAgentInput( + prompt="What's the weather in Tokyo?", model=model + ), + id=f"openai-compatible-{uuid.uuid4()}", + task_queue=task_queue, + ) + for model in (TOGETHER_MODEL, FIREWORKS_MODEL) + ) ) - assert result == "Sunny over Tokyo" - # The Agents SDK's defaults for OpenAI's default model (a reasoning effort - # and verbosity) must not reach a provider that rejects them. - assert len(model.settings) == 2 - for settings in model.settings: - assert settings.reasoning is None - assert settings.verbosity is None + assert results == ["Sunny over Tokyo", "Clear Tokyo sky"] + # Each Workflow's two model calls (the tool-call turn and the answer) were + # routed by its own model name. + assert sorted(provider.requested, key=str) == sorted( + [TOGETHER_MODEL, TOGETHER_MODEL, FIREWORKS_MODEL, FIREWORKS_MODEL] + ) + for model in provider.models.values(): + assert len(model.settings) == 2 + for settings in model.settings: + # The agent names its model, so the Agents SDK sends none of the + # settings meant for OpenAI's default model. + assert settings.reasoning is None + assert settings.verbosity is None + + +async def test_a_provider_without_a_key_fails_the_workflow_without_retrying( + client: Client, monkeypatch: pytest.MonkeyPatch +) -> None: + client = _with_provider(client, _provider_with_keys(monkeypatch, "together")) + task_queue = f"openai-agents-openai-compatible-{uuid.uuid4()}" + + async with Worker( + client, + task_queue=task_queue, + workflows=[OpenAICompatibleAgentWorkflow], + ): + with pytest.raises(WorkflowFailureError) as err: + await client.execute_workflow( + OpenAICompatibleAgentWorkflow.run, + OpenAICompatibleAgentInput(prompt="Hello", model=FIREWORKS_MODEL), + id=f"openai-compatible-{uuid.uuid4()}", + task_queue=task_queue, + ) + + activity_error = err.value.cause + assert isinstance(activity_error, ActivityError) + assert activity_error.retry_state == RetryState.NON_RETRYABLE_FAILURE + assert isinstance(activity_error.cause, ApplicationError) + assert "FIREWORKS_API_KEY" in activity_error.cause.message