cursus.steps.scripts.slipbox_knowledge_routing

Slipbox Knowledge Routing — Cursus ProcessingStep script (PROPOSAL scaffold).

Reads a mounted domain knowledge+ruleset corpus and runs a compile → index → route pipeline, emitting to the downstream BedrockProcessing step:

• prompt_rulesetthe compiled prompt ruleset (prompts.json, in the {ruleset, rules}

  envelope the consumer expects; output schema embedded in ruleset)

• routed_records : the input records + selected_rule_names + routing_confidence

Pipeline stages (each is a scaffold TODO — fill in the domain’s source function):

COMPILE read knowledge_corpus/rule_*.md -> prompts.json (in memory)

[TODO: port the domain’s rule-compilation function]

INDEX read knowledge_corpus/pattern_*.md (+ behavior_*.md)

-> SentenceTransformer.encode -> in-memory routing index [TODO: port the domain’s index-build function]; the encoder is overridden to the offline embedding_model input path so no HuggingFace-hub download occurs.

ROUTE read records parquet -> build_query_text -> cosine-match

-> activation top-k -> routed rule names + routing_confidence [TODO: port the domain’s batch-route + activation-scoring functions]

Internal consistency gate: the set of rules linked from the routing index MUST be a subset of the compiled rule_names in prompts.json (otherwise routing could emit a rule name the ruleset does not define).

NOTE (PROPOSAL scaffold): the routing logic below is a faithful skeleton with explicit TODOs pointing at the source functions. The Cursus contract surface — the main(input_paths, output_paths, environ_vars, job_args) signature, the I/O container paths, the env-var reads, and the __main__ argparse — is complete and correct so that validate/preflight pass and the step is constructible.

compile_prompt_ruleset(knowledge_dir, log)

Compile the rule_*.md knowledge corpus into an in-memory prompt ruleset.

TODO: port the domain’s rule-compilation function here.

Returns a dict shaped like the emitted prompts.json — the {ruleset, rules} envelope the downstream Bedrock consumer’s _adapt_ruleset_templates expects (ruleset = the shared prompt layer + embedded output_schema; rules = a LIST of per-rule dicts each carrying at least rule_name):

{

“ruleset”: {“system_prompt”: str, “input_placeholders”: […],

“output_schema”: {…}},

“rules”: [{“rule_name”: str, “description”: str, “metadata”: {…}}, …], “rule_names”: [rule_name, …], # convenience mirror of rules[*].rule_name

}

Parameters:

• knowledge_dir (str) – Path to the mounted knowledge corpus.

• log (Callable[[str], None]) – Logging function.

Returns:

The compiled prompt-ruleset dict.

Return type:

Dict[str, Any]

build_routing_index(knowledge_dir, embedding_model_dir, model_name, log)

Read pattern_*.md (+ behavior_*.md) and encode them into an in-memory routing index.

Build the in-memory routing index (TODO: port the domain index-build function).

Returns a dict shaped like:

{

“pattern_names”: [str, …], “embeddings”: np.ndarray (n_patterns, dim), “linked_rules”: {pattern_name: [rule_name, …], …},

}

Parameters:

• knowledge_dir (str) – Path to the mounted knowledge corpus.

• embedding_model_dir (str | None) – Optional path to offline encoder weights.

• model_name (str) – Fallback SentenceTransformer model name.

• log (Callable[[str], None]) – Logging function.

Returns:

The in-memory routing index dict.

Return type:

Dict[str, Any]

build_query_text(row)

Build the query text for a single record used to match against the pattern index.

Query-assembly half of the batch-route stage.

score_rules_by_activation(query_embedding, index, threshold, top_k)

Score rules by activation and return the top-k routed rule names + confidence.

TODO: port the domain’s activation-scoring function here.

Parameters:

• query_embedding – The (dim,) normalized query embedding.

• index (Dict[str, Any]) – The in-memory routing index from build_routing_index.

• threshold (float) – Minimum cosine similarity for a pattern to activate its rules.

• top_k (int) – Maximum number of routed rules to keep.

Returns:

(routed_rule_names, routing_confidence)

Return type:

Tuple[List[str], float]

route_records(records_dir, index, threshold, top_k, log, encode_batch_size=256)

Read the input records and route each one to a set of rule names + confidence.

Batch-route the records (TODO: port the domain batch-route function).

Parameters:

• records_dir (str) – Path to the mounted records (parquet shards).

• index (Dict[str, Any]) – The in-memory routing index.

• threshold (float) – Activation threshold.

• top_k (int) – Max routed rules per record.

• log (Callable[[str], None]) – Logging function.

Returns:

• selected_rule_names : list[str] (the routed rule names; this is the column name the downstream Bedrock consumer reads by default — BEDROCK_ROUTED_RULES_COLUMN, default selected_rule_names)

• routing_confidence : float

Return type:

The records DataFrame with two added columns

assert_index_rules_subset_of_ruleset(index, ruleset, log)

Internal consistency gate: every rule the routing index can emit MUST be defined in the compiled prompt ruleset (index linked_rules ⊆ prompts.json rule_names).

write_prompt_ruleset(ruleset, output_dir, log)

Write prompts.json in the {ruleset, rules} envelope the Bedrock consumer expects.

The consumer’s gate requires BOTH a top-level ruleset object AND a rules list (bedrock_processing.py load_prompt_templates → _adapt_ruleset_templates); the output schema travels inside ruleset.output_schema so no separate schema channel is needed.

write_routed_records(df, output_dir, log)

Write the routed records (records + selected_rule_names + confidence) as parquet.

main(input_paths, output_paths, environ_vars, job_args, logger=None)

Main logic for slipbox knowledge routing, refactored for testability.

Parameters:

• input_paths (Dict[str, str]) – Dict of input container paths keyed by logical name (‘records’, ‘knowledge_corpus’, ‘embedding_model’).

• output_paths (Dict[str, str]) – Dict of output container paths keyed by logical name (‘prompt_ruleset’, ‘routed_records’).

• environ_vars (Dict[str, str]) – Dict of environment variables.

• job_args (Namespace) – Parsed command-line arguments (carries –job_type).

• logger (Callable[[str], None] | None) – Optional logging function (defaults to print).

Returns:

A small summary dict describing what was compiled/indexed/routed.

Return type:

Dict[str, Any]