Output Validation Post-Hook
Use post-hooks to validate a Team's output for comprehensiveness, collaboration, consistency, and safety, raising OutputCheckError on failure.
Validate a team's completed output with a post-hook, then display it only when the run status is RunStatus.completed. OutputCheckError marks the run as RunStatus.error; the returned content may still contain the rejected answer.
Use nonstreaming arun() as below when validation must finish before display. A streaming post-hook runs after content chunks have already been emitted and cannot retract them.
These curated examples check the returned run status and the validator's structured result. A validator call makes an additional model request; its assessment is probabilistic. The example treats an unavailable or malformed assessment as a failed check.
Code
import asyncio
from agno.agent import Agent
from agno.exceptions import CheckTrigger, OutputCheckError
from agno.models.openai import OpenAIResponses
from agno.run.team import TeamRunOutput
from agno.team import Team
from agno.run import RunStatus
from pydantic import BaseModel
class TeamOutputValidationResult(BaseModel):
is_comprehensive: bool
shows_collaboration: bool
is_consistent: bool
is_professional: bool
is_safe: bool
concerns: list[str]
confidence_score: float
def validate_team_response_quality(run_output: TeamRunOutput, team: Team) -> None:
"""Validate team output quality and collaboration consistency."""
if not isinstance(run_output.content, str) or len(run_output.content.strip()) < 20:
raise OutputCheckError(
"Team response is too short or empty",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
team_context = f"Team '{team.name}' with {len(team.members)} members: "
team_context += ", ".join(
[
f"{member.name} ({getattr(member, 'description', 'No description')})"
for member in team.members
]
)
validator_agent = Agent(
name="Team Output Validator",
model=OpenAIResponses(id="gpt-5.2"),
instructions=[
"You are a team output quality validator. Analyze team responses for:",
"1. COMPREHENSIVENESS: Response covers multiple areas of expertise relevant to the question",
"2. COLLABORATION: Response integrates multiple perspectives into a coherent answer.",
" A well-synthesized unified response DOES count as collaboration - it does NOT need explicit member attribution or handoffs.",
" If the response covers topics from different domains (e.g. legal, tax, risk), that shows collaboration.",
"3. CONSISTENCY: Different perspectives are coherent and don't contradict each other",
"4. PROFESSIONALISM: Language is professional and appropriate",
"5. SAFETY: Content is safe and doesn't contain harmful advice",
"",
"Provide a confidence score (0.0-1.0) for overall quality.",
"List any specific concerns.",
"",
"Be lenient - a comprehensive, multi-perspective response should pass even if it reads as a unified document.",
],
output_schema=TeamOutputValidationResult,
)
try:
validation_result = validator_agent.run(
input=f"""
{team_context}
Validate this team response: '{run_output.content}'
Consider:
- Does it show multiple perspectives working together?
- Is it more valuable than a single agent response would be?
- Are the different viewpoints consistent and complementary?
"""
)
except Exception as exc:
raise OutputCheckError("Validator unavailable; validation is required.") from exc
if validation_result.status != RunStatus.completed or not isinstance(
validation_result.content, TeamOutputValidationResult
):
raise OutputCheckError("Validator did not return a valid assessment.")
result = validation_result.content
if not 0.0 <= result.confidence_score <= 1.0:
raise OutputCheckError("Validator returned an invalid confidence score.")
if not result.is_comprehensive:
raise OutputCheckError(
f"Team response lacks comprehensiveness. Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if not result.shows_collaboration:
raise OutputCheckError(
f"Response doesn't show effective team collaboration. Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if not result.is_consistent:
raise OutputCheckError(
f"Team response contains inconsistencies between member perspectives. Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if not result.is_professional:
raise OutputCheckError(
f"Team response lacks professional tone. Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if not result.is_safe:
raise OutputCheckError(
f"Team response contains potentially unsafe content. Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if result.confidence_score < 0.7:
raise OutputCheckError(
f"Team response quality score too low ({result.confidence_score:.2f}). Concerns: {', '.join(result.concerns)}",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
def simple_team_coordination_check(run_output: TeamRunOutput, team: Team) -> None:
"""Apply lightweight checks for evidence of team collaboration."""
if not isinstance(run_output.content, str):
raise OutputCheckError("Expected a text response")
content = run_output.content.strip()
team_indicators = [
"we recommend",
"our analysis",
"team",
"collectively",
"different perspectives",
"combined",
"consensus",
"coordinate",
]
member_mentions = sum(
1 for member in team.members if member.name.lower() in content.lower()
)
has_team_language = any(
indicator in content.lower() for indicator in team_indicators
)
if not has_team_language and member_mentions < 2:
raise OutputCheckError(
"Response doesn't show evidence of team collaboration or multiple perspectives",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
if len(content) < 100:
raise OutputCheckError(
"Team response is too brief to demonstrate collaborative value",
check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
)
team_with_validation = Team(
name="Legal Advisory Team",
members=[
Agent(
name="Corporate Lawyer",
model=OpenAIResponses(id="gpt-5.2"),
description="Expert in corporate law, contracts, and compliance",
),
Agent(
name="Tax Attorney",
model=OpenAIResponses(id="gpt-5.2"),
description="Specialist in tax law, regulations, and planning",
),
Agent(
name="Risk Analyst",
model=OpenAIResponses(id="gpt-5.2"),
description="Expert in legal risk assessment and mitigation",
),
],
post_hooks=[validate_team_response_quality],
instructions=[
"Collaborate to provide comprehensive legal guidance:",
"Corporate Lawyer: Address legal structure, compliance, and contracts",
"Tax Attorney: Cover tax implications and optimization strategies",
"Risk Analyst: Identify and assess legal risks and mitigation approaches",
"",
"Work together to provide coordinated legal advice that leverages all expertise areas.",
],
)
team_simple = Team(
name="Content Creation Team",
members=[
Agent(name="Writer", model=OpenAIResponses(id="gpt-5.2")),
Agent(name="Editor", model=OpenAIResponses(id="gpt-5.2")),
],
post_hooks=[simple_team_coordination_check],
instructions=[
"Collaborate to create high-quality content with proper writing and editing coordination."
],
)
async def main() -> None:
for team, request in [
(team_with_validation, "Compare the legal, tax, and risk considerations for LLC and C-Corp structures."),
(team_simple, "Draft and edit a blog post about remote work."),
]:
response = await team.arun(request)
if response.status != RunStatus.completed:
# Rejected content can still be present in response.content.
print("Output validation or generation failed; answer withheld.")
continue
print(response.content)
if __name__ == "__main__":
asyncio.run(main())Usage
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activateInstall dependencies
uv pip install -U agno openaiExport your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"Run example
Save the code above as output_validation_post_hook.py, then run:
python output_validation_post_hook.py