Roadmap Phase 3: native structured outputs for all LLM call sites

.env.example

@@ -46,6 +46,11 @@ RA_LLM__MODEL=auto
 # With or without /v1 — Ollama providers normalize via normalize_openai_base_url()
 RA_LLM__BASE_URL=http://localhost:11434/v1
 RA_LLM__API_KEY=ollama
+
+# Schema-enforced LLM output (native structured outputs). Disable only for
+# backends that cannot honor response schemas; the legacy JSON-repair path
+# is then used instead.
+# RA_LLM__STRUCTURED_OUTPUTS=false
 # OLLAMA_API_KEY=ollama
 
 # Synthesis — see config/ollama_models.yaml for per-model hints

config/default.yaml

@@ -4,6 +4,7 @@ llm:
   base_url: http://localhost:11434
   temperature: 0.2
   timeout_seconds: 120
+  structured_outputs: true
 
 embedding:
   model: BAAI/bge-small-en-v1.5

docs/configuration/environment-variables.md

@@ -30,6 +30,7 @@ Boolean env values accept standard truthy strings (`true`, `1`, `yes`).
 | Variable | Default | Description |
 |----------|---------|-------------|
 | `RA_LLM__PROVIDER` | `ollama` | `ollama`, `openai`, or `anthropic` |
+| `RA_LLM__STRUCTURED_OUTPUTS` | `true` | Schema-enforced LLM output; disable to use the legacy JSON-repair path |
 | `RA_LLM__MODEL` | `auto` | Model name; `auto` selects from `config/ollama_models.yaml` (Ollama only) |
 | `RA_LLM__BASE_URL` | `http://localhost:11434` | API base URL (Ollama OpenAI-compatible endpoint) |
 

docs/llm/heuristic-vs-llm.md

@@ -124,6 +124,18 @@ RA_QUERY_EXPANSION__LLM_ENABLED=true
 
 YAML equivalents under `synthesis:` and `query_expansion:` in `config/default.yaml`. See [Stage toggles](../configuration/stage-toggles.md).
 
+## Structured outputs
+
+When the LLM path runs, output schemas are enforced natively by default
+(`RA_LLM__STRUCTURED_OUTPUTS=true`): the pydantic model is attached to the
+agent (`output_type`), pydantic-ai validates the response and retries with
+validation feedback on schema violations, and the caller receives a typed
+object or falls back to heuristics. Disable the flag to route through the
+legacy prose-JSON path with the repair/retry machinery in `src/utils/` —
+useful only for backends that cannot honor response schemas. Identity
+fields (paper ID, title) are always taken from retrieval metadata, never
+from model output.
+
 ## Verify LLM is active
 
 Check pipeline logs or debug JSON in `logs/debug/`:

src/analysis/gap_analysis.py

@@ -109,13 +109,6 @@ async def analyze_gaps(
 
         llm_config = get_settings().llm
 
-    agent = AgentFactory(llm_config).create_agent(AgentRole.GAP_ANALYSIS, config=llm_config)
-    response_handler = handler or EnhancedResponseHandler(ResponseHandlerConfig())
-    context = RequestContext(
-        user_query=query,
-        model_name=llm_config.model,
-        session_id=session_id,
-    )
     prompt = _build_gap_prompt(query, synthesis, clusters)
 
     from ..utils.progress_reporter import get_progress_reporter
@@ -124,6 +117,27 @@ async def analyze_gaps(
     if reporter is not None:
         reporter.set_activity("Analyzing research gaps with AI…")
 
+    if llm_config.structured_outputs:
+        from ..models.structured import try_run_structured
+
+        gap_result = await try_run_structured(
+            AgentRole.GAP_ANALYSIS,
+            prompt,
+            GapAnalysisResult,
+            llm_config,
+        )
+        if gap_result is not None:
+            return gap_result
+        return _heuristic_gap_analysis(synthesis, query, clusters)
+
+    agent = AgentFactory(llm_config).create_agent(AgentRole.GAP_ANALYSIS, config=llm_config)
+    response_handler = handler or EnhancedResponseHandler(ResponseHandlerConfig())
+    context = RequestContext(
+        user_query=query,
+        model_name=llm_config.model,
+        session_id=session_id,
+    )
+
     result = await response_handler.process_structured_response(
         agent,
         prompt,

src/analysis/synthesis.py

@@ -343,6 +343,43 @@ def recover_synthesis_output(
     return synthesis
 
 
+def _use_structured_outputs(llm_config: LLMConfig | None) -> bool:
+    if llm_config is not None:
+        return llm_config.structured_outputs
+    from ..config.settings import get_settings
+
+    return get_settings().llm.structured_outputs
+
+
+async def _extract_single_paper_structured(
+    paper: RankedPaper,
+    query: str,
+    llm_config: LLMConfig | None,
+    passages: list[str] | None = None,
+) -> tuple[PaperExtraction, bool]:
+    """Schema-enforced extraction; falls back to heuristics on failure."""
+    from ..models.structured import try_run_structured
+
+    prompt = _build_extraction_prompt(paper, query, passages)
+    extraction = await try_run_structured(
+        AgentRole.EXTRACTION,
+        prompt,
+        PaperExtraction,
+        llm_config,
+    )
+    if extraction is None:
+        return _heuristic_extraction(paper, passages), False
+
+    # Identity fields come from our metadata, never from model output.
+    updates: dict[str, object] = {
+        "paper_id": paper.paper.paper_id,
+        "title": paper.paper.title,
+    }
+    if not extraction.evidence and passages:
+        updates["evidence"] = _evidence_from_passages(passages)
+    return extraction.model_copy(update=updates), True
+
+
 async def _extract_single_paper(
     paper: RankedPaper,
     query: str,
@@ -412,15 +449,21 @@ async def extract_papers(
     llm_targets = ranked_papers[:max_llm_papers]
     heuristic_targets = ranked_papers[max_llm_papers:]
 
-    agent = AgentFactory(llm_config).create_agent(AgentRole.EXTRACTION, config=llm_config)
-    response_handler = handler or EnhancedResponseHandler(
-        _synthesis_handler_config(synthesis_config.extraction_max_retries)
-    )
-    context = RequestContext(
-        user_query=query,
-        model_name=llm_config.model if llm_config else "default",
-        session_id=session_id,
-    )
+    use_structured = _use_structured_outputs(llm_config)
+    agent = None
+    response_handler = None
+    context = None
+    if not use_structured:
+        # Legacy prose-JSON path with the repair/retry machinery.
+        agent = AgentFactory(llm_config).create_agent(AgentRole.EXTRACTION, config=llm_config)
+        response_handler = handler or EnhancedResponseHandler(
+            _synthesis_handler_config(synthesis_config.extraction_max_retries)
+        )
+        context = RequestContext(
+            user_query=query,
+            model_name=llm_config.model if llm_config else "default",
+            session_id=session_id,
+        )
 
     from ..utils.progress_reporter import get_progress_reporter
 
@@ -447,14 +490,22 @@ async def _extract_with_circuit(index: int, paper: RankedPaper) -> PaperExtracti
                     f"Analyzing paper {index}/{len(llm_targets)}: {title_preview}…"
                 )
 
-            extraction, llm_success = await _extract_single_paper(
-                paper,
-                query,
-                agent,
-                response_handler,
-                context,
-                passages=paper_passages,
-            )
+            if use_structured:
+                extraction, llm_success = await _extract_single_paper_structured(
+                    paper,
+                    query,
+                    llm_config,
+                    passages=paper_passages,
+                )
+            else:
+                extraction, llm_success = await _extract_single_paper(
+                    paper,
+                    query,
+                    agent,
+                    response_handler,
+                    context,
+                    passages=paper_passages,
+                )
 
         if llm_success:
             breaker["consecutive_failures"] = 0
@@ -519,15 +570,6 @@ async def synthesize_collective(
 
         llm_config = get_settings().llm
 
-    agent = AgentFactory(llm_config).create_agent(AgentRole.SYNTHESIS, config=llm_config)
-    response_handler = handler or EnhancedResponseHandler(
-        _synthesis_handler_config(synthesis_config.collective_max_retries)
-    )
-    context = RequestContext(
-        user_query=query,
-        model_name=llm_config.model,
-        session_id=session_id,
-    )
     prompt = _build_synthesis_prompt(query, extractions, clusters, ranked_papers)
 
     logger.info("Running LLM collective synthesis across %d paper(s)", len(extractions))
@@ -539,6 +581,30 @@ async def synthesize_collective(
             f"Synthesizing insights across {len(extractions)} paper(s)…"
         )
 
+    if _use_structured_outputs(llm_config):
+        from ..models.structured import try_run_structured
+
+        synthesis = await try_run_structured(
+            AgentRole.SYNTHESIS,
+            prompt,
+            SynthesisResult,
+            llm_config,
+        )
+        if synthesis is not None:
+            return synthesis
+        logger.warning("LLM collective synthesis failed; using heuristic fallback")
+        return _heuristic_synthesis(query, extractions, clusters, ranked_papers=ranked_papers)
+
+    agent = AgentFactory(llm_config).create_agent(AgentRole.SYNTHESIS, config=llm_config)
+    response_handler = handler or EnhancedResponseHandler(
+        _synthesis_handler_config(synthesis_config.collective_max_retries)
+    )
+    context = RequestContext(
+        user_query=query,
+        model_name=llm_config.model,
+        session_id=session_id,
+    )
+
     result = await response_handler.process_structured_response(
         agent,
         prompt,

src/config/settings.py

@@ -66,6 +66,7 @@ class LLMConfig(BaseModel):
     api_key: str | None = None
     temperature: float = 0.2
     timeout_seconds: int = 120
+    structured_outputs: bool = True
 
 
 class EmbeddingConfig(BaseModel):

src/models/__init__.py

@@ -2,6 +2,12 @@
 """LLM provider abstraction (not paper domain models)."""
 
 from .base import ROLE_SYSTEM_PROMPTS, AgentRole, LLMProvider, normalize_openai_base_url
+from .structured import (
+    STRUCTURED_ROLE_PROMPTS,
+    create_structured_agent,
+    run_structured,
+    try_run_structured,
+)
 from .factory import (
     AgentFactory,
     create_llm_agent,
@@ -14,6 +20,10 @@
 from .anthropic import AnthropicProviderImpl
 
 __all__ = [
+    "STRUCTURED_ROLE_PROMPTS",
+    "create_structured_agent",
+    "run_structured",
+    "try_run_structured",
     "AgentFactory",
     "AgentRole",
     "AnthropicProviderImpl",

src/models/structured.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""Schema-enforced LLM calls via pydantic-ai structured outputs.
+
+Instead of asking the model for JSON in prose and repairing whatever comes
+back, the schema is attached to the agent (``output_type``) and pydantic-ai
+enforces it natively — invalid output triggers the framework's own retry
+with validation feedback, and the caller receives a typed object or an
+exception. This replaces the string-repair path for providers that support
+structured outputs; heuristic fallbacks remain the safety net.
+"""
+
+from __future__ import annotations
+
+from typing import TypeVar
+
+from pydantic import BaseModel
+from pydantic_ai import Agent
+from pydantic_ai.models import Model
+
+from ..utils.logging_system import logger
+from .base import ROLE_SYSTEM_PROMPTS, AgentRole
+from .factory import AgentFactory, create_llm_provider
+
+OutputT = TypeVar("OutputT", bound=BaseModel)
+
+# Slim prompts for the structured path: the schema is enforced by the
+# framework, so the "respond with ONLY JSON ..." scaffolding the legacy
+# prompts carry is unnecessary and only adds noise.
+STRUCTURED_ROLE_PROMPTS: dict[AgentRole, str] = {
+    AgentRole.EXPANSION: (
+        "You expand research queries into concise academic search phrases "
+        "and focused sub-questions."
+    ),
+    AgentRole.EXTRACTION: (
+        "You are a research paper analyst. Extract the paper's methodology, "
+        "datasets, benchmarks, limitations, and findings. When full-text "
+        "passages are provided, ground the extraction in them and quote short "
+        "verbatim evidence snippets; otherwise leave evidence empty."
+    ),
+    AgentRole.SYNTHESIS: (
+        "You are a research synthesis expert. From structured per-paper "
+        "extractions and thematic clusters, derive cross-paper agreements, "
+        "disagreements, trends, gaps, datasets, and methodologies."
+    ),
+    AgentRole.GAP_ANALYSIS: (
+        "You are a research strategist. From cross-paper synthesis, identify "
+        "prioritized research gaps, actionable opportunities, and "
+        "underexplored areas."
+    ),
+}
+
+
+def create_structured_agent(
+    role: AgentRole,
+    output_type: type[OutputT],
+    llm_config=None,
+    *,
+    retries: int = 2,
+    model: Model | None = None,
+) -> Agent:
+    """Create an agent whose output is validated against ``output_type``."""
+    if model is None:
+        resolved = AgentFactory(llm_config).config
+        model = create_llm_provider(resolved).create_model(resolved)
+    prompt = STRUCTURED_ROLE_PROMPTS.get(role) or ROLE_SYSTEM_PROMPTS[role]
+    return Agent(
+        model=model,
+        system_prompt=prompt,
+        output_type=output_type,
+        retries=retries,
+    )
+
+
+async def run_structured(
+    role: AgentRole,
+    prompt: str,
+    output_type: type[OutputT],
+    llm_config=None,
+    *,
+    retries: int = 2,
+    model: Model | None = None,
+) -> OutputT:
+    """Run a structured agent and return the validated output. Raises on failure."""
+    agent = create_structured_agent(
+        role,
+        output_type,
+        llm_config,
+        retries=retries,
+        model=model,
+    )
+    result = await agent.run(prompt)
+    return result.output
+
+
+async def try_run_structured(
+    role: AgentRole,
+    prompt: str,
+    output_type: type[OutputT],
+    llm_config=None,
+    *,
+    retries: int = 2,
+    model: Model | None = None,
+) -> OutputT | None:
+    """Like :func:`run_structured`, but returns None on any failure."""
+    try:
+        return await run_structured(
+            role,
+            prompt,
+            output_type,
+            llm_config,
+            retries=retries,
+            model=model,
+        )
+    except Exception as exc:
+        logger.warning("Structured %s call failed: %s", role.value, exc)
+        return None