Each cookbook is a complete, runnable example. No API keys required for the scripted cookbooks. The Org Knowledge Console connects to live Slack, Gmail, and Google Docs.
Connect Slack + Gmail + Google Docs via OAuth. Ingest your company's knowledge, explore it in a graph, and ask questions — all from a browser.
Ingest literature + ChatGPT + Slack. Bridge prediction finds candidates — 4/10 validated in published research.
Model architecture from K8s manifests, capture incident patterns from Slack. "What breaks if Redis goes down?" — answered with sources.
Track models, datasets, and features with provenance. "What was tried for churn prediction?" — every answer traces to an experiment run.
Your company's most valuable knowledge lives in Slack threads, email chains, and Google Docs that nobody can find six months later. This cookbook shows how to connect those sources via OAuth, ingest everything into a Attest knowledge base, and explore it from a browser dashboard.
$ pip install attest-console $ attest-console my_company.db
This opens http://localhost:8877 in your browser. From there:
# The console does this for you, but you can also do it from Python: import attestdb from attest_console.org_vocabulary import register_org_vocabulary db = attestdb.quickstart("my_company.db") register_org_vocabulary(db) db.configure_curator("groq") # Free tier, ~$0.05/M tokens # Ingest a Slack export results = db.ingest_slack("team_export.zip", extraction="smart") for r in results: print(f" #{r.conversation_id}: {r.claims_ingested} claims") # Or ingest text directly (from an email, a doc, anything) db.ingest_text( "The auth team decided to migrate from JWT to session tokens. " "Sarah proposed this and the platform team approved it.", source_id="eng-all-hands-2024-02", ) # Query what you know frame = db.query("auth-service", depth=2) print(frame.narrative) # Find who knows what for rel in frame.direct_relationships: print(f" {rel.predicate} → {rel.target.name} ({rel.n_independent_sources} sources)") # Ask natural-language questions # The Ask page does entity extraction + context assembly + LLM synthesis # "Who is responsible for the auth migration?" # → sarah proposed auth-migration, platform team approved auth-migration # (2 claims, sources: slack #eng-decisions, eng-all-hands-2024-02)
The console registers an organizational vocabulary with entity types and predicates designed for company knowledge:
| Entity Types | Predicate Types |
|---|---|
person, team, channel, document, email_thread |
authored, mentioned, decided, discussed_in |
project, decision, tool, process, meeting |
responsible_for, member_of, reports_to, depends_on |
related_to, proposed, approved, uses |
Smart extraction on Groq (free tier, ~$0.05/M tokens):
| Source | Items | Smart Skip | Est. Cost |
|---|---|---|---|
| 10K Slack messages | 10,000 | 60% | $0.02 |
| 2K email threads | 2,000 | 50% | $0.025 |
| 50 Google Docs | 50 | 30% | $0.004 |
| 100 NL queries | 100 | — | $0.01 |
| Total | $0.06 |
Heuristic mode (no LLM) is completely free. Good for structured text like meeting notes and project updates.
Stats overview: entity/claim counts, source breakdown, health score, quick actions.
Search with HTMX typeahead, filter by type and min claims, click through to detail pages.
Full-screen Cytoscape.js canvas. Node size = claims, edge thickness = confidence. Focus on any entity.
Knowledge health: multi-source %, confidence alerts, bridge predictions, knowledge gaps.
Pick a source, choose extraction mode, watch SSE progress. Job history with costs.
Natural-language questions with grounded answers. Evidence panel shows exact claims used.
A cancer biology lab uses ChatGPT and Slack daily. This cookbook captures claims from published literature, ChatGPT research sessions, and Slack channel discussions into a unified knowledge base. Then it discovers novel drug targets, finds knowledge gaps, and tracks research questions.
# One-line setup db = attestdb.quickstart("cancer_lab.db", vocabs=["bio"], curator="heuristic") # Ingest from multiple sources db.ingest_batch(literature_claims) # PubMed, PDB, Reactome db.ingest_chat(chatgpt_session, extraction="heuristic") # ChatGPT conversation db.ingest_slack("lab_slack.zip", extraction="heuristic") # Slack workspace # Query unified knowledge frame = db.query("BRCA1", depth=2) print(frame.narrative) # Discover hidden connections db.generate_structural_embeddings(dim=32) bridges = db.find_bridges(top_k=10) # Track research questions db.ingest_inquiry(question="Can Olaparib treat breast cancer?", subject=("Olaparib", "compound"), object=("Breast Cancer", "disease"))
Knowledge base: 21 entities, 29 claims
BRCA1: 24 claims, 23 relationships
--[associated_with]--> Breast Cancer (disease, conf=0.91)
--[binds]--> RAD51 (protein, conf=0.85)
--[involved_in]--> DNA Repair Pathway (pathway, conf=0.70)
Bridge predictions:
olaparib <-> talazoparib (similarity=0.816)
Total runtime: 1.98s
#!/usr/bin/env python3
"""Biomedical Research Cookbook — No API keys required. Runs in ~2 seconds."""
from __future__ import annotations
import json, logging, os, sys, tempfile, time, zipfile
logging.disable(logging.WARNING)
import attestdb
from attestdb import ClaimInput
def section(title):
print(f"\n{'─'*60}\n {title}\n{'─'*60}\n")
def main():
start = time.perf_counter()
with tempfile.TemporaryDirectory() as tmp:
db = attestdb.quickstart(os.path.join(tmp, "cancer_lab"), vocabs=["bio"], curator="heuristic")
section("Ingest published literature")
literature_claims = [
ClaimInput(subject=("BRCA1","gene"), predicate=("associated_with","associated_with"),
object=("Breast Cancer","disease"), provenance={"source_type":"database_import","source_id":"pmid:20301425"}, confidence=0.95),
ClaimInput(subject=("BRCA2","gene"), predicate=("associated_with","associated_with"),
object=("Breast Cancer","disease"), provenance={"source_type":"database_import","source_id":"pmid:20301425"}, confidence=0.93),
ClaimInput(subject=("BRCA1","gene"), predicate=("associated_with","associated_with"),
object=("Ovarian Cancer","disease"), provenance={"source_type":"database_import","source_id":"pmid:24677121"}, confidence=0.91),
ClaimInput(subject=("BRCA1","protein"), predicate=("binds","binds"),
object=("RAD51","protein"), provenance={"source_type":"experimental","source_id":"pdb:1n0w"}, confidence=0.92),
ClaimInput(subject=("TP53","protein"), predicate=("interacts","interacts"),
object=("BRCA1","protein"), provenance={"source_type":"experimental","source_id":"pmid:19837678"}, confidence=0.88),
ClaimInput(subject=("BRCA1","gene"), predicate=("involved_in","involved_in"),
object=("DNA Repair Pathway","pathway"), provenance={"source_type":"database_import","source_id":"reactome:R-HSA-73894"}, confidence=0.94),
ClaimInput(subject=("RAD51","protein"), predicate=("involved_in","involved_in"),
object=("Homologous Recombination","pathway"), provenance={"source_type":"database_import","source_id":"reactome:R-HSA-5693532"}, confidence=0.96),
ClaimInput(subject=("Tamoxifen","compound"), predicate=("treats","treats"),
object=("Breast Cancer","disease"), provenance={"source_type":"clinical_trial","source_id":"nct:00003140"}, confidence=0.97),
ClaimInput(subject=("Olaparib","compound"), predicate=("treats","treats"),
object=("Ovarian Cancer","disease"), provenance={"source_type":"clinical_trial","source_id":"nct:01874353"}, confidence=0.89),
ClaimInput(subject=("Olaparib","compound"), predicate=("inhibits","inhibits"),
object=("PARP1","protein"), provenance={"source_type":"experimental","source_id":"pmid:16912195"}, confidence=0.95),
ClaimInput(subject=("PARP1","protein"), predicate=("involved_in","involved_in"),
object=("DNA Repair Pathway","pathway"), provenance={"source_type":"database_import","source_id":"reactome:R-HSA-73894"}, confidence=0.93),
ClaimInput(subject=("Tamoxifen","compound"), predicate=("inhibits","inhibits"),
object=("Estrogen Receptor","protein"), provenance={"source_type":"experimental","source_id":"pmid:15928335"}, confidence=0.93),
ClaimInput(subject=("Estrogen Receptor","protein"), predicate=("associated_with","associated_with"),
object=("Breast Cancer","disease"), provenance={"source_type":"experimental","source_id":"pmid:18202748"}, confidence=0.94),
]
batch = db.ingest_batch(literature_claims)
print(f" Ingested {batch.ingested} claims from literature")
# Corroboration
db.ingest(subject=("BRCA1","gene"), predicate=("associated_with","associated_with"),
object=("Breast Cancer","disease"), provenance={"source_type":"database_import","source_id":"disgenet:C0006142"}, confidence=0.90)
print(f" Corroboration: {len(db.claims_by_content_id(db.claims_for('brca1')[0].content_id))} sources confirm BRCA1 ~ Breast Cancer")
section("Extract from ChatGPT")
result = db.ingest_chat([
{"role":"user","content":"What genes are mutated in triple-negative breast cancer?"},
{"role":"assistant","content":"BRCA1 is associated with Triple Negative Breast Cancer. TP53 is associated with Triple Negative Breast Cancer. PIK3CA is associated with Triple Negative Breast Cancer. PTEN is associated with Triple Negative Breast Cancer."},
{"role":"user","content":"Treatment options for BRCA-mutated TNBC?"},
{"role":"assistant","content":"Olaparib treats Triple Negative Breast Cancer. Talazoparib treats Triple Negative Breast Cancer. Talazoparib inhibits PARP1. Carboplatin treats Triple Negative Breast Cancer. Pembrolizumab treats Triple Negative Breast Cancer."},
], conversation_id="tnbc-research", extraction="heuristic")
print(f" {result.claims_ingested} claims from ChatGPT session")
section("Import Slack channels")
slack_zip = os.path.join(tmp, "lab_slack.zip")
with zipfile.ZipFile(slack_zip, "w") as zf:
zf.writestr("channels.json", json.dumps([{"id":"C1","name":"brca-project"},{"id":"C2","name":"journal-club"}]))
zf.writestr("users.json", json.dumps([{"id":"U1","name":"sarah","profile":{"real_name":"Sarah"}}]))
zf.writestr("brca-project/2024-01-20.json", json.dumps([
{"type":"message","user":"U1","text":"PALB2 connection to BRCA2?","ts":"1705700000.000"},
{"type":"message","bot_id":"B1","text":"PALB2 binds BRCA2. PALB2 is associated with Breast Cancer. PALB2 is involved in Homologous Recombination.","ts":"1705700060.000"},
]))
zf.writestr("journal-club/2024-01-22.json", json.dumps([
{"type":"message","user":"U1","text":"ATM inhibitor paper","ts":"1705900000.000"},
{"type":"message","bot_id":"B1","text":"ATM is involved in DNA Repair Pathway. AZD0156 inhibits ATM. ATM interacts with BRCA1.","ts":"1705900060.000"},
]))
slack_results = db.ingest_slack(slack_zip, extraction="heuristic")
print(f" {sum(r.claims_ingested for r in slack_results)} claims from Slack")
section("Query unified knowledge")
stats = db.stats()
print(f" {stats['entity_count']} entities, {stats['total_claims']} claims")
frame = db.query("BRCA1", depth=2)
print(f" BRCA1: {frame.claim_count} claims, {len(frame.direct_relationships)} relationships")
for rel in frame.direct_relationships[:8]:
print(f" --[{rel.predicate}]--> {rel.target.name} ({rel.target.entity_type}, conf={rel.confidence:.2f})")
section("Discover hidden connections")
db.generate_structural_embeddings(dim=32)
for b in db.find_bridges(top_k=5):
print(f" {b.entity_a} <-> {b.entity_b} (similarity={b.similarity:.3f})")
gaps = db.find_gaps({"gene":{"associated_with","involved_in"},"compound":{"treats","inhibits"}}, min_claims=1)
print(f" {len(gaps)} knowledge gaps found")
section("Research questions")
db.ingest_inquiry(question="Can Olaparib treat breast cancer?", subject=("Olaparib","compound"), object=("Breast Cancer","disease"), predicate_hint="treats")
db.ingest(subject=("Olaparib","compound"), predicate=("treats","treats"), object=("Breast Cancer","disease"),
provenance={"source_type":"clinical_trial","source_id":"nct:02000622"}, confidence=0.88)
print(f" Inquiry matches: {len(db.check_inquiry_matches(subject_id='Olaparib', object_id='Breast Cancer'))}")
section("Source retraction")
cascade = db.retract_cascade("pmid:20301425", reason="Data fabrication")
print(f" Retracted: {cascade.source_retract.retracted_count} claims")
print(f" Downstream degraded: {cascade.degraded_count}")
active = [c for c in db.claims_for("brca1") if c.status.name == "ACTIVE"]
print(f" BRCA1 still has {len(active)} active claims (corroboration preserved)")
db.close()
print(f"\n Total runtime: {time.perf_counter() - start:.2f}s")
return 0
if __name__ == "__main__":
sys.exit(main())
An infrastructure team models service dependencies from Kubernetes manifests, captures incident response claims from ChatGPT debugging sessions, and pulls tribal knowledge from Slack ops channels. Query "what depends on Redis?" — answered with every source that says so.
db = attestdb.quickstart("infra.db", vocabs=["devops"], curator="heuristic") # Model architecture db.ingest_batch(architecture_claims) # K8s manifests, org chart, monitoring db.ingest_chat(incident_chat, ...) # Post-incident ChatGPT sessions db.ingest_slack("ops_slack.zip", ...) # Slack ops channels # Impact analysis redis_frame = db.query("Redis Cache", depth=2) has_path = db.path_exists("PagerDuty Alert", "PostgreSQL", max_depth=3)
#!/usr/bin/env python3
"""DevOps Knowledge Base Cookbook — No API keys required. Runs in ~2 seconds."""
from __future__ import annotations
import json, logging, os, sys, tempfile, time, zipfile
logging.disable(logging.WARNING)
import attestdb
from attestdb import ClaimInput
def section(title):
print(f"\n{'─'*60}\n {title}\n{'─'*60}\n")
def main():
start = time.perf_counter()
with tempfile.TemporaryDirectory() as tmp:
db = attestdb.quickstart(os.path.join(tmp, "infra_kb"), vocabs=["devops"], curator="heuristic")
section("Model service architecture")
batch = db.ingest_batch([
ClaimInput(subject=("API Gateway","service"), predicate=("depends_on","depends_on"), object=("Auth Service","service"),
provenance={"source_type":"config_management","source_id":"k8s:api-gateway"}, confidence=1.0),
ClaimInput(subject=("API Gateway","service"), predicate=("depends_on","depends_on"), object=("User Service","service"),
provenance={"source_type":"config_management","source_id":"k8s:api-gateway"}, confidence=1.0),
ClaimInput(subject=("API Gateway","service"), predicate=("depends_on","depends_on"), object=("Redis Cache","service"),
provenance={"source_type":"config_management","source_id":"k8s:api-gateway"}, confidence=1.0),
ClaimInput(subject=("Auth Service","service"), predicate=("depends_on","depends_on"), object=("PostgreSQL","service"),
provenance={"source_type":"config_management","source_id":"k8s:auth-service"}, confidence=1.0),
ClaimInput(subject=("Auth Service","service"), predicate=("depends_on","depends_on"), object=("Redis Cache","service"),
provenance={"source_type":"config_management","source_id":"k8s:auth-service"}, confidence=1.0),
ClaimInput(subject=("User Service","service"), predicate=("depends_on","depends_on"), object=("PostgreSQL","service"),
provenance={"source_type":"config_management","source_id":"k8s:user-service"}, confidence=1.0),
ClaimInput(subject=("Datadog Agent","service"), predicate=("monitors","monitors"), object=("API Gateway","service"),
provenance={"source_type":"monitoring","source_id":"datadog:config"}, confidence=0.95),
ClaimInput(subject=("Datadog Agent","service"), predicate=("monitors","monitors"), object=("PostgreSQL","service"),
provenance={"source_type":"monitoring","source_id":"datadog:config"}, confidence=0.95),
ClaimInput(subject=("PagerDuty Alert","alert"), predicate=("monitors","monitors"), object=("Redis Cache","service"),
provenance={"source_type":"monitoring","source_id":"pagerduty:redis-alert"}, confidence=0.9),
ClaimInput(subject=("Platform Team","team"), predicate=("owns","owns"), object=("API Gateway","service"),
provenance={"source_type":"config_management","source_id":"org-chart"}, confidence=1.0),
ClaimInput(subject=("Platform Team","team"), predicate=("owns","owns"), object=("Auth Service","service"),
provenance={"source_type":"config_management","source_id":"org-chart"}, confidence=1.0),
ClaimInput(subject=("Data Team","team"), predicate=("owns","owns"), object=("PostgreSQL","service"),
provenance={"source_type":"config_management","source_id":"org-chart"}, confidence=1.0),
ClaimInput(subject=("Platform Team","team"), predicate=("owns","owns"), object=("Redis Cache","service"),
provenance={"source_type":"config_management","source_id":"org-chart"}, confidence=1.0),
])
print(f" {batch.ingested} architecture claims")
section("Incident response chat")
result = db.ingest_chat([
{"role":"user","content":"Redis Cache outage at 3am. API Gateway returning 502s. Blast radius?"},
{"role":"assistant","content":"API Gateway depends on Redis Cache for session caching. Auth Service depends on Redis Cache for token storage. PagerDuty Alert monitors Redis Cache. The runbook RB-REDIS-001 mitigates Redis Cache outages."},
{"role":"user","content":"How to prevent this?"},
{"role":"assistant","content":"Redis Sentinel monitors Redis Cache for failover. Platform Team owns Redis Cache. Datadog Agent monitors API Gateway. Consider a circuit breaker in API Gateway that falls back to PostgreSQL."},
], conversation_id="inc-2024-redis-outage", extraction="heuristic")
print(f" {result.claims_ingested} claims from incident chat")
section("Slack ops channels")
slack_zip = os.path.join(tmp, "ops_slack.zip")
with zipfile.ZipFile(slack_zip, "w") as zf:
zf.writestr("channels.json", json.dumps([{"id":"C1","name":"ops-incidents"},{"id":"C2","name":"architecture"}]))
zf.writestr("users.json", json.dumps([{"id":"U1","name":"eng","profile":{"real_name":"Engineer"}}]))
zf.writestr("ops-incidents/2024-02-01.json", json.dumps([
{"type":"message","user":"U1","text":"PostgreSQL connection limits again","ts":"1706800000.000"},
{"type":"message","bot_id":"B1","text":"Auth Service depends on PostgreSQL. User Service depends on PostgreSQL. Data Team owns PostgreSQL.","ts":"1706800060.000"},
]))
zf.writestr("architecture/2024-02-05.json", json.dumps([
{"type":"message","user":"U1","text":"Redis to KeyDB migration impact?","ts":"1707100000.000"},
{"type":"message","bot_id":"B1","text":"Redis Cache is used by API Gateway for rate limiting. Auth Service depends on Redis Cache for JWT blacklisting.","ts":"1707100060.000"},
]))
slack_results = db.ingest_slack(slack_zip, extraction="heuristic")
print(f" {sum(r.claims_ingested for r in slack_results)} claims from Slack")
section("What depends on Redis?")
redis_frame = db.query("Redis Cache", depth=2)
print(f" Redis Cache: {redis_frame.claim_count} claims")
for rel in redis_frame.direct_relationships:
print(f" {rel.predicate} <-- {rel.target.name}")
section("Postgres blast radius")
pg_frame = db.query("PostgreSQL", depth=2)
deps = [r for r in pg_frame.direct_relationships if r.predicate == "depends_on"]
print(f" {len(deps)} services depend on PostgreSQL")
print(f" PagerDuty -> PostgreSQL path: {db.path_exists('PagerDuty Alert', 'PostgreSQL', max_depth=3)}")
section("Post-incident questions")
db.ingest_inquiry(question="Does Redis Sentinel monitor Redis Cache?",
subject=("Redis Sentinel","service"), object=("Redis Cache","service"), predicate_hint="monitors")
print(f" Open inquiries: {len(db.open_inquiries())}")
final = db.stats()
print(f"\n Final: {final['entity_count']} entities, {final['total_claims']} claims")
db.close()
print(f" Runtime: {time.perf_counter() - start:.2f}s")
return 0
if __name__ == "__main__":
sys.exit(main())
An ML team tracks experiments, model comparisons, and feature engineering decisions. Register experiment results as claims with provenance, extract insights from model discussion chats, and query "what was tried for churn prediction?" — every answer traces back to its source.
db = attestdb.quickstart("ml.db", vocabs=["ml"], curator="heuristic") # Track experiments with performance payloads db.ingest( subject=("XGBoost V2", "model"), predicate=("trained_on", "trained_on"), object=("Churn Dataset Q4", "dataset"), provenance={"source_type": "experiment_log", "source_id": "exp:003"}, payload={"accuracy": 0.87, "auc": 0.90}, ) # Query what was tried frame = db.query("Churn Dataset Q4", depth=2) # Model lineage claims = db.claims_for("xgboost v2", predicate_type="derived_from")
#!/usr/bin/env python3
"""ML Experiment Tracker Cookbook — No API keys required. Runs in ~2 seconds."""
from __future__ import annotations
import json, logging, os, sys, tempfile, time, zipfile
logging.disable(logging.WARNING)
import attestdb
from attestdb import ClaimInput
def section(title):
print(f"\n{'─'*60}\n {title}\n{'─'*60}\n")
def main():
start = time.perf_counter()
with tempfile.TemporaryDirectory() as tmp:
db = attestdb.quickstart(os.path.join(tmp, "ml_tracker"), vocabs=["ml"], curator="heuristic")
section("Register experiments")
batch = db.ingest_batch([
ClaimInput(subject=("XGBoost V1","model"), predicate=("trained_on","trained_on"), object=("Churn Dataset Q4","dataset"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-001"}, confidence=1.0, payload={"accuracy":0.82,"auc":0.85}),
ClaimInput(subject=("XGBoost V1","model"), predicate=("uses_feature","uses_feature"), object=("Tenure Months","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-001"}, confidence=0.95),
ClaimInput(subject=("XGBoost V1","model"), predicate=("uses_feature","uses_feature"), object=("Monthly Charges","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-001"}, confidence=0.92),
ClaimInput(subject=("XGBoost V1","model"), predicate=("uses_feature","uses_feature"), object=("Contract Type","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-001"}, confidence=0.88),
ClaimInput(subject=("Random Forest V1","model"), predicate=("trained_on","trained_on"), object=("Churn Dataset Q4","dataset"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-002"}, confidence=1.0, payload={"accuracy":0.79,"auc":0.81}),
ClaimInput(subject=("XGBoost V1","model"), predicate=("outperforms","outperforms"), object=("Random Forest V1","model"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-compare-001"}, confidence=0.88),
ClaimInput(subject=("XGBoost V2","model"), predicate=("trained_on","trained_on"), object=("Churn Dataset Q4","dataset"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-003"}, confidence=1.0, payload={"accuracy":0.87,"auc":0.90}),
ClaimInput(subject=("XGBoost V2","model"), predicate=("derived_from","derived_from"), object=("XGBoost V1","model"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-003"}, confidence=1.0),
ClaimInput(subject=("XGBoost V2","model"), predicate=("uses_feature","uses_feature"), object=("Support Tickets 30D","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-003"}, confidence=0.96),
ClaimInput(subject=("XGBoost V2","model"), predicate=("uses_feature","uses_feature"), object=("NPS Score","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-003"}, confidence=0.91),
ClaimInput(subject=("XGBoost V2","model"), predicate=("outperforms","outperforms"), object=("XGBoost V1","model"),
provenance={"source_type":"experiment_log","source_id":"exp:churn-compare-002"}, confidence=0.92),
ClaimInput(subject=("Neural Net V1","model"), predicate=("trained_on","trained_on"), object=("Fraud Dataset 2024","dataset"),
provenance={"source_type":"experiment_log","source_id":"exp:fraud-001"}, confidence=1.0, payload={"accuracy":0.94,"f1":0.72}),
ClaimInput(subject=("Neural Net V1","model"), predicate=("uses_feature","uses_feature"), object=("Transaction Amount","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:fraud-001"}, confidence=0.97),
ClaimInput(subject=("Neural Net V1","model"), predicate=("uses_feature","uses_feature"), object=("Merchant Category","feature"),
provenance={"source_type":"experiment_log","source_id":"exp:fraud-001"}, confidence=0.85),
ClaimInput(subject=("Fraud Dataset 2024","dataset"), predicate=("derived_from","derived_from"), object=("Fraud Dataset 2023","dataset"),
provenance={"source_type":"experiment_log","source_id":"data:fraud-v2"}, confidence=1.0),
])
print(f" {batch.ingested} experiment claims")
section("ChatGPT model discussion")
result = db.ingest_chat([
{"role":"user","content":"XGBoost V2 is at 0.90 AUC. Should we try LightGBM?"},
{"role":"assistant","content":"XGBoost V2 outperforms Random Forest V1. LightGBM Churn V1 was trained on Churn Dataset Q4 with 0.88 AUC. XGBoost V2 outperforms LightGBM Churn V1. Key features: Support Tickets 30D and NPS Score."},
{"role":"user","content":"What about fraud detection?"},
{"role":"assistant","content":"Neural Net V1 was evaluated on Fraud Dataset 2024 with 0.72 F1. Isolation Forest V1 was trained on Fraud Dataset 2024 as baseline. Neural Net V1 outperforms Isolation Forest V1 on precision."},
], conversation_id="ml-review-weekly", extraction="heuristic")
print(f" {result.claims_ingested} claims from chat")
section("Slack #ml-experiments")
slack_zip = os.path.join(tmp, "ml_slack.zip")
with zipfile.ZipFile(slack_zip, "w") as zf:
zf.writestr("channels.json", json.dumps([{"id":"C1","name":"ml-experiments"}]))
zf.writestr("users.json", json.dumps([{"id":"U1","name":"ds","profile":{"real_name":"Data Scientist"}}]))
zf.writestr("ml-experiments/2024-03-01.json", json.dumps([
{"type":"message","user":"U1","text":"Trying ensemble for churn","ts":"1709300000.000"},
{"type":"message","bot_id":"B1","text":"Stacked Ensemble V1 was trained on Churn Dataset Q4. Stacked Ensemble V1 outperforms XGBoost V2 with 0.92 AUC. Stacked Ensemble V1 uses feature Contract Type. Stacked Ensemble V1 was derived from XGBoost V2.","ts":"1709300060.000"},
]))
slack_results = db.ingest_slack(slack_zip, extraction="heuristic")
print(f" {sum(r.claims_ingested for r in slack_results)} claims from Slack")
section("What was tried for churn prediction?")
churn_frame = db.query("Churn Dataset Q4", depth=2)
print(f" Models trained on Churn Dataset Q4:")
for rel in churn_frame.direct_relationships:
if rel.predicate == "trained_on":
print(f" {rel.target.name}")
section("Feature importance")
for feat in db.list_entities(entity_type="feature"):
claims = db.claims_for(feat.id, predicate_type="uses_feature")
if claims:
models = set(c.subject.display_name or c.subject.id for c in claims)
print(f" {feat.name}: {len(models)} model(s)")
section("Model lineage")
for c in db.claims_for("xgboost v2", predicate_type="derived_from"):
print(f" XGBoost V2 derived from {c.object.display_name or c.object.id}")
final = db.stats()
print(f"\n Final: {final['entity_count']} entities, {final['total_claims']} claims")
db.close()
print(f" Runtime: {time.perf_counter() - start:.2f}s")
return 0
if __name__ == "__main__":
sys.exit(main())
Nine new methods that answer questions only a provenance-tracking database can answer. Every example below works on any Attest database — no API keys, no external services.
What happens if a source turns out to be wrong?
# "If we retract paper_42, what breaks?" report = db.impact("paper_42") print(f"Direct claims: {report.direct_claims}") print(f"Downstream claims: {report.downstream_claims}") print(f"Affected entities: {report.affected_entities}")
Where is your knowledge vulnerable?
# Find entities backed by only a single source blindspots = db.blindspots(min_claims=5) print(f"Single-source entities: {blindspots.single_source_entities}") print(f"Low-confidence areas: {len(blindspots.low_confidence_areas)}")
Do multiple sources agree?
# "How much agreement is there about BRCA1?" report = db.consensus("BRCA1") print(f"Sources: {report.unique_sources}, Agreement: {report.agreement_ratio:.0%}") print(f"Claims by source: {report.claims_by_source}")
What's your weakest knowledge?
# Find claims backed by only one source fragile = db.fragile(max_sources=1) for c in fragile[:5]: print(f" {c.subject.id} —[{c.predicate.id}]→ {c.object.id} ({c.provenance.source_type})")
What hasn't been updated recently?
# Claims older than 90 days without corroboration stale = db.stale(days=90) print(f"{len(stale)} claims need refreshing")
Full provenance chain for any claim.
# "Where did this claim come from? What depends on it?" trail = db.audit(claim_id) print(f"Source: {trail.source_type}:{trail.source_id}") print(f"Corroborating claims: {len(trail.corroborating_claims)}") print(f"Downstream dependents: {trail.downstream_dependents}")
How is your knowledge changing over time?
# "What changed in the last 30 days?" drift = db.drift(days=30) print(f"New claims: {drift.new_claims}, New entities: {drift.new_entities}") print(f"Retracted: {drift.retracted_claims}") print(f"Confidence delta: {drift.confidence_delta:+.3f}")
Which sources can you trust?
# Per-source corroboration and retraction rates reliability = db.source_reliability() for source_id, metrics in reliability.items(): print(f" {source_id}: {metrics['corroboration_rate']:.0%} corroborated, " f"{metrics['retraction_rate']:.0%} retracted")
Would a hypothetical claim add value?
from attestdb import ClaimInput # "If someone claims Olaparib treats breast cancer, what happens?" hyp = ClaimInput( subject=("Olaparib", "compound"), predicate=("treats", "treats"), object=("Breast Cancer", "disease"), provenance={"source_type": "review", "source_id": "hypothetical"}, ) report = db.hypothetical(hyp) print(f"Would corroborate: {report.would_corroborate}") print(f"Fills gap: {report.fills_gap}") print(f"Related entities: {report.related_entities}")