Extracting structured records from messy text is the backbone of most LLM integrations. This langchain extraction chain pydantic v2 tutorial walks through building a typed pipeline that turns a blob of text into validated Pydantic objects, with runnable code at each step.
Prerequisites
- Python 3.10 or newer
langchain,langchain-openai, andpydantic(v2.x) installed- An OpenAI API key, or an OpenAI-compatible endpoint. If you’re using n4n.ai, its single OpenAI-compatible endpoint exposes 240+ models and forwards provider cache-control hints, so
ChatOpenAIworks unchanged.
pip install langchain langchain-openai pydantic
export OPENAI_API_KEY="sk-..."
Define the Pydantic v2 Schema
Pydantic v2 changed internals but the surface for schema definition is stable. Use BaseModel and Field to give the model hints that improve extraction quality.
from pydantic import BaseModel, Field
class Person(BaseModel):
name: str = Field(description="Full name of the person")
age: int | None = Field(default=None, description="Estimated age in years")
hobbies: list[str] = Field(description="List of known hobbies")
# Expected schema in JSON:
# {"name": "Ada Lovelace", "age": 36, "hobbies": ["math", "music"]}
The descriptions are not cosmetic. LangChain serializes them into the function-calling schema sent to the model.
Wire Up the Chat Model
LangChain’s ChatOpenAI supports with_structured_output, which binds the Pydantic model as a function and parses the response back into an instance.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
extractor = llm.with_structured_output(Person)
If you point at a compatible gateway, pass base_url:
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
base_url="https://api.n4n.ai/v1", # optional: OpenAI-compatible
api_key="your-key",
)
Run the Extraction
Feed unstructured text and inspect the validated object.
text = """
Ada Lovelace was a mathematician who loved music and writing.
She was around 36 when she wrote her famous notes.
"""
person = extractor.invoke(text)
print(person)
Expected output:
name='Ada Lovelace' age=36 hobbies=['music', 'writing']
The returned value is a Person instance. If the model emits a number as string, Pydantic coerces it. If a field is missing and has no default, the chain raises a ValidationError.
Extract Multiple Records
Real inputs often contain several entities. Wrap the model in a container.
from typing import List
class People(BaseModel):
people: List[Person]
multi_extractor = llm.with_structured_output(People)
doc = """
Alan Turing enjoyed chess and cryptography. He was about 41.
Grace Hopper liked sailing and compilers. She was 79.
"""
result = multi_extractor.invoke(doc)
for p in result.people:
print(p.name, p.age, p.hobbies)
Expected output:
Alan Turing 41 ['chess', 'cryptography']
Grace Hopper 79 ['sailing', 'compilers']
Add a Prompt Template
Binding a schema does not give you control over instructions. Compose a prompt with ChatPromptTemplate to set the extraction context.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a precise extraction engine. Pull only facts present in the text."),
("human", "{input}")
])
chain = prompt | llm.with_structured_output(People)
out = chain.invoke({"input": doc})
print(out.people[0].name)
This pattern is the standard langchain extraction chain pydantic v2 tutorial architecture: prompt | model.with_structured_output(schema).
Handle Validation and Partial Data
Models hallucinate or omit. Make fields optional and catch errors.
from pydantic import ValidationError
class SafePerson(BaseModel):
name: str
age: int | None = None
hobbies: List[str] = []
safe_extractor = llm.with_structured_output(SafePerson)
try:
obj = safe_extractor.invoke("The cat sat on the mat.")
print(obj)
except ValidationError as e:
print("Failed:", e)
With the above, the model may return name='The cat' and empty hobbies. To force “no extraction” you can add a sentinel or use a union type.
Use Enum and Constraints
Pydantic v2 supports Field constraints and Enum for closed vocabularies.
from enum import Enum
from pydantic import Field, conint
class Skill(str, Enum):
MATH = "math"
MUSIC = "music"
SPORTS = "sports"
class StrictPerson(BaseModel):
name: str
skill: Skill
age: conint(gt=0, lt=130)
strict_extractor = llm.with_structured_output(StrictPerson)
If the model returns an out-of-range age, Pydantic raises before the data hits your DB.
Batch Processing
LangChain chains are runnable over lists with batch.
texts = ["Ada loves math.", "Bob plays sports."]
# Define a simpler schema for demo
class Mini(BaseModel):
name: str
interest: str
mini_extractor = llm.with_structured_output(Mini)
results = mini_extractor.batch(texts)
print([r.name for r in results])
Expected output:
['Ada', 'Bob']
Debug the Raw Payload
When parsing fails, inspect the model’s raw message. Disable structured output binding temporarily.
raw = llm.invoke("Extract: Ada, 36, music")
print(raw.content)
You’ll see a JSON string or function call arguments. This is useful when the schema is too strict for the model’s tendency.
Pydantic v1 vs v2 Migration Notes
In v1 you used class Config:. In v2, use model_config. Also validator becomes field_validator. If you copy old schemas, they will break.
# v2 style
from pydantic import field_validator
class CleanPerson(BaseModel):
name: str
@field_validator("name")
@classmethod
def strip_name(cls, v):
return v.strip()
Choosing JSON Mode vs Function Calling
with_structured_output defaults to function calling for models that support it. You can pass method="json_mode" to use OpenAI’s JSON mode, but then you must supply a strict schema and the model may not fill all fields. For the langchain extraction chain pydantic v2 tutorial, function calling is simpler.
extractor_json = llm.with_structured_output(Person, method="json_mode")
Add Retries
Wrap the extractor in RunnableRetry from LangChain to absorb transient validation or parser failures.
from langchain_core.runnables import RunnableRetry
from langchain_core.exceptions import OutputParserException
retry_extractor = RunnableRetry(
bound=extractor,
retry_exception_types=(ValidationError, OutputParserException),
max_attempts=3,
)
This reduces silent failures in production loops.
Full Script
A minimal end-to-end file combining the pieces:
import os
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
class Person(BaseModel):
name: str = Field(description="Full name")
age: int | None = Field(default=None)
hobbies: List[str] = Field(default_factory=list)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "Extract people from text."),
("human", "{text}")
])
chain = prompt | llm.with_structured_output(Person)
if __name__ == "__main__":
out = chain.invoke({"text": "Linus likes coding and chess. He is 55."})
print(out)
Expected output:
name='Linus' age=55 hobbies=['coding', 'chess']
Closing Notes on the langchain extraction chain pydantic v2 tutorial
You now have a typed extraction pipeline: define schema, bind with with_structured_output, compose with prompts, and validate. The same pattern scales to nested models, enums, and batch jobs. For production, add retries on ValidationError and meter token usage per call.