Adapt Instructor¶

Adapt Instructor — mvp.adapt_instructor

Cluster: Data Processing | Type: component | MCP Tools: 31

Overview¶

Structured LLM output extraction block that takes a source text and a JSON Schema definition, calls an LLM, and returns a validated Python dict conforming to that schema — retrying up to a configurable number of times on validation failure. Supports claude-code, ollama, openrouter, and an explicit offline demo backend, making it usable in both local-only and cloud-connected pilot workflows. Latency and token counts are reported where the selected backend exposes them.

Launch status: pilot-ready for local MCP workflows. It is useful for first-user extraction, classification, schema storage, validation, and demo workflows, but should not be presented as fully polished GA infrastructure.

Production caveats:

• Backend fallback is explicit, not a general automatic chain. Use backend="offline" for deterministic demo extraction, or configure ollama/openrouter/claude-code directly. Multimodal extraction can fall back to text-only extraction and reports degraded: true when that happens.
• The MCP store writes extraction audit records to SQLite. By default the MCP server uses ~/.instructor_mcp/instructor.db; records include input snippets, extracted output JSON, errors, latency, and token counts. For healthcare, legal, finance, or other sensitive workflows, set INSTRUCTOR_DB_PATH deliberately, control filesystem access, and avoid sending regulated data to cloud backends without your own compliance review.
• Current imports emit Pydantic warnings because the field name schema_json shadows a BaseModel attribute. The warnings are noisy but do not prevent the component or MCP tools from running.
• tokens_used is 0 for the claude-code path because that subprocess path does not expose token usage.
• Native adapter operations expose backend_capabilities, schema-guarded extraction/completion, and provisional streaming. Unsupported provider params, arbitrary tools, unbounded retries/timeouts, and schema bypass are blocked with stable G6_E_* codes.
• The Tier-1 infer() path now carries the canonical reliability envelope (completion_state, warning_card, evidence, request_id, run_id) with an honest three-way state. A provider-backed call (openrouter/ollama/claude-code) whose output validates against the schema on the first attempt reports verified. The offline deterministic fallback and any retry-recovered output report qualified-draft (never verified). Dependency, configuration, provider, schema, and retry-exhaustion failures fail closed as Result.fail carrying completion_state=blocked-escalated plus a stable G6_E_* code (G6_E_DEPENDENCY, G6_E_CONFIG, G6_E_PROVIDER, G6_E_VALIDATION, G6_E_RETRY). The openrouter credential state is enforced via backend_capabilities() before any provider call. The component stays pilot/beta; verified is reachable only on a real executed-and-validated provider call.
• The shorthand schema form shown below is supported for quick starts. Full JSON Schema remains the preferred form for production validation because it supports required fields, nested objects, enums, and constraints explicitly.

When to use:

• Extracting typed, schema-validated fields from free-text documents (invoices, reports, emails)
• Converting LLM responses into structured data for downstream Pydantic models or database writes
• Running repeated extraction tasks with retry logic to ensure schema compliance
• Building ETL pipelines that parse unstructured text into structured records

Example:

from mvp.adapt_instructor import InstructorBlock, InstructorInput

block = InstructorBlock(name="instructor")
result = block.infer(InstructorInput(
    text="Order #1042 placed by Alice Smith on 2026-03-15 for $249.99.",
    schema_json={"order_id": "string", "customer": "string", "amount": "number"},
    backend="offline",
    max_retries=3,
))
# result.value.extracted → {"order_id": "1042", "customer": "Alice Smith", "amount": 249.99}

For production validation, prefer full JSON Schema:

schema_json={
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "customer": {"type": "string"},
        "amount": {"type": "number"},
    },
    "required": ["order_id", "customer", "amount"],
}

Works well with: ctx_langextract, adapt_memory, database

Public API¶

InstructorBlock(AIBlock[InstructorInput, InstructorOutput, None])¶

Extract structured data from text using LLM + Pydantic validation.

Methods:

infer(data: InstructorInput) -> Result[InstructorOutput]¶

InstructorInput(BaseModel)¶

Input for structured extraction from text via LLM.

Methods:

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

InstructorOutput(BaseModel)¶

Output from structured extraction.

InstructorMCPBlock(AIBlock[MCPInstructorInput, MCPInstructorOutput, dict])¶

Dispatcher for 27 adapt_instructor MCP operations.

Methods:

store() -> InstructorStore¶

infer(data: MCPInstructorInput) -> Result[MCPInstructorOutput]¶

MCPInstructorInput(BaseModel)¶

Input to InstructorMCPBlock — 27 ops.

Methods:

schema_json() -> dict¶

MCPInstructorOutput(BaseModel)¶

Output from InstructorMCPBlock.

InstructorStore¶

SQLite-backed store for the adapt_instructor MCP sub-package.

Constructor:

Methods:

store_schema(name: str, schema_json: dict, description: str = '', tags: str = '', version: int = 0) -> dict¶

Store a JSON schema. version=0 auto-increments.

retrieve_schema(name: str) -> dict | None¶

Retrieve latest version of a schema by name.

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

List schemas, optionally filtered by tags.

delete_schema(name: str) -> int¶

Delete all versions of a schema. Returns count deleted.

search_schemas(query: str, top_k: int = 10) -> list[dict]¶

TF-IDF search across schemas (falls back to substring).

store_template(name: str, template: str, description: str = '', tags: str = '') -> dict¶

Store a prompt template. Returns id + name.

retrieve_template(name: str) -> dict | None¶

Retrieve a template by name.

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

List templates, optionally filtered by tags.

delete_template(name: str) -> int¶

Delete a template by name. Returns count deleted.

increment_template_use(name: str) -> None¶

render_template(name: str, variables: dict) -> str | None¶

Render a template with variables. Returns None if not found.

store_backend(backend: str, model: str = '', max_retries: int = 3, api_key_env: str = '', is_default: bool = False, priority: int = 0, config_json: str = '{}') -> dict¶

Store a backend configuration.

list_backends() -> list[dict]¶

List all configured backends.

get_default_backend() -> dict | None¶

Get the default backend config.

log_extraction(provider: str = '', schema_name: str = '', template_name: str = '', input_text: str = '', output_json: dict | None = None, success: bool = True, error_message: str = '', latency_ms: int = 0, tokens_used: int = 0) -> str¶

Log an extraction attempt. Returns the extraction id.

log_hook_event(extraction_id: str, stage: str, payload: dict | None = None) -> None¶

Log a hook callback event.

get_hook_events(extraction_id: str) -> list[dict]¶

Get all hook events for an extraction.

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

store_retry_config(max_retries: int = 3, strategy: str = 'reask') -> dict¶

Store/update retry configuration.

get_retry_config() -> dict¶

Retrieve retry configuration.

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

MCP Tools¶