.

Author: kritinv

deepeval/openai_agents/extractors.py

@@ -35,6 +35,7 @@
         SpanData,
         HandoffSpanData,
         CustomSpanData,
+        GuardrailSpanData,
     )
     openai_agents_available = True
 except ImportError:
@@ -66,6 +67,8 @@ def custom_update_span_attributes(span: BaseSpan, span_data: SpanData):
         update_attributes_from_handoff_span_data(span, span_data)
     elif isinstance(span_data, CustomSpanData):
         update_attributes_from_custom_span_data(span, span_data)
+    elif isinstance(span_data, GuardrailSpanData):
+        update_attributes_from_guardrail_span_data(span, span_data)
 
 ########################################################
 ### LLM Span ###########################################
@@ -213,6 +216,14 @@ def update_attributes_from_custom_span_data(
     span.name = custom_span_data.name
     span.metadata = {"data": custom_span_data.data}
 
+def update_attributes_from_guardrail_span_data(
+    span: BaseSpan, 
+    guardrail_span_data: GuardrailSpanData
+):
+    # Update Span
+    span.name = "Guardrail: " + guardrail_span_data.name 
+    span.metadata = {"data": guardrail_span_data.triggered, "type": guardrail_span_data.type}
+
 ########################################################
 ### Parse Input Utils ##################################
 ########################################################

tests/integrations/openai_agents/output_guardrails_agent.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import asyncio
+import json
+
+from pydantic import BaseModel, Field
+
+from agents import (
+    Agent,
+    GuardrailFunctionOutput,
+    OutputGuardrailTripwireTriggered,
+    RunContextWrapper,
+    Runner,
+    output_guardrail,
+)
+
+"""
+This example shows how to use output guardrails.
+
+Output guardrails are checks that run on the final output of an agent.
+They can be used to do things like:
+- Check if the output contains sensitive data
+- Check if the output is a valid response to the user's message
+
+In this example, we'll use a (contrived) example where we check if the agent's response contains
+a phone number.
+"""
+
+
+# The agent's output type
+class MessageOutput(BaseModel):
+    reasoning: str = Field(description="Thoughts on how to respond to the user's message")
+    response: str = Field(description="The response to the user's message")
+    user_name: str | None = Field(description="The name of the user who sent the message, if known")
+
+
+@output_guardrail
+async def sensitive_data_check(
+    context: RunContextWrapper, agent: Agent, output: MessageOutput
+) -> GuardrailFunctionOutput:
+    phone_number_in_response = "650" in output.response
+    phone_number_in_reasoning = "650" in output.reasoning
+
+    return GuardrailFunctionOutput(
+        output_info={
+            "phone_number_in_response": phone_number_in_response,
+            "phone_number_in_reasoning": phone_number_in_reasoning,
+        },
+        tripwire_triggered=phone_number_in_response or phone_number_in_reasoning,
+    )
+
+
+agent = Agent(
+    name="Assistant",
+    instructions="You are a helpful assistant.",
+    output_type=MessageOutput,
+    output_guardrails=[sensitive_data_check],
+)
+
+
+async def output_guardrails_agent():
+    # This should be ok
+    await Runner.run(agent, "What's the capital of California?")
+    print("First message passed")
+
+    # This should trip the guardrail
+    try:
+        result = await Runner.run(
+            agent, "My phone number is 650-123-4567. Where do you think I live?"
+        )
+        print(
+            f"Guardrail didn't trip - this is unexpected. Output: {json.dumps(result.final_output.model_dump(), indent=2)}"
+        )
+
+    except OutputGuardrailTripwireTriggered as e:
+        print(f"Guardrail tripped. Info: {e.guardrail_result.output.output_info}")
+

tests/integrations/openai_agents/streaming_agent.py

@@ -7,7 +7,6 @@ async def streaming_agent():
         name="Joker",
         instructions="You are a helpful assistant.",
     )
-
     result = Runner.run_streamed(agent, input="Please tell me 5 jokes.")
     async for event in result.stream_events():
         if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):

tests/integrations/openai_agents/test_openai_agent.py

@@ -10,6 +10,7 @@
 
 # Import agents
 from tests.integrations.openai_agents.streaming_guardrails_agent import streaming_guardrails_agent
+from tests.integrations.openai_agents.output_guardrails_agent import output_guardrails_agent
 from tests.integrations.openai_agents.code_interpreter_agent import code_interpreter_agent
 from tests.integrations.openai_agents.customer_service_agent import customer_service_agent
 from tests.integrations.openai_agents.streaming_agent import streaming_agent
@@ -21,17 +22,20 @@
 
 # Run agents
 if __name__ == "__main__":
-    # asyncio.run(customer_service_agent())
-    # asyncio.run(research_agent())
     # if not shutil.which("uvx"):
     #     raise RuntimeError("uvx is not installed. Please install it with `pip install uvx`.")
     # asyncio.run(git_agent())
+
+    # asyncio.run(customer_service_agent())
+    # asyncio.run(research_agent())
     # asyncio.run(code_interpreter_agent())
     # asyncio.run(remote_agent())
     # asyncio.run(streaming_agent())
+    # asyncio.run(streaming_guardrails_agent())
+    asyncio.run(output_guardrails_agent())
 
-    # Run streaming agent 10 times
-    async def gather_streaming_agents():
-        tasks = [streaming_agent() for _ in range(10)]
-        await asyncio.gather(*tasks)
-    asyncio.run(gather_streaming_agents())
+    ## Run streaming agent 10 times
+    # async def gather_streaming_agents():
+    #     tasks = [streaming_agent() for _ in range(10)]
+    #     await asyncio.gather(*tasks)
+    # asyncio.run(gather_streaming_agents())