Ctx Ace¶

Ctx ACE — mvp.ctx_ace

Cluster: Context & Retrieval | Type: component | MCP Tools: 26

Overview¶

Agent Cognition Engine providing a stateful working memory with attention-driven retrieval. Observations are tagged, scored, and stored in memory; the attend operation ranks them by TF-IDF cosine similarity to a query (Jaccard word-overlap fallback when sklearn is unavailable). The forget operation prunes by age or tag, keeping memory lean during long agent runs.

Production caveat: Treat ctx_ace as playbook-based adaptive context, not autonomous model self-training. The MCP tools can persist playbooks and run a Generator → Reflector → Curator loop that proposes and applies localized playbook updates, but they do not fine-tune model weights, prove that future answers improve, or remove the need for evaluation and human review. In launch copy and customer docs, describe this component as "adaptive context/playbook management" rather than a fully self-training system.

When to use:

• Giving a long-running agent a bounded, queryable working memory
• Retrieving the most relevant prior observations before generating a response
• Pruning stale observations to stay within token budget constraints
• Maintaining a scoped playbook of lessons from evaluated runs, with human-reviewed updates

Example:

from mvp.ctx_ace import CtxACEBlock, ACEInput

block = CtxACEBlock(name="ace")
block.infer(ACEInput(operation="observe", content="The gripper failed on object A", tags=["error"]))
result = block.infer(ACEInput(operation="attend", query="gripper failure", top_k=3))
# result.ok → True; result.value.observations → top-3 most relevant entries

Works well with: context_engine, ctx_recursive, ctx_mnm

Public API¶

CtxACEBlock(AIBlock[ACEInput, ACEOutput, dict])¶

Agent Cognition Engine with working memory.

Methods:

infer(data: ACEInput) -> Result[ACEOutput]¶

Observation(BaseModel)¶

ACEInput(BaseModel)¶

ACEOutput(BaseModel)¶

CtxACEMCPBlock(AIBlock[MCPACEInput, MCPACEOutput, dict])¶

26-op ACE context engineering block with SQLite persistence.

Methods:

infer(data: MCPACEInput) -> Result[MCPACEOutput]¶

MCPACERecord(BaseModel)¶

MCPACEInput(BaseModel)¶

MCPACEOutput(BaseModel)¶

ACEStore¶

Sync SQLite store with 6 tables.

Constructor:

Methods:

create_playbook(name: str, content: str = '') -> str¶

get_playbook(name: str) -> dict[str, Any] | None¶

get_playbook_by_id(pid: str) -> dict[str, Any] | None¶

list_playbooks() -> list[dict[str, Any]]¶

delete_playbook(name: str) -> bool¶

update_playbook_content(name: str, content: str) -> bool¶

increment_bullet_id(name: str) -> int¶

add_bullet(playbook_id: str, bullet_id: str, section: str, content: str) -> str¶

get_bullets(playbook_id: str) -> list[dict[str, Any]]¶

update_bullet_counts(bullet_id: str, helpful_delta: int = 0, harmful_delta: int = 0, playbook_id: str | None = None) -> bool¶

delete_bullet(bullet_id: str, playbook_id: str | None = None) -> bool¶

add_reflection(playbook_id: str, question: str, reasoning_trace: str, predicted_answer: str, ground_truth: str, reflection_content: str, bullet_tags_json: str, is_correct: bool) -> str¶

list_reflections(playbook_id: str, limit: int = 50) -> list[dict[str, Any]]¶

save_session(name: str, playbook_id: str, config_json: str) -> str¶

load_session(name: str) -> dict[str, Any] | None¶

list_sessions() -> list[dict[str, Any]]¶

update_session(name: str, best_accuracy: float | None = None, total_steps: int | None = None, status: str | None = None) -> bool¶

add_training_sample(session_id: str, step: int, question: str, context: str, target: str, pre_train_answer: str, post_train_answer: str, pre_train_correct: bool, post_train_correct: bool, reflection_json: str, curator_ops_json: str) -> str¶

list_training_samples(session_id: str, limit: int = 50) -> list[dict[str, Any]]¶

set_config(key: str, value_json: str) -> str¶

get_config(key: str) -> dict[str, Any] | None¶

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

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

MCP Tools¶