Examples
Runnable recipes. All examples mirror the repo's test suite (tests/test_e2e.py et al.).
Contradiction lowers confidence — and stays visible
kb.endorse(claim_id, agent_b, "contradict", basis="cold deploys fine now")
disputed = kb.brief(scope, task="deploying pricing").items[0]
assert disputed.confidence < base.confidence
assert disputed.contradictions == 1
assert claim_id in kb.brief(scope, task="deploying pricing").conflicts
The claim is not removed — it ranks lower, carries its contradiction count, and appears
in briefing.conflicts so a human can see the dispute.
Corroboration resets staleness and outvotes
kb.endorse(claim_id, agent_b, "corroborate", basis="reproduced today")
kb.endorse(claim_id, agent_c, "corroborate", basis="confirmed in staging")
item = kb.brief(scope, task="deploying pricing").items[0]
# two distinct corroborators: confidence up, staleness reset
An identity's most recent endorsement is its position — corroborating twice from the same key does not stack.
Provenance-linked invalidation
from datetime import datetime, timezone
# the artifact a claim references has changed
kb.invalidate(scope, "file://pricing-model.md", datetime.now(timezone.utc))
item = kb.brief(scope, task="pricing").items[0]
assert item.suspect # flips the moment the artifact changes
assert item.verify_first # briefing tells the consumer to re-check
# a later corroboration clears suspect
kb.endorse(item.claim.claim_id, agent_b, "corroborate", basis="still holds post-change")
Governed promotion with protected topics
kb.set_policy(scope, Policy(mode="auto", protected_topics=("security",)))
# ordinary claims land immediately
p1 = kb.propose(scope, agent_a, "semantic", "CI runs on every push")
assert p1.status == "accepted"
# protected topics queue for human review even in auto mode
p2 = kb.propose(scope, agent_a, "procedural",
"rotate credentials weekly", topics=("security",))
assert p2.status == "pending"
p2 = kb.review(p2.id, human_keys, "approve", basis="matches security policy")
Trust weighting (the SourceWeight seam)
from fg_agent_knowledge import confidence
from fg_agent_knowledge.scoring import weighted_support
weight = {"alice": 3.0, "bob": 0.0} # address -> weight
corr, contra = weighted_support(ends, lambda a: weight[a])
assert (corr, contra) == (3.0, 0.0)
assert confidence(corr, contra) > confidence(2.0, 0.0)
Pass the same callable to kb.brief(..., source_weight=...) to apply it to ranking.
Revision and retirement
# revision: a new claim that names its predecessor (author-only)
p = kb.propose(scope, agent_a, "semantic",
"staging DB is MySQL", supersedes=old_claim_id)
# the superseded claim disappears from briefings; history remains queryable
# retirement: the only exit from briefings (author-only)
kb.retire(claim_id, agent_a, reason="no longer applicable")
assert kb.claim(claim_id) is not None # the record is never deleted
A custom retriever (bring your own embeddings)
from fg_agent_knowledge import RetrievalQuery, Ranking
from fg_agent_knowledge.retrievers import BaseRetriever
class EmbeddingRetriever(BaseRetriever):
def rank(self, query: RetrievalQuery, claims):
scores = {c.claim_id: cosine(embed(query.task), embed(c.body.statement))
for c in claims}
return Ranking(scores=scores)
kb = KnowledgeBase(SQLiteStore("team.db"), retriever=EmbeddingRetriever())
The core still owns trust and assembly — your retriever only supplies relevance.
An LLM consolidator (as a consumer, never part of the protocol)
The pattern from examples/consolidator.md:
- Read recent episodes from your capture feed (the store exposes episodes by id only — keep your own list of what you captured).
- Distill them with your model of choice — the LLM lives here, outside the standard.
kb.propose(...)the distilled claim, citing the episode ids as pedigree.- On re-encountering a known claim,
kb.endorse(..., "corroborate")instead of re-proposing.
Content addressing makes naive re-derivation harmless — proposing an identical body is idempotent — and scope policy protects what needs protecting.
Things that fail (by design)
kb.review(promo.id, proposer_keys, "approve") # PolicyError — self-review
kb.propose(scope_b, keys, "semantic", "...",
episodes=(episode_from_scope_a,)) # ScopeError — cross-space pedigree
kb.endorse(claim_id, keys, "corroborate", basis=0.9) # ValidationError — basis must be a string