Mari documentation

Install

Python 3.11–3.13 is supported. NumPy is the only runtime dependency.

pip install mari-components

# Optional runtime adapters
pip install 'mari-components[openai-agents]'
pip install 'mari-components[langchain]'

How it works

The base wheel installs the domain, connector, synchronization, retrieval, knowledge, trajectory, and verification modules without a model SDK or database client. Extras add adapter imports only; applications still inject model calls, HTTP transport, persistence, clocks, and authorization decisions. This is package behavior, not a research-derived algorithm.

Architecture

Values cross explicit boundaries. Mari plans and validates; the application performs side effects.

How it works

Provider data is normalized into immutable values. Pure functions transform those values into plans, candidates, reports, or index payloads. The caller validates the return value and commits it through its own transaction boundary. Because network and storage operations are injected, the same input values can be replayed in tests before any side effect occurs.

Documents, identity, and ACLs

KnowledgeDocument is the canonical provider-owned record. Its stable ID is {source_id}/{external_id}. Domain values are frozen dataclasses.

How it works

source_id names one configured source; external_id is the provider’s stable object key. Their pair prevents two providers from colliding. revision identifies content version, while updated_at is descriptive metadata and is never used as identity. ACL visibility and principals travel with the document so an allowed-ID set can be computed before retrieval scoring. Frozen values prevent an indexed object from changing behind its recorded revision.

from mari_components import DocumentACL, KnowledgeDocument, Principal

doc = KnowledgeDocument(
    source_id="github:acme/product",
    external_id="file:docs/refunds.md",
    title="Refund policy",
    body="## Enterprise\nRefunds close after 30 days.",
    revision="8f31c2a", updated_at="2026-08-31T10:00:00Z",
    source_url="https://github.com/acme/product/blob/main/docs/refunds.md",
    acl=DocumentACL(visibility="restricted", principals=(
        Principal(kind="team", identifier="support"),
    )),
    metadata={"path": "docs/refunds.md"},
)
assert doc.document_id == "github:acme/product/file:docs/refunds.md"

Polling and streaming connectors

Every connector defines a frozen configuration object, validation, and polling. GitHub, Slack, Google Drive, and Confluence also accept verified provider events. Network calls use an injected HttpTransport.

How it works

A polling connector starts from the caller’s cursor, requests bounded pages, normalizes provider objects, emits explicit tombstones, and returns the next cursor/checkpoint. A streaming connector verifies the raw delivery before parsing it, reduces provider payloads to bounded ChangeHint keys, coalesces duplicates, and canonically refetches the object. Both routes produce PollPage, so event order, partial webhook payloads, and provider retry behavior cannot bypass synchronization invariants.

Provider examples

All polling functions accept the same PollRequest and injected HttpTransport, and return an iterator of PollPage values.

from mari_components.connectors import GitHubConfig, poll_github
cfg = GitHubConfig(token=token, repository="acme/product",
    branch="main", paths=("docs/**",),
    content_types=("files", "issues", "pull_requests"))
pages = poll_github(cfg, request, http=http)

from mari_components.connectors import SlackConfig, poll_slack
cfg = SlackConfig(bot_token=bot_token,
    history_token=history_token, channels=("C0123",))
pages = poll_slack(cfg, request, http=http)

from mari_components.connectors import GoogleDriveConfig, poll_google_drive
cfg = GoogleDriveConfig(access_token=token, folder_id="folder-id")
pages = poll_google_drive(cfg, request, http=http)
# poll_google_drive_changes(...) and start_google_drive_watch(...)

from mari_components.connectors import ConfluenceConfig, poll_confluence
cfg = ConfluenceConfig(site_url="https://acme.atlassian.net/wiki",
    email="bot@acme.com", api_token=token, space_key="ENG")
pages = poll_confluence(cfg, request, http=http)

from mari_components.connectors import DropboxConfig, poll_dropbox
cfg = DropboxConfig(token=token, path="/Knowledge")
pages = poll_dropbox(cfg, request, http=http)

from mari_components.connectors import NotionConfig, poll_notion
cfg = NotionConfig(token=token)
pages = poll_notion(cfg, request, http=http)

from mari_components.connectors import AirtableConfig, poll_airtable
cfg = AirtableConfig(token=token, base_id="appABC123")
pages = poll_airtable(cfg, request, http=http)

from mari_components.connectors import AsanaConfig, poll_asana
cfg = AsanaConfig(token=token, workspace_gid="workspace-gid",
    project_gid="project-gid")
pages = poll_asana(cfg, request, http=http)

from mari_components.connectors import JiraConfig, poll_jira
cfg = JiraConfig(site_url="https://acme.atlassian.net",
    email="bot@acme.com", api_token=token, project_key="SUP")
pages = poll_jira(cfg, request, http=http)

from mari_components.connectors import LinearConfig, poll_linear
cfg = LinearConfig(api_key=api_key, team_id="team-id")
pages = poll_linear(cfg, request, http=http)

from mari_components.connectors import TrelloConfig, poll_trello
cfg = TrelloConfig(api_key=api_key, token=token)
pages = poll_trello(cfg, request, http=http)

from mari_components.connectors import ZendeskConfig, poll_zendesk
cfg = ZendeskConfig(subdomain="acme",
    email="bot@acme.com", api_token=token)
pages = poll_zendesk(cfg, request, http=http)

from mari_components import PollRequest
from mari_components.connectors import GitHubConfig, poll_github, validate_github

config = GitHubConfig(token=token, repository="acme/product",
    paths=("docs/**",), content_types=("files", "issues"))
validation = validate_github(config, http=http)
request = PollRequest(cursor=saved_cursor, page_size=100, page_limit=20)

for page in poll_github(config, request, http=http):
    consume(page)

Streaming

stream_pages requires a verifier, rejects oversized deliveries and batches, parses provider-specific hints, coalesces repeated aggregate keys, and calls an injected hydration function. The application owns the webhook server, queue, acknowledgement, and retries.

from mari_components.connectors import StreamEvent, stream_pages

event = StreamEvent(provider="slack", raw_body=raw_body, headers=headers)

def hydrate(hint):
    document, complete = fetch_slack_thread_by_id(config,
        hint.metadata["channel"], hint.metadata["thread_timestamp"], http=http)
    return (PollPage(upserts=(document,) if document else (),
        snapshot_complete=complete),)

for page in stream_pages((event,), verify=verify_signature, hydrate=hydrate):
    consume(page)

Connector-specific capabilities

• All twelve connectors: polling, validation, pagination limits, normalized documents, and explicit deletion handling.
• GitHub, Slack, Google Drive, and Confluence: verified streaming change hints plus canonical refetch.
• Slack: canonical thread fetch by ID.
• Google Drive: native Changes polling and push-watch registration.
• Confluence: direct canonical page fetch.
• ConnectorDefinition.supports(ConnectorMode.POLL | STREAM) exposes mode capabilities for setup UIs.

Synchronization

plan_sync compares a durable SyncState with one PollPage and returns a side-effect-free SyncPlan. stream_sync applies the same rules across pages.

How it works

For each upsert, Mari validates source ownership and compares a deterministic content fingerprint with the manifest: equal means unchanged; unequal means upsert. Explicit tombstones always become deletes. Absence becomes deletion only after the terminal page of an authoritative full snapshot. The returned plan carries the prior generation as a compare-and-swap precondition and the next manifest/cursor as proposed state; persistence must atomically commit both data and state.

from mari_components import SyncMode
from mari_components.sync import SyncState, plan_sync

state = load_state() or SyncState()
for page in provider_pages:
    plan = plan_sync(state, page,
        source_id="github:acme/product", mode=SyncMode.FULL)
    store.commit(upserts=plan.upserts, deletes=plan.deletes,
        state=plan.state, expected_generation=plan.expected_generation)
    state = plan.state

Enforced invariants

• Page replay is idempotent through content fingerprints and manifests.
• Only terminal, authoritative full pages reconcile absence.
• Explicit tombstones apply in full and incremental modes.
• Incomplete full sync cannot resume as incremental.
• Generation compare-and-swap prevents concurrent state loss.
• Foreign source IDs, duplicate IDs, and upsert/delete overlap are rejected.

Multi-vector retrieval

Mari implements MUVERA fixed-dimensional candidate generation, PolarQuant compression, and exact normalized MaxSim reranking in one retrieval path.

from mari_components.retrieval import FDEConfig, build_index, search_index

index = build_index({doc.document_id: token_vectors},
    config=FDEConfig(repetitions=20, projection_dimension=16))
hits = search_index(index, query_token_vectors, limit=8,
    allowed_document_ids=authorized_document_ids)

serialize_index and deserialize_index use versioned, checksummed payloads. exact_maxsim is public for direct scoring.

How it works and backing algorithms

Mari's current path uses token-level late interaction: each query token takes its maximum similarity to any document token, and the maxima are summed. MUVERA maps those multi-vector sets to fixed-dimensional encodings for fast candidate generation; Mari then reranks the candidates with exact MaxSim. The packed Polar codec is an implementation-level compression of candidate encodings, not an alternative relevance model.

Proposed index interface

The common protocol should describe capabilities rather than a vendor. Index selection then becomes pipeline configuration and can be evaluated against recall, latency, memory, freshness, and ACL-filter behavior.

indexes = {
    "exact": DenseFlatIndex(metric="cosine"),
    "dense": HNSWIndex(metric="cosine", m=32, ef_search=128),
    "compressed": IVFPQIndex(partitions=4096, subquantizers=32),
    "lexical": BM25Index(k1=1.2, b=0.75),
    "sparse": SparseVectorIndex(model="splade"),
    "late": LateInteractionIndex(candidate="muvera", rerank="maxsim"),
}

hybrid = HybridIndex(arms=[indexes["lexical"], indexes["dense"], indexes["late"]],
    fusion=ReciprocalRankFusion(k=60))

Rank fusion, graph recall, and diverse packing

from mari_components.retrieval import (
    maximal_marginal_relevance, personalized_pagerank,
    project_graph_scores, reciprocal_rank_fusion,
)

fused = reciprocal_rank_fusion(
    {"muvera": dense_ids, "lexical": lexical_ids, "recent": recent_ids},
    weights={"recent": 0.25}, rank_constant=60,
    eligible=authorized_document_ids.__contains__, limit=40)

nodes = personalized_pagerank(graph, query_seeds,
    allowed_node_ids=authorized_graph_nodes, damping=0.85)
passages = project_graph_scores(nodes.hits, node_passages, limit=20)

context = maximal_marginal_relevance(
    {hit.document_id: hit.score for hit in fused},
    similarity=passage_similarity, relevance_weight=0.65, limit=12)
assert nodes.converged

Memory segmentation and mutation plans

hybrid_topic_segments splits a stream only where an attention-boundary peak and a semantic-similarity valley agree. The application extracts candidates from those bounded groups and classifies each one as add, update, delete, or no-op. plan_memory_mutations validates the decisions without writing storage.

How it works

Normalize boundary and adjacent-similarity arrays to the n−1 gaps between n turns. A gap is eligible only when its attention score is a local peak above the configured boundary threshold and its adjacent semantic similarity is below the valley threshold. Eligible gaps split consecutive, non-overlapping segments. Mutation planning then requires exactly one decision per candidate, validates update/delete targets against current IDs, rejects duplicate adds and conflicting operations on one target, and returns a deterministic plan.

from mari_components.knowledge import (
    MemoryDecision, MemoryOperation, hybrid_topic_segments,
    plan_memory_mutations,
)

segments = hybrid_topic_segments(turns,
    attention_boundaries=attention, adjacent_similarities=similarity,
    similarity_threshold=0.40)

plan = plan_memory_mutations(existing, candidates, {
    "new-role": MemoryDecision(operation=MemoryOperation.UPDATE,
        target_id="role", reason="newer explicit statement"),
    "unchanged": MemoryDecision(operation=MemoryOperation.NOOP),
})
store.commit(plan, expected_generation=generation)

Knowledge parsers

Models return JSON-like values. Parsers resolve all evidence against supplied document and section revisions and return immutable typed values. Research establishes each task formulation; Mari implements a deterministic validation boundary rather than the cited model.

How it works

Each parser first requires the recipe’s top-level collection, then validates every required field and enum, resolves evidence through the exact contract below, derives deterministic audit signals, and constructs frozen result types. It never repairs a claim’s meaning. Batch claim assessment is the exception to fail-fast parsing: rows are keyed back to caller order, absent rows become uncertain, and individually malformed rows do not erase valid siblings.

from mari_components.knowledge import parse_answer

raw = model(question, documents)
answer = parse_answer(question, documents, raw)
print(answer.disposition)         # grounded | insufficient_evidence
print(answer.grounding_coverage)  # deterministic text coverage
print(answer.evidence[0].quote)  # exact source text

Additional deterministic helpers include normalize_claim, deduplicate_fact_candidates, grounding_coverage, and excerpt. Recoverable batch drift is handled conservatively: assessment rows are restored to caller order, missing rows become uncertain, and good rows survive alongside invalid ones. Structured fact qualifiers preserve subject, relation, object, scope, validity, and conditions.

Evidence contracts

An evidence record is a byte-for-byte quotation bound to the exact document and section revision that was supplied to a parser. It is provenance, not a model confidence score.

How it works

1. Restrict the corpus. Build an allowed map from only the KnowledgeDocument values passed by the caller. A model cannot cite an ID outside that map.
2. Resolve the document. Require document_id. If exactly one allowed document contains the quote, Mari may recover a missing ID; zero or multiple holders is rejected.
3. Match exact text. Require a non-empty quote and test literal containment in the canonical document body. Fuzzy, normalized, or semantic matches are not accepted.
4. Resolve one section. Split the document into current sections and find sections containing the quote. A repeated quote spanning multiple sections is rejected unless section_id selects exactly one.
5. Derive coordinates. Mari computes start = section.start + section.body.index(quote) and end = start + len(quote); it does not trust model-supplied offsets or revisions.
6. Bind revisions. The accepted record receives the current document.revision, stable section_id, and content-derived section.revision. These become the invalidation key.

from mari_components.knowledge import parse_facts

raw = {"facts": [{
    "claim": "Enterprise refunds close after 30 days.",
    "evidence": [{"document_id": doc.document_id,
                  "section_id": "enterprise",
                  "quote": "30 days"}],
}]}
fact = parse_facts([doc], raw)[0]
e = fact.evidence[0]
assert doc.body[e.start:e.end] == e.quote

# Rejected: unknown document, absent quote, or a repeated quote
# whose section cannot be selected unambiguously.

Dependency conversion

evidence_dependencies projects each record to (document_id, document_revision, section_id, section_revision), deduplicated by (document_id, section_id) and returned in stable order. Two records naming different revisions of the same key raise ValueError; silently choosing one would make reuse nondeterministic.

Freshness and impact

Freshness is an exact dependency comparison. It answers “did an input revision change?”—not “is the answer still semantically correct?”

How it works

1. Record dependencies. Every derived artifact stores the document or section revision used to build it.
2. Select comparison granularity. If a dependency names a section and the caller supplies a section-revision map, compare section hashes. Otherwise compare the containing document revision as a conservative fallback.
3. Classify every key. Missing document/section → missing; empty expected/current revision → unversioned; unequal revisions → stale; otherwise → current.
4. Reduce deterministically. Overall precedence is missing > unversioned > stale > current. Changes and IDs are sorted, so the same inputs produce the same report.
5. Propagate impact. impacted_artifacts evaluates each artifact independently and returns only non-current artifacts. Mari reports the set; the application chooses whether to regenerate, review, or retire them.

from mari_components.knowledge import (
    FreshnessStatus, assess_dependencies, assess_freshness,
    impacted_artifacts,
)

report = assess_freshness(answer.evidence, current_revisions,
    current_section_revisions=current_sections)
if not report.reusable:
    refresh(report.changes, report.missing_dependency_ids)

stale = impacted_artifacts(dependencies_by_artifact, current_revisions,
    current_section_revisions=current_sections)

Document edit versus affected section

# The document changed v1 → v2, but the cited section is still s1.
current_revisions = {doc_id: "v2"}
current_sections = {(doc_id, "refund-window"): "s1"}

fine = assess_dependencies(deps, current_revisions,
    current_section_revisions=current_sections)
assert fine.status == FreshnessStatus.CURRENT

coarse = assess_dependencies(deps, current_revisions)
assert coarse.status == FreshnessStatus.STALE  # safe fallback

Sections and incremental fact scans

document_sections maps Markdown headings to stable section IDs and content revisions. section_revisions builds the current revision map. Fact scans can then skip unchanged sections.

How it works

Scan Markdown heading lines, treating content before the first heading as a preamble. Normalize each heading into a slug and suffix collisions deterministically. Store absolute body offsets and hash the section body into its revision. pending_fact_sections compares (document_id, section_id) → revision with the last committed scan and yields new or changed sections only. Persist new scan revisions only after extracted facts commit, or a failed run would incorrectly suppress retry.

from mari_components.knowledge import (
    document_sections, fact_scan_revisions, pending_fact_sections,
)

sections = document_sections(document)
pending = pending_fact_sections([document], previous_scan_revisions)
facts = [parse_facts([document], model(section.body)) for section in pending]
next_revisions = fact_scan_revisions(pending)  # persist only after facts commit

Tags and links

How it works

Tag keys are normalized before add/remove set operations; definitions validate that assignments refer to known tags. Search weight combines assigned tag weights through deterministic policy rather than rewriting source relevance. Explicit-link extraction recognizes source references first. Similarity linking scores only caller-supplied candidate IDs, removes self-links, applies a threshold and limit, and sorts ties stably. Returned links are proposals; committing and interpreting them remains application policy.

from mari_components.knowledge import (
    TagAssignments, TagDefinition, assign_tags, derive_links, search_weight,
)

definitions = {"canonical": TagDefinition(key="canonical", label="Canonical",
    kind="canonical", search_weight=2.0, behaviors=("Wins conflicts",))}
assignments = assign_tags(TagAssignments(), doc.document_id, definitions,
    add=("canonical",))
weight = search_weight(doc.document_id, assignments, definitions)
links = derive_links(doc.document_id, candidate_ids,
    score=lambda source, target: similarity_scores[source, target])

Reviewed workflows and cached answers

Reviewed workflow indexes match new requests to approved intents. Policy thresholds independently control speculative retrieval and direct cached-response reuse.

How it works

Build an index from reviewed workflow intent vectors. At query time, compute normalized similarity, retain only workflows whose dependencies are authorized, and choose the best match with stable ties. Crossing the lower threshold may start retrieval speculatively; crossing the higher threshold only makes reuse eligible. A cached response is returned only after its exact evidence dependencies pass freshness checks. Similarity never overrides ACL or revision failure.

from mari_components.trajectories import (
    WorkflowPolicy, build_reviewed_workflow_index, decide_reviewed_workflow,
)

index = build_reviewed_workflow_index(reviewed_workflows)
decision = decide_reviewed_workflow(query_vectors, index, current_revisions,
    policy=WorkflowPolicy(speculation_threshold=0.72, cache_threshold=0.95),
    allowed_document_ids=authorized_document_ids)

Related APIs: match_reviewed_workflow, start_speculative_retrieval, match_cached_response, and workflow_freshness. Reuse requires a strong match plus fresh, authorized dependencies.

Trajectories and agent evaluation

normalize_steps converts runtime records into privacy-bounded TrajectoryStep values. parse_trajectory_analysis validates model-proposed phases. Mari provides adapters, not an agent loop.

How it works

Adapters map framework events into ordered AgentEvent values. Normalization assigns stable step positions, keeps allowlisted metadata, and redacts sensitive argument names and transport fields. Tool evaluation compares observed names and counts against expectations; outcome evaluation compares terminal paths and completion. A proposed phase analysis must cover every observed event exactly once with contiguous, non-overlapping ranges and known tool families.

from mari_components.agents import evaluate_outcome, evaluate_tools
from mari_components.trajectories import normalize_steps

steps = normalize_steps(runtime_events)
tools = evaluate_tools(events, expected_tools=("search_knowledge",))
outcome = evaluate_outcome(paths=("resolved",),
    expected_paths=("resolved",), completed=True)

AgentEvent and EventKind are framework-neutral. Optional adapters cover OpenAI Agents and LangChain/LangGraph. Normalization redacts common sensitive arguments; phase validation requires the returned ranges to cover observed events exactly.

from mari_components.trajectories import parse_trajectory_analysis

analysis = parse_trajectory_analysis(normalized_events, model_labels,
    family_map={"search_product_knowledge": "inspect",
                "answer": "answer"})

Verification portfolios

Verification functions score already-parsed values and retain all successful attempts and failures for audit.

How it works

best_of_n calls the producer up to n times, parses each output, records parse failures without discarding successful siblings, scores valid candidates, and selects the highest score with stable first-wins ties; it may stop once the threshold is met. verdict_consensus counts typed verdicts rather than free text. Grounding scores combine declared deterministic components; they are used for ranking or abstention, never calibrated as probabilities.

from mari_components.verification import best_of_n, score_grounded

result = best_of_n(
    lambda: model(question, documents),
    lambda raw: parse_answer(question, documents, raw),
    lambda answer: score_grounded(answer,
        required_ideas=("eligibility", "time limit")),
    attempts=3, threshold=0.90)

audit(result.selected, result.attempts, result.failures, result.stopped_early)

Errors and deliberate boundaries

How it works

Exceptions classify which boundary failed and whether repeating the same request can help. Connector adapters translate provider responses into authentication, transient, or permanent failures. Snapshot validation raises IncompleteSnapshot before absence can become deletion. Knowledge parsers raise MalformedModelOutput before an invalid value crosses into typed state. Mari never retries automatically because retry budgets, clocks, credentials, and side effects belong to the host.

Safe representations and connector contracts

Credentials are excluded from connector configuration representations. HttpRequest representations redact authorization headers, bodies, URL userinfo, and common sensitive query parameters. check_connector_contract provides an executable contract check for third-party connector implementations.

from mari_components import SyncMode
from mari_components.testing import check_connector_contract

pages = tuple(my_connector(config, request, http=fake_http))
report = check_connector_contract(pages, mode=SyncMode.FULL,
    starting_cursor=request.cursor)
assert report.pages == len(pages)

Not included: model client, prompt framework, database, scheduler, credential store, authorization engine, agent runtime, or worker queue.