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..998cd640 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. 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 | Example model name | +|---|---|---| +| 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` | + +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 +``` + +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 --model together/openai/gpt-oss-120b +``` + +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 #### 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..43b4faea --- /dev/null +++ b/openai_agents/model_providers/openai_compatible_providers.py @@ -0,0 +1,113 @@ +"""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. 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 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. The + # starter runs it when no model is given. + default_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", + default_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", + default_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", + default_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): + """Routes ``/`` names to the matching OpenAI-compatible endpoint. + + 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, 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, 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 new file mode 100644 index 00000000..9f54e84c --- /dev/null +++ b/openai_agents/model_providers/run_openai_compatible_worker.py @@ -0,0 +1,66 @@ +"""Worker that serves Baseten, Fireworks AI, and Together AI from one process.""" + +import asyncio +import logging +import sys +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() -> 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 + # 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=[ + 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=model_provider, + ), + ], + ) + + worker = Worker( + client, + task_queue="openai-agents-model-providers-task-queue", + workflows=[OpenAICompatibleAgentWorkflow], + ) + await worker.run() + # @@@SNIPEND + + +if __name__ == "__main__": + 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 new file mode 100644 index 00000000..25ba7387 --- /dev/null +++ b/openai_agents/model_providers/run_openai_compatible_workflow.py @@ -0,0 +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, 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 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(), + ], + ) + # 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__": + 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 new file mode 100644 index 00000000..5884d6dd --- /dev/null +++ b/openai_agents/model_providers/workflows/openai_compatible_workflow.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +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, 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 + async def get_weather(city: str) -> str: + workflow.logger.debug(f"Getting weather for {city}") + return f"The weather in {city} is sunny." + + 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=input.model, + tools=[get_weather], + ) + + result = await Runner.run(agent, input.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..f9453cf9 --- /dev/null +++ b/tests/openai_agents/model_providers_test.py @@ -0,0 +1,235 @@ +"""Offline tests for the OpenAI-compatible model provider sample.""" + +import asyncio +import uuid +from collections.abc import AsyncIterator +from datetime import timedelta +from typing import Any + +import pytest +from agents import ( + AgentOutputSchemaBase, + Handoff, + Model, + ModelProvider, + ModelResponse, + ModelSettings, + ModelTracing, + OpenAIChatCompletionsModel, + Tool, + TResponseInputItem, +) +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 +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 ( + 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.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()) + + +def test_routes_a_prefixed_model_name_to_its_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + 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 + + 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_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): + """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) + + def stream_response( + self, *args: Any, **kwargs: Any + ) -> AsyncIterator[TResponseStreamEvent]: + raise NotImplementedError() + + +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"], + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=30) + ), + model_provider=provider, + ), + ] + 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( + client, + task_queue=task_queue, + workflows=[OpenAICompatibleAgentWorkflow], + ): + 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 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