Ctx Langextract¶

ctx_langextract — mvp.ctx_langextract

Cluster: Context & Retrieval | Type: component | MCP Tools: 26

Overview¶

Structured information extractor that pulls typed field values from free-form text using a backend-aware strategy: google/langextract is attempted when available, and regex heuristics provide the dependency-free fallback. Within regex extraction, caller-supplied patterns take priority, then built-in heuristics cover common types (email, URL, date, phone, currency amount, number, person name, ZIP code, IP address), and finally a field_name: value label-colon pattern catches anything else. Returns a flat dict of extracted values alongside the source text and a canonical degradation envelope.

Capabilities: info exposes source_capabilities for regex_heuristic, google_langextract, G6 LLM, SQLite persistence, search backend, returned-set rerank status, and the degradation taxonomy.

Quality: evaluate_quality reports quality_method: exact_match. It compares strings exactly with no semantic normalization, so date-format variants or synonyms can fail even when a human would consider them equivalent.

When to use:

• Parsing structured fields (emails, dates, amounts) from unstructured documents or LLM outputs
• Lightweight extraction where a full NLP pipeline would be overkill
• Post-processing scraped or converted text to populate a schema

Example:

from mvp.ctx_langextract import CtxLangExtractBlock, ExtractInput

block = CtxLangExtractBlock(name="extract")
result = block.infer(ExtractInput(
    text="Invoice date: 2026-01-15. Contact: [email protected]. Total: $1,200.00",
    schema_fields=["date", "email", "amount"],
))
# result.value.extracted → {"date": "2026-01-15", "email": "[email protected]", "amount": "$1,200.00"}

Works well with: ctx_markitdown, ctx_scrapling, align_specs

Public API¶

LangExtractRerankDecision¶

Validated advisory rerank verdict over a returned search-record set.

LLMLangExtractRerankRuntime¶

Provider-neutral search-result-rerank runtime backed by G6's LLM caller.

Constructor:

Methods:

rerank(query: str, records: list[Any]) -> LangExtractRerankDecision¶

CtxLangExtractRerankPatternRuntime¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

enforce_eligibility(decision: LangExtractRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxLangExtractPlanner¶

Runtime-first advisory result-rerank facade with returned-order fallback.

Constructor:

Methods:

rerank(query: str, records: list[Any]) -> list[Any]¶

CtxLangExtractBlock(AIBlock[ExtractInput, ExtractOutput, None])¶

Structured extraction from free text.

Methods:

infer(data: ExtractInput) -> Result[ExtractOutput]¶

capabilities() -> dict[str, Any]¶

bias() -> dict¶

ExtractInput(BaseModel)¶

Input to CtxLangExtractBlock.

ExtractOutput(BaseModel)¶

Output from CtxLangExtractBlock.

CtxLangExtractMCPBlock(AIBlock[MCPLangExtractInput, MCPLangExtractOutput, dict])¶

Full-featured extraction block with SQLite persistence.

Methods:

infer(data: MCPLangExtractInput) -> Result[MCPLangExtractOutput]¶

MCPLangExtractRecord(BaseModel)¶

MCPLangExtractInput(BaseModel)¶

MCPLangExtractOutput(BaseModel)¶

LangExtractStore¶

Sync SQLite store with 6 tables.

Constructor:

Methods:

upsert_schema(name: str, fields_json: str, patterns_json: str, description: str) -> str¶

get_schema(name: str) -> dict[str, Any] | None¶

list_schemas(limit: int = 50) -> list[dict[str, Any]]¶

delete_schema(name: str) -> bool¶

upsert_session(name: str, state_json: str) -> str¶

get_session(name: str) -> dict[str, Any] | None¶

list_sessions(limit: int = 50) -> list[dict[str, Any]]¶

delete_session(name: str) -> bool¶

add_extraction(schema_name: str, text: str, extracted_json: str, n_fields_found: int, n_fields_total: int, source: str) -> str¶

query_extractions(schema_name: str = '', source: str = '', limit: int = 50) -> list[dict[str, Any]]¶

upsert_template(name: str, fields_json: str, patterns_json: str, description: str, tags: list[str]) -> str¶

get_template(name: str) -> dict[str, Any] | None¶

list_templates(limit: int = 50) -> list[dict[str, Any]]¶

add_quality_eval(extraction_id: str, ground_truth_json: str, field_scores_json: str, overall_score: float) -> str¶

query_quality_evals(extraction_id: str = '', limit: int = 50) -> list[dict[str, Any]]¶

upsert_pattern(field_name: str, pattern: str, description: str) -> str¶

get_pattern(field_name: str) -> dict[str, Any] | None¶

list_patterns(limit: int = 100) -> list[dict[str, Any]]¶

text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]¶

TF-IDF search across schemas, templates, extractions, and custom patterns.

count_all() -> dict[str, int]¶

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

Decide whether the agentic search-result-rerank planner should be used.

eligible_fingerprint(records: list[Any]) -> str¶

sha256 over the returned record texts (order-sensitive).

validate_langextract_rerank_decision(decision: LangExtractRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

Returned-set / anti-injection guard for a search-result-rerank decision.

planner_is_llm_trusted(planner: Any) -> bool¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶