Structured Output and Function Calling

What structured output means

Structured output is when you tell a model "I don't want free-form text—I want JSON with these exact fields in this exact schema." Instead of letting the model generate whatever it wants, you constrain its output to a specific format.

Why this matters for RAG: Most RAG systems need more than just text. They need the answer plus citations (which chunks was this based on?), confidence scores, and sometimes structured answers (e.g., a product recommendation with price, availability, rating).

A typical RAG answer should include:

• The actual answer (text)
• Which source documents it came from (citations)
• A confidence score (is this grounded in retrieved chunks, or did the model guess?)

Free-form text answers can't guarantee all of this. Structured output can.

Function calling: the round-trip

Function calling (also called tool use) is how you make models reliable at calling external functions. The model doesn't actually call the function—your code does—but the model learns to structure its output as a function call, and your code interprets it.

The round-trip:

1. You define a function schema (name, parameters, description)
2. You include that schema in the model's instructions
3. The model decides to call a function and emits structured JSON with the function name and arguments
4. Your code parses that JSON and executes the real function
5. You send the result back to the model as a new message
6. The model continues based on the result

Here's a concrete example:

import anthropic
import json

client = anthropic.Anthropic(api_key="your-api-key")

# Step 1: Define your function schemas
tools = [
    {
        "name": "get_product_info",
        "description": "Get details about a product in the catalog",
        "input_schema": {
            "type": "object",
            "properties": {
                "product_id": {
                    "type": "string",
                    "description": "The unique ID of the product"
                }
            },
            "required": ["product_id"]
        }
    },
    {
        "name": "check_inventory",
        "description": "Check if a product is in stock",
        "input_schema": {
            "type": "object",
            "properties": {
                "product_id": {
                    "type": "string",
                    "description": "The ID of the product to check"
                },
                "location": {
                    "type": "string",
                    "description": "Warehouse location (e.g., 'NYC', 'LA')"
                }
            },
            "required": ["product_id", "location"]
        }
    }
]

# Step 2: Define the actual implementations
def get_product_info(product_id: str) -> dict:
    """Real implementation - hits your database."""
    # In reality, this queries a product database
    if product_id == "P001":
        return {"name": "Widget A", "price": 29.99, "rating": 4.5}
    elif product_id == "P002":
        return {"name": "Widget B", "price": 39.99, "rating": 4.8}
    return {"error": "Product not found"}

def check_inventory(product_id: str, location: str) -> dict:
    """Real implementation - checks warehouse system."""
    # In reality, this queries a warehouse system
    return {"product_id": product_id, "location": location, "in_stock": True, "quantity": 15}

# Step 3: Run the model with tool-calling enabled
def rag_answer_with_tools(user_query: str):
    """Answer a question using function calling to retrieve structured data."""

    messages = [
        {
            "role": "user",
            "content": user_query
        }
    ]

    # Step 4: Call the model, requesting tool use
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    # Step 5: Handle the model's response
    # The model might emit tool calls or regular text
    while response.stop_reason == "tool_use":
        # Extract tool call from the response
        tool_use_block = None
        for block in response.content:
            if block.type == "tool_use":
                tool_use_block = block
                break

        if not tool_use_block:
            break

        tool_name = tool_use_block.name
        tool_input = tool_use_block.input

        # Step 6: Execute the real function
        if tool_name == "get_product_info":
            result = get_product_info(tool_input["product_id"])
        elif tool_name == "check_inventory":
            result = check_inventory(tool_input["product_id"], tool_input["location"])
        else:
            result = {"error": f"Unknown tool: {tool_name}"}

        # Step 7: Send the result back to the model
        messages.append({"role": "assistant", "content": response.content})
        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tool_use_block.id,
                    "content": json.dumps(result)
                }
            ]
        })

        # Continue the conversation
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

    # Step 8: Extract the final text response
    final_answer = ""
    for block in response.content:
        if hasattr(block, "text"):
            final_answer = block.text
            break

    return final_answer

# Usage
answer = rag_answer_with_tools("Is Widget A in stock at our LA warehouse? If so, how much does it cost?")
print(answer)
# Output might be: "Widget A is in stock at the LA warehouse with 15 units available.
# It costs $29.99 per unit."

Key insight: The model never actually calls your function. It just outputs JSON like:

{"type": "tool_use", "id": "abcd1234", "name": "get_product_info", "input": {"product_id": "P001"}}

Your code interprets that JSON, runs the real function, and tells the model the result. This gives you:

1. Reliability: The model is constrained to emit valid function calls (no arbitrary text)
2. Auditability: You can log exactly what functions were called and why
3. Composability: One function call can trigger another (multi-step reasoning)

Structured output for RAG citations

The most important use of structured output in RAG is enforcing citation format. A RAG system should always return something like:

{
  "answer": "The return policy allows returns within 30 days if unopened...",
  "citations": [
    {"text": "30 days unopened", "source": "return_policy.pdf", "page": 1},
    {"text": "14 days if opened", "source": "return_policy.pdf", "page": 1}
  ],
  "confidence": 0.92
}

Without structured output, the model might return citations in freeform text ("As mentioned on page 1 of the return policy..."), making it hard to validate or link back to sources.

Here's how to enforce this with Claude's structured output feature:

import anthropic
import json

client = anthropic.Anthropic(api_key="your-api-key")

# Define the exact output schema you want
citation_schema = {
    "type": "object",
    "properties": {
        "answer": {
            "type": "string",
            "description": "The answer to the user's question, grounded in retrieved documents"
        },
        "citations": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "claim": {"type": "string", "description": "The specific claim being cited"},
                    "source_id": {"type": "string", "description": "The ID of the source document"},
                    "relevance_score": {"type": "number", "description": "How relevant this citation is (0-1)"}
                },
                "required": ["claim", "source_id"]
            },
            "description": "List of citations supporting the answer"
        },
        "confidence": {
            "type": "number",
            "description": "How confident the answer is (0-1), based on whether it's grounded in retrieved documents"
        }
    },
    "required": ["answer", "citations", "confidence"]
}

def rag_answer_with_citations(user_query: str, retrieved_chunks: list[str]) -> dict:
    """
    Answer a RAG question with structured citations.
    Enforces that the model must return JSON with specific fields.
    """

    # Format retrieved chunks for the model
    context_text = "Retrieved context:\n"
    for i, chunk in enumerate(retrieved_chunks):
        context_text += f"\n[Source {i}]\n{chunk}\n"

    prompt = f"""{context_text}

User question: {user_query}

Answer the question using ONLY the provided context.
For each claim you make, cite the source by its ID.
If the context doesn't contain the answer, say so explicitly."""

    # Call the model with structured output
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": prompt
            }
        ],
        # This tells Claude to output JSON matching our schema
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "rag_answer",
                "schema": citation_schema,
                "strict": True  # Enforce strict schema matching
            }
        }
    )

    # Extract the JSON response
    response_text = response.content[0].text
    result = json.loads(response_text)

    return result

# Usage
chunks = [
    "Return policy: Items can be returned within 30 days if unopened.",
    "Opened items can be returned within 14 days.",
    "Returns require proof of purchase."
]

result = rag_answer_with_citations(
    "Can I return an opened item?",
    chunks
)

print(json.dumps(result, indent=2))
# Output:
# {
#   "answer": "Yes, you can return an opened item within 14 days.",
#   "citations": [
#     {
#       "claim": "opened items can be returned within 14 days",
#       "source_id": "1",
#       "relevance_score": 0.98
#     }
#   ],
#   "confidence": 0.95
# }

Validation and retry when models emit malformed structures

Even with strict: True and a schema, models sometimes fail to produce valid JSON. You need defensive code:

import anthropic
import json
from typing import Optional

def rag_with_validation_and_retry(
    user_query: str,
    retrieved_chunks: list[str],
    max_retries: int = 3
) -> Optional[dict]:
    """
    Answer with structured output, with validation and retry logic.
    If the model produces invalid JSON or violates the schema, retry with clarification.
    """

    client = anthropic.Anthropic()

    citation_schema = {
        "type": "object",
        "properties": {
            "answer": {"type": "string"},
            "citations": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "claim": {"type": "string"},
                        "source_id": {"type": "string"}
                    },
                    "required": ["claim", "source_id"]
                }
            },
            "confidence": {"type": "number", "minimum": 0, "maximum": 1}
        },
        "required": ["answer", "citations", "confidence"]
    }

    context_text = "Retrieved context:\n"
    for i, chunk in enumerate(retrieved_chunks):
        context_text += f"[Source {i}]\n{chunk}\n"

    for attempt in range(max_retries):
        try:
            # If this is a retry, ask for JSON output more explicitly
            instruction = "Return a JSON object with 'answer', 'citations', and 'confidence' fields."
            if attempt > 0:
                instruction = f"Retry attempt {attempt}. Return ONLY valid JSON. Do not include markdown code blocks. {instruction}"

            prompt = f"""{context_text}

User question: {user_query}

{instruction}"""

            response = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}],
                response_format={
                    "type": "json_schema",
                    "json_schema": {
                        "name": "rag_answer",
                        "schema": citation_schema,
                        "strict": True
                    }
                }
            )

            response_text = response.content[0].text

            # Attempt to parse JSON
            result = json.loads(response_text)

            # Validate required fields
            if not all(k in result for k in ["answer", "citations", "confidence"]):
                raise ValueError("Missing required fields in response")

            # Validate citation format
            for citation in result.get("citations", []):
                if not ("claim" in citation and "source_id" in citation):
                    raise ValueError(f"Invalid citation format: {citation}")

            # Validate confidence is a number between 0 and 1
            if not isinstance(result["confidence"], (int, float)):
                raise ValueError("Confidence must be a number")
            if not 0 <= result["confidence"] <= 1:
                raise ValueError("Confidence must be between 0 and 1")

            print(f"✓ Valid response on attempt {attempt + 1}")
            return result

        except (json.JSONDecodeError, ValueError) as e:
            print(f"✗ Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                # Last retry failed; return a graceful error
                return {
                    "answer": "I encountered an error formatting my response. Please try again.",
                    "citations": [],
                    "confidence": 0.0
                }

    return None

# Usage
chunks = [
    "Return policy: 30 days unopened.",
    "Opened items: 14 days."
]

result = rag_with_validation_and_retry("Can I return opened items?", chunks)
print(json.dumps(result, indent=2))

Why this matters: Even Claude (which is very good at structured output) occasionally produces:

• Invalid JSON (missing quotes, trailing commas)
• JSON that doesn't match the schema (extra fields, wrong types)
• JSON wrapped in markdown code blocks

Defensive validation and retry logic ensures your application doesn't crash when this happens. You can either retry or gracefully degrade to a fallback response.

Worked example: a full RAG pipeline with structured output

Putting it all together: retrieve chunks, call the model with structured output and citations, validate the result:

import anthropic
import json

def full_rag_pipeline(user_query: str, knowledge_base: list[str]) -> dict:
    """
    Complete RAG with retrieval, structured output, and citation.
    """
    from sentence_transformers import SentenceTransformer
    import numpy as np

    # Step 1: Embed query and retrieve relevant chunks
    embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
    query_embedding = embedding_model.encode(user_query)

    # Embed all chunks and find the most similar
    chunk_embeddings = [embedding_model.encode(chunk) for chunk in knowledge_base]
    similarities = [np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb))
                    for emb in chunk_embeddings]

    # Get top-3 chunks
    top_3_indices = np.argsort(similarities)[-3:][::-1]
    retrieved_chunks = [(knowledge_base[i], i) for i in top_3_indices]

    # Step 2: Format context for the model
    context_text = "Retrieved context:\n"
    for chunk_text, source_id in retrieved_chunks:
        context_text += f"[Source {source_id}]\n{chunk_text}\n"

    # Step 3: Call model with structured output
    client = anthropic.Anthropic()

    schema = {
        "type": "object",
        "properties": {
            "answer": {"type": "string"},
            "citations": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "claim": {"type": "string"},
                        "source_id": {"type": "integer"}
                    }
                }
            },
            "grounded": {"type": "boolean"}
        },
        "required": ["answer", "citations", "grounded"]
    }

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"{context_text}\nQuestion: {user_query}\n\nAnswer using the retrieved context."
        }],
        response_format={"type": "json_schema", "json_schema": {"name": "answer", "schema": schema}}
    )

    # Step 4: Parse and return
    result = json.loads(response.content[0].text)
    result["retrieved_sources"] = [knowledge_base[i] for _, i in retrieved_chunks]

    return result

# Usage
kb = [
    "Return policy: 30 days if unopened, 14 days if opened",
    "Shipping: $5 flat or free over $50",
    "Warranty: 1 year parts and labor",
    "Contact: [email protected]"
]

answer = full_rag_pipeline("Can I return something I opened?", kb)
print(json.dumps(answer, indent=2))

Common mistake

Treating structured output as a magic bullet. A JSON schema doesn't make the model more truthful; it just makes the format predictable. A model can still hallucinate claims that aren't in the retrieved context, or cite a source that doesn't actually support the claim.

Structured output + validation + citations creates verifiability, not correctness. You still need retrieval quality, grounding in context, and evaluation metrics to know if your answers are actually right. The schema just makes it easy to check whether the answer references the sources at all.