Chapter 8 — The Auto-Ontology Paradigm¶

from trails import capability

@capability
def import_employees(ctx) -> dict:
    """Load employee data with no schema at all."""
    employees = [
        {"name": "Alice", "department": "Engineering", "salary": 95000},
        {"name": "Bob", "department": "Marketing", "salary": 82000},
        {"name": "Carol", "department": "Engineering", "salary": 91000,
         "manager": "Alice"},
    ]
    count = 0
    for emp in employees:
        ctx.kg.node(labels=["Employee"], properties=emp)
        count += 1
    return {"imported": count}

from trails.rml import run_mapping
from trails.testing import fresh_context

ctx = fresh_context()
result = run_mapping(ctx, "mappings/employees.ttl",
                     employees="employees.csv")
print(f"Loaded {result.triples_added} triples")

from trails.onto_infer import infer_schema, generate_code

schema = infer_schema(
    ctx.kg._store,
    trace_id="employee-analysis",
    min_instances=3,
    confidence=0.8,
)

print(f"Analyzed {schema.total_triples_analyzed} triples")
print(f"Found {len(schema.candidates)} candidate types")

for candidate in schema.candidates:
    print(f"\n{candidate.name} ({candidate.instance_count} instances, "
          f"confidence: {candidate.confidence:.2f})")
    for prop in candidate.properties:
        marker = "required" if prop.required else "optional"
        print(f"  {prop.name}: {prop.python_type} ({marker}, "
              f"confidence: {prop.confidence:.2f})")

# Generate the Python module
code = generate_code(schema)
print(code)

"""Auto-generated @node_type declarations.

Source: trails onto infer (deterministic)
"""
from __future__ import annotations

from trails.orm import node_type

# 47 instance(s), confidence: 0.94
# source rdf:type: https://myapp.example/Employee
@node_type("Employee", fields={
    "name": str,         # confidence: 1.00 | e.g. 'Alice', 'Bob', 'Carol'
    "department": str,   # confidence: 1.00 | e.g. 'Engineering', 'Marketing'
    "salary": int,       # confidence: 0.96 | e.g. '95000', '82000'
    "manager": str,      # TODO: verify | confidence: 0.72 | e.g. 'https://...'
})
class Employee:
    pass

# Before (generated):
@node_type("Employee", fields={
    "name": str,
    "department": str,
    "salary": int,
    "manager": str,  # TODO: verify | confidence: 0.72
})
class Employee:
    pass

# After (reviewed):
@node_type("Employee", fields={
    "name": str,
    "department": str,
    "salary": int,
    "manager": "Employee | None",  # optional self-reference
})
class Employee: ...

from trails.onto_generate import generate_schema, estimate_cost

# Always estimate cost first
est = estimate_cost(
    "Clinical trial management with patients, studies, sites, "
    "and adverse events.",
    model="haiku",
)
print(f"Estimated cost: ${est.estimated_cost_usd:.6f}")

# Generate
result = generate_schema(
    "Clinical trial management with patients, studies, sites, "
    "and adverse events. Each study has multiple sites. Patients "
    "enroll at a site. Adverse events link to patients and studies.",
    model="haiku",
    max_types=10,
)

print(f"Generated {len(result.types)} types: {result.types}")
print(f"Actual cost: ${result.cost_usd:.6f}")
print(result.code)

from trails.onto_generate import refine_schema

refined = refine_schema(
    code=result.code,
    feedback="Add a Randomization type linking Patient to StudyArm. "
             "Make patient.email optional.",
    model="haiku",
)
print(refined.code)

from trails.onto_refine import UsageCollector, SchemaAnalyzer

# Start collecting usage data (register as observability hook)
collector = UsageCollector()
collector.start()

# ... your app runs, writes data, queries data ...

analyzer = SchemaAnalyzer(collector)
report = analyzer.analyze(
    min_samples=10,
    null_threshold=0.90,
    violation_threshold=0.10,
    co_query_threshold=0.80,
)

for s in report.suggestions:
    print(f"[{s.kind}] {s.node_type}.{s.field}: {s.reason} "
          f"(confidence: {s.confidence:.2f}, samples: {s.sample_count})")

from trails.onto_refine import generate_migration_code

code = generate_migration_code(report.suggestions)
print(code)

from fastapi import FastAPI
from trails_admin.routes.ontology import router

app = FastAPI()
app.include_router(router)
# Dashboard available at /admin/ontology

from trails.rml import run_mapping
from trails.testing import fresh_context

ctx = fresh_context()
result = run_mapping(ctx, "mappings/employees.ttl",
                     employees="employees.csv")
print(f"Loaded {result.triples_added} triples in {result.duration_ms:.0f}ms")
# Loaded 20 triples in 45ms

"""Auto-generated @node_type declarations.

Source: trails onto infer (deterministic)
"""
from __future__ import annotations

import datetime

from trails.orm import node_type

# 5 instance(s), confidence: 0.92
# source rdf:type: https://myapp.example/Employee
@node_type("Employee", fields={
    "name": str,       # confidence: 1.00 | e.g. 'Alice', 'Bob', 'Carol'
    "dept": str,       # confidence: 1.00 | e.g. 'Engineering', 'Marketing'
    "salary": int,     # confidence: 1.00 | e.g. '95000', '82000'
    "hire_date": str,  # confidence: 1.00 | e.g. '2023-01-15'
})
class Employee:
    pass

import datetime
from trails import node_type

@node_type("Employee", fields={
    "name": str,
    "department": str,       # renamed from "dept" for clarity
    "salary": int,
    "hire_date": datetime.datetime,  # upgraded from str
    "manager": "Employee | None",    # added from domain knowledge
})
class Employee: ...

from trails import capability

@capability
def top_earners(ctx, min_salary: int = 90000) -> list:
    hits = Employee.where(salary__gte=min_salary).fetch(ctx)
    return [
        {"name": e.name, "department": e.department, "salary": e.salary}
        for e in hits
    ]

@capability
def team_report(ctx, department: str) -> dict:
    members = Employee.where(department=department).fetch(ctx)
    total = sum(m.salary for m in members)
    return {
        "department": department,
        "headcount": len(members),
        "total_salary": total,
        "members": [{"name": m.name, "salary": m.salary} for m in members],
    }

from trails.onto_refine import UsageCollector, SchemaAnalyzer

collector = UsageCollector()
collector.start()

# App runs for a week...

analyzer = SchemaAnalyzer(collector)
report = analyzer.analyze(min_samples=50)
for s in report.suggestions:
    print(f"[{s.kind}] {s.node_type}.{s.field}: {s.reason}")

# Example output:
# [add_index] Employee.department: Fields [name, department]
#   co-queried in 87% of cases — consider a composite index.
# [make_optional] Employee.manager: Field 'manager' is null in
#   92% of 500 writes — consider making it optional.