Adapt Trm¶

Adapt TRM — mvp.adapt_trm

Cluster: Multimodal | Type: component | MCP Tools: 43

Overview¶

Tiny Recursive Model (TRM) block that composes input token sequences into a hierarchical representation through recursive averaging and simple linear projections — a pure-Python approximation of Samsung SAIL Montreal's tiny recursive neural architectures. Supports three tasks: embed (return a fixed-size representation), classify (return class probabilities), and generate (reconstruct or extend the sequence). No neural framework dependency is required; all computation is plain Python arithmetic.

Role in G6 reliability workflows

adapt_trm is a deterministic local utility for sequence embeddings, grid transforms, sequence recall, search, clustering, and similarity checks. It can help a G6 workflow compare structured states, persist examples, and inspect lightweight representations, but it is not a trainable reliability layer by itself. Product claims about reliability, self-improvement, or production assurance should refer to the broader G6 harness, verification, grounding, evaluation, and feedback loop around this component.

Not a transformer or cross-modal foundation model

Despite the name, this component does not ship a trained transformer, CLIP-style multimodal encoder, or learned cross-modal alignment. It accepts numeric sequences and grids supplied by upstream components. For production image, audio, or text embedding quality, pair it with a purpose-built model or adapter and treat adapt_trm as a lightweight deterministic tool inside the pipeline.

When to use:

Producing lightweight sequence embeddings for similarity or clustering tasks without a large transformer
Classifying short token sequences (log lines, sensor readings) with a minimal compute footprint
Exploring recursive composition as an inductive bias in a goal-engine or solver loop
Embedding structured data sequences before passing them to a downstream learning block
Persisting and searching examples through the MCP sequence store during local debugging or pilot workflows

Example:

from mvp.adapt_trm import AdaptTRMBlock, TRMInput

block = AdaptTRMBlock(name="trm")
result = block.infer(TRMInput(
    sequence=[[0.1, 0.2, 0.3], [0.4, 0.5, 0.6], [0.7, 0.8, 0.9]],
    task="embed",
    hidden_dim=16,
    max_depth=3,
))
# result.value.embedding → 16-dim vector; result.value.depth_reached → actual recursion depth

Works well with: adapt_learning, ctx_rag, adapt_pytorch

Public API¶

`TRMInput(BaseModel)`¶

Input to AdaptTRMBlock.

Field	Type	Default
`sequence`	`list[list[float]]`	required
`task`	`Literal['classify', 'embed', 'generate']`	'embed'
`n_classes`	`int`	2
`hidden_dim`	`int`	16
`max_depth`	`int`	3

`TRMOutput(BaseModel)`¶

Output from AdaptTRMBlock.

Field	Type	Default
`predictions`	`list[float]`	required
`embedding`	`list[float]`	required
`depth_reached`	`int`	required
`n_recursive_calls`	`int`	required
`task`	`str`	required
`degraded`	`bool`	False
`degradation_reason`	`str`	''

`AdaptTRMBlock(AIBlock[TRMInput, TRMOutput, None])`¶

Tiny Recursive Model block.

Field	Type	Default
`name`	`str`	'adapt_trm'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: TRMInput) -> Result[TRMOutput]`¶

`MCPTRMInput(BaseModel)`¶

Input for adapt_trm MCP v2.0 (27 ops: 25 core + 2 advisory).

Field	Type	Default
`op`	`TRMOp`	required
`sequence`	`list[list[float]] \\| None`	None
`sequences`	`list[list[list[float]]] \\| None`	None
`sequence_a`	`list[list[float]] \\| None`	None
`sequence_b`	`list[list[float]] \\| None`	None
`hidden_dim`	`int`	Field(default=16, ge=1)
`max_depth`	`int`	Field(default=3, ge=1)
`n_classes`	`int`	Field(default=2, ge=2)
`h_cycles`	`int`	Field(default=2, ge=1)
`l_cycles`	`int`	Field(default=3, ge=1)
`iterations`	`int`	Field(default=5, ge=1)
`method`	`Literal['mean', 'max', 'min']`	'mean'
`grid`	`list[list[int]] \\| None`	None
`grid_a`	`list[list[int]] \\| None`	None
`grid_b`	`list[list[int]] \\| None`	None
`n_colors`	`int`	Field(default=10, ge=1)
`transform_id`	`int`	Field(default=0, ge=0, le=7)
`name`	`str \\| None`	None
`embedding`	`list[float] \\| None`	None
`tags`	`str \\| None`	None
`notes`	`str \\| None`	None
`tags_csv`	`str \\| None`	None
`limit`	`int`	Field(default=50, ge=1)
`query`	`str \\| None`	None
`top_k`	`int`	Field(default=5, ge=1)
`k`	`int`	Field(default=3, ge=1)
`input_modality`	`str`	'sequence'
`reasoning_structure`	`str`	''
`backend_id`	`str \\| None`	None
`target_op`	`str \\| None`	None
`payload`	`dict \\| None`	None
`dry_run`	`bool`	False
`allow_gpu`	`bool`	False
`request_id`	`str \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None

`MCPTRMOutput(BaseModel)`¶

Output from adapt_trm MCP v2.0.

Field	Type	Default
`op`	`str`	required
`message`	`str`	required
`embedding`	`list[float] \\| None`	None
`embeddings`	`list[list[float]] \\| None`	None
`predictions`	`list[float] \\| None`	None
`predicted_class`	`int \\| None`	None
`confidence`	`float \\| None`	None
`depth_reached`	`int \\| None`	None
`n_recursive_calls`	`int \\| None`	None
`vectors`	`list[list[float]] \\| None`	None
`shape`	`list[int] \\| None`	None
`similarity`	`float \\| None`	None
`convergence_deltas`	`list[float] \\| None`	None
`rationale`	`str \\| None`	None
`records`	`list[dict] \\| None`	None
`data`	`dict \\| None`	None
`review_status`	`str`	''
`calibration_status`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	''
`warning_card`	`dict \\| None`	None
`evidence`	`dict \\| None`	None
`request_id`	`str \\| None`	None
`task_id`	`str \\| None`	None
`run_id`	`str \\| None`	None

`MCPTRMRecord(BaseModel)`¶

A stored sequence record from the TRMStore.

Field	Type	Default
`id`	`int`	required
`name`	`str`	required
`vectors_json`	`str`	required
`embedding_json`	`str`	required
`config_json`	`str`	required
`tags`	`str`	required
`notes`	`str`	required
`timestamp`	`str`	required

`TRMStore`¶

Manages the 5-table TRM SQLite database.

Constructor:

Parameter	Type	Default
`db_path`	`str \\| None`	None

Methods:

`upsert_sequence(name: str, vectors: list[list[float]], embedding: list[float], config: dict | None = None, tags: str = '', notes: str = '') -> int`¶

`get_sequence(name: str) -> dict | None`¶

`list_sequences(tags_csv: str | None = None, limit: int = 50) -> list[dict]`¶

`all_sequences_with_embeddings() -> list[dict]`¶

`upsert_pattern(name: str, grid: list[list[int]], embedding: list[float], label: str = '', tags: str = '') -> int`¶

`log_analysis(name: str, op: str, config: dict, result: dict) -> None`¶

`log_session(op: str, params: dict, input_shape: str, output_shape: str, duration_ms: float) -> None`¶

`log_error(op: str, error_message: str, params: dict) -> None`¶

`table_counts() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 5) -> list[dict]`¶

Search sequences + patterns by name/tags/notes (TF-IDF or substring).

`AdaptTRMMCPBlock(AIBlock[MCPTRMInput, MCPTRMOutput, None])`¶

27-op MCP dispatcher for adapt_trm v2.0 (25 core + 2 advisory).

Field	Type	Default
`name`	`str`	'adapt_trm_mcp'
`db_path`	`str \\| None`	None
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`EmbeddingPlanner \\| None`	None

Methods:

`infer(input_data: MCPTRMInput) -> Result[MCPTRMOutput]`¶

MCP Tools¶

Operation	Source
`embed`	trm_mcp
`classify`	trm_mcp
`generate`	trm_mcp
`analyse`	trm_mcp
`capabilities`	trm_mcp
`batch_embed`	trm_mcp
`compare`	trm_mcp
`pool`	trm_mcp
`hrm_embed`	trm_mcp
`hrm_classify`	trm_mcp
`refine`	trm_mcp
`attention_embed`	trm_mcp
`adaptive_embed`	trm_mcp
`grid_encode`	trm_mcp
`grid_embed`	trm_mcp
`grid_compare`	trm_mcp
`dihedral_transform`	trm_mcp
`sequence_store`	trm_mcp
`sequence_retrieve`	trm_mcp
`sequence_list`	trm_mcp
`nearest_neighbors`	trm_mcp
`cluster`	trm_mcp
`search`	trm_mcp
`pattern_match`	trm_mcp
`trm_info`	trm_mcp
`recommend_embedding_strategy`	trm_mcp
`list_patterns`	trm_mcp
`backend_capabilities`	trm_mcp
`backend_op_schema`	trm_mcp
`backend_validate`	trm_mcp
`backend_run`	trm_mcp
`backend_load`	trm_mcp
`backend_unload`	trm_mcp
`backend_health`	trm_mcp
`backend_infer`	trm_mcp
`backend_train`	trm_mcp
`backend_load_checkpoint`	trm_mcp
`backend_export`	trm_mcp
`backend_probe_state`	trm_mcp
`backend_native_call`	trm_mcp
`mean`	trm_mcp
`max`	trm_mcp
`min`	trm_mcp

Adapt Trm¶

Overview¶

Public API¶

TRMInput(BaseModel)¶

TRMOutput(BaseModel)¶

AdaptTRMBlock(AIBlock[TRMInput, TRMOutput, None])¶

infer(data: TRMInput) -> Result[TRMOutput]¶

MCPTRMInput(BaseModel)¶

MCPTRMOutput(BaseModel)¶

MCPTRMRecord(BaseModel)¶

TRMStore¶

upsert_sequence(name: str, vectors: list[list[float]], embedding: list[float], config: dict | None = None, tags: str = '', notes: str = '') -> int¶

get_sequence(name: str) -> dict | None¶

list_sequences(tags_csv: str | None = None, limit: int = 50) -> list[dict]¶

all_sequences_with_embeddings() -> list[dict]¶

upsert_pattern(name: str, grid: list[list[int]], embedding: list[float], label: str = '', tags: str = '') -> int¶

log_analysis(name: str, op: str, config: dict, result: dict) -> None¶

log_session(op: str, params: dict, input_shape: str, output_shape: str, duration_ms: float) -> None¶

log_error(op: str, error_message: str, params: dict) -> None¶

table_counts() -> dict[str, int]¶

text_search(query: str, top_k: int = 5) -> list[dict]¶

AdaptTRMMCPBlock(AIBlock[MCPTRMInput, MCPTRMOutput, None])¶

infer(input_data: MCPTRMInput) -> Result[MCPTRMOutput]¶

MCP Tools¶

`TRMInput(BaseModel)`¶

`TRMOutput(BaseModel)`¶

`AdaptTRMBlock(AIBlock[TRMInput, TRMOutput, None])`¶

`infer(data: TRMInput) -> Result[TRMOutput]`¶

`MCPTRMInput(BaseModel)`¶

`MCPTRMOutput(BaseModel)`¶

`MCPTRMRecord(BaseModel)`¶

`TRMStore`¶

`upsert_sequence(name: str, vectors: list[list[float]], embedding: list[float], config: dict | None = None, tags: str = '', notes: str = '') -> int`¶

`get_sequence(name: str) -> dict | None`¶

`list_sequences(tags_csv: str | None = None, limit: int = 50) -> list[dict]`¶

`all_sequences_with_embeddings() -> list[dict]`¶

`upsert_pattern(name: str, grid: list[list[int]], embedding: list[float], label: str = '', tags: str = '') -> int`¶

`log_analysis(name: str, op: str, config: dict, result: dict) -> None`¶

`log_session(op: str, params: dict, input_shape: str, output_shape: str, duration_ms: float) -> None`¶

`log_error(op: str, error_message: str, params: dict) -> None`¶

`table_counts() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 5) -> list[dict]`¶

`AdaptTRMMCPBlock(AIBlock[MCPTRMInput, MCPTRMOutput, None])`¶

`infer(input_data: MCPTRMInput) -> Result[MCPTRMOutput]`¶