primus.rubric_memory package

class primus.rubric_memory.BiblicusRubricEvidenceRetriever(corpus_root: str | Path | None = None, *, corpus_sources: Sequence[LocalRubricMemorySource | S3RubricMemorySource] | None = None, retriever_id: str = 'scan', max_total_items: int = 16, maximum_total_characters: int = 60000, source_window_characters: int = 6000, query_planner: RubricMemoryQueryPlanner | None = None, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None)

Bases: object

Retrieve rubric-memory evidence from one prepared Biblicus corpus.

__init__(corpus_root: str | Path | None = None, *, corpus_sources: Sequence[LocalRubricMemorySource | S3RubricMemorySource] | None = None, retriever_id: str = 'scan', max_total_items: int = 16, maximum_total_characters: int = 60000, source_window_characters: int = 6000, query_planner: RubricMemoryQueryPlanner | None = None, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None)

classmethod from_local_score(*, scorecard_name: str, score_name: str, retriever_id: str = 'scan', max_total_items: int = 16, maximum_total_characters: int = 60000, source_window_characters: int = 6000, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None) → BiblicusRubricEvidenceRetriever

classmethod from_score(*, scorecard_name: str, score_name: str, retriever_id: str = 'scan', max_total_items: int = 16, maximum_total_characters: int = 60000, source_window_characters: int = 6000, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None, s3_client: Any | None = None) → BiblicusRubricEvidenceRetriever

async retrieve(request: RubricEvidencePackRequest) → Sequence[EvidenceSnippet]

class primus.rubric_memory.ConfidenceInputs(*, score_version_id: Annotated[str, MinLen(min_length=1)], total_evidence_count: Annotated[int, Ge(ge=0)], score_scope_evidence_count: Annotated[int, Ge(ge=0)], prefix_scope_evidence_count: Annotated[int, Ge(ge=0)] = 0, scorecard_scope_evidence_count: Annotated[int, Ge(ge=0)], unknown_scope_evidence_count: Annotated[int, Ge(ge=0)], high_authority_evidence_count: Annotated[int, Ge(ge=0)], low_authority_evidence_count: Annotated[int, Ge(ge=0)], conflicting_or_stale_evidence_count: Annotated[int, Ge(ge=0)], chronological_evidence_count: Annotated[int, Ge(ge=0)], suggested_confidence: ConfidenceLevel)

Bases: BaseModel

Deterministic inputs used by Python to constrain confidence.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

chronological_evidence_count: int

conflicting_or_stale_evidence_count: int

high_authority_evidence_count: int

low_authority_evidence_count: int

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prefix_scope_evidence_count: int

score_scope_evidence_count: int

score_version_id: str

scorecard_scope_evidence_count: int

suggested_confidence: ConfidenceLevel

total_evidence_count: int

unknown_scope_evidence_count: int

class primus.rubric_memory.ConfidenceLevel(value)

Bases: str, Enum

HIGH = 'high'

LOW = 'low'

MEDIUM = 'medium'

class primus.rubric_memory.EvidenceClassification(value)

Bases: str, Enum

HISTORICAL_CONTEXT = 'historical_context'

POSSIBLE_STALE_RUBRIC = 'possible_stale_rubric'

RUBRIC_CONFLICTING = 'rubric_conflicting'

RUBRIC_GAP = 'rubric_gap'

RUBRIC_SUPPORTED = 'rubric_supported'

class primus.rubric_memory.EvidenceSnippet(*, snippet_text: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], source_uri: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], scope_level: str = 'unknown', source_type: str = 'unknown', authority_level: str = 'unknown', source_timestamp: ~datetime.datetime | None = None, author: str | None = None, retrieval_score: float = 0.0, policy_concepts: list[str] = <factory>, evidence_classification: ~primus.rubric_memory.models.EvidenceClassification = EvidenceClassification.HISTORICAL_CONTEXT)

Bases: BaseModel

Corpus evidence with provenance retained from Biblicus retrieval.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

author: str | None

authority_level: str

evidence_classification: EvidenceClassification

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

policy_concepts: list[str]

retrieval_score: float

scope_level: str

snippet_text: str

source_timestamp: datetime | None

source_type: str

source_uri: str

class primus.rubric_memory.LocalRubricMemoryCorpusPaths(scorecard_root: Path, scorecard_knowledge_base: Path, prefix_knowledge_bases: list[Path], score_knowledge_base: Path)

Bases: object

Convention-derived local rubric-memory paths for one score.

__init__(scorecard_root: Path, scorecard_knowledge_base: Path, prefix_knowledge_bases: list[Path], score_knowledge_base: Path) → None

prefix_knowledge_bases: list[Path]

score_knowledge_base: Path

scorecard_knowledge_base: Path

scorecard_root: Path

property sources: list[LocalRubricMemorySource]

class primus.rubric_memory.LocalRubricMemoryCorpusResolver

Bases: object

Resolve rubric-memory folders using the existing pulled-score convention.

resolve(*, scorecard_name: str, score_name: str) → LocalRubricMemoryCorpusPaths

class primus.rubric_memory.LocalRubricMemorySource(root: Path, scope_level: str)

Bases: object

A local rubric-memory corpus folder with its score hierarchy scope.

__init__(root: Path, scope_level: str) → None

root: Path

scope_level: str

class primus.rubric_memory.PreparedRubricMemoryCorpus(corpus_root: Path, prepared_root: Path, manifest_path: Path, fingerprint: str, retriever_id: str, status: str, source_file_count: int, sources: list[dict[str, Any]])

Bases: object

A prepared Biblicus corpus built from rubric-memory sources.

__init__(corpus_root: Path, prepared_root: Path, manifest_path: Path, fingerprint: str, retriever_id: str, status: str, source_file_count: int, sources: list[dict[str, Any]]) → None

corpus_root: Path

fingerprint: str

manifest_path: Path

prepared_root: Path

retriever_id: str

source_file_count: int

sources: list[dict[str, Any]]

status: str

class primus.rubric_memory.RubricAuthority(*, score_version_id: Annotated[str, MinLen(min_length=1)], rubric_text: Annotated[str, MinLen(min_length=1)], score_code: str = '')

Bases: BaseModel

Storage-boundary projection of ScoreVersion authority into rubric terms.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rubric_text: str

score_code: str

score_version_id: str

exception primus.rubric_memory.RubricAuthorityError

Bases: RuntimeError

Raised when official ScoreVersion rubric authority cannot be resolved.

class primus.rubric_memory.RubricAuthorityResolver(api_client: Any)

Bases: object

Resolve the canonical rubric and score code from the champion ScoreVersion.

Primus storage still names the rubric field guidelines. This class is the storage adapter boundary: callers receive rubric_text and do not need to know about the legacy field name.

__init__(api_client: Any)

async resolve(score_id: str) → RubricAuthority

async resolve_score_version(score_version_id: str) → RubricAuthority

class primus.rubric_memory.RubricEvidencePack(*, score_version_id: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], rubric_reading: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], evidence_classification: ~primus.rubric_memory.models.EvidenceClassification, supporting_evidence: list[~primus.rubric_memory.models.EvidenceSnippet] = <factory>, conflicting_evidence: list[~primus.rubric_memory.models.EvidenceSnippet] = <factory>, history_of_change: list[~primus.rubric_memory.models.RubricHistoryEvent] = <factory>, likely_reason_for_disagreement: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], confidence: ~primus.rubric_memory.models.ConfidenceLevel, confidence_inputs: ~primus.rubric_memory.models.ConfidenceInputs, open_questions: list[str] = <factory>)

Bases: BaseModel

Structured answer for one disputed classification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

confidence: ConfidenceLevel

confidence_inputs: ConfidenceInputs

conflicting_evidence: list[EvidenceSnippet]

evidence_classification: EvidenceClassification

history_of_change: list[RubricHistoryEvent]

likely_reason_for_disagreement: str

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

open_questions: list[str]

rubric_reading: str

score_version_id: str

supporting_evidence: list[EvidenceSnippet]

class primus.rubric_memory.RubricEvidencePackContextFormatter(*, max_snippet_characters: int = 1400)

Bases: object

Render a rubric evidence pack as deterministic sub-agent context.

__init__(*, max_snippet_characters: int = 1400)

format(pack: RubricEvidencePack) → str

class primus.rubric_memory.RubricEvidencePackRequest(*, scorecard_identifier: Annotated[str, MinLen(min_length=1)], score_identifier: Annotated[str, MinLen(min_length=1)], score_version_id: Annotated[str, MinLen(min_length=1)], rubric_text: Annotated[str, MinLen(min_length=1)], score_code: str = '', transcript_text: str = '', model_value: str = '', model_explanation: str = '', feedback_value: str = '', feedback_comment: str = '', topic_hint: str | None = None)

Bases: BaseModel

Inputs needed to explain one disputed classification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

feedback_comment: str

feedback_value: str

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_explanation: str

model_value: str

rubric_text: str

score_code: str

score_identifier: str

score_version_id: str

scorecard_identifier: str

topic_hint: str | None

transcript_text: str

class primus.rubric_memory.RubricEvidencePackService(*, retriever: RubricEvidenceRetriever, synthesizer: RubricEvidenceSynthesizer)

Bases: object

Generate rubric evidence packs from official rubric authority and corpus evidence.

Python owns retrieval, provenance, dedupe, chronology, confidence policy, and final response shaping. The synthesizer is responsible only for interpreting the shaped evidence into prose and structured labels.

__init__(*, retriever: RubricEvidenceRetriever, synthesizer: RubricEvidenceSynthesizer)

async generate(request: RubricEvidencePackRequest) → RubricEvidencePack

class primus.rubric_memory.RubricEvidenceRetriever(*args, **kwargs)

Bases: Protocol

__init__(*args, **kwargs)

async retrieve(request: RubricEvidencePackRequest) → Sequence[EvidenceSnippet]: Return candidate evidence snippets for the disputed item.

class primus.rubric_memory.RubricEvidenceSynthesizer(*args, **kwargs)

Bases: Protocol

__init__(*args, **kwargs)

async synthesize(*, request: RubricEvidencePackRequest, evidence: Sequence[EvidenceSnippet], history: Sequence[RubricHistoryEvent], confidence_inputs: ConfidenceInputs) → RubricEvidencePack: Interpret shaped evidence and return a structured pack.

class primus.rubric_memory.RubricHistoryEvent(*, source_timestamp: datetime | None = None, source_uri: Annotated[str, MinLen(min_length=1)], scope_level: str = 'unknown', authority_level: str = 'unknown', summary: Annotated[str, MinLen(min_length=1)], evidence_classification: EvidenceClassification = EvidenceClassification.HISTORICAL_CONTEXT)

Bases: BaseModel

Chronological policy-memory event derived from corpus evidence.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

authority_level: str

evidence_classification: EvidenceClassification

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scope_level: str

source_timestamp: datetime | None

source_uri: str

summary: str

class primus.rubric_memory.RubricMemoryCitation(*, id: Annotated[str, MinLen(min_length=1)], kind: Literal['official_rubric', 'corpus_evidence'], excerpt: Annotated[str, MinLen(min_length=1)], source_uri: str | None = None, scope_level: str = 'unknown', source_timestamp: datetime | None = None, authority_level: str = 'unknown', score_version_id: Annotated[str, MinLen(min_length=1)], evidence_classification: str = 'unknown')

Bases: BaseModel

Stable citation handle for official rubric authority or corpus evidence.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

authority_level: str

evidence_classification: str

excerpt: str

id: str

kind: Literal['official_rubric', 'corpus_evidence']

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

scope_level: str

score_version_id: str

source_timestamp: datetime | None

source_uri: str | None

class primus.rubric_memory.RubricMemoryCitationContext(*, markdown_context: str, citation_index: list[RubricMemoryCitation] = <factory>, machine_context: dict[str, ~typing.Any]=<factory>, diagnostics: list[dict[str, ~typing.Any]]=<factory>)

Bases: BaseModel

Human-readable rubric-memory context plus machine-readable citations.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

citation_ids() → set[str]

citation_index: list[RubricMemoryCitation]

diagnostics: list[dict[str, Any]]

machine_context: dict[str, Any]

markdown_context: str

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class primus.rubric_memory.RubricMemoryCitationFormatter(*, max_excerpt_characters: int = 900)

Bases: object

Convert a RubricEvidencePack into deterministic citation context.

__init__(*, max_excerpt_characters: int = 900)

from_pack(pack: RubricEvidencePack) → RubricMemoryCitationContext

from_recent_evidence(*, request: RubricEvidencePackRequest, evidence: Sequence[EvidenceSnippet], metadata: dict[str, Any]) → RubricMemoryCitationContext: Format recency-biased retrieved evidence for optimizer briefings.

from_retrieved_evidence(*, request: RubricEvidencePackRequest, evidence: Sequence[EvidenceSnippet]) → RubricMemoryCitationContext: Format retrieved evidence as LLM input context without synthesis.

class primus.rubric_memory.RubricMemoryCitationValidation(*, supplied_ids: list[str] = <factory>, valid_ids: list[str] = <factory>, missing_ids: list[str] = <factory>, unused_ids: list[str] = <factory>, omitted_citations: bool = False, warnings: list[str] = <factory>)

Bases: BaseModel

Non-blocking diagnostics for citation use by an LLM consumer.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

missing_ids: list[str]

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

omitted_citations: bool

supplied_ids: list[str]

unused_ids: list[str]

valid_ids: list[str]

warnings: list[str]

class primus.rubric_memory.RubricMemoryContextProvider(*, api_client: Any, citation_formatter: RubricMemoryCitationFormatter | None = None, model_attempt_authority: Any = None)

Bases: object

Shared service that produces citation-ready rubric-memory context.

__init__(*, api_client: Any, citation_formatter: RubricMemoryCitationFormatter | None = None, model_attempt_authority: Any = None)

async generate_for_request(request: RubricEvidencePackRequest) → RubricMemoryCitationContext

async generate_for_score_item(*, scorecard_identifier: str, score_identifier: str, score_id: str, score_version_id: str | None = None, transcript_text: str = '', model_value: str = '', model_explanation: str = '', feedback_value: str = '', feedback_comment: str = '', topic_hint: str | None = None) → RubricMemoryCitationContext

local_corpus_status(*, scorecard_identifier: str, score_identifier: str) → dict[str, Any]

async retrieve_for_score_item(*, scorecard_identifier: str, score_identifier: str, score_id: str, score_version_id: str | None = None, transcript_text: str = '', model_value: str = '', model_explanation: str = '', feedback_value: str = '', feedback_comment: str = '', topic_hint: str | None = None) → RubricMemoryCitationContext

async retrieve_for_score_items(*, scorecard_identifier: str, score_identifier: str, score_id: str, item_contexts: Sequence[dict[str, str]], score_version_id: str | None = None, topic_hint: str | None = None) → dict[str, RubricMemoryCitationContext]: Retrieve citation context for existing LLM consumers without synthesis.

class primus.rubric_memory.RubricMemoryGatedSMEQuestion(*, id: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], original_text: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], final_text: str = '', action: ~primus.rubric_memory.sme_question_gate.SMEQuestionGateAction, answer_status: ~primus.rubric_memory.sme_question_gate.SMEQuestionAnswerStatus, rationale: str = '', citation_ids: list[str] = <factory>, citation_validation: ~primus.rubric_memory.citations.RubricMemoryCitationValidation | None = None)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

action: SMEQuestionGateAction

answer_status: SMEQuestionAnswerStatus

citation_ids: list[str]

citation_validation: RubricMemoryCitationValidation | None

final_text: str

id: str

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

original_text: str

rationale: str

class primus.rubric_memory.RubricMemoryPreparedCorpusManager(cache_root: str | Path | None = None)

Bases: object

Prepare rubric-memory sources into a reusable Biblicus corpus.

SIDECAR_SCHEMA_VERSION = 'rubric-memory-sidecar-v1'

__init__(cache_root: str | Path | None = None)

infer_source_timestamp(relative_path: Path | PurePosixPath) → datetime | None

prepare(*, corpus_sources: Sequence[LocalRubricMemorySource | S3RubricMemorySource], retriever_id: str = 'scan', force: bool = False) → PreparedRubricMemoryCorpus

class primus.rubric_memory.RubricMemoryQueryPlan(expanded_query_text: 'str', retrieval_phrases: 'list[str]', important_tokens: 'list[str]')

Bases: object

__init__(expanded_query_text: str, retrieval_phrases: list[str], important_tokens: list[str]) → None

expanded_query_text: str

important_tokens: list[str]

retrieval_phrases: list[str]

class primus.rubric_memory.RubricMemoryQueryPlanner(*, max_phrases: int = 80)

Bases: object

Build runtime retrieval hints from existing rubric/item context.

__init__(*, max_phrases: int = 80)

plan(request: RubricEvidencePackRequest) → RubricMemoryQueryPlan

class primus.rubric_memory.RubricMemoryRecentBriefingProvider(*, api_client: Any, citation_formatter: RubricMemoryCitationFormatter | None = None, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None, s3_client: Any | None = None, reference_date: date | None = None)

Bases: object

Build recency-biased rubric-memory citation context for one score.

DEFAULT_DAYS = 30

DEFAULT_QUERY = 'recent SME stakeholder policy update rubric guideline change clarification score scorecard scoring decision'

__init__(*, api_client: Any, citation_formatter: RubricMemoryCitationFormatter | None = None, prepared_corpus_manager: RubricMemoryPreparedCorpusManager | None = None, s3_client: Any | None = None, reference_date: date | None = None)

async retrieve_recent(*, scorecard_identifier: str, score_identifier: str, score_id: str, score_version_id: str | None = None, query: str = '', days: int = 30, since: date | str | None = None, limit: int = 16) → RubricMemoryCitationContext

class primus.rubric_memory.RubricMemorySMEQuestion(*, id: Annotated[str, MinLen(min_length=1)], text: Annotated[str, MinLen(min_length=1)])

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

id: str

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

text: str

class primus.rubric_memory.RubricMemorySMEQuestionGateRequest(*, scorecard_identifier: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], score_identifier: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], score_version_id: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], rubric_memory_context: ~primus.rubric_memory.citations.RubricMemoryCitationContext, candidate_agenda_items: list[~primus.rubric_memory.sme_question_gate.RubricMemorySMEQuestion] = <factory>, optimizer_context: str = '')

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

candidate_agenda_items: list[RubricMemorySMEQuestion]

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

optimizer_context: str

rubric_memory_context: RubricMemoryCitationContext

score_identifier: str

score_version_id: str

scorecard_identifier: str

class primus.rubric_memory.RubricMemorySMEQuestionGateResult(*, score_version_id: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], final_agenda_markdown: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], final_items: list[~primus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, suppressed_items: list[~primus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, transformed_items: list[~primus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, kept_items: list[~primus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, citation_diagnostics: list[dict[str, ~typing.Any]] = <factory>, summary_counts: dict[str, int] = <factory>)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

citation_diagnostics: list[dict[str, Any]]

final_agenda_markdown: str

final_items: list[RubricMemoryGatedSMEQuestion]

kept_items: list[RubricMemoryGatedSMEQuestion]

model_config = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

score_version_id: str

summary_counts: dict[str, int]

suppressed_items: list[RubricMemoryGatedSMEQuestion]

transformed_items: list[RubricMemoryGatedSMEQuestion]

class primus.rubric_memory.RubricMemorySMEQuestionGateService(*, synthesizer: RubricMemorySMEQuestionGateSynthesizer | None = None)

Bases: object

Typed, citation-validating wrapper around SME question gate synthesis.

__init__(*, synthesizer: RubricMemorySMEQuestionGateSynthesizer | None = None)

async gate(request: RubricMemorySMEQuestionGateRequest) → RubricMemorySMEQuestionGateResult

class primus.rubric_memory.S3RubricMemoryCorpusPaths(bucket_name: str, scorecard_prefix: str, scorecard_knowledge_base_prefix: str, prefix_knowledge_base_prefixes: list[str], score_knowledge_base_prefix: str, sources: list[S3RubricMemorySource])

Bases: object

Convention-derived S3 rubric-memory prefixes for one score.

__init__(bucket_name: str, scorecard_prefix: str, scorecard_knowledge_base_prefix: str, prefix_knowledge_base_prefixes: list[str], score_knowledge_base_prefix: str, sources: list[S3RubricMemorySource]) → None

bucket_name: str

prefix_knowledge_base_prefixes: list[str]

score_knowledge_base_prefix: str

scorecard_knowledge_base_prefix: str

scorecard_prefix: str

sources: list[S3RubricMemorySource]

class primus.rubric_memory.S3RubricMemoryCorpusResolver(*, bucket_name: str | None = None, s3_client: Any | None = None)

Bases: object

Resolve rubric-memory S3 prefixes using the name-based hierarchy.

__init__(*, bucket_name: str | None = None, s3_client: Any | None = None)

resolve(*, scorecard_name: str, score_name: str) → S3RubricMemoryCorpusPaths

class primus.rubric_memory.S3RubricMemoryObject(key: str, size: int, etag: str, last_modified: datetime | None)

Bases: object

One raw S3 object in a rubric-memory corpus source.

__init__(key: str, size: int, etag: str, last_modified: datetime | None) → None

etag: str

key: str

last_modified: datetime | None

size: int

class primus.rubric_memory.S3RubricMemorySource(bucket_name: str, prefix: str, scope_level: str, objects: tuple[S3RubricMemoryObject, ...])

Bases: object

An S3 rubric-memory prefix with its score hierarchy scope.

__init__(bucket_name: str, prefix: str, scope_level: str, objects: tuple[S3RubricMemoryObject, ...]) → None

bucket_name: str

objects: tuple[S3RubricMemoryObject, ...]

prefix: str

scope_level: str

class primus.rubric_memory.SMEQuestionAnswerStatus(value)

Bases: str, Enum

ANSWERED_BY_CORPUS = 'answered_by_corpus'

ANSWERED_BY_RUBRIC = 'answered_by_rubric'

CONFLICTING_EVIDENCE = 'conflicting_evidence'

PARTIALLY_ANSWERED = 'partially_answered'

TRUE_OPEN_QUESTION = 'true_open_question'

class primus.rubric_memory.SMEQuestionGateAction(value)

Bases: str, Enum

KEEP = 'keep'

SUPPRESS = 'suppress'

TRANSFORM = 'transform'

class primus.rubric_memory.TactusRubricEvidenceSynthesizer(*, provider: str = 'openai', model: str = 'gpt-5-mini-2025-08-07', procedure_id: str = 'rubric_evidence_pack_synthesis', max_tokens: int = 16000, max_input_tokens: int = 48000, model_attempt_max_attempts: int = 3, model_attempt_authority=None)

Bases: object

Run the repo-owned Tactus synthesis procedure for evidence packs.

__init__(*, provider: str = 'openai', model: str = 'gpt-5-mini-2025-08-07', procedure_id: str = 'rubric_evidence_pack_synthesis', max_tokens: int = 16000, max_input_tokens: int = 48000, model_attempt_max_attempts: int = 3, model_attempt_authority=None)

async synthesize(*, request: RubricEvidencePackRequest, evidence: Sequence[EvidenceSnippet], history: Sequence[RubricHistoryEvent], confidence_inputs: ConfidenceInputs) → RubricEvidencePack

class primus.rubric_memory.TactusRubricMemorySMEQuestionGateSynthesizer(*, provider: str = 'openai', model: str = 'gpt-5-mini-2025-08-07', procedure_id: str = 'rubric_memory_sme_question_gate', max_tokens: int = 16000, max_input_tokens: int = 48000, model_attempt_max_attempts: int = 2, model_attempt_authority: Any = None)

Bases: object

Run the repo-owned Tactus procedure for SME question gating.

__init__(*, provider: str = 'openai', model: str = 'gpt-5-mini-2025-08-07', procedure_id: str = 'rubric_memory_sme_question_gate', max_tokens: int = 16000, max_input_tokens: int = 48000, model_attempt_max_attempts: int = 2, model_attempt_authority: Any = None)

async synthesize(*, request: RubricMemorySMEQuestionGateRequest) → dict[str, Any]

primus.rubric_memory.candidate_agenda_items_from_markdown(markdown: str) → list[RubricMemorySMEQuestion]: Split optimizer SME agenda Markdown into deterministic candidate items.

primus.rubric_memory.format_gated_sme_agenda(items: Sequence[RubricMemoryGatedSMEQuestion]) → str

primus.rubric_memory.validate_rubric_memory_citations(supplied_ids: Iterable[str] | None, context: RubricMemoryCitationContext | dict[str, Any] | None, *, require_citation: bool = False) → RubricMemoryCitationValidation

primus.rubric_memory package

Submodules