┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐

│  Decompose  │──▶│   Search    │──▶│  Extract    │──▶│ Synthesize  │

│  Question   │   │   Planning  │   │   Facts     │   │  Answer     │

│             │   │             │   │             │   │             │

│ In: Query   │   │ In: Plan    │   │ In: Results │   │ In: Facts   │

│ Out: SubQ[] │   │ Out: Steps  │   │ Out: Fact[] │   │ Out: Answer │

└──────┬──────┘   └──────┬──────┘   └──────┬──────┘   └──────┬──────┘

       │                 │                 │                 │

       ▼                 ▼                 ▼                 ▼

  [Schema]          [Schema]           [Schema]           [Schema]

  [Guardrail]       [Guardrail]        [Guardrail]        [Guardrail]

  [Eval: 0.9]       [Eval: 0.85]       [Eval: 0.9]        [Eval: 0.95]

# agent_eval/schemas.py

from pydantic import BaseModel, Field

from typing import Literal, Any



class DecomposeInput(BaseModel):

    user_query: str

    context: dict = Field(default_factory=dict)



class DecomposeOutput(BaseModel):

    sub_questions: list[str] = Field(min_length=1, max_length=5)

    requires_tools: bool

    reasoning: str



class PlanInput(BaseModel):

    sub_questions: list[str]

    available_tools: list[str]



class Step(BaseModel):

    query: str

    source: Literal["web", "internal", "api"]

    priority: int = Field(ge=1, le=3)



class PlanOutput(BaseModel):

    steps: list[Step] = Field(min_length=1)

    estimated_confidence: float = Field(ge=0, le=1)



class Fact(BaseModel):

    claim: str

    evidence: str

    source_url: str

    confidence: float = Field(ge=0, le=1)



class ExtractInput(BaseModel):

    tool_results: list[Any]

    original_query: str



class ExtractOutput(BaseModel):

    facts: list[Fact] = Field(min_length=1)

    gaps: list[str] = Field(default_factory=list)

    confidence: float



class Citation(BaseModel):

    text: str

    source_url: str



class SynthesizeInput(BaseModel):

    facts: list[Fact]

    user_query: str

    tone: Literal["professional", "casual", "technical"] = "professional"



class SynthesizeOutput(BaseModel):

    answer: str

    citations: list[Citation]

    confidence: float

    warnings: list[str] = Field(default_factory=list)

# agent_eval/agent.py

class StructuredAgent:

    def __init__(self, steps: list[AgentStep], guardrails: list[Guardrail], evaluator: Evaluator):

        self.steps = steps

        self.guardrails = guardrails

        self.evaluator = evaluator



    async def run(self, input: BaseModel) -> AgentResult:

        context = input.model_dump()

        step_results = []



        for step in self.steps:

            # 1. Execute step with structured output extraction

            output = await self._execute_step(step, context)



            # 2. Validate schema

            validated = step.output_model.model_validate(output)



            # 3. Run guardrails (blocking)

            for guardrail in self.guardrails:

                if not await guardrail.check(validated, context):

                    return AgentResult(blocked=True, reason=guardrail.violation)



            # 4. Evaluate step quality (non-blocking, for observability)

            eval_result = await self.evaluator.evaluate_step(step.name, context, validated)



            step_results.append(StepResult(step=step.name, output=validated, eval=eval_result))

            context.update(validated.model_dump())



        return AgentResult(steps=step_results, final_output=context)

# agent_eval/guardrails.py

from abc import ABC, abstractmethod



class Guardrail(ABC):

    @abstractmethod

    async def check(self, output: BaseModel, context: dict) -> bool: ...



class CitationValidator(Guardrail):

    """Every claim in answer must have a citation from retrieved docs."""

    async def check(self, output: SynthesizeOutput, context: dict) -> bool:

        retrieved_docs = context.get("retrieved_docs", [])

        doc_text = " ".join(d.text for d in retrieved_docs)



        for citation in output.citations:

            if citation.text not in doc_text:

                return False  # Hallucinated citation

        return True



class ConfidenceGate(Guardrail):

    """Block low-confidence outputs."""

    def __init__(self, threshold: float = 0.7):

        self.threshold = threshold



    async def check(self, output: BaseModel, context: dict) -> bool:

        return getattr(output, "confidence", 1.0) >= self.threshold



class FormatEnforcer(Guardrail):

    """Ensure structured output matches schema exactly."""

    async def check(self, output: BaseModel, context: dict) -> bool:

        try:

            type(output).model_validate(output.model_dump())

            return True

        except ValidationError:

            return False



class SafetyGuardrail(Guardrail):

    """PII, harmful content, policy violations."""

    def __init__(self):

        self.detector = instructor.from_openai(AsyncOpenAI())



    async def check(self, output: BaseModel, context: dict) -> bool:

        class SafetyCheck(BaseModel):

            safe: bool

            violations: list[str]



        result = await self.detector.chat.completions.create(

            model="gpt-4o-mini",

            response_model=SafetyCheck,

            messages=[{

                "role": "user",

                "content": f"Check for PII, harmful content, policy violations:\n{output.model_dump_json()}"

            }],

            temperature=0.0,

        )

        return result.safe

# agent_eval/evaluation.py

class StepEvaluator:

    def __init__(self, judges: list[Judge]):

        self.judges = judges



    async def evaluate_step(self, step_name: str, input: dict, output: BaseModel) -> StepEvalResult:

        results = {}

        for judge in self.judges:

            if judge.applies_to(step_name):

                result = await judge.evaluate(input, output)

                results[judge.name] = result



        return StepEvalResult(step=step_name, judge_results=results)



# Judges per step type

DECOMPOSE_JUDGES = [

    LLMJudge("completeness", "All aspects of query covered?", threshold=0.8),

    LLMJudge("no_hallucination", "Sub-questions answerable from available tools?", threshold=0.9),

]



PLAN_JUDGES = [

    LLMJudge("feasibility", "Plan executable with available tools?", threshold=0.85),

    LLMJudge("efficiency", "Minimal steps to answer?", threshold=0.7),

]



EXTRACT_JUDGES = [

    LLMJudge("faithfulness", "Facts supported by tool results?", threshold=0.9),

    LLMJudge("completeness", "All relevant info extracted?", threshold=0.8),

]



SYNTHESIZE_JUDGES = [

    LLMJudge("accuracy", "Answer matches extracted facts?", threshold=0.9),

    LLMJudge("citation_quality", "Citations precise and relevant?", threshold=0.85),

    LLMJudge("tone_adherence", "Matches requested tone?", threshold=0.8),

]

# agent_eval/structured_output.py

import instructor

from openai import AsyncOpenAI

from pydantic import BaseModel, ValidationError



class StructuredExtractor:

    def __init__(self, model="gpt-4o-mini", max_retries=3):

        self.client = instructor.from_openai(AsyncOpenAI())

        self.model = model

        self.max_retries = max_retries



    async def extract(self, response_model: type[BaseModel], prompt: str, 

                      system: str = None, context: dict = None) -> BaseModel:

        messages = []

        if system:

            messages.append({"role": "system", "content": system})

        if context:

            messages.append({"role": "system", "content": f"Context:\n{json.dumps(context)}"})

        messages.append({"role": "user", "content": prompt})



        last_error = None

        for attempt in range(self.max_retries):

            try:

                return await self.client.chat.completions.create(

                    model=self.model,

                    response_model=response_model,

                    messages=messages,

                    temperature=0.0,

                )

            except ValidationError as e:

                last_error = e

                messages.append({"role": "assistant", "content": f"Validation failed: {e}"})

                messages.append({"role": "user", "content": "Fix the validation errors. Output ONLY valid JSON."})



        raise last_error

pip install agent-eval-framework

from agent_eval import StructuredAgent, DecomposeStep, PlanStep, ExtractStep, SynthesizeStep

from agent_eval.guardrails import CitationValidator, ConfidenceGate, SafetyGuardrail

from agent_eval.judges import create_judge_ensemble



agent = StructuredAgent(

    steps=[

        DecomposeStep(),

        PlanStep(),

        ExtractStep(),

        SynthesizeStep(),

    ],

    guardrails=[

        CitationValidator(),

        ConfidenceGate(0.7),

        SafetyGuardrail(),

    ],

    evaluator=StepEvaluator(judges=create_judge_ensemble()),

)



result = await agent.run(DecomposeInput(user_query="How do I reset my 2FA?"))