docs / agent-knowledge / examples

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:

  1. Read recent episodes from your capture feed (the store exposes episodes by id only — keep your own list of what you captured).
  2. Distill them with your model of choice — the LLM lives here, outside the standard.
  3. kb.propose(...) the distilled claim, citing the episode ids as pedigree.
  4. 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