Copy OpenAI Agent samples from openai/openai-agents-python

Author: AnatoliB

samples-v2/openai_agents/__init__.py

@@ -0,0 +1,3 @@
+# Make the examples directory into a package to avoid top-level module name collisions.
+# This is needed so that mypy treats files like examples/customer_service/main.py and
+# examples/researcher_app/main.py as distinct modules rather than both named "main".

samples-v2/openai_agents/agent_patterns/README.md

@@ -0,0 +1,54 @@
+# Common agentic patterns
+
+This folder contains examples of different common patterns for agents.
+
+## Deterministic flows
+
+A common tactic is to break down a task into a series of smaller steps. Each task can be performed by an agent, and the output of one agent is used as input to the next. For example, if your task was to generate a story, you could break it down into the following steps:
+
+1. Generate an outline
+2. Generate the story
+3. Generate the ending
+
+Each of these steps can be performed by an agent. The output of one agent is used as input to the next.
+
+See the [`deterministic.py`](./deterministic.py) file for an example of this.
+
+## Handoffs and routing
+
+In many situations, you have specialized sub-agents that handle specific tasks. You can use handoffs to route the task to the right agent.
+
+For example, you might have a frontline agent that receives a request, and then hands off to a specialized agent based on the language of the request.
+See the [`routing.py`](./routing.py) file for an example of this.
+
+## Agents as tools
+
+The mental model for handoffs is that the new agent "takes over". It sees the previous conversation history, and owns the conversation from that point onwards. However, this is not the only way to use agents. You can also use agents as a tool - the tool agent goes off and runs on its own, and then returns the result to the original agent.
+
+For example, you could model the translation task above as tool calls instead: rather than handing over to the language-specific agent, you could call the agent as a tool, and then use the result in the next step. This enables things like translating multiple languages at once.
+
+See the [`agents_as_tools.py`](./agents_as_tools.py) file for an example of this.
+
+## LLM-as-a-judge
+
+LLMs can often improve the quality of their output if given feedback. A common pattern is to generate a response using a model, and then use a second model to provide feedback. You can even use a small model for the initial generation and a larger model for the feedback, to optimize cost.
+
+For example, you could use an LLM to generate an outline for a story, and then use a second LLM to evaluate the outline and provide feedback. You can then use the feedback to improve the outline, and repeat until the LLM is satisfied with the outline.
+
+See the [`llm_as_a_judge.py`](./llm_as_a_judge.py) file for an example of this.
+
+## Parallelization
+
+Running multiple agents in parallel is a common pattern. This can be useful for both latency (e.g. if you have multiple steps that don't depend on each other) and also for other reasons e.g. generating multiple responses and picking the best one.
+
+See the [`parallelization.py`](./parallelization.py) file for an example of this. It runs a translation agent multiple times in parallel, and then picks the best translation.
+
+## Guardrails
+
+Related to parallelization, you often want to run input guardrails to make sure the inputs to your agents are valid. For example, if you have a customer support agent, you might want to make sure that the user isn't trying to ask for help with a math problem.
+
+You can definitely do this without any special Agents SDK features by using parallelization, but we support a special guardrail primitive. Guardrails can have a "tripwire" - if the tripwire is triggered, the agent execution will immediately stop and a `GuardrailTripwireTriggered` exception will be raised.
+
+This is really useful for latency: for example, you might have a very fast model that runs the guardrail and a slow model that runs the actual agent. You wouldn't want to wait for the slow model to finish, so guardrails let you quickly reject invalid inputs.
+
+See the [`input_guardrails.py`](./input_guardrails.py) and [`output_guardrails.py`](./output_guardrails.py) files for examples.

samples-v2/openai_agents/agent_patterns/agents_as_tools.py

@@ -0,0 +1,79 @@
+import asyncio
+
+from agents import Agent, ItemHelpers, MessageOutputItem, Runner, trace
+
+"""
+This example shows the agents-as-tools pattern. The frontline agent receives a user message and
+then picks which agents to call, as tools. In this case, it picks from a set of translation
+agents.
+"""
+
+spanish_agent = Agent(
+    name="spanish_agent",
+    instructions="You translate the user's message to Spanish",
+    handoff_description="An english to spanish translator",
+)
+
+french_agent = Agent(
+    name="french_agent",
+    instructions="You translate the user's message to French",
+    handoff_description="An english to french translator",
+)
+
+italian_agent = Agent(
+    name="italian_agent",
+    instructions="You translate the user's message to Italian",
+    handoff_description="An english to italian translator",
+)
+
+orchestrator_agent = Agent(
+    name="orchestrator_agent",
+    instructions=(
+        "You are a translation agent. You use the tools given to you to translate."
+        "If asked for multiple translations, you call the relevant tools in order."
+        "You never translate on your own, you always use the provided tools."
+    ),
+    tools=[
+        spanish_agent.as_tool(
+            tool_name="translate_to_spanish",
+            tool_description="Translate the user's message to Spanish",
+        ),
+        french_agent.as_tool(
+            tool_name="translate_to_french",
+            tool_description="Translate the user's message to French",
+        ),
+        italian_agent.as_tool(
+            tool_name="translate_to_italian",
+            tool_description="Translate the user's message to Italian",
+        ),
+    ],
+)
+
+synthesizer_agent = Agent(
+    name="synthesizer_agent",
+    instructions="You inspect translations, correct them if needed, and produce a final concatenated response.",
+)
+
+
+async def main():
+    msg = input("Hi! What would you like translated, and to which languages? ")
+
+    # Run the entire orchestration in a single trace
+    with trace("Orchestrator evaluator"):
+        orchestrator_result = await Runner.run(orchestrator_agent, msg)
+
+        for item in orchestrator_result.new_items:
+            if isinstance(item, MessageOutputItem):
+                text = ItemHelpers.text_message_output(item)
+                if text:
+                    print(f"  - Translation step: {text}")
+
+        synthesizer_result = await Runner.run(
+            synthesizer_agent, orchestrator_result.to_input_list()
+        )
+
+    print(f"\n\nFinal response:\n{synthesizer_result.final_output}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

samples-v2/openai_agents/agent_patterns/deterministic.py

@@ -0,0 +1,80 @@
+import asyncio
+
+from pydantic import BaseModel
+
+from agents import Agent, Runner, trace
+
+"""
+This example demonstrates a deterministic flow, where each step is performed by an agent.
+1. The first agent generates a story outline
+2. We feed the outline into the second agent
+3. The second agent checks if the outline is good quality and if it is a scifi story
+4. If the outline is not good quality or not a scifi story, we stop here
+5. If the outline is good quality and a scifi story, we feed the outline into the third agent
+6. The third agent writes the story
+"""
+
+story_outline_agent = Agent(
+    name="story_outline_agent",
+    instructions="Generate a very short story outline based on the user's input.",
+)
+
+
+class OutlineCheckerOutput(BaseModel):
+    good_quality: bool
+    is_scifi: bool
+
+
+outline_checker_agent = Agent(
+    name="outline_checker_agent",
+    instructions="Read the given story outline, and judge the quality. Also, determine if it is a scifi story.",
+    output_type=OutlineCheckerOutput,
+)
+
+story_agent = Agent(
+    name="story_agent",
+    instructions="Write a short story based on the given outline.",
+    output_type=str,
+)
+
+
+async def main():
+    input_prompt = input("What kind of story do you want? ")
+
+    # Ensure the entire workflow is a single trace
+    with trace("Deterministic story flow"):
+        # 1. Generate an outline
+        outline_result = await Runner.run(
+            story_outline_agent,
+            input_prompt,
+        )
+        print("Outline generated")
+
+        # 2. Check the outline
+        outline_checker_result = await Runner.run(
+            outline_checker_agent,
+            outline_result.final_output,
+        )
+
+        # 3. Add a gate to stop if the outline is not good quality or not a scifi story
+        assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
+        if not outline_checker_result.final_output.good_quality:
+            print("Outline is not good quality, so we stop here.")
+            exit(0)
+
+        if not outline_checker_result.final_output.is_scifi:
+            print("Outline is not a scifi story, so we stop here.")
+            exit(0)
+
+        print("Outline is good quality and a scifi story, so we continue to write the story.")
+
+        # 4. Write the story
+        story_result = await Runner.run(
+            story_agent,
+            outline_result.final_output,
+        )
+        print(f"Story: {story_result.final_output}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

samples-v2/openai_agents/agent_patterns/forcing_tool_use.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Literal
+
+from pydantic import BaseModel
+
+from agents import (
+    Agent,
+    FunctionToolResult,
+    ModelSettings,
+    RunContextWrapper,
+    Runner,
+    ToolsToFinalOutputFunction,
+    ToolsToFinalOutputResult,
+    function_tool,
+)
+
+"""
+This example shows how to force the agent to use a tool. It uses `ModelSettings(tool_choice="required")`
+to force the agent to use any tool.
+
+You can run it with 3 options:
+1. `default`: The default behavior, which is to send the tool output to the LLM. In this case,
+    `tool_choice` is not set, because otherwise it would result in an infinite loop - the LLM would
+    call the tool, the tool would run and send the results to the LLM, and that would repeat
+    (because the model is forced to use a tool every time.)
+2. `first_tool_result`: The first tool result is used as the final output.
+3. `custom`: A custom tool use behavior function is used. The custom function receives all the tool
+    results, and chooses to use the first tool result to generate the final output.
+
+Usage:
+python examples/agent_patterns/forcing_tool_use.py -t default
+python examples/agent_patterns/forcing_tool_use.py -t first_tool
+python examples/agent_patterns/forcing_tool_use.py -t custom
+"""
+
+
+class Weather(BaseModel):
+    city: str
+    temperature_range: str
+    conditions: str
+
+
+@function_tool
+def get_weather(city: str) -> Weather:
+    print("[debug] get_weather called")
+    return Weather(city=city, temperature_range="14-20C", conditions="Sunny with wind")
+
+
+async def custom_tool_use_behavior(
+    context: RunContextWrapper[Any], results: list[FunctionToolResult]
+) -> ToolsToFinalOutputResult:
+    weather: Weather = results[0].output
+    return ToolsToFinalOutputResult(
+        is_final_output=True, final_output=f"{weather.city} is {weather.conditions}."
+    )
+
+
+async def main(tool_use_behavior: Literal["default", "first_tool", "custom"] = "default"):
+    if tool_use_behavior == "default":
+        behavior: Literal["run_llm_again", "stop_on_first_tool"] | ToolsToFinalOutputFunction = (
+            "run_llm_again"
+        )
+    elif tool_use_behavior == "first_tool":
+        behavior = "stop_on_first_tool"
+    elif tool_use_behavior == "custom":
+        behavior = custom_tool_use_behavior
+
+    agent = Agent(
+        name="Weather agent",
+        instructions="You are a helpful agent.",
+        tools=[get_weather],
+        tool_use_behavior=behavior,
+        model_settings=ModelSettings(
+            tool_choice="required" if tool_use_behavior != "default" else None
+        ),
+    )
+
+    result = await Runner.run(agent, input="What's the weather in Tokyo?")
+    print(result.final_output)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-t",
+        "--tool-use-behavior",
+        type=str,
+        required=True,
+        choices=["default", "first_tool", "custom"],
+        help="The behavior to use for tool use. Default will cause tool outputs to be sent to the model. "
+        "first_tool_result will cause the first tool result to be used as the final output. "
+        "custom will use a custom tool use behavior function.",
+    )
+    args = parser.parse_args()
+    asyncio.run(main(args.tool_use_behavior))