Store Instructor-extracted structured outputs in CortexDB as typed experiences.

Instructor Integration

Instructor extracts typed Pydantic objects from LLM responses. CortexDB stores them as durable, queryable memory — either as JSON content or directly as fact triples.

Install

pip install cortexdbai[instructor]

Extraction cache

import os
from cortexdb import Cortex
from cortexdb.integrations.instructor import CortexDBCache

client = Cortex(
    api_url="https://api-v1.cortexdb.ai",
    actor="user:alice",
    bearer=os.environ["CORTEX_TOKEN"],
)

cache = CortexDBCache(client=client, scope="org:acme/user:alice")

# Serve identical/similar extractions from memory instead of re-calling the LLM.
cached = cache.get("Extract the user info", UserInfo)
if cached is None:
    result = instructor_client.chat.completions.create(...)  # your Instructor call
    cache.set("Extract the user info", result)

remember_extraction / recall_extractions store past extractions to reuse as few-shot examples.

Prefer manual control?

import os, instructor
from datetime import datetime, timezone
from uuid import uuid4
from openai import OpenAI
from pydantic import BaseModel
from cortexdb.v1 import V1Client

llm     = instructor.from_openai(OpenAI())
cortex  = V1Client(api_url="https://api-v1.cortexdb.ai", actor="user:alice",
                   bearer=os.environ["CORTEX_TOKEN"])
SCOPE   = "org:acme/user:alice"


class DealUpdate(BaseModel):
    customer:   str
    deal_stage: str
    seats:      int

extracted = llm.chat.completions.create(
    model="gpt-4o",
    response_model=DealUpdate,
    messages=[{"role": "user", "content": "Acme just bumped to 200 seats and signed."}],
)

# Capture as JSON for inspection
cortex.experience(
    scope=SCOPE,
    modality="tool_result",
    content_kind="json",
    content_data=extracted.model_dump(),
    observed_at=datetime.now(timezone.utc).isoformat(),
    idempotency_key=f"inst-{uuid4()}",
)

# Or capture as a fact triple
cortex.experience(
    scope=SCOPE,
    modality="observation",
    content_kind="triple",
    triple={
        "subject":   {"type": "entity", "id": f"ent_{extracted.customer}"},
        "predicate": "deal_stage",
        "object":    {"datatype": "string", "value": extracted.deal_stage},
    },
    observed_at=datetime.now(timezone.utc).isoformat(),
    idempotency_key=f"inst-fact-{uuid4()}",
)

Recall structured context before extraction

Use recall() to ground the extraction prompt in known account context:

pack = cortex.recall(
    scope=SCOPE,
    view="holistic",
    query=f"What do we already know about {extracted.customer}?",
    include=["facts", "beliefs", "episodes"],
    budgets={"max_tokens": 2000},
)

context_block = pack.get("context_block") or ""

See also