Review dell'OpenAI Agents SDK

1
from agents import Agent, FileSearchTool, Runner, WebSearchTool
2

3
agent = Agent(
4
    name="Assistant",
5
    tools=[
6
        WebSearchTool(),
7
        FileSearchTool(
8
            max_num_results=3,
9
            vector_store_ids=["VECTOR_STORE_ID"],
10
        ),
11
    ],
12
)
13

14
async def main():
15
    result = await Runner.run(agent, "Which coffee shop should I go to, taking into account my preferences and the weather today in SF?")
16
    print(result.final_output)

1
import json
2

3
from typing_extensions import TypedDict, Any
4

5
from agents import Agent, FunctionTool, RunContextWrapper, function_tool
6

7
class Location(TypedDict):
8
    lat: float
9
    long: float
10

11
@function_tool
12
async def fetch_weather(location: Location) -> str:
13

14
    """Fetch the weather for a given location.
15

16
    Args:
17
        location: The location to fetch the weather for.
18
    """
19
    # In real life, we'd fetch the weather from a weather API
20
    return "sunny"
21

22
@function_tool(name_override="fetch_data")
23
def read_file(ctx: RunContextWrapper[Any], path: str, directory: str | None = None) -> str:
24
    """Read the contents of a file.
25

26
    Args:
27
        path: The path to the file to read.
28
        directory: The directory to read the file from.
29
    """
30
    # In real life, we'd read the file from the file system
31
    return "<file contents>"
32

33
agent = Agent(
34
    name="Assistant",
35
    tools=[fetch_weather, read_file],
36
)
37

38
for tool in agent.tools:
39
    if isinstance(tool, FunctionTool):
40
        print(tool.name)
41
        print(tool.description)
42
        print(json.dumps(tool.params_json_schema, indent=2))
43
        print()

1
from agents import Agent, Runner
2
import asyncio
3

4
spanish_agent = Agent(
5
    name="Spanish agent",
6
    instructions="You translate the user's message to Spanish",
7
)
8

9
french_agent = Agent(
10
    name="French agent",
11
    instructions="You translate the user's message to French",
12
)
13

14
orchestrator_agent = Agent(
15
    name="orchestrator_agent",
16
    instructions=(
17
        "You are a translation agent. You use the tools given to you to translate."
18
        "If asked for multiple translations, you call the relevant tools."
19
    ),
20
    tools=[
21
        spanish_agent.as_tool(
22
            tool_name="translate_to_spanish",
23
            tool_description="Translate the user's message to Spanish",
24
        ),
25
        french_agent.as_tool(
26
            tool_name="translate_to_french",
27
            tool_description="Translate the user's message to French",
28
        ),
29
    ],
30
)
31

32
async def main():
33
    result = await Runner.run(orchestrator_agent, input="Say 'Hello, how are you?' in Spanish.")
34
    print(result.final_output)

1
from agents import Agent, handoff
2

3
billing_agent = Agent(name="Billing agent")
4
refund_agent = Agent(name="Refund agent")
5

6
triage_agent = Agent(name="Triage agent", handoffs=[billing_agent, handoff(refund_agent)])

1
from agents import Agent
2
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
3

4
billing_agent = Agent(
5
    name="Billing agent",
6
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
7
    <Fill in the rest of your prompt here>.""",
8
)

1
import asyncio
2
import random
3
from agents import Agent, ItemHelpers, Runner, function_tool
4

5
@function_tool
6
def how_many_jokes() -> int:
7
    return random.randint(1, 10)
8

9
async def main():
10
    agent = Agent(
11
        name="Joker",
12
        instructions="First call the `how_many_jokes` tool, then tell that many jokes.",
13
        tools=[how_many_jokes],
14
    )
15

16
    result = Runner.run_streamed(
17
        agent,
18
        input="Hello",
19
    )
20
    print("=== Run starting ===")
21

22
    async for event in result.stream_events():
23
        # We'll ignore the raw responses event deltas
24
        if event.type == "raw_response_event":
25
            continue
26
        # When the agent updates, print that
27
        elif event.type == "agent_updated_stream_event":
28
            print(f"Agent updated: {event.new_agent.name}")
29
            continue
30
        # When items are generated, print them
31
        elif event.type == "run_item_stream_event":
32
            if event.item.type == "tool_call_item":
33
                print("-- Tool was called")
34
            elif event.item.type == "tool_call_output_item":
35
                print(f"-- Tool output: {event.item.output}")
36
            elif event.item.type == "message_output_item":
37
                print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}")
38
            else:
39
                pass  # Ignore other event types
40

41
    print("=== Run complete ===")
42

43
if __name__ == "__main__":
44
    asyncio.run(main())

1
from agents import Agent, Runner, AsyncOpenAI, OpenAIChatCompletionsModel
2
import asyncio
3

4
spanish_agent = Agent(
5
    name="Spanish agent",
6
    instructions="You only speak Spanish.",
7
    model="o3-mini",
8
)
9

10
english_agent = Agent(
11
    name="English agent",
12
    instructions="You only speak English",
13
    model=OpenAIChatCompletionsModel(
14
        model="gpt-4o",
15
        openai_client=AsyncOpenAI()
16
    ),
17
)
18

19
triage_agent = Agent(
20
    name="Triage agent",
21
    instructions="Handoff to the appropriate agent based on the language of the request.",
22
    handoffs=[spanish_agent, english_agent],
23
    model="gpt-4o",
24
)
25

26
async def main():
27
    result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
28
    print(result.final_output)

1
external_client = AsyncOpenAI(
2
    api_key="EXTERNAL_API_KEY",
3
    base_url="https://api.external.com/v1/",
4
)
5

6
spanish_agent = Agent(
7
    name="Spanish agent",
8
    instructions="You only speak Spanish.",
9
    model=OpenAIChatCompletionsModel(
10
        model="EXTERNAL_MODEL_NAME",
11
        openai_client=external_client,
12
    ),
13
    model_settings=ModelSettings(temperature=0.5),
14
)