FilesExpand file tree

 main

/

agent_lifecycle_example.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

140 lines (114 loc) · 4.34 KB

 main

/

agent_lifecycle_example.py

Copy path

Top

File metadata and controls

• Code
• Blame

140 lines (114 loc) · 4.34 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

import asyncio

import random

from typing import Any

from pydantic import BaseModel

from agents import (

Agent,

AgentHookContext,

AgentHooks,

RunContextWrapper,

Runner,

Tool,

function_tool,

)

from examples.auto_mode import input_with_fallback, is_auto_mode

class CustomAgentHooks(AgentHooks):

def __init__(self, display_name: str):

self.event_counter = 0

self.display_name = display_name

async def on_start(self, context: AgentHookContext, agent: Agent) -> None:

self.event_counter += 1

# Access the turn_input from the context to see what input the agent received

print(

f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} started with turn_input: {context.turn_input}"

)

async def on_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None:

self.event_counter += 1

print(

f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} ended with output {output}"

)

async def on_handoff(self, context: RunContextWrapper, agent: Agent, source: Agent) -> None:

self.event_counter += 1

print(

f"### ({self.display_name}) {self.event_counter}: Agent {source.name} handed off to {agent.name}"

)

# Note: The on_tool_start and on_tool_end hooks apply only to local tools.

# They do not include hosted tools that run on the OpenAI server side,

# such as WebSearchTool, FileSearchTool, CodeInterpreterTool, HostedMCPTool,

# or other built-in hosted tools.

async def on_tool_start(self, context: RunContextWrapper, agent: Agent, tool: Tool) -> None:

self.event_counter += 1

print(

f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} started tool {tool.name}"

)

async def on_tool_end(

self, context: RunContextWrapper, agent: Agent, tool: Tool, result: object

) -> None:

self.event_counter += 1

print(

f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} ended tool {tool.name} with result {result}"

)

###

@function_tool

def random_number(max: int) -> int:

"""

Generate a random number from 0 to max (inclusive).

"""

if is_auto_mode():

if max <= 0:

print("[debug] auto mode returning deterministic value 0")

return 0

value = min(max, 37)

if value % 2 == 0:

value = value - 1 if value > 1 else 1

print(f"[debug] auto mode returning deterministic odd number {value}")

return value

return random.randint(0, max)

@function_tool

def multiply_by_two(x: int) -> int:

"""Simple multiplication by two."""

return x * 2

class FinalResult(BaseModel):

number: int

multiply_agent = Agent(

name="Multiply Agent",

instructions="Multiply the number by 2 and then return the final result.",

tools=[multiply_by_two],

output_type=FinalResult,

hooks=CustomAgentHooks(display_name="Multiply Agent"),

)

start_agent = Agent(

name="Start Agent",

instructions="Generate a random number. If it's even, stop. If it's odd, hand off to the multiply agent.",

tools=[random_number],

output_type=FinalResult,

handoffs=[multiply_agent],

hooks=CustomAgentHooks(display_name="Start Agent"),

)

async def main() -> None:

user_input = input_with_fallback("Enter a max number: ", "50")

try:

max_number = int(user_input)

await Runner.run(

start_agent,

input=f"Generate a random number between 0 and {max_number}.",

)

except ValueError:

print("Please enter a valid integer.")

return

print("Done!")

if __name__ == "__main__":

asyncio.run(main())

"""

$ python examples/basic/agent_lifecycle_example.py

Enter a max number: 250

### (Start Agent) 1: Agent Start Agent started

### (Start Agent) 2: Agent Start Agent started tool random_number

### (Start Agent) 3: Agent Start Agent ended tool random_number with result 37

### (Start Agent) 4: Agent Start Agent handed off to Multiply Agent

### (Multiply Agent) 1: Agent Multiply Agent started

### (Multiply Agent) 2: Agent Multiply Agent started tool multiply_by_two

### (Multiply Agent) 3: Agent Multiply Agent ended tool multiply_by_two with result 74

### (Multiply Agent) 4: Agent Multiply Agent ended with output number=74

Done!

"""