Every method maps to one lifecycle

Open or create an Attest database. The primary entry point for all operations.

Parameters

Returns

AttestDB — database handle (also usable as context manager)

Example

import attestdb

# Open or create a database
db = attestdb.open("my_knowledge.db")

# Context manager closes automatically
with attestdb.open("my_knowledge.db") as db:
    db.ingest(...)

One-line setup: create a database, register vocabularies, and configure the curator in a single call.

Parameters

Returns

AttestDB — fully configured database handle

Example

import attestdb

db = attestdb.quickstart("bio.db", vocabs=["bio"], curator="gemini")

Add a single claim with full provenance. The atomic write operation — every claim must have a source.

Parameters

Returns

str — the claim_id (SHA-256 hash)

Example

claim_id = db.ingest(
    subject=("api-gateway", "service"),
    predicate=("depends_on", "dependency"),
    object=("redis", "service"),
    provenance={"source_type": "k8s_manifest", "source_id": "deploy/prod"},
    confidence=0.95,
)

Bulk-ingest many claims at once. Faster than individual ingest() calls — warms caches and skips per-claim corroboration tracking.

Parameters

Returns

BatchResult — .ingested (int), .duplicates (int), .errors (list)

Example

from attestdb import ClaimInput

claims = [
    ClaimInput(
        subject=("api-gateway", "service"),
        predicate=("depends_on", "dependency"),
        object=("redis", "service"),
        provenance={"source_type": "config", "source_id": "k8s"},
    ),
    # ... more claims
]
result = db.ingest_batch(claims)
print(f"Ingested {result.ingested}, skipped {result.duplicates} dupes")

Extract claims from unstructured text using LLM-powered extraction, then ingest them.

Parameters

Returns

list — extracted and ingested claims

Example

db.ingest_text(
    "Redis is used as the primary cache for the API gateway. "
    "The gateway also depends on PostgreSQL for persistent storage.",
    source_id="architecture_doc_v2",
)

Extract claims from a conversation. Accepts OpenAI/Anthropic message format ([{role, content}, ...]).

Parameters

Returns

ChatIngestionResult — per-turn breakdown of extracted claims

Example

messages = [
    {"role": "user", "content": "What cache does our API use?"},
    {"role": "assistant", "content": "The API gateway uses Redis for caching."},
]
result = db.ingest_chat(messages, extraction="heuristic")

Extract claims from a Slack workspace export ZIP. Optionally filter by channel name.

Parameters

Returns

list[ChatIngestionResult] — one result per channel/thread with bot interaction

Example

results = db.ingest_slack(
    "slack_export.zip",
    channels=["engineering", "incidents"],
    extraction="smart",
)

Create a connector for an external data source. Returns a Connector instance — call .run(db) to fetch and ingest. 30 connectors available: slack, teams, gmail, gdocs, gdrive, zoho, postgres, mysql, mssql, notion, confluence, sharepoint, csv, sqlite, github, jira, linear, hubspot, salesforce, zendesk, servicenow, pagerduty, http, airtable, mongodb, elasticsearch, s3, google_sheets, box, dsi.

Parameters

Returns

Connector — call .run(db) to execute

Examples

# Slack: live channel history
conn = db.connect("slack", token="xoxb-...", channels=["general"])
result = conn.run(db)

# PostgreSQL: map query columns to claims
conn = db.connect("postgres",
    dsn="postgresql://user:pass@host/db",
    query="SELECT gene, relation, target FROM interactions",
    mapping={"subject": "gene", "predicate": "relation", "object": "target"},
)
result = conn.run(db)

# Notion: ingest pages as text
conn = db.connect("notion", api_key="ntn_...", save=True)
result = conn.run(db)

Get a full picture of an entity: relationships, narrative summary, confidence scores, contradictions, and topic membership.

Parameters

Returns

ContextFrame — focal entity, relationships, claim count, narrative, confidence range, contradictions, topic membership

Example

frame = db.query("redis")
print(frame.focal_entity.name, "—", frame.claim_count, "claims")
print(frame.narrative)
for rel in frame.direct_relationships:
    print(f"  {rel.predicate} → {rel.target.name} ({rel.confidence:.0%})")

Semantic search via embeddings. Returns the top-k closest claims by vector similarity.

Parameters

Returns

list[tuple[str, float]] — list of (claim_id, distance) pairs

Same as query() but also returns timing and candidate counts for performance profiling.

Returns

tuple[ContextFrame, QueryProfile] — the frame plus profiling data (elapsed_ms, total_candidates, after_scoring)

Example

frame, profile = db.explain("redis")
print(f"Query took {profile.elapsed_ms:.1f}ms, {profile.total_candidates} candidates")

Find the top-k paths between two entities with per-hop edge details, sorted by total confidence.

Parameters

Returns

list[PathResult] — each with .steps (list of PathStep), .total_confidence, .length

Example

paths = db.find_paths("api-gateway", "postgresql")
for p in paths:
    hops = " → ".join(s.entity_id for s in p.steps)
    print(f"{hops}  (confidence: {p.total_confidence:.2f})")

Health metrics for the knowledge base: single-source entities, source distribution, knowledge density, gap counts.

Parameters

Returns

QualityReport — total_claims, total_entities, single_source_entity_count, avg_claims_per_entity, source_type_distribution, predicate_distribution

Example

report = db.quality_report()
print(f"{report.total_entities} entities, {report.total_claims} claims")
print(f"{report.single_source_entity_count} single-source entities")
print(f"Avg claims/entity: {report.avg_claims_per_entity:.1f}")

Quantified health score (0–100) with weighted metrics: multi-source ratio, corroboration, freshness, source diversity, and confidence trend.

Returns

KnowledgeHealth — health_score (0–100), multi_source_ratio, corroboration_ratio, freshness_score, source_diversity, confidence_trend, knowledge_density

Example

health = db.knowledge_health()
print(f"Health: {health.health_score:.0f}/100")
print(f"Multi-source ratio: {health.multi_source_ratio:.0%}")
print(f"Corroboration: {health.corroboration_ratio:.0%}")

Vocabulary-driven gap identification. Compares each entity's relationship profile against expected predicate patterns for its type.

Parameters

Returns

list[GapResult] — entities missing expected relationships

Example

gaps = db.find_gaps({
    "gene": {"associated_with", "interacts_with", "expressed_in"},
    "drug": {"treats", "targets"},
})
for gap in gaps:
    print(f"{gap.entity_id} missing: {gap.missing_predicate_types}")

Predict potential connections between currently-unlinked entities using embedding similarity and common-neighbor scoring.

Parameters

Returns

list[BridgePrediction] — predicted connections with confidence scores

Find entities with reliability concerns: single-source dependencies, stale data, or wide confidence spreads.

Parameters

Returns

list[EntityConfidenceAlert] — entities with provenance or confidence issues

Retract all claims from a source and mark anything that depended on them as degraded. The nuclear option for bad sources.

Parameters

Returns

CascadeResult — .source_retract (RetractResult), .degraded_claim_ids (list), .degraded_count (int)

Example

result = db.retract_cascade("unreliable_vendor", "Data quality issues found in audit")
print(f"Retracted {result.source_retract.retracted_count} claims")
print(f"Degraded {result.degraded_count} downstream dependents")

Time-travel: query the knowledge base as it was at a specific point in time. Returns a read-only snapshot view.

Parameters

Returns

AttestDBSnapshot — read-only view supporting query(), claims_for(), list_entities()

Example

import time

# View knowledge base as of yesterday
yesterday = int((time.time() - 86400) * 1_000_000_000)
snapshot = db.at(yesterday)
frame = snapshot.query("redis")
print(f"Yesterday: {frame.claim_count} claims")

If this source is retracted, how many claims and entities are affected? Preview the blast radius before acting.

Parameters

Returns

ImpactReport — direct_claims, downstream_claims, affected_entities, claim_ids

Example

report = db.impact("vendor_api_v2")
print(f"Direct: {report.direct_claims}, Downstream: {report.downstream_claims}")
print(f"Affects {len(report.affected_entities)} entities")

Find entities backed by only a single source, knowledge gaps, and low-confidence areas.

Parameters

Returns

BlindspotMap — single_source_entities, knowledge_gaps, low_confidence_areas

Example

blind = db.blindspots()
print(f"{len(blind.single_source_entities)} entities rely on a single source")
for entity in blind.single_source_entities[:5]:
    print(f"  {entity}")

How many independent sources agree about an entity? Returns agreement ratio, claims by source, and corroborated content IDs.

Parameters

Returns

ConsensusReport — total_claims, unique_sources, avg_confidence, agreement_ratio, claims_by_source, corroborated_content_ids

Example

report = db.consensus("redis")
print(f"Agreement: {report.agreement_ratio:.0%} across {report.unique_sources} sources")

Per-source corroboration and retraction rates. Pass a source_id for one source, or omit for all sources.

Parameters

Returns

dict — per-source metrics: total_claims, active, retracted, degraded, corroboration_rate, retraction_rate

Example

reliability = db.source_reliability()
for src, metrics in reliability.items():
    print(f"{src}: {metrics['corroboration_rate']:.0%} corroborated")

What-if analysis: would this claim corroborate existing knowledge? Does it fill a gap between known entities?

Parameters

Returns

HypotheticalReport — would_corroborate, existing_corroborations, fills_gap, content_id, related_entities

Example

from attestdb import ClaimInput

report = db.hypothetical(ClaimInput(
    subject=("redis", "service"),
    predicate=("depends_on", "dependency"),
    object=("sentinel", "service"),
    provenance={"source_type": "test", "source_id": "test"},
))
print(f"Would corroborate: {report.would_corroborate}")
print(f"Fills gap: {report.fills_gap}")

Discover novel regulatory predictions via causal composition. Follows causal edges through intermediaries and composes predicates (inhibits + inhibits = activates). Returns predictions ranked by convergent evidence — genuine gaps first. No LLM calls. Validated at 47% precision across 4 genes (8/17 confirmed, 0 contradicted).

Parameters

Returns

list[Prediction] — target, predicted_predicate, supporting_paths, opposing_paths, consensus, is_gap, evidence

Example

predictions = db.predict("gene_7157")  # TP53
for p in predictions[:5]:
    print(f"{p.predicted_predicate} -> {p.target}")
    print(f"  {p.supporting_paths} supporting, gap={p.is_gap}")

Test a hypothesis against the knowledge graph. Returns causal evidence for/against with multi-hop composition paths, contradiction detection, gap analysis, and follow-up suggestions. No LLM calls.

Parameters

Returns

SandboxVerdict — verdict (supported/contradicted/plausible/insufficient_data), confidence_score, direct + indirect evidence, gaps, follow-ups

Example

verdict = db.what_if(
    ("gene_940", "gene"),
    ("upregulates", "relation"),
    ("gene_29126", "gene"),
)
print(verdict.verdict)       # "plausible"
print(verdict.explanation)   # "12 causal path(s) supporting"

Proactive hypothesis generation from graph structure. Three signals: bridge predictions (ensemble-scored pairs with composed predicates), cross-domain insights (topology bridge entities), and chain completion (2-hop pairs missing direct connections). Pure computation, no LLM.

Parameters

Returns

list[Discovery] — hypothesis, predicted_predicate, confidence, novelty_score, evidence_summary, supporting_paths, suggested_action

Example

for d in db.discover(top_k=5):
    print(d.hypothesis)
    print(f"  {d.predicted_predicate} (conf={d.confidence:.2f}, novelty={d.novelty_score:.2f})")
    print(f"  → {d.suggested_action}")

Find structural analogies: A:B :: C:D. Uses structural embeddings to find entities similar to A and B, then predicts the C:D pair and relationship.

Parameters

Returns

list[Analogy] — entity_a, entity_b, entity_c, entity_d, predicted_predicate, score, explanation

Example

for a in db.analogies("BRCA1", "apoptosis"):
    print(f"{a.entity_c} : {a.entity_d} (score={a.score:.2f})")
    print(f"  {a.explanation}")

Evaluate a natural-language hypothesis against the knowledge base. Parses entities, finds multi-hop evidence chains, and returns a verdict with supporting/contradicting evidence.

Parameters

Returns

HypothesisVerdict — verdict (supported/contradicted/partial/unsupported), verdict_confidence, supporting_chains, contradicting_chains, confidence_gaps, suggested_next_steps

Example

verdict = db.test_hypothesis("aspirin reduces inflammation via COX-2")
print(f"{verdict.verdict} (confidence={verdict.verdict_confidence:.2f})")
for chain in verdict.supporting_chains:
    print(f"  {chain.summary}")

Knowledge diff — like git diff for knowledge. Shows what beliefs formed, strengthened, weakened, or contradicted between two time periods. No other database tracks how beliefs evolve over time.

Parameters

Returns

KnowledgeDiff — new_beliefs, strengthened, weakened, new_contradictions, new_entities, new_sources, total_new_claims, summary

Example

diff = db.diff(since="2025-01-01")
print(diff.summary)
# "47 new beliefs; 12 strengthened; 3 new contradictions; 8 new sources"
for b in diff.new_beliefs[:5]:
    print(f"  {b.subject} {b.predicate} {b.object} (conf={b.confidence_after:.2f})")

Self-healing contradictions. Finds all opposing claims (via OPPOSITE_PREDICATES), scores evidence quality on each side (corroboration, source diversity, recency, confidence), and picks winners. Optionally ingests resolution meta-claims. No other database can reason about its own conflicts.

Parameters

Returns

ContradictionReport — total_found, resolved, ambiguous, analyses (with evidence weights and margins), claims_added

Example

report = db.resolve_contradictions(auto_resolve=True)
print(f"Found {report.total_found}, resolved {report.resolved}")
for a in report.analyses:
    print(f"  {a.subject} ↔ {a.object}: {a.resolution} (margin={a.margin:.2f})")

Counterfactual what-if analysis — compute cascading effects without modifying the database. "What if this paper is retracted? 47 claims affected, 3 drug mechanisms break." No other database can simulate scenarios on its own integrity.

Parameters

Returns

SimulationReport — claims_affected, claims_removed, claims_degraded, entities_now_orphaned, connection_losses, confidence_shifts, risk_score, risk_level, summary

Example

# What if our main source is wrong?
sim = db.simulate(retract_source="paper_2024_nature")
print(f"{sim.claims_removed} claims affected, risk: {sim.risk_level}")
for loss in sim.connection_losses:
    print(f"  {loss.entity_a} ↔ {loss.entity_b}: lost {loss.lost_predicates}")

Knowledge compilation — generate a structured research brief with citations, confidence levels, contradictions, and gaps. An automated literature review from the graph. No other database can produce a structured document with provenance-tracked evidence chains.

Parameters

Returns

KnowledgeBrief — sections (title, key_findings, citations, contradictions, gaps), executive_summary, total_entities, total_claims_cited, strongest_findings, weakest_areas

Example

brief = db.compile("sickle cell treatment")
print(brief.executive_summary)
for section in brief.sections:
    print(f"\n## {section.title}")
    for f in section.key_findings:
        print(f"  • {f}")

Full provenance-traced reasoning chain between two entities. Traces the best path with source citations at every hop, flags contradictions, computes reliability, and generates a human-readable narrative.

Parameters

Returns

Explanation — connected, steps (with source_summary, evidence_text), chain_confidence, narrative, alternative_paths, source_count

Example

exp = db.explain_why("aspirin", "inflammation")
print(exp.narrative)
# Connection: aspirin → inflammation (2 hops, confidence=0.72)
#   1. aspirin —[inhibits]→ cox-2 (conf=0.90) [2 source(s): paper_1, trial_5]
#   2. cox-2 —[promotes]→ inflammation (conf=0.80) [1 source(s): textbook]

Predict next connections for an entity. Uses 2-hop structural analysis and historical growth patterns to predict which entities are most likely to become connected next.

Parameters

Returns

Forecast — predictions (target_entity, predicted_predicate, score, reason, evidence_entities), growth_rate, trajectory

Example

fc = db.forecast("BRCA1")
print(f"Trajectory: {fc.trajectory}, {fc.growth_rate:.1f} connections/month")
for p in fc.predictions[:5]:
    print(f"  → {p.target_entity} via {p.predicted_predicate} (score={p.score:.2f})")

Diff two knowledge bases. Shows what each knows that the other doesn’t, shared beliefs, entity coverage gaps, and contradictions between them. No other database can structurally compare two knowledge bases.

Parameters

Returns

MergeReport — self_unique_beliefs, other_unique_beliefs, shared_beliefs, conflicts, self_unique_entities, other_unique_entities, summary

Example

team_a = attestdb.open("team_a.db")
team_b = attestdb.open("team_b.db")
report = team_a.merge_report(team_b)
print(f"Team A knows {report.self_unique_beliefs} things Team B doesn't")
print(f"{len(report.conflicts)} disagreements")

Full gap-closing loop: detect blindspots, formulate questions, research each via LLM, ingest validated claims, and measure improvement.

Parameters

Returns

InvestigationReport — questions_generated, questions_researched, claims_ingested, blindspot_before, blindspot_after

Example

report = db.investigate(max_questions=10)
print(f"Researched {report.questions_researched} questions")
print(f"Ingested {report.claims_ingested} new claims")
print(f"Blindspots: {report.blindspot_before} → {report.blindspot_after}")

Research a single question. The LLM generates structured claims, which are validated and ingested.

Parameters

Returns

ResearchResult — claims_ingested, claims_rejected, inquiry_resolved, source

Example

result = db.research_question(
    "What databases does the API gateway depend on?",
    entity_id="api-gateway",
    entity_type="service",
)
print(f"Ingested {result.claims_ingested} claims")