How to Perform Structured Data Extraction Using `needle.extract()` with Pydantic Models

needle.extract() converts unstructured LLM output into validated Python objects by accepting a Pydantic model as a schema argument, automatically generating JSON system prompts and enforcing type safety.

The needle library from cactus-compute provides a streamlined bridge between free-form text generation and structured data. By integrating Pydantic directly into its extraction pipeline, needle eliminates the need for manual JSON parsing and validation when working with language model responses.

Understanding needle.extract() Architecture

The extract() method, implemented in needle/__init__.py (lines 299-365), orchestrates five distinct operations to transform raw LLM output into typed data:

Step-by-Step Extraction Flow

Step	Operation	Implementation Location
1	Prompt Construction – Builds a system prompt embedding the Pydantic model's JSON schema	needle/__init__.py, lines 299-326
2	LLM Invocation – Calls the underlying model via self.run() with token limits	needle/__init__.py, lines 327-340
3	JSON Extraction – Cleans and parses the raw response with json.loads()	needle/__init__.py, lines 341-350
4	Pydantic Validation – Instantiates the model using schema.parse_obj() for automatic validation	needle/__init__.py, lines 351-361
5	Typed Return – Returns the populated Pydantic instance or dictionary	needle/__init__.py, lines 362-365

This architecture ensures that any deviation from the expected schema triggers a ValidationError, providing immediate feedback on malformed LLM outputs.

extract() Method Signature

def extract(
    self,
    text: str,
    schema: type | dict,
    max_new_tokens: int = 256,
    system: str | None = None,
) -> Any:
    ...

• text – The raw string to parse, typically direct LLM output.
• schema – A Pydantic BaseModel subclass or a plain dictionary describing JSON structure.
• max_new_tokens – Hard limit on generation length for structured responses.
• system – Optional override for the system prompt template.

Basic Pydantic Extraction Example

The following demonstrates complete workflow from model definition to validated output:

from pydantic import BaseModel, Field
from needle import Needle


class WeatherReport(BaseModel):
    location: str = Field(..., description="City name")
    temperature_c: float = Field(..., description="Temperature in Celsius")
    condition: str = Field(..., description="Short weather description")


# Initialize with any supported LLM backend

needle = Needle(model="gpt-4o-mini")

# Generate unstructured text

prompt = "Give me the current weather for Paris in JSON format."
raw_response = needle.run(prompt)

# Extract into typed object

report: WeatherReport = needle.extract(
    text=raw_response,
    schema=WeatherReport,
    max_new_tokens=150
)

print(report.location)       # → "Paris"

print(report.temperature_c)  # → 17.3

print(report.condition)      # → "Partly cloudy"

Handling Validation Failures

When LLM output violates the schema, extract() propagates Pydantic's standard ValidationError:

from pydantic import ValidationError

try:
    report = needle.extract(
        text=raw_response,
        schema=WeatherReport
    )
except ValidationError as e:
    # e.errors() provides granular field-level failure details

    for error in e.errors():
        print(f"Field '{error['loc'][0]}': {error['msg']}")
    # Implement retry logic or fallback parsing here

This pattern enables robust error recovery without losing type safety.

Dictionary Schema Alternative

For lightweight use cases, extract() accepts plain dictionaries instead of Pydantic models:

schema = {
    "name": "str",
    "age": "int",
    "email": "str"
}

data = needle.extract(
    text=raw_response,
    schema=schema
)

# Returns: dict with parsed values, no Pydantic validation

Trade-off: Dictionary schemas skip automatic type coercion and validation, returning raw parsed JSON. Use Pydantic models when field types, constraints, or nested structures matter.

Advanced Pydantic Features Supported

needle.extract() respects all Pydantic capabilities through its parse_obj() invocation:

• Type coercion – Automatic conversion of compatible types (e.g., "42" → 42 for int fields).
• Custom validators – @validator and @field_validator decorators execute during extraction.
• Nested models – Complex hierarchies resolve recursively through standard Pydantic parsing.
• Default values – Optional fields populate with defaults when absent from LLM output.
• Field aliases – Field(alias="alternate_name") maps JSON keys to Python attributes.

from pydantic import BaseModel, validator, Field
from datetime import datetime


class Event(BaseModel):
    title: str
    start_time: datetime = Field(..., alias="startTime")
    attendees: list[str] = []
    
    @validator('attendees', pre=True)
    def split_comma_attendees(cls, v):
        if isinstance(v, str):
            return [x.strip() for x in v.split(',')]
        return v


# Extraction automatically applies validator and alias handling

event = needle.extract(raw_json, schema=Event)

Integration with Needle's Core Pipeline

The extract() method leverages needle/model/run.py for actual LLM communication. This separation allows:

• Model-agnostic operation – Swap backends without changing extraction code.
• Consistent retry handling – Inherited from the base run() implementation.
• Token accounting – max_new_tokens applies specifically to structured generation, distinct from initial prompt processing.

Summary

• needle.extract() in needle/__init__.py (lines 299-365) provides type-safe structured data extraction from LLM outputs.
• Pydantic models passed as schema enable automatic validation, coercion, and nested data handling.
• Dictionary schemas offer lighter-weight parsing without validation overhead.
• Validation errors surface as standard Pydantic exceptions for programmatic handling.
• All Pydantic features—validators, aliases, defaults, nested models—function normally through the parse_obj() integration point.

Frequently Asked Questions

What happens if the LLM returns malformed JSON?

needle.extract() attempts to clean and parse the response with json.loads() at lines 341-350 of needle/__init__.py. If parsing fails, a JSONDecodeError propagates. If parsing succeeds but validation against the Pydantic model fails, a ValidationError is raised instead.

Can I use Pydantic v2 with needle.extract()?

The source implementation uses parse_obj(), which is the Pydantic v1 method. For Pydantic v2 compatibility, check the needle repository version—cactus-compute may have updated to model_validate() in newer releases. The schema parameter type remains type | dict regardless of Pydantic version.

How does max_new_tokens differ from the main generation limit?

The max_new_tokens parameter in extract() specifically constrains the LLM's output length for the structured JSON response (lines 327-340). This is separate from any token limits applied during the initial needle.run() call that generated the raw text, allowing tighter control over extraction-phase generation.

Is there performance overhead to Pydantic validation in extract()?

Validation occurs once per extraction via schema.parse_obj() (lines 351-361). For latency-sensitive applications, dictionary schemas bypass this overhead entirely. In practice, Pydantic validation adds negligible cost compared to LLM inference time, and the safety guarantees typically justify the minimal overhead.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}