`mnesis.operators.agentic_map`¶

agentic_map ¶

AgenticMap operator — parallel sub-agent sessions.

AgentMapResult ¶

Bases: BaseModel

The result of running a single sub-agent through AgenticMap.

AgentMapBatch ¶

Bases: BaseModel

Aggregate result of a full AgenticMap.run_all() call.

AgenticMap ¶

AgenticMap(
    config: OperatorConfig,
    event_bus: EventBus | None = None,
    model: str | None = None,
)

Spawn independent sub-agent sessions per input item.

Each item gets a full MnesisSession with multi-turn reasoning, tool access, and automatic context management. Sub-sessions are isolated — they do not inherit the parent's conversation history.

The parent session pays only O(1) context cost per item: it sees only the final output text, regardless of how many turns the sub-session needed.

Permission restrictions (always applied): - Sub-sessions cannot spawn further sub-agents. - read_only=True is not yet implemented; passing it raises NotImplementedError.

Example::

agentic_map = AgenticMap(config.operators, model="anthropic/claude-opus-4-6")
async for result in agentic_map.run(
    inputs=repositories,
    agent_prompt_template="Analyze this repository and report quality issues:\n{{ item }}",
    read_only=False,
    max_turns=20,
):
    print(f"Repo: {result.input}\nFindings: {result.output_text[:200]}")

run `async` ¶

run(
    inputs: list[Any],
    agent_prompt_template: str,
    *,
    model: str | None = None,
    concurrency: int | None = None,
    read_only: bool = True,
    tools: list[Any] | None = None,
    max_turns: int = 20,
    continuation_message: str = "",
    parent_session_id: str | None = None,
    agent: str = "general",
    lcm_config: MnesisConfig | None = None,
    db_path: str | None = None,
    pool: StorePool | None = None,
) -> AsyncGenerator[AgentMapResult, None]

Spawn sub-agent sessions in parallel, one per input item.

This is an async generator — iterate with async for, not await.

Parameters:

Name	Type	Description	Default
`inputs`	`list[Any]`	List of items. Each is rendered into `{{ item }}`.	required
`agent_prompt_template`	`str`	Jinja2 template for the initial user message. Must reference the `item` variable.	required
`model`	`str \| None`	LLM model string for all sub-sessions. Falls back to the model set on `__init__` if not provided here.	`None`
`concurrency`	`int \| None`	Maximum concurrent sub-sessions.	`None`
`read_only`	`bool`	Reserved. `True` raises `NotImplementedError` until enforcement is implemented.	`True`
`tools`	`list[Any] \| None`	Optional tool definitions for sub-sessions.	`None`
`max_turns`	`int`	Maximum turns per sub-session before stopping.	`20`
`continuation_message`	`str`	Message sent as the user turn after turn 0. If an empty string (default), the sub-session runs only the initial turn and then stops, regardless of `max_turns`.	`''`
`are not needed for typical use`			required
`parent_session_id`	`str \| None`	Optional parent session ID for lineage tracking.	`None`
`agent`	`str`	Agent role for sub-sessions.	`'general'`
`lcm_config`	`MnesisConfig \| None`	Mnesis config for sub-sessions. Defaults to `MnesisConfig()`.	`None`
`db_path`	`str \| None`	Override database path (useful for testing).	`None`
`pool`	`StorePool \| None`	Shared `StorePool` so all sub-sessions use one connection. If omitted a fresh pool is created and closed automatically at the end of this call.	`None`

Yields:

Type	Description
`AsyncGenerator[AgentMapResult, None]`	AgentMapResult objects as sub-sessions complete (not in input order).

Raises:

Type	Description
`NotImplementedError`	If `read_only=True` is passed (not yet implemented).
`ValueError`	If neither `model` nor the constructor `model` is set, or if `agent_prompt_template` does not reference `item`.

run_all `async` ¶

run_all(
    inputs: list[Any],
    agent_prompt_template: str,
    *,
    model: str | None = None,
    concurrency: int | None = None,
    read_only: bool = True,
    tools: list[Any] | None = None,
    max_turns: int = 20,
    continuation_message: str = "",
    parent_session_id: str | None = None,
    agent: str = "general",
    lcm_config: MnesisConfig | None = None,
    db_path: str | None = None,
    pool: StorePool | None = None,
) -> AgentMapBatch

Spawn all sub-agents and return an aggregate AgentMapBatch.

Convenience alternative to run() for callers who do not need streaming. Collects all results before returning.

Returns:

Type	Description
`AgentMapBatch`	AgentMapBatch with `.successes`, `.failures`, and `.total_attempts`.

mnesis.operators.agentic_map¶