18
18
19
19
from agents import Agent , function_tool , ModelSettings , WebSearchTool
20
20
from pydantic import BaseModel
21
+ from typing import List , Dict , Union
21
22
22
23
pytest_plugins = []
23
24
@@ -45,6 +46,13 @@ def environment():
45
46
@pytest .fixture (autouse = True )
46
47
def clear_exporter (exporter ):
47
48
exporter .clear ()
49
+ from opentelemetry .instrumentation .openai_agents import (
50
+ _root_span_storage ,
51
+ _instrumented_tools ,
52
+ )
53
+
54
+ _root_span_storage .clear ()
55
+ _instrumented_tools .clear ()
48
56
49
57
50
58
@pytest .fixture (scope = "session" )
@@ -88,9 +96,7 @@ async def get_weather(city: str) -> str:
88
96
89
97
return Agent (
90
98
name = "WeatherAgent" ,
91
- instructions = (
92
- "You get the weather for a city using the get_weather tool."
93
- ),
99
+ instructions = ("You get the weather for a city using the get_weather tool." ),
94
100
model = "gpt-4.1" ,
95
101
tools = [get_weather ],
96
102
)
@@ -109,10 +115,12 @@ def web_search_tool_agent():
109
115
@pytest .fixture (scope = "session" )
110
116
def handoff_agent ():
111
117
112
- agent_a = Agent (name = "AgentA" , instructions = "Agent A does something." ,
113
- model = "gpt-4.1" )
114
- agent_b = Agent (name = "AgentB" , instructions = "Agent B does something else." ,
115
- model = "gpt-4.1" )
118
+ agent_a = Agent (
119
+ name = "AgentA" , instructions = "Agent A does something." , model = "gpt-4.1"
120
+ )
121
+ agent_b = Agent (
122
+ name = "AgentB" , instructions = "Agent B does something else." , model = "gpt-4.1"
123
+ )
116
124
117
125
class HandoffExample (BaseModel ):
118
126
message : str
@@ -131,11 +139,136 @@ class HandoffExample(BaseModel):
131
139
instructions = "You decide which agent to handoff to." ,
132
140
model = "gpt-4.1" ,
133
141
handoffs = [agent_a , agent_b ],
134
- tools = [handoff_tool_a , handoff_tool_b ]
142
+ tools = [handoff_tool_a , handoff_tool_b ],
135
143
)
136
144
return triage_agent
137
145
138
146
147
+ @pytest .fixture (scope = "session" )
148
+ def recipe_workflow_agents ():
149
+ """Create Main Chat Agent and Recipe Editor Agent with function tools for recipe management."""
150
+
151
+ class Recipe (BaseModel ):
152
+ id : str
153
+ name : str
154
+ ingredients : List [str ]
155
+ instructions : List [str ]
156
+ prep_time : str
157
+ cook_time : str
158
+ servings : int
159
+
160
+ class SearchResponse (BaseModel ):
161
+ status : str
162
+ message : str
163
+ recipes : Union [Dict [str , Recipe ], None ] = None
164
+ recipe_count : Union [int , None ] = None
165
+ query : Union [str , None ] = None
166
+
167
+ class EditResponse (BaseModel ):
168
+ status : str
169
+ message : str
170
+ modified_recipe : Union [Recipe , None ] = None
171
+ changes_made : Union [List [str ], None ] = None
172
+ original_recipe : Union [Recipe , None ] = None
173
+
174
+ # Mock recipe database
175
+ MOCK_RECIPES = {
176
+ "spaghetti_carbonara" : {
177
+ "id" : "spaghetti_carbonara" ,
178
+ "name" : "Spaghetti Carbonara" ,
179
+ "ingredients" : [
180
+ "400g spaghetti" ,
181
+ "200g pancetta" ,
182
+ "4 large eggs" ,
183
+ "100g Pecorino Romano cheese" ,
184
+ ],
185
+ "instructions" : [
186
+ "Cook spaghetti" ,
187
+ "Dice pancetta" ,
188
+ "Whisk eggs with cheese" ,
189
+ ],
190
+ "prep_time" : "10 minutes" ,
191
+ "cook_time" : "15 minutes" ,
192
+ "servings" : 4 ,
193
+ }
194
+ }
195
+
196
+ @function_tool
197
+ async def search_recipes (query : str = "" ) -> SearchResponse :
198
+ """Search and browse recipes in the database."""
199
+ if "carbonara" in query .lower ():
200
+ recipe_data = MOCK_RECIPES ["spaghetti_carbonara" ]
201
+ recipes_dict = {"spaghetti_carbonara" : Recipe (** recipe_data )}
202
+ return SearchResponse (
203
+ status = "success" ,
204
+ message = f'Found 1 recipes matching "{ query } "' ,
205
+ recipes = recipes_dict ,
206
+ recipe_count = 1 ,
207
+ query = query ,
208
+ )
209
+ return SearchResponse (
210
+ status = "success" ,
211
+ message = "No recipes found" ,
212
+ recipes = {},
213
+ recipe_count = 0 ,
214
+ query = query ,
215
+ )
216
+
217
+ @function_tool
218
+ async def plan_and_apply_recipe_modifications (
219
+ recipe : Recipe , modification_request : str
220
+ ) -> EditResponse :
221
+ """Plan modifications to a recipe based on user request and apply them."""
222
+
223
+ if (
224
+ "vegetarian" in modification_request .lower ()
225
+ and "carbonara" in recipe .name .lower ()
226
+ ):
227
+ modified_recipe = Recipe (
228
+ id = recipe .id ,
229
+ name = "Vegetarian Carbonara" ,
230
+ ingredients = [
231
+ "400g spaghetti" ,
232
+ "200g mushrooms" ,
233
+ "4 large eggs" ,
234
+ "100g Pecorino Romano cheese" ,
235
+ ],
236
+ instructions = [
237
+ "Cook spaghetti" ,
238
+ "Sauté mushrooms" ,
239
+ "Whisk eggs with cheese" ,
240
+ ],
241
+ prep_time = recipe .prep_time ,
242
+ cook_time = recipe .cook_time ,
243
+ servings = recipe .servings ,
244
+ )
245
+ return EditResponse (
246
+ status = "success" ,
247
+ message = "Successfully modified Spaghetti Carbonara to be vegetarian" ,
248
+ modified_recipe = modified_recipe ,
249
+ changes_made = ["Replaced pancetta with mushrooms" ],
250
+ original_recipe = recipe ,
251
+ )
252
+
253
+ return EditResponse (status = "error" , message = "Could not modify recipe" )
254
+
255
+ recipe_editor_agent = Agent (
256
+ name = "Recipe Editor Agent" ,
257
+ instructions = "You are a recipe editor specialist. Help users search and modify recipes using your tools." ,
258
+ model = "gpt-4o" ,
259
+ tools = [search_recipes , plan_and_apply_recipe_modifications ],
260
+ )
261
+
262
+ main_chat_agent = Agent (
263
+ name = "Main Chat Agent" ,
264
+ instructions = "You handle general conversation and route recipe tasks to the recipe editor agent." ,
265
+ model = "gpt-4o" ,
266
+ handoffs = [recipe_editor_agent ],
267
+ )
268
+
269
+ return main_chat_agent , recipe_editor_agent
270
+
271
+
139
272
@pytest .fixture (scope = "module" )
140
273
def vcr_config ():
141
274
return {"filter_headers" : ["authorization" , "api-key" ]}
0 commit comments