Structured Output Skill

Role

You are an elite AI output engineer. You know that raw LLM text is unreliable for

production code. You use Pydantic, instructor, and JSON Schema to enforce types,

validate ranges, and retry on failure — making LLM outputs as reliable as database

queries.

The Core Problem

❌ Bad:  response = llm("Extract the brand name")  →  "The brand is Nike!"
✅ Good: response = extract(BrandInfo)              →  BrandInfo(name="Nike", confidence=0.95)

Part 1: Pydantic Models for LLM Outputs

from pydantic import BaseModel, Field, field_validator
from typing import Optional, Literal
from enum import Enum

# --- Fan Segment Model ---
class FanSegment(str, Enum):
    TRIBAL = "tribal"
    PASSIONATE = "passionate"
    CASUAL = "casual"
    DISTANT = "distant"
    CORPORATE = "corporate"


class FanProfile(BaseModel):
    """Structured fan profile extracted from survey response text."""
    segment: FanSegment = Field(..., description="Primary fan segment")
    confidence: float = Field(..., ge=0.0, le=1.0,
                               description="Confidence score 0-1")
    reasoning: str = Field(..., description="One-sentence explanation of classification")
    engagement_score: Optional[int] = Field(None, ge=0, le=100,
                                             description="Engagement score 0-100")
    key_indicators: list[str] = Field(default_factory=list,
                                       description="Top 3 behavioural indicators")

    @field_validator("key_indicators")
    @classmethod
    def limit_indicators(cls, v):
        return v[:3]  # Enforce max 3


class BrandSentimentExtraction(BaseModel):
    """Structured sentiment extraction from brand mention text."""
    brand_name: str
    sentiment: Literal["positive", "negative", "neutral", "mixed"]
    sentiment_score: float = Field(..., ge=-1.0, le=1.0)
    key_themes: list[str] = Field(..., max_length=5)
    notable_quote: Optional[str] = None
    requires_action: bool = False


class ResearchInsight(BaseModel):
    """A single structured research insight."""
    headline: str = Field(..., max_length=100)
    finding: str = Field(..., description="2-3 sentence evidence-based finding")
    implication: str = Field(..., description="Strategic implication for the brand")
    confidence: Literal["high", "medium", "low"]
    data_sources: list[str] = Field(default_factory=list)
    priority: int = Field(..., ge=1, le=5, description="Priority 1=highest")


class IMIReport(BaseModel):
    """Structured IMI research report."""
    brand: str
    report_date: str
    executive_summary: str = Field(..., max_length=500)
    top_insights: list[ResearchInsight] = Field(..., min_length=1, max_length=5)
    fan_index_score: float = Field(..., ge=0, le=100)
    recommended_actions: list[str] = Field(..., max_length=3)

Part 2: instructor — The Production Standard

instructor wraps any LLM client to return Pydantic models directly.

Used by Anthropic engineers, OpenAI teams, and every serious AI product.

# pip install instructor
import instructor
import anthropic
from openai import OpenAI

# --- Claude via instructor ---
claude_client = instructor.from_anthropic(
    anthropic.Anthropic(),
    mode=instructor.Mode.ANTHROPIC_TOOLS  # Use tool-use mode for reliability
)

def extract_claude(text: str, model_class: type[BaseModel],
                    system: str = None,
                    model: str = "claude-haiku-4-5-20251001",
                    max_retries: int = 3) -> BaseModel:
    """
    Extract structured data from text using Claude + instructor.
    Returns a validated Pydantic model instance.
    """
    messages = [{"role": "user", "content": text}]

    return claude_client.chat.completions.create(
        model=model,
        response_model=model_class,
        messages=messages,
        system=system or "Extract structured information accurately.",
        max_retries=max_retries,
    )


# --- OpenAI via instructor ---
oai_client_structured = instructor.from_openai(OpenAI())

def extract_openai(text: str, model_class: type[BaseModel],
                    model: str = "gpt-4o-mini",
                    max_retries: int = 3) -> BaseModel:
    """Extract structured data using OpenAI + instructor."""
    return oai_client_structured.chat.completions.create(
        model=model,
        response_model=model_class,
        messages=[
            {"role": "system", "content": "Extract structured information accurately."},
            {"role": "user", "content": text}
        ],
        max_retries=max_retries,
    )

Part 3: Manual JSON Parsing with Retry

For when you can't use instructor (e.g., local models):

import json, re
from typing import Type, TypeVar
T = TypeVar("T", bound=BaseModel)


def extract_json_from_response(text: str) -> str | None:
    """Extract JSON from LLM response that may have surrounding text."""
    # Try direct parse first
    try:
        json.loads(text)
        return text
    except json.JSONDecodeError:
        pass

    # Try to find JSON block in markdown code fences
    fence_match = re.search(r"```(?:json)?\n([\s\S]*?)\n```", text)
    if fence_match:
        return fence_match.group(1)

    # Try to find raw JSON object
    brace_match = re.search(r"\{[\s\S]*\}", text)
    if brace_match:
        return brace_match.group(0)

    return None


def llm_extract(
    prompt: str,
    schema: Type[T],
    llm_fn: callable,
    max_retries: int = 3,
    system: str = None
) -> T:
    """
    Extract a Pydantic model from LLM with retry on validation failure.
    Works with any LLM function.
    """
    schema_json = json.dumps(schema.model_json_schema(), indent=2)
    full_prompt = f"""{prompt}

Respond with a JSON object matching this exact schema:
{schema_json}

Return ONLY the JSON object. No explanation, no markdown fences."""

    last_error = None
    for attempt in range(max_retries):
        try:
            response = llm_fn(full_prompt, system=system)
            json_str = extract_json_from_response(response)

            if not json_str:
                raise ValueError(f"No JSON found in response: {response[:200]}")

            data = json.loads(json_str)
            return schema.model_validate(data)

        except (json.JSONDecodeError, ValueError) as e:
            last_error = e
            logger.warning(f"Extraction attempt {attempt+1} failed: {e}")
            if attempt < max_retries - 1:
                full_prompt += f"\n\nPrevious attempt failed: {e}. Please fix and try again."

    raise ValueError(f"Failed to extract {schema.__name__} after {max_retries} attempts. Last error: {last_error}")

Part 4: Streaming Structured Output

For long extractions where you want partial results as they arrive:

from instructor import Partial

def stream_fan_profile(text: str) -> Generator[FanProfile, None, None]:
    """
    Stream a partial FanProfile as The AI generates it.
    Returns progressively more complete objects.
    """
    for partial_response in claude_client.chat.completions.create_partial(
        model="claude-haiku-4-5-20251001",
        response_model=Partial[FanProfile],
        messages=[{"role": "user", "content": text}],
        system="Classify this fan profile."
    ):
        yield partial_response

Part 5: Batch Extraction Pipeline

from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm

def batch_extract(
    texts: list[str],
    schema: Type[T],
    llm_fn: callable,
    max_workers: int = 5,
    show_progress: bool = True
) -> list[T | None]:
    """
    Extract structured data from multiple texts in parallel.
    Returns list aligned with input (None for failures).
    """
    results = [None] * len(texts)

    def _extract_one(idx: int, text: str) -> tuple[int, T | None]:
        try:
            return idx, llm_extract(text, schema, llm_fn)
        except Exception as e:
            logger.error(f"Extraction failed for item {idx}: {e}")
            return idx, None

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(_extract_one, i, text): i
            for i, text in enumerate(texts)
        }

        iterator = as_completed(futures)
        if show_progress:
            iterator = tqdm(iterator, total=len(texts), desc="Extracting")

        for future in iterator:
            idx, result = future.result()
            results[idx] = result

    success_rate = sum(1 for r in results if r is not None) / len(results)
    logger.info(f"Batch extraction: {success_rate:.1%} success rate ({len(texts)} items)")

    return results

Part 6: IMI-Specific Extractors

def extract_fan_segment(survey_response: str) -> FanProfile:
    """Classify a survey response into an IMI fan segment."""
    prompt = f"""Classify this fan survey response into the IMI five-segment model.

IMI Segments:
- tribal: Long-term, emotionally invested, habitual attendance, identity tied to club
- passionate: High engagement, attends regularly, social media active, merchandise buyer
- casual: Watches on TV, occasional match-goer, modest spending
- distant: Low engagement, follows results only, minimal commercial value
- corporate: Attends for hospitality/networking, not emotionally attached

Survey response:
{survey_response}"""

    return extract_claude(prompt, FanProfile)


def extract_brand_sentiment(mention_text: str) -> BrandSentimentExtraction:
    """Extract structured sentiment from a brand mention."""
    prompt = f"""Extract structured sentiment analysis from this brand mention text.

Text: {mention_text}"""

    return extract_claude(prompt, BrandSentimentExtraction)


def extract_research_insights(report_text: str,
                                brand: str,
                                n_insights: int = 3) -> list[ResearchInsight]:
    """Extract top research insights from a report."""

    class InsightList(BaseModel):
        insights: list[ResearchInsight] = Field(
            ..., description=f"Top {n_insights} insights"
        )

    prompt = f"""Extract the top {n_insights} strategic insights for {brand} from this research.

Research text:
{report_text[:4000]}"""

    result = extract_claude(prompt, InsightList)
    return result.insights

Output Standards

• Always use Pydantic models for any LLM output feeding downstream code
• Use instructor for all Claude/OpenAI extractions (most reliable)
• Fall back to manual JSON parsing with retry for local models
• Set max_retries=3 minimum on all extractions
• Log extraction success rates in production — target >95%
• Use Literal types for constrained choices (sentiment, segment, etc.)
• Test schemas with edge cases: empty input, mixed languages, very long text

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)