plexus.rubric_memory.sme_question_gate module

class plexus.rubric_memory.sme_question_gate.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: ~plexus.rubric_memory.sme_question_gate.SMEQuestionGateAction, answer_status: ~plexus.rubric_memory.sme_question_gate.SMEQuestionAnswerStatus, rationale: str = '', citation_ids: list[str] = <factory>, citation_validation: ~plexus.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 plexus.rubric_memory.sme_question_gate.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 plexus.rubric_memory.sme_question_gate.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: ~plexus.rubric_memory.citations.RubricMemoryCitationContext, candidate_agenda_items: list[~plexus.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 plexus.rubric_memory.sme_question_gate.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[~plexus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, suppressed_items: list[~plexus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, transformed_items: list[~plexus.rubric_memory.sme_question_gate.RubricMemoryGatedSMEQuestion] = <factory>, kept_items: list[~plexus.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 plexus.rubric_memory.sme_question_gate.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 plexus.rubric_memory.sme_question_gate.RubricMemorySMEQuestionGateSynthesizer(*args, **kwargs)

Bases: Protocol

__init__(*args, **kwargs)

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

Classify candidate SME agenda items against rubric-memory evidence.

class plexus.rubric_memory.sme_question_gate.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 plexus.rubric_memory.sme_question_gate.SMEQuestionGateAction(value)

Bases: str, Enum

KEEP = 'keep'

SUPPRESS = 'suppress'

TRANSFORM = 'transform'

class plexus.rubric_memory.sme_question_gate.TactusRubricMemorySMEQuestionGateSynthesizer(*, provider: str = 'openai', model: str = 'gpt-5-mini', procedure_id: str = 'rubric_memory_sme_question_gate', max_tokens: int = 16000)

Bases: object

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

__init__(*, provider: str = 'openai', model: str = 'gpt-5-mini', procedure_id: str = 'rubric_memory_sme_question_gate', max_tokens: int = 16000)

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

plexus.rubric_memory.sme_question_gate.candidate_agenda_items_from_markdown(markdown: str) → list[RubricMemorySMEQuestion]

Split optimizer SME agenda Markdown into deterministic candidate items.

plexus.rubric_memory.sme_question_gate.format_gated_sme_agenda(items: Sequence[RubricMemoryGatedSMEQuestion]) → str