| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240 |
- ---
- title: "Output Guardrails"
- description: "Validate agent responses before they reach users."
- icon: "arrow-right-from-bracket"
- ---
- Output guardrails validate agent responses **before** they reach users or other agents. When a guardrail trips, the agent receives feedback and retries.
- ## Function Signature
- Each output guardrail receives three parameters:
- ```python
- from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail
- from pydantic import BaseModel
- @output_guardrail
- async def my_output_guardrail(
- context: RunContextWrapper,
- agent: Agent,
- response_text: str | BaseModel,
- ) -> GuardrailFunctionOutput:
- """Validate agent output."""
- return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)
- ```
- **Parameters:**
- - `context`: Run context wrapper with access to shared state.
- - `agent`: The Agent instance generating the response.
- - `response_text`: The agent response as a string, or a structured model when `output_type` is set.
- **Return:**
- - `GuardrailFunctionOutput` with:
- - `tripwire_triggered` (bool): `True` if validation failed.
- - `output_info` (str): Feedback message sent to the agent when `tripwire_triggered=True`.
- ## Basic Output Guardrail
- ```python
- from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail
- @output_guardrail
- async def response_content_guardrail(
- context: RunContextWrapper, agent: Agent, response_text: str
- ) -> GuardrailFunctionOutput:
- tripwire_triggered = "bad word" in response_text.lower()
- output_info = "Please avoid using inappropriate language." if tripwire_triggered else ""
- return GuardrailFunctionOutput(output_info=output_info, tripwire_triggered=tripwire_triggered)
- agent = Agent(
- name="CustomerSupportAgent",
- instructions="You are a helpful customer support agent.",
- output_guardrails=[response_content_guardrail],
- )
- ```
- ## Practical Example: Preventing Sensitive Information Leaks
- ```python
- from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail
- @output_guardrail(name="ForbidSensitiveEmail")
- async def forbid_sensitive_email(
- context: RunContextWrapper, agent: Agent, response_text: str
- ) -> GuardrailFunctionOutput:
- if "@" in response_text:
- return GuardrailFunctionOutput(
- output_info="Do not share email addresses. Offer to connect via the support portal instead.",
- tripwire_triggered=True,
- )
- return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)
- support_agent = Agent(
- name="SupportPilot",
- instructions="You handle customer support. Official email: support@example.com.",
- model="gpt-5.4-mini",
- output_guardrails=[forbid_sensitive_email],
- validation_attempts=1,
- )
- ```
- See the full example at [`examples/guardrails_output.py`](https://github.com/VRSEN/agency-swarm/blob/main/examples/guardrails_output.py).
- ## Example: Simple Format Enforcement
- ```python
- import json
- from agency_swarm import GuardrailFunctionOutput, RunContextWrapper, output_guardrail
- @output_guardrail(name="RequireJSONFormat")
- async def require_json_format(
- context: RunContextWrapper, agent: Agent, response_text: str
- ) -> GuardrailFunctionOutput:
- try:
- json.loads(response_text)
- return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)
- except json.JSONDecodeError:
- return GuardrailFunctionOutput(
- output_info="Response must be valid JSON. Wrap your response in curly braces.",
- tripwire_triggered=True,
- )
- ```
- ## Output Guardrail Retry Flow
- When an output guardrail trips, the agent gets multiple chances to fix its response. The `validation_attempts` parameter controls this behavior.
- ### How Retry Works
- <Steps>
- <Step title="Agent generates response">
- The agent produces its initial response.
- </Step>
- <Step title="Output guardrail checks response">
- Each output guardrail validates the response.
- </Step>
- <Step title="If validation fails">
- The agent receives a **system message** containing the guardrail `output_info`.
- </Step>
- <Step title="Agent retries">
- The agent generates a new response, informed by that message.
- </Step>
- <Step title="Repeat until success or limit reached">
- This cycle continues up to `validation_attempts` times.
- </Step>
- <Step title="If all attempts fail">
- `OutputGuardrailTripwireTriggered` is raised.
- </Step>
- </Steps>
- ## Configure `validation_attempts`
- ```python
- agent = Agent(
- name="CustomerSupportAgent",
- instructions="You are a helpful customer support agent.",
- output_guardrails=[response_content_guardrail],
- validation_attempts=2,
- )
- ```
- | Setting | Behavior |
- |---------|----------|
- | `validation_attempts=0` | Fail-fast (no retry, immediate exception) |
- | `validation_attempts=1` | Default (one retry after initial failure) |
- | `validation_attempts=2+` | Multiple retries for more complex validations |
- <Note>
- Each retry sends the guardrail `output_info` message to the agent as a system message, giving the agent context to adjust its response.
- </Note>
- ## Handling Validation Failures
- ```python
- from agency_swarm import OutputGuardrailTripwireTriggered
- try:
- response = await agency.get_response("Hello!")
- except OutputGuardrailTripwireTriggered as exc:
- print(f"Validation failed: {exc.guardrail_result.output_info}")
- ```
- ## Message History
- Output guardrail failures are stored as system messages with `message_origin="output_guardrail_error"`.
- For most use cases, `role`, `content`, and `message_origin` are enough. Extra metadata is mainly for debugging and run tracing.
- | Origin | Meaning |
- |--------|---------|
- | `output_guardrail_error` | Output guardrail failure (system message) |
- <Accordion title="Example history entry (with debug metadata)">
- ```json
- {
- "role": "system",
- "content": "You are not allowed to include your email address in your response. Ask agent to redirect user to the contact page: https://www.example.com/contact",
- "message_origin": "output_guardrail_error",
- "agent": "DatabaseAgent",
- "callerAgent": "CustomerSupportAgent",
- "agent_run_id": "agent_run_id",
- "parent_run_id": "call_id",
- "timestamp": 1758103770629217,
- "type": "message"
- }
- ```
- </Accordion>
- ## Agent-to-Agent Validation
- Use guardrails to control how agents communicate with each other. When adding communication flows between agents, the recipient agent's guardrails define the message format.
- <Accordion title="Example: Task/Response contract between agents">
- ```python
- from agency_swarm import Agency, Agent, GuardrailFunctionOutput, RunContextWrapper, input_guardrail, output_guardrail
- @input_guardrail(name="RequireTaskPrefix")
- async def require_task_prefix(
- context: RunContextWrapper, agent: Agent, agent_input: str | list[str]
- ) -> GuardrailFunctionOutput:
- text = agent_input if isinstance(agent_input, str) else " ".join(agent_input)
- blocked = not text.startswith("Task:")
- return GuardrailFunctionOutput(
- output_info="ERROR: Requests to this agent must begin with 'Task:'" if blocked else "",
- tripwire_triggered=blocked,
- )
- @output_guardrail(name="RequireResponsePrefix")
- async def require_response_prefix(
- context: RunContextWrapper, agent: Agent, response_text: str
- ) -> GuardrailFunctionOutput:
- blocked = not response_text.startswith("Response:")
- return GuardrailFunctionOutput(
- output_info="ERROR: Responses must start with 'Response:'" if blocked else "",
- tripwire_triggered=blocked,
- )
- ceo = Agent(name="CEO", instructions="You are the CEO agent.")
- worker = Agent(
- name="Worker",
- instructions="You are the worker agent.",
- input_guardrails=[require_task_prefix],
- output_guardrails=[require_response_prefix],
- raise_input_guardrail_error=True,
- )
- agency = Agency(ceo, communication_flows=[(ceo, worker)])
- ```
- </Accordion>
- In this example:
- - If the CEO sends a message that does not start with `Task:`, the worker input guardrail triggers.
- - The CEO receives an error and adjusts its message.
- - The worker output guardrail enforces `Response:` in returned messages.
- <Note>
- Agent-to-agent messages are always single strings, so input guardrails for inter-agent communication receive a string (not a list).
- </Note>
|