Workloads

batch = await kb.remember_batch(docs, namespace=ns_id,
                                entity_types=["ORGANIZATION", "PRODUCT", "CONCEPT", "PERSON"],
                                relationship_types=["AFFECTS", "ASSIGNED_TO", "RELATES_TO", "MENTIONS"])

# vector shape:
await kb.recall("tickets about login failures", namespace=ns_id, limit=5)

# graph shape — walk outward from the customer entity:
acme = _find_entity(await kb.list_entities(namespace=ns_id, entity_type="ORGANIZATION"), "Acme")
related = await kb.find_related_entities(acme.id, namespace=ns_id, max_depth=2, limit=25)

v1 = await kb.create_namespace()
stable_id, v1_row = v1.namespace_id, v1.id

await kb.remember(text, namespace=stable_id, ...)            # facade → active version

v2 = await kb.storage.create_namespace_version(previous_version=v1)  # atomically flips active
active_row = await kb.storage.resolve_namespace(stable_id)   # stable id → active row id
v1_entities = await kb.storage.list_entities(v1_row)         # read a frozen older version

from khora import ExpertiseConfig, EntityTypeConfig, RelationshipTypeConfig
from khora.engines.vectorcypher.engine import VectorCypherConfig

expertise = ExpertiseConfig(
    name="recruiter_demo",
    description="Recruiter-style ontology: candidates, employers, skills.",
    # The system prompt does the canonicalization that dedup alone can't,
    # which matters most on the embedded backend (no entity-vector index).
    system_prompt=(
        "Extract from short candidate résumés. Rules:\n"
        "1. Companies: canonicalize to the brand, no legal suffix or URL "
        "('Stripe Inc.', 'Stripe, Inc.', 'stripe.com' all become 'Stripe').\n"
        "2. Skills: canonicalize aliases ('k8s'/'K8s' to 'Kubernetes', "
        "'Postgres' to 'PostgreSQL', 'Golang' to 'Go').\n"
        "3. Negation: no HAS_SKILL edge for a skill the candidate explicitly lacks.\n"
        "4. One CANDIDATE per person, by full name."
    ),
    entity_types=[
        EntityTypeConfig(name="CANDIDATE", description="A job candidate, by name.",
                         identifiers=["name"]),
        EntityTypeConfig(name="COMPANY", description="A current or past employer.",
                         identifiers=["name"], aliases=["EMPLOYER", "ORGANIZATION"]),
        EntityTypeConfig(name="SKILL", description="A technical skill or technology.",
                         identifiers=["name"], aliases=["TECHNOLOGY", "TOOL"]),
    ],
    relationship_types=[
        RelationshipTypeConfig(name="WORKED_AT", description="CANDIDATE employed by COMPANY.",
                               source_types=["CANDIDATE"], target_types=["COMPANY"],
                               properties=["start_year", "end_year", "role"]),
        RelationshipTypeConfig(name="HAS_SKILL", description="CANDIDATE has experience with SKILL.",
                               source_types=["CANDIDATE"], target_types=["SKILL"]),
    ],
)

# Full extraction (every chunk; résumés have no filler) + the ontology.
engine_kwargs = {"vectorcypher_config": VectorCypherConfig(skeleton_core_ratio=1.0,
                                                           min_extraction_tokens=0)}
batch = await kb.remember_batch(
    resumes, namespace=ns_id, expertise=expertise,
    entity_types=expertise.get_entity_type_names(),
    relationship_types=expertise.get_relationship_type_names(),
)
# Cross-document unification (the expansion phase) then collapses surface-form
# variants into one node per entity, keyed off each type's `identifiers`.
# A two-set graph intersection answers "Stripe alumni ∩ Kubernetes users".

# 1. Describe each embedded figure with a vision model (app-level; alt text steers it),
#    then ingest text sections and figure descriptions in one batch. Each chunk keeps a
#    stable external_id, and the image path rides in metadata.
figure_desc = await describe_image(fig_path, alt_text, client=vision_client)
batch = [
    {"content": section_text, "external_id": f"{doc}#{heading}",
     "metadata": {"kind": "text"}},
    {"content": f"figure: {figure_desc}", "external_id": f"{doc}#figure-{fig}",
     "metadata": {"kind": "image", "image_path": str(fig_path)}},
]
await kb.remember_batch(
    batch, namespace=ns_id,
    entity_types=["SPACECRAFT", "INSTRUMENT", "LOCATION", "MEASUREMENT", "MISSION"],
    relationship_types=["RELATES_TO", "PART_OF", "LOCATED_IN"],
)

# 2. RAG: recall the top chunks, answer from that context only, cite each source.
result = await kb.recall(question, namespace=ns_id, limit=10)
sources = [d.external_id for d in result.documents]   # what the answer is grounded in