SAIDEX:LangChain 原生结构化数据抽取库,支持小模型与智能体工具循环
mlauf-labs/SAIDEX
SAIDEX 是一个 LangChain 原生的结构化数据抽取库,可将任意 LangChain 兼容的聊天模型变成可靠的抽取器:通过 tool-calling 或 JSON 模式驱动 LLM,用 Pydantic 校验输出,并把字段级错误反馈给模型自我纠正。
SAIDEX — Structured AI Data EXtraction
LangChain-native · async · Pydantic validated · tool-calling + JSON modes · agentic tool loop · auto-retry · fallback model · multimodal · small & local models
Docs · Install · Quick start · Examples · Contributing · Changelog
Extract validated Pydantic models from LLM responses — with automatic retry, field-level error feedback, an optional fallback model, and an agentic tool loop for extractions that need lookups or side effects first.
saidex turns any LangChain-compatible chat model into a reliable structured-data extractor. It drives the LLM via tool-calling (or raw JSON for models without tool-calling support), validates the output with Pydantic, and feeds detailed error messages back to the model for self-correction — all transparently.
Works with the models you already have — from GPT-4o down to small open-weight models running locally via Ollama or vLLM. If the model doesn't support tool calling, ExtractionMode.JSON handles it without any code changes on your side.
Why use this library?
Key advantages
| Advantage | Detail |
|---|---|
| LangChain-native | Plugs directly into any LangChain chat model — no wrapper, no adapter, no glue code. If you already use LangChain, you're set. |
| Small & local models supported | Works with compact open-weight models (Llama, Mistral, Phi, Qwen, …) served locally via Ollama or vLLM — no cloud dependency required. |
| No tool-calling requirement | Models that don't support function/tool calling (most local models) work via ExtractionMode.JSON — same validation and retry logic, zero extra code. |
| Cost-optimised | Pair a cheap small model as primary with a powerful one as fallback — pay for the big model only when the small one can't deliver. |
| Agentic tool loop | Give the LLM your own tools (lookups, API calls, side effects) — it calls them in a loop and finishes with a schema-validated final answer. |
Problems it solves
| Problem | What this library does |
|---|---|
| LLMs sometimes return malformed JSON | Detects invalid tool calls and retries automatically |
| LLM output passes JSON parsing but fails schema validation | Sends a detailed, field-level error report back to the LLM |
| One model isn't reliable enough | Configure a fallback model that takes over after the primary exhausts its retries |
| Network errors / rate limits mid-call | Built-in configurable retry with exponential back-off |
| Your model doesn't support tool calling | Switch to ExtractionMode.JSON — works with any chat model |
| Debugging is hard | Structured ExtractDataStats return value — know exactly how many retries each phase needed |
| LangGraph agents trip over malformed tool calls | Swap in SaidexToolNode — a drop-in ToolNode that repairs, validates and corrects tool calls before execution |
Installation
pip install saidex
If you use OpenAI models and want automatic rate-limit handling:
pip install "saidex[openai]"Quick start
Extract from a text string
import asyncio from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI from saidex import extract_data_from_text class PersonInfo(BaseModel): name: str age: int occupation: str async def main(): llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) person, stats = await extract_data_from_text( llm, PersonInfo, "Alice Müller, 34, works as a software engineer in Munich.", ) print(person) # name='Alice Müller' age=34 occupation='software engineer' print(stats.total_retries) # 0 — first attempt was valid asyncio.run(main())
Extract from a chat history
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from saidex import extract_data messages = [ SystemMessage(content="Analyse the following support conversation."), HumanMessage(content="Here is the transcript:"), AIMessage(content="Customer: My order is late!\nAgent: We'll ship a replacement today."), HumanMessage(content="Provide the structured summary."), ] summary, stats = await extract_data(llm, SupportTicketSummary, messages)
Core concepts
extract_data — the main function
async def extract_data( llm_model: Any, schema: type[ModelT], messages: list[BaseMessage], *, mode: ExtractionMode = ExtractionMode.TOOL_CALLING, callbacks: list[Any] | None = None, fallback_llm_model: Any = None, max_primary_retries: int = 3, max_fallback_retries: int = 3, retry_config: RetryConfig | None = None, ) -> tuple[ModelT | None, ExtractDataStats]:
Takes a list of BaseMessage objects (the full conversation context) and returns a validated (instance, stats) tuple.
extract_data_from_text — convenience wrapper
async def extract_data_from_text( llm_model: Any, schema: type[ModelT], text: str, *, system_prompt: str | None = None, ... ) -> tuple[ModelT | None, ExtractDataStats]:
Converts a plain string into [SystemMessage, HumanMessage] and calls extract_data. You can supply a custom system_prompt; a sensible default is used otherwise.
extract_data_list — batch extraction → list[ModelT]
When one document contains several repeated records (invoice line items, multiple
people in a transcript, products on a page), extract_data_list and
extract_data_list_from_text return a precisely typed list[ModelT] from a
single LLM call. Define the schema for one item:
from saidex import extract_data_list_from_text class InvoiceLine(BaseModel): description: str quantity: int unit_price: float lines, stats = await extract_data_list_from_text(llm, InvoiceLine, document) # lines: list[InvoiceLine] | None print(f"Extracted {stats.item_count} items")
The per-item schema is wrapped in a one-field container internally, so both
extraction modes, per-item validation (errors point at items -> 2 -> quantity),
retries, and fallback all work unchanged. See
Batch Extraction.
Extraction modes — with or without tool calling
By default the library forces the model to call a bound tool. If your model
does not support tool calling (e.g. many local models served by Ollama or
llama.cpp), switch to ExtractionMode.JSON: the schema's JSON Schema is
injected into the prompt and the model's raw JSON reply is parsed and
validated instead.
from saidex import ExtractionMode, extract_data_from_text # Default — tool calling result, stats = await extract_data_from_text(llm, MySchema, text) # No tool calling required — model replies with raw JSON result, stats = await extract_data_from_text( llm, MySchema, text, mode=ExtractionMode.JSON )
Both modes share the same validation, error-feedback, retry, and fallback behaviour. JSON mode tolerates markdown code fences and stray text around the JSON object. See Extraction Modes for details.
OpenAI-compatible gateways — ToolCallConfig
Tool-calling mode binds the schema with OpenAI's most reliable flag set: a
forced tool_choice, strict=True, and parallel_tool_calls=False. Some
OpenAI-compatible gateways (Kimi/Moonshot, some Qwen or DeepSeek deployments)
reject those flags with an HTTP 400. SAIDEX notices such a rejection and
re-binds the tool once with relaxed flags — no configuration needed, and no
validation retry consumed. You can also state the flags up front:
from saidex import ToolCallConfig, extract_data_from_text result, stats = await extract_data_from_text( llm, MySchema, text, tool_config=ToolCallConfig.COMPATIBLE, # tool_choice="auto", no strict/parallel flags )
Every flag is individually settable; None omits the keyword argument entirely
(which differs from sending false). See
Extraction Modes.
Retry strategy
┌───────────────────────────────────────────────┐
│ Primary model │
│ Attempt 1 → Attempt 2 → ... → Attempt N │
│ (validation errors fed back as messages) │
└──────────────────────┬────────────────────────┘
│ all attempts failed
▼
┌───────────────────────────────────────────────┐
│ Fallback model (optional) │
│ Attempt 1 → Attempt 2 → ... → Attempt M │
│ (fresh context + hint about previous failure) │
└───────────────────────────────────────────────┘
When a validation attempt fails, the library appends a detailed error message to the conversation — covering missing fields, type errors, enum violations, and constraint failures — so the model knows exactly what to fix.
ExtractDataStats
result, stats = await extract_data(...) stats.success # bool — did the run produce a validated instance? stats.failure_reason # FailureReason | None — why it failed (None on success) stats.primary_retries # int — retries against the primary model stats.fallback_retries # int — retries against the fallback model stats.total_retries # int — sum of both stats.fallback_used # bool — was the fallback model invoked? stats.salvaged # bool — result produced by an on_exhausted hook stats.item_count # int — items returned by extract_data_list (else 0) stats.format_errors # int — pure parse / tool-call failures stats.problem_fields # tuple[str, ...] — fields that ever failed validation stats.field_issues # tuple[FieldIssue, ...] — structured per-field problems stats.last_candidate # last schema-valid but rejected instance # (None on a clean success; kept when salvaged) stats.last_rejection # tuple[FieldIssue, ...] — the issues that rejected it if not stats.success: print(stats.failure_reason, stats.problem_fields)
Breaking change:
int(stats)and the comparison shims were removed — usestats.total_retries.field_issuesare kept even on a successful run when an earlier attempt was self-corrected.
Sometimes an extraction needs more than the text in front of the model — a
database lookup, an API call, or a resource that must be created first.
extract_data_with_tools runs an agent loop: the LLM may call your tools any
number of times, then delivers a final answer validated against your schema.
from pydantic import BaseModel, Field from saidex import Tool, extract_data_with_tools # 1. Describe the tool's arguments with a Pydantic model class OrderStatusArgs(BaseModel): order_id: str = Field(description="Order id, e.g. ORD-1042") # 2. Provide an async handler — its result is shown to the LLM async def order_status_handler(order_id: str) -> dict: return {"order_id": order_id, "status": "shipped", "eta": "2026-06-15"} order_status = Tool( name="get_order_status", description="Look up the current status and ETA of an order.", parameters=OrderStatusArgs, handler=order_status_handler, ) # 3. Define the final-answer schema and run the loop class TicketResolution(BaseModel): order_id: str current_status: str customer_reply: str = Field(description="Friendly reply for the customer") resolution, stats = await extract_data_with_tools( llm, TicketResolution, "Customer asks: where is my order ORD-1042?", tools=[order_status], ) print(resolution.customer_reply) print(stats.iterations, stats.tool_calls) # e.g. 2 1
How it works:
- Your
Tools are bound to the model alongside a final-answer tool built fromschema. - The LLM calls helper tools as needed; every result is appended to the conversation as a
ToolMessage. - When the LLM submits the final answer, it is validated with Pydantic — on failure the field-level error report is fed back and the loop continues.
- The loop ends with a validated instance, or
Noneoncemax_iterations/max_validation_retriesare exhausted (optionally retried on a fallback model first).
Good to know:
- Handler exceptions and invalid tool arguments don't crash the loop — they are returned to the LLM as tool output so it can self-correct.
- Models without tool-calling support can deliver the final answer as raw JSON via
final_answer_mode=ExtractionMode.JSON(the helper tools themselves always require tool calling). - Need full control over the conversation (multi-turn, prior context)? Use
run_extractor_agent(llm, schema, messages, tools=...)— same behaviour, but takes a ready-built message list.
ExtractorRunStats
stats.iterations # int — LLM invocations in the loop stats.tool_calls # int — helper-tool executions stats.validation_retries # int — final answers that failed validation stats.fallback_used # bool — was the fallback model invoked?
LangGraph integration — SaidexToolNode
Building an agent with LangGraph
instead? SaidexToolNode is a drop-in replacement for langgraph.prebuilt.ToolNode
that guarantees every tool-call id receives exactly one ToolMessage — the
stock node silently ignores malformed tool calls, which leaves the next model
turn with an unanswered tool-call id that OpenAI-compatible providers reject.
pip install "saidex[langgraph]"from saidex.langgraph import SaidexToolNode graph.add_node("tools", SaidexToolNode(tools)) # instead of ToolNode(tools)
It repairs malformed JSON, validates arguments against each tool's Pydantic schema before execution, and — by policy — answers bad calls with structured field-level feedback, runs a bounded LLM correction cycle, or fails fast.
Full guide:
docs/langgraph-toolnode.md— runnable example:examples/16_langgraph_toolnode.py
Examples
With a fallback model
Use a cheaper primary model and fall back to a more capable one only when needed:
from langchain_openai import ChatOpenAI from saidex import extract_data_from_text primary = ChatOpenAI(model="gpt-4o-mini", temperature=0) fallback = ChatOpenAI(model="gpt-4o", temperature=0) result, stats = await extract_data_from_text( primary, FinancialReport, report_text, fallback_llm_model=fallback, max_primary_retries=2, max_fallback_retries=3, ) if stats.fallback_used: print(f"Fallback needed after {stats.primary_retries} primary retries.")
Custom retry configuration
from saidex import RetryConfig, extract_data_from_text config = RetryConfig( max_retries=2, retry_delays=[2.0, 5.0], # exponential-ish back-off rate_limit_retry_interval=30.0, rate_limit_max_duration_seconds=300.0, ) result, stats = await extract_data_from_text( llm, MySchema, text, retry_config=config )
Using create_instance_safe standalone
Need safe Pydantic instantiation with rich error messages but without an LLM?
from saidex import create_instance_safe instance, error = create_instance_safe(MySchema, **data) if instance is None: print(error) # structured, human-readable validation report
LangChain callbacks (Langfuse, LangSmith, …)
Any LangChain BaseCallbackHandler can be passed via callbacks:
from langfuse.callback import CallbackHandler result, stats = await extract_data( llm, MySchema, messages, callbacks=[CallbackHandler()], )
Multimodal: Extract from images
The library works with any vision-capable LLM — cloud models such as GPT-4o as well as locally hosted models via vLLM. Images are passed as standard LangChain multimodal messages.
Helper functions
Two small helpers (shown below, also in examples/05_image_analysis.py)
build the right message format for you:
def image_message_from_url(url: str, prompt: str = "...", *, detail: str = "high") -> HumanMessage: ... def image_message_from_file(path: str | Path, prompt: str = "...", *, media_type: str | None = None) -> HumanMessage: ...
Both return a HumanMessage with a text block and an image block that any
vision model understands.
Extract from an image URL (OpenAI GPT-4o)
import asyncio from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field from saidex import extract_data class ReceiptItem(BaseModel): description: str total: float class Receipt(BaseModel): vendor: str date: str | None = None items: list[ReceiptItem] total: float currency: str = "EUR" def image_message_from_url(url: str, prompt: str) -> HumanMessage: return HumanMessage(content=[ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": url, "detail": "high"}}, ]) async def main(): llm = ChatOpenAI(model="gpt-4o", temperature=0) messages = [ SystemMessage(content="Extract all data from the receipt image precisely."), image_message_from_url( "https://example.com/receipt.jpg", prompt="Extract the receipt data.", ), ] receipt, stats = await extract_data(llm, Receipt, messages) print(receipt) asyncio.run(main())
Extract from a local image file (base64)
import base64 from pathlib import Path def image_message_from_file(path: str | Path, prompt: str) -> HumanMessage: path = Path(path) mime_map = {".jpg": "image/jpeg", ".png": "image/png", ".webp": "image/webp"} mime = mime_map.get(path.suffix.lower(), "image/jpeg") b64 = base64.b64encode(path.read_bytes()).decode() return HumanMessage(content=[ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}, ]) messages = [ SystemMessage(content="Extract contact details from the business card."), image_message_from_file("card.jpg", prompt="Read this business card."), ] result, stats = await extract_data(llm, BusinessCard, messages)
Use a local vLLM vision model
vLLM exposes an OpenAI-compatible API, so you simply
point ChatOpenAI at your local server.
1. Start the vLLM server
pip install vllm vllm serve Qwen/Qwen2-VL-7B-Instruct --port 8000
Popular vision models supported by vLLM:
| Model | Strengths |
|---|---|
Qwen/Qwen2-VL-7B-Instruct |
Strong OCR, multilingual, recommended |
Qwen/Qwen2.5-VL-7B-Instruct |
Updated Qwen VL, better reasoning |
llava-hf/llava-1.5-7b-hf |
Lightweight, fast |
microsoft/Phi-3.5-vision-instruct |
Compact, efficient |
OpenGVLab/InternVL2-8B |
Good at charts and documents |
2. Connect and extract
from langchain_openai import ChatOpenAI from saidex import extract_data vllm = ChatOpenAI( model="Qwen/Qwen2-VL-7B-Instruct", # must match the loaded model base_url="http://localhost:8000/v1", # vLLM default endpoint api_key="EMPTY", # vLLM needs no real API key temperature=0, max_tokens=1024, ) messages = [ SystemMessage(content="Extract the chart data precisely."), image_message_from_url(chart_url, prompt="Analyse this chart."), ] result, stats = await extract_data(vllm, ChartData, messages)
Multiple images in one request
def multi_image_message(urls: list[str], prompt: str) -> HumanMessage: content = [{"type": "text", "text": prompt}] for url in urls: content.append({"type": "image_url", "image_url": {"url": url}}) return HumanMessage(content=content) messages = [ SystemMessage(content="Compare the two product labels."), multi_image_message([url1, url2], prompt="Analyse both labels."), ] result, stats = await extract_data(llm, ComparisonSchema, messages)
See
examples/05_image_analysis.pyfor the complete runnable example including a fallback model and all helper functions.
Schema design guide
A well-designed Pydantic schema is the single most important factor for reliable extraction. Two rules cover most of the ground:
- Use the right types — the schema becomes a JSON Schema that the LLM must fill in; unclear types produce unclear output.
- Describe every field — the LLM reads
Field(description=...)to understand what to put there.
Supported field types
| Type | Example | Notes |
|---|---|---|
str |
name: str |
Plain text |
int |
age: int |
Whole number |
float |
price: float |
Decimal number |
bool |
active: bool |
true / false |
str | None |
nickname: str | None = None |
Nullable — use None as default |
list[str] |
tags: list[str] |
Zero-or-more strings |
list[MyModel] |
items: list[LineItem] |
Nested model list |
dict[str, str] |
metadata: dict[str, str] |
Simple string map |
dict[str, list[str]] |
groups: dict[str, list[str]] |
Map with list values |
StrEnum / str Enum |
status: Status |
Constrains value to fixed set |
Literal["a", "b"] |
unit: Literal["kg", "g"] |
Inline allowed-values list |
Nested BaseModel |
address: Address |
Structured sub-object |
Field(ge=0, le=100) |
score: int = Field(ge=0, le=100) |
Numeric constraints |
Field(min_length=1) |
title: str = Field(min_length=1) |
String length constraints |
⚠️ Types to avoid
Never use
object, baredict, orAnyas a field type.
# BAD — the LLM has no idea what structure to produce. # Pydantic accepts anything, so validation never catches wrong output. class Bad(BaseModel): data: object # ❌ extra: dict # ❌ value: Any # ❌ payload: dict[str, Any] # ❌ — the Any part is the problem # GOOD — replace with a typed model or a concrete dict type class Good(BaseModel): data: MySubModel # ✅ explicit structure extra: dict[str, str] # ✅ both key and value typed count: int # ✅ concrete type payload: dict[str, list[str]] # ✅ fully typed
Avoid datetime, date, and time objects — many models struggle with
Python's ISO 8601 serialisation. Use str with a format hint in the
description instead:
# BAD from datetime import date due: date # ❌ — serialisation surprises # GOOD due: str | None = Field(default=None, description="Due date as YYYY-MM-DD, or null") # ✅
Field descriptions are essential
The description string is included in the JSON Schema that the LLM receives.
It is the primary way to tell the model exactly what to put in a field.
# BAD — the LLM has to guess the meaning, unit, and format class Invoice(BaseModel): amount: float # ❌ — which currency? incl. tax? total or subtotal? date: str # ❌ — which format? invoice date or due date? # GOOD — the LLM knows exactly what is expected class Invoice(BaseModel): amount: float = Field( description="Total invoice amount in EUR including VAT, e.g. 119.00" ) date: str = Field( description="Invoice issue date in ISO 8601 format: YYYY-MM-DD" )
Tips for writing good descriptions:
- State the unit (
EUR,kg,%,seconds) - State the format for strings (
YYYY-MM-DD,HH:MM,E.164 phone) - Give a concrete example (
e.g. 42.50,e.g. "2026-01-31") - Clarify ambiguous names (
"subtotal before tax"vs"total including VAT") - For optional fields, state what
nullmeans ("null if not mentioned")
Complete well-designed schema example
from enum import Enum from pydantic import BaseModel, Field class Priority(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class Subtask(BaseModel): title: str = Field(description="Short action item, max one sentence") done: bool = Field(default=False, description="true if already completed") class Task(BaseModel): title: str = Field(description="Short task title, max 80 characters") description: str = Field(description="Full description of what needs to be done") priority: Priority = Field(description="Urgency level: low / medium / high / critical") due_date: str | None = Field( default=None, description="Deadline as YYYY-MM-DD, or null if no deadline is mentioned", ) estimated_hours: float | None = Field( default=None, ge=0, description="Estimated effort in hours, or null if unknown", ) subtasks: list[Subtask] = Field( default_factory=list, description="List of smaller steps; empty list if none mentioned", ) tags: list[str] = Field( default_factory=list, description="Short lowercase labels, e.g. ['backend', 'urgent']", )
For a complete reference including edge cases see
docs/schema-design.md.
Pydantic validators — clean and verify LLM output
LLMs are inconsistent: a price might come back as "€ 1.234,56", a date as
"05.01.2026", a boolean as "ja". Pydantic validators let you fix this
transparently — before your application code sees the data and without
needing extra retries.
| Validator | When it runs | Typical use |
|---|---|---|
@field_validator(mode='before') |
Before type coercion | Strip €, normalise "05.01.2026" → "2026-01-05", map "ja" → True |
@field_validator(mode='after') |
After type coercion | Lowercase emails, uppercase country codes, prepend https:// |
@model_validator(mode='before') |
Before any field | Parse a flat string into a dict structure |
@model_validator(mode='after') |
After all fields | Cross-field rules: end_date > start_date, total matches line items |
@computed_field |
Read-only property | Derive subtotal, price_gross — never asked from the LLM |
Example — clean messy LLM amounts and dates
from typing import Any import re from pydantic import BaseModel, Field, field_validator class Invoice(BaseModel): amount: float = Field(ge=0, description="Total in EUR") date: str = Field(description="Issue date as YYYY-MM-DD") paid: bool @field_validator("amount", mode="before") @classmethod def clean_amount(cls, v: Any) -> Any: if isinstance(v, str): v = re.sub(r"[€$£\s]", "", v).replace(",", ".") return v @field_validator("date", mode="before") @classmethod def normalise_date(cls, v: Any) -> Any: if isinstance(v, str): m = re.match(r"^(\d{1,2})[./](\d{1,2})[./](\d{4})$", v.strip()) if m: return f"{m.group(3)}-{m.group(2).zfill(2)}-{m.group(1).zfill(2)}" return v @field_validator("paid", mode="before") @classmethod def normalise_bool(cls, v: Any) -> Any: if isinstance(v, str): return v.lower() in {"yes", "ja", "true", "1", "paid"} return v # All these LLM outputs are accepted and normalised: Invoice(amount="€ 1.190,00", date="05.01.2026", paid="ja") Invoice(amount="$42.50", date="31/12/2026", paid="yes") Invoice(amount=99.0, date="2026-03-15", paid=False)
Example — cross-field consistency (@model_validator)
from pydantic import BaseModel, model_validator, computed_field class LineItem(BaseModel): quantity: int unit_price: float @computed_field @property def subtotal(self) -> float: return round(self.quantity * self.unit_price, 2) class Order(BaseModel): order_date: str delivery_date: str | None = None items: list[LineItem] declared_total: float @model_validator(mode="after") def check_dates(self) -> "Order": if self.delivery_date and self.delivery_date < self.order_date: raise ValueError("delivery_date must not be before order_date") return self @model_validator(mode="after") def check_total(self) -> "Order": calculated = round(sum(i.subtotal for i in self.items), 2) if abs(self.declared_total - calculated) / max(calculated, 0.01) > 0.01: raise ValueError( f"declared_total {self.declared_total} differs from " f"calculated {calculated} by more than 1 %" ) return self
When a validator raises ValueError, create_instance_safe catches it,
formats it into a structured error message, and — inside
extract_data — sends it back to the LLM so it can self-correct.
Full example with all four validator types:
examples/06_pydantic_validators.py— detailed docs
For common cases the library ships ready-made field types so you don't have
to write a validator at all — e.g. IsoDateStr for yyyy-mm-dd dates, plus
IbanStr, VatIdStr, CountryCodeStr, CurrencyCodeStr, IsinStr,
PhoneStr, and LanguageCodeStr (and RRuleStr for RFC 5545 recurrence rules,
via the optional saidex[rrule] extra):
from saidex import IsoDateStr, IbanStr, CountryCodeStr class Payment(BaseModel): due_date: IsoDateStr | None = Field(None, description="Due date as yyyy-mm-dd") iban: IbanStr | None = Field(None, description="Payee IBAN") country: CountryCodeStr | None = Field(None, description="ISO 3166-1 alpha-2 code")
Full list and a guide to building your own:
docs/built-in-types.md
For checks that span fields, hit a database, or apply a business rule, pass an
external validator callable (sync or async) to any extraction function. It
runs after Pydantic validation; raising — or returning an error string — feeds the
message back into the same retry loop so the model can self-correct:
def validate_invoice(inv: Invoice) -> None: if inv.total != sum(line.amount for line in inv.lines): raise ValueError("Line items do not sum to the stated total") invoice, stats = await extract_data_from_text(llm, Invoice, text, validator=validate_invoice)
To report several findings at once, return a list of ValidationIssues — each gets
its own field_path and rule id in stats.field_issues, and on_mismatch="flag"
records a soft finding without consuming a retry.
To vary a validator by retry round — strict early, lenient on the last attempt —
wrap it with with_context, which passes a ValidatorContext as second argument
(details).
When every retry fails, an on_exhausted hook can still keep the part that
passed — e.g. drop the rejected items — and the run succeeds with
stats.salvaged=True
(details,
examples/17_salvage_partial_results.py).
Cross-field example and full guide:
examples/12_external_validator.py— detailed docs
Source grounding — reject hallucinated values
To make sure the model only returns values that are actually in the document, mark a field as grounded. After Pydantic validation, SAIDEX checks that the value appears in the source text; if it doesn't, the same retry loop asks the model to correct it. Both surfaces are visible right in the schema:
from typing import Annotated from saidex import Grounded, GroundedField class Invoice(BaseModel): vendor: Annotated[str, Grounded()] # marker form invoice_no: str = GroundedField(description="No.") # field-helper form country: str total: float = GroundedField(locale_field="country") # locale-aware matching
Matching is case/whitespace/diacritics-insensitive and locale-aware: a
total of 1234.5 is matched against 1.234,50 (de) or 1,234.50 (en), with
the locale taken from a sibling field (locale_field) or a fixed locale.
It is built on a reusable FieldCheck pattern, so you can add your own
source-aware checks the same way.
Full guide and the custom-check pattern:
examples/13_source_grounding.py— detailed docs
API reference
extract_data
| Parameter | Type | Default | Description |
|---|---|---|---|
llm_model |
Any |
— | LangChain chat model (.bind_tools() required for tool-calling mode) |
schema |
type[ModelT] |
— | Pydantic BaseModel subclass |
messages |
list[BaseMessage] |
— | Conversation context |
mode |
ExtractionMode |
TOOL_CALLING |
TOOL_CALLING or JSON (no tool calling) |
tool_config |
ToolCallConfig | None |
ToolCallConfig.OPENAI |
Which flags bind_tools sends in tool-calling mode (docs) |
callbacks |
list[Any] | None |
None |
LangChain callback handlers |
fallback_llm_model |
Any | None |
None |
Second model tried on primary failure |
max_primary_retries |
int |
3 |
Validation retries for primary model |
max_fallback_retries |
int |
3 |
Validation retries for fallback model |
retry_config |
RetryConfig | None |
DEFAULT_RETRY_CONFIG |
Network retry settings |
validator |
Validator[ModelT] | None |
None |
External sync/async check run after Pydantic validation; rejecting re-enters the retry loop (docs) |
Returns: tuple[ModelT | None, ExtractDataStats]
extract_data_from_text
All parameters of extract_data plus:
| Parameter | Type | Default | Description |
|---|---|---|---|
text |
str |
— | Text to analyse |
mode |
ExtractionMode |
TOOL_CALLING |
TOOL_CALLING or JSON (no tool calling) |
system_prompt |
str | None |
auto | System instruction (auto-generated if omitted) |
ExtractionMode
| Value | Description |
|---|---|
ExtractionMode.TOOL_CALLING |
Bind the schema as a tool and force the model to call it. Requires a tool-calling-capable model. Default. |
ExtractionMode.JSON |
Inject the JSON Schema into the prompt and parse the model's raw JSON reply. Works with any chat model. |
ToolCallConfig
Controls the keyword arguments sent to bind_tools in tool-calling mode. None
means the keyword argument is omitted from the request — not sent as false.
| Field | Type | Default | Description |
|---|---|---|---|
tool_choice |
"forced" | "auto" | None |
"forced" |
Force the schema tool by name, let the model choose, or send no tool_choice |
strict |
bool | None |
True |
Value for strict (OpenAI structured outputs), or None to omit |
parallel_tool_calls |
bool | None |
False |
Value for parallel_tool_calls, or None to omit |
auto_relax |
bool |
True |
On a flag rejection (HTTP 400 or TypeError), re-bind once with COMPATIBLE instead of failing |
Presets: ToolCallConfig.OPENAI (the defaults above) and
ToolCallConfig.COMPATIBLE (tool_choice="auto", no strict, no
parallel_tool_calls).
extract_data_with_tools
| Parameter | Type | Default | Description |
|---|---|---|---|
llm_model |
Any |
— | LangChain chat model with bind_tools support |
schema |
type[ModelT] |
— | Pydantic model for the final answer |
text |
str |
— | Task description (becomes the HumanMessage) |
tools |
list[Tool] |
— | Helper tools the LLM may call freely |
final_answer_mode |
ExtractionMode |
TOOL_CALLING |
How the final answer is collected |
system_prompt |
str | None |
auto | System instruction (auto-generated if omitted) |
callbacks |
list[Any] | None |
None |
LangChain callback handlers |
fallback_llm_model |
Any | None |
None |
Second model tried when the loop budget is exhausted |
max_iterations |
int |
12 |
Max LLM invocations per model |
max_validation_retries |
int |
3 |
Max final answers that may fail validation |
retry_config |
RetryConfig | None |
DEFAULT_RETRY_CONFIG |
Network retry settings |
validator |
Validator[ModelT] | None |
None |
External sync/async check run on the final answer; rejecting keeps the agent loop running (docs) |
Returns: tuple[ModelT | None, ExtractorRunStats]
run_extractor_agent accepts the same parameters but takes a full
messages: list[BaseMessage] instead of text / system_prompt.
Tool
| Field | Type | Description |
|---|---|---|
name |
str |
Function name exposed to the LLM |
description |
str |
What the tool does — shown to the LLM |
parameters |
type[BaseModel] |
Pydantic schema describing the tool's arguments |
handler |
async callable | Receives validated arguments as kwargs; the result is serialised into a ToolMessage |
RetryConfig
| Field | Type | Default | Description |
|---|---|---|---|
max_retries |
int |
3 |
Max transient-error retries |
retry_delays |
list[float] |
[1.0, 2.0, 4.0] |
Per-attempt delays in seconds |
retryable_exceptions |
tuple[type[Exception], ...] |
httpx + openai errors | Exceptions that trigger a transient retry |
rate_limit_exceptions |
tuple[type[Exception], ...] |
openai.RateLimitError |
Exceptions that trigger rate-limit retry |
rate_limit_retry_interval |
float |
10.0 |
Seconds between rate-limit retries |
rate_limit_max_duration_seconds |
float |
900.0 |
Maximum total wait for rate-limit retries |
Requirements
- Python 3.11+
pydantic >= 2.0langchain-core >= 0.2- Any LangChain chat model. Tool-calling mode (default) needs a tool/function-calling-capable model (OpenAI, Anthropic, Gemini, Mistral, …);
ExtractionMode.JSONworks with any chat model, including local models without tool-calling support.
Project structure
saidex/
├── src/saidex/ # library source
│ ├── __init__.py # public API
│ ├── extractor.py # extract_data, extract_data_from_text + agent loop
│ ├── models.py # ExtractDataStats, ExtractorRunStats, ExtractionMode
│ ├── retry.py # RetryConfig + with_retry
│ ├── tools.py # Tool dataclass for the agent loop
│ ├── utils.py # create_instance_safe
│ └── py.typed # PEP 561 marker (enables type checking by consumers)
├── examples/ # runnable usage examples
├── tests/ # pytest test suite
├── docs/ # topic-specific documentation (see docs/index.md)
├── pyproject.toml # project metadata + tool config
├── uv.lock # pinned dependency tree (committed to git)
└── .python-version # Python version pin for uv (3.12)
Development
This project uses uv for dependency and virtual-environment management.
uv is a fast, modern replacement for pip + venv + pip-tools, written in Rust.
First-time setup
# 1. Install uv (if not already installed) pip install uv # or on macOS/Linux: curl -LsSf https://astral.sh/uv/install.sh | sh # 2. Clone the repository git clone https://github.com/mlauf-labs/saidex cd saidex # 3. Create the virtual environment and install all dependencies uv sync
uv sync reads uv.lock and installs the exact pinned versions into .venv.
The virtual environment is created automatically — no manual python -m venv needed.
Daily commands
uv run pytest # run the full test suite uv run pytest tests/test_utils.py -v # run a single test file uv run ruff check src/ tests/ # lint uv run ruff check src/ tests/ --fix # lint + auto-fix uv run ruff format src/ tests/ # format code uv run mypy src/ # static type checking
Managing dependencies
# Add a runtime dependency (updates pyproject.toml + uv.lock) uv add langchain-openai # Add a dev-only dependency uv add --group dev ipython # Add an optional extra (e.g. openai rate-limit support) uv add --optional openai openai # Remove a dependency uv remove langchain-openai # Upgrade all dependencies to their latest allowed versions uv lock --upgrade uv sync
Build and publish
# Build source distribution + wheel into dist/ uv build # Publish to PyPI (requires PYPI_TOKEN or interactive login) uv publish
About uv.lock
uv.lock is a machine-generated file that pins the exact version of every
direct and transitive dependency. It is committed to the repository so that
every developer and CI run installs the identical package tree.
To update all dependencies to their latest allowed versions:
uv lock --upgrade # re-solves the dependency graph uv sync # applies the new lock file to .venv
Contributing
Contributions are welcome! Please open an issue first to discuss significant changes.
See CONTRIBUTING.md for the development setup, the test/lint/type-check commands, and the pull-request checklist.
Planned work and feature ideas are tracked as GitHub issues.
Have a use case SAIDEX doesn't cover yet? Open an issue — we'd love to hear about it.
Security
If you discover a security vulnerability, please do not open a public issue. Report it privately via GitHub Security Advisories — see SECURITY.md for details.
License
Copyright 2026 Martin Lauff. Licensed under the Apache License, Version 2.0.
来源:GitHub · github.com