Skip to content

Architectures

Every public architecture class — auto-generated from docstrings.

Foundational patterns

Reflection 

Reflection(max_iterations: int = 3, target_score: int = 9, **kwargs: Any)

Bases: Architecture

Self-Refine style reflection loop.

Source code in src/agentic_architectures/architectures/reflection.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    max_iterations: int = 3,
    target_score: int = 9,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.max_iterations = max_iterations
    self.target_score = target_score
    self._judge: LLMJudge[_ReflectionCritique] = LLMJudge(
        schema=_ReflectionCritique,
        rubric=(
            "Score the candidate (1-10) and provide concrete, actionable critique. "
            "Focus on what is missing, wrong, unclear, or could be stronger. "
            "Reserve scores ≥9 for genuinely excellent work."
        ),
        llm=self.llm,
    )

ToolUse 

ToolUse(tools: list[Any] | None = None, max_rounds: int = 6, system_prompt: str | None = None, **kwargs: Any)

Bases: Architecture

Tool-calling loop. The agent decides when to use a tool and when to stop.

Source code in src/agentic_architectures/architectures/tool_use.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self,
    tools: list[Any] | None = None,
    max_rounds: int = 6,
    system_prompt: str | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if not provider_supports_tools():
        raise RuntimeError(
            "Tool Use requires a provider with tool-calling support. "
            "Switch LLM_PROVIDER to one of: nebius, openai, anthropic, groq, "
            "together, fireworks, mistralai, google, ollama."
        )

    if tools is None:
        from agentic_architectures.tools.search import web_search_tool

        tools = [web_search_tool(max_results=5)]
    self.tools = tools
    self.max_rounds = max_rounds
    self.system_prompt = system_prompt or self.DEFAULT_SYSTEM_PROMPT
    self._llm_with_tools = self.llm.bind_tools(self.tools)

ReAct 

ReAct(tools: list[Any] | None = None, max_rounds: int = 5, system_prompt: str | None = None, **kwargs: Any)

Bases: Architecture

Explicit two-node ReAct loop: think → act → (tools → think)* → END.

Source code in src/agentic_architectures/architectures/react.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    tools: list[Any] | None = None,
    max_rounds: int = 5,
    system_prompt: str | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if not provider_supports_tools():
        raise RuntimeError(
            "ReAct requires a provider with tool-calling support. Switch LLM_PROVIDER to one that does."
        )

    if tools is None:
        from agentic_architectures.tools.search import web_search_tool

        tools = [web_search_tool(max_results=5)]
    self.tools = tools
    self.max_rounds = max_rounds
    self.system_prompt = system_prompt or self.DEFAULT_SYSTEM_PROMPT
    self._llm_with_tools = self.llm.bind_tools(self.tools)

Planning 

Planning(tools: list[Any] | None = None, max_replans: int = 2, executor_rounds: int = 3, **kwargs: Any)

Bases: Architecture

Plan-Execute-Replan loop.

Source code in src/agentic_architectures/architectures/planning.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def __init__(
    self,
    tools: list[Any] | None = None,
    max_replans: int = 2,
    executor_rounds: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if tools is None:
        from agentic_architectures.tools.search import web_search_tool

        tools = [web_search_tool(max_results=4)]
    self.tools = tools
    self.max_replans = max_replans
    self.executor_rounds = executor_rounds

    self._planner = self.llm.with_structured_output(Plan)
    self._replanner = self.llm.with_structured_output(ReplanDecision)

    # Executor is a fresh ReAct sub-agent — composability in action.
    from agentic_architectures.architectures.tool_use import ToolUse

    self._executor = ToolUse(
        llm=self.llm.bind_tools(tools) if False else self.llm,  # passed through
        tools=tools,
        max_rounds=executor_rounds,
    )

MultiAgent 

MultiAgent(specialists: dict[str, str] | None = None, tools: list[Any] | None = None, specialist_rounds: int = 3, **kwargs: Any)

Bases: Architecture

Supervisor + N specialists + Writer.

Source code in src/agentic_architectures/architectures/multi_agent.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def __init__(
    self,
    specialists: dict[str, str] | None = None,
    tools: list[Any] | None = None,
    specialist_rounds: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.specialists = specialists or DEFAULT_SPECIALISTS

    if tools is None:
        from agentic_architectures.tools.search import web_search_tool

        tools = [web_search_tool(max_results=4)]
    self.tools = tools
    self.specialist_rounds = specialist_rounds

    # Each specialist is a ToolUse instance with its own role prompt.
    # IMPORTANT: append the cap-search rule so the role prompt doesn't
    # silently disable ToolUse's runaway-search protection.
    _CAP_RULE = (
        "\n\nRULES:\n"
        "  - Use the search tool at most 1-2 times.\n"
        "  - After your second search, STOP searching and answer with what you have.\n"
        "  - Always cite source URLs."
    )
    from agentic_architectures.architectures.tool_use import ToolUse

    self._specialist_agents = {
        name: ToolUse(
            llm=self.llm,
            tools=self.tools,
            max_rounds=specialist_rounds,
            system_prompt=prompt + _CAP_RULE,
        )
        for name, prompt in self.specialists.items()
    }

    self._SupervisorDecision = _build_decision_schema(list(self.specialists.keys()))
    self._supervisor_llm = self.llm.with_structured_output(self._SupervisorDecision)

PEV 

PEV(tools: list[Any] | None = None, max_retries_per_step: int = 2, executor_rounds: int = 4, **kwargs: Any)

Bases: Architecture

Plan-Execute-Verify with per-step retry-with-critique.

Source code in src/agentic_architectures/architectures/pev.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def __init__(
    self,
    tools: list[Any] | None = None,
    max_retries_per_step: int = 2,
    executor_rounds: int = 4,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if tools is None:
        from agentic_architectures.tools.search import web_search_tool

        tools = [web_search_tool(max_results=4)]
    self.tools = tools
    self.max_retries_per_step = max_retries_per_step
    self.executor_rounds = executor_rounds

    from agentic_architectures.architectures.tool_use import ToolUse

    # Strict per-step prompt: PEV steps are atomic, so 1-2 searches suffice.
    self._executor = ToolUse(
        llm=self.llm,
        tools=tools,
        max_rounds=executor_rounds,
        system_prompt=(
            "You are executing one atomic step of a larger plan. "
            "Use the search tool AT MOST ONCE per step (twice only if the "
            "first result was empty). After your search, immediately answer "
            "with the specific fact/value requested and a source URL. "
            "Do NOT preamble, do NOT explain — just the requested fact and a URL."
        ),
    )
    self._planner = self.llm.with_structured_output(Plan)
    self._verifier: LLMJudge[_StepVerification] = LLMJudge(
        schema=_StepVerification,
        rubric=(
            "Judge whether the executed step's result fully addresses the "
            "step's intent given the original task. "
            "A satisfactory result must (1) contain the specific fact / "
            "computation / artifact the step asks for, and (2) be grounded "
            "(cite a URL OR show the explicit computation)."
        ),
        llm=self.llm,
    )

Blackboard 

Blackboard(knowledge_sources: dict[str, str] | None = None, max_rounds: int = 6, min_confidence: int = 3, **kwargs: Any)

Bases: Architecture

Distributed multi-agent system with opportunistic, bid-driven contribution.

Source code in src/agentic_architectures/architectures/blackboard.py 
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(
    self,
    knowledge_sources: dict[str, str] | None = None,
    max_rounds: int = 6,
    min_confidence: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.knowledge_sources = knowledge_sources or DEFAULT_KNOWLEDGE_SOURCES
    self.max_rounds = max_rounds
    self.min_confidence = min_confidence
    self._bid_llm = self.llm.with_structured_output(_AgentBid)

EpisodicSemanticAgent 

EpisodicSemanticAgent(episodic: EpisodicMemory | None = None, semantic: SemanticMemory | None = None, recall_k_episodes: int = 3, recall_facts_depth: int = 1, **kwargs: Any)

Bases: Architecture

Dual-memory agent: FAISS episodes + NetworkX/Neo4j facts; persistent across run() calls.

Source code in src/agentic_architectures/architectures/episodic_semantic.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(
    self,
    episodic: EpisodicMemory | None = None,
    semantic: SemanticMemory | None = None,
    recall_k_episodes: int = 3,
    recall_facts_depth: int = 1,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.episodic = episodic if episodic is not None else EpisodicMemory()
    self.semantic = semantic if semantic is not None else SemanticMemory()
    self.recall_k_episodes = recall_k_episodes
    self.recall_facts_depth = recall_facts_depth
    self._fact_extractor = self.llm.with_structured_output(_ExtractedFacts)

TreeOfThoughts 

TreeOfThoughts(branching: int = 3, beam_width: int = 2, max_depth: int = 3, **kwargs: Any)

Bases: Architecture

Beam-search over a tree of LLM-generated reasoning steps.

Source code in src/agentic_architectures/architectures/tree_of_thoughts.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __init__(
    self,
    branching: int = 3,
    beam_width: int = 2,
    max_depth: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.branching = branching
    self.beam_width = beam_width
    self.max_depth = max_depth
    self._generator = self.llm.with_structured_output(_ThoughtCandidates)
    self._evaluator: LLMJudge[_ThoughtScore] = LLMJudge(
        schema=_ThoughtScore,
        rubric=(
            "Score this candidate thought on a STRICT 1-5 scale (see schema). "
            "You are running beam search — equal scores destroy the signal. "
            "Be DISCRIMINATING: most thoughts should be 2-4. Reserve 5 for rare "
            "excellence. If a thought is just a paraphrase of a sibling, score it 2."
        ),
        llm=self.llm,
    )

MentalLoop 

MentalLoop(n_candidates: int = 3, scoring_fn: Callable[[float], int] | None = None, **kwargs: Any)

Bases: Architecture

Generate → simulate → score → pick best → explain.

Create a Mental Loop architecture.

Parameters:

Name Type Description Default
n_candidates int

how many candidate actions to generate.

 3
scoring_fn Callable[[float], int] | None

optional deterministic Python function that takes the LLM's predicted_metric (a float) and returns an int 1-5 score. When set, the LLM's own overall_score is OVERRIDDEN by this function — the canonical fix for LLM-as-Scorer flatness. When None (default), the LLM's overall_score is used as-is.

 None
Source code in src/agentic_architectures/architectures/mental_loop.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def __init__(
    self,
    n_candidates: int = 3,
    scoring_fn: Callable[[float], int] | None = None,
    **kwargs: Any,
) -> None:
    """Create a Mental Loop architecture.

    Args:
        n_candidates: how many candidate actions to generate.
        scoring_fn: optional deterministic Python function that takes the
            LLM's `predicted_metric` (a float) and returns an int 1-5 score.
            When set, the LLM's own `overall_score` is OVERRIDDEN by this
            function — the canonical fix for LLM-as-Scorer flatness.
            When None (default), the LLM's `overall_score` is used as-is.
    """
    super().__init__(**kwargs)
    self.n_candidates = n_candidates
    self.scoring_fn = scoring_fn
    self._generator = self.llm.with_structured_output(_CandidateActions)
    self._simulator = self.llm.with_structured_output(_SimulatedOutcome)

MetaController 

MetaController(roster: dict[str, Architecture] | None = None, **kwargs: Any)

Bases: Architecture

LLM router that picks the right architecture per task.

Source code in src/agentic_architectures/architectures/meta_controller.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def __init__(
    self,
    roster: dict[str, Architecture] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.roster: dict[str, Architecture] = roster or _default_roster(self.llm)

    # Dynamic Literal type from roster keys so router output is constrained.
    names = tuple(self.roster.keys())
    next_t = Literal[names]  # type: ignore[valid-type]
    self._RouterDecision = create_model(
        "RouterDecision",
        chosen_arch=(
            next_t,
            Field(
                description=(
                    "Which architecture is the best fit for this task? "
                    "Pick exactly one from the roster. See the descriptions "
                    "below to choose."
                )
            ),
        ),
        reason=(
            str,
            Field(description="One sentence explaining the routing choice."),
        ),
    )
    self._router_llm = self.llm.with_structured_output(self._RouterDecision)

GraphMemoryAgent 

GraphMemoryAgent(semantic: SemanticMemory | None = None, traversal_depth: int = 2, **kwargs: Any)

Bases: Architecture

Knowledge-graph world model + traversal-based Q&A.

Source code in src/agentic_architectures/architectures/graph_memory.py 
70
71
72
73
74
75
76
77
78
79
def __init__(
    self,
    semantic: SemanticMemory | None = None,
    traversal_depth: int = 2,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.semantic = semantic if semantic is not None else SemanticMemory()
    self.traversal_depth = traversal_depth
    self._extractor = self.llm.with_structured_output(_IngestionResult)

ingest 

ingest(text: str) -> list[_IngestionTriple]

Extract triples from text and add them to the graph.

Source code in src/agentic_architectures/architectures/graph_memory.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
def ingest(self, text: str) -> list[_IngestionTriple]:
    """Extract triples from text and add them to the graph."""
    prompt = (
        "Extract a comprehensive set of (subject, predicate, object) triples "
        "from the text below. Each triple must use SPECIFIC named entities and "
        "a SHORT snake_case relation verb. Capture every concrete fact, "
        "skip generic statements.\n\n"
        f"--- TEXT ---\n{text}"
    )
    result = self._extractor.invoke(prompt)
    for t in result.triples:
        self.semantic.add_fact(t.subject, t.predicate, t.object)
    return result.triples

run 

run(task: str, **kwargs: Any) -> ArchitectureResult

Answer a question by traversing the current graph.

Source code in src/agentic_architectures/architectures/graph_memory.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def run(self, task: str, **kwargs: Any) -> ArchitectureResult:
    """Answer a question by traversing the current graph."""
    query = task
    entities = self._entities_in_query(query)
    # Fall back to all entities for short / vague queries
    if not entities and len(query.split()) <= 12:
        entities = self._entities_in_graph()[:20]

    facts_block, facts = self._facts_block(entities)
    prompt = (
        "You are answering a question using ONLY the facts in the knowledge "
        "graph. Do NOT use parametric knowledge — if the graph doesn't have "
        "the answer, say so explicitly.\n\n"
        f"## Question\n{query}\n\n"
        f"## Relevant graph facts (from {len(entities)} matched entit(y/ies))\n"
        f"{facts_block}\n\n"
        "Answer concisely. Cite the specific (s, p, o) triples you used."
    )
    answer = str(self.llm.invoke(prompt).content)

    return ArchitectureResult(
        output=answer,
        state={
            "matched_entities": entities,
            "facts_used": len(facts),
            "total_entities_in_graph": len(self._entities_in_graph()),
        },
        trace=[
            {"type": "matched_entities", "items": entities},
            {"type": "facts_retrieved", "items": facts[:30]},
            {"type": "answer", "content": answer},
        ],
        metadata={
            "matched_entities": len(entities),
            "facts_retrieved": len(facts),
            "total_entities_in_graph": len(self._entities_in_graph()),
            "graph_backend": type(self.semantic.backend).__name__,
        },
    )

Ensemble 

Ensemble(voters: dict[str, str] | None = None, aggregator_mode: Literal['llm_synth', 'majority_vote', 'highest_confidence'] = 'llm_synth', **kwargs: Any)

Bases: Architecture

N voters with diverse prompts run on the same task; aggregator combines.

Source code in src/agentic_architectures/architectures/ensemble.py 
132
133
134
135
136
137
138
139
140
141
def __init__(
    self,
    voters: dict[str, str] | None = None,
    aggregator_mode: Literal["llm_synth", "majority_vote", "highest_confidence"] = "llm_synth",
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.voters = voters or DEFAULT_VOTERS
    self.aggregator_mode = aggregator_mode
    self._voter_llm = self.llm.with_structured_output(_VoterOpinion)

DryRun 

DryRun(irreversibility_threshold: int = 4, require_human_approval_above_severity: Literal['low', 'medium', 'high'] | None = 'high', **kwargs: Any)

Bases: Architecture

Propose → simulate → approve → execute-or-skip safety pattern.

Parameters:

Name Type Description Default
irreversibility_threshold int

Python hard-cap. Any predicted irreversibility

= this value BLOCKS execution regardless of the LLM approval. Default 4.

 4
require_human_approval_above_severity Literal['low', 'medium', 'high'] | None

Severity levels at or above this require human-in-the-loop in production (in this demo, marked but auto-mocked).

 'high'
Source code in src/agentic_architectures/architectures/dry_run.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(
    self,
    irreversibility_threshold: int = 4,
    require_human_approval_above_severity: Literal["low", "medium", "high"] | None = "high",
    **kwargs: Any,
) -> None:
    """
    Args:
        irreversibility_threshold: Python hard-cap. Any predicted irreversibility
            >= this value BLOCKS execution regardless of the LLM approval. Default 4.
        require_human_approval_above_severity: Severity levels at or above this
            require human-in-the-loop in production (in this demo, marked but auto-mocked).
    """
    super().__init__(**kwargs)
    self.irreversibility_threshold = irreversibility_threshold
    self.require_human_approval_above_severity = require_human_approval_above_severity
    self._proposer = self.llm.with_structured_output(_ProposedAction)
    self._dry_runner = self.llm.with_structured_output(_DryRunOutcome)
    self._approver = self.llm.with_structured_output(_ApprovalDecision)

RLHFSelfImprovement 

RLHFSelfImprovement(max_iterations: int = 2, target_score: int = 8, word_count_range: tuple[int, int] = (30, 100), **kwargs: Any)

Bases: Architecture

Generate → critique → revise → maybe archive; archive persists across runs.

Parameters:

Name Type Description Default
max_iterations int

max refinement rounds per task.

 2
target_score int

minimum COMPOSITE Python-score (0-10) for archive acceptance.

 8
word_count_range tuple[int, int]

(min, max) word count to receive the word-count points.

 (30, 100)
Source code in src/agentic_architectures/architectures/rlhf.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def __init__(
    self,
    max_iterations: int = 2,
    target_score: int = 8,
    word_count_range: tuple[int, int] = (30, 100),
    **kwargs: Any,
) -> None:
    """
    Args:
        max_iterations: max refinement rounds per task.
        target_score: minimum COMPOSITE Python-score (0-10) for archive acceptance.
        word_count_range: (min, max) word count to receive the word-count points.
    """
    super().__init__(**kwargs)
    self.max_iterations = max_iterations
    self.target_score = target_score
    self.word_count_range = word_count_range
    self._editor = self.llm.with_structured_output(_EditorCritique)
    # Persistent archive across run() calls on the same instance.
    self.archive: list[dict[str, Any]] = []

CellularAutomata 

CellularAutomata(rule_prompt: str, allowed_states: list[str], height: int = 4, width: int = 4, max_steps: int = 3, **kwargs: Any)

Bases: Architecture

Grid of cells, each updated by an LLM per step.

Parameters:

Name Type Description Default
rule_prompt str

A short description of the LOCAL update rule the LLM should follow.

 required
allowed_states list[str]

List of valid state labels each cell can hold (e.g. ['fire', 'tree', 'empty']).

 required
height, width

Grid dimensions. Default 4x4 = 16 cells; cap aggressively.

 required
max_steps int

How many time steps to simulate.

 3
Source code in src/agentic_architectures/architectures/cellular_automata.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(
    self,
    rule_prompt: str,
    allowed_states: list[str],
    height: int = 4,
    width: int = 4,
    max_steps: int = 3,
    **kwargs: Any,
) -> None:
    """
    Args:
        rule_prompt: A short description of the LOCAL update rule the LLM should follow.
        allowed_states: List of valid state labels each cell can hold (e.g. ['fire', 'tree', 'empty']).
        height, width: Grid dimensions. Default 4x4 = 16 cells; cap aggressively.
        max_steps: How many time steps to simulate.
    """
    super().__init__(**kwargs)
    self.rule_prompt = rule_prompt
    self.allowed_states = allowed_states
    self.height = height
    self.width = width
    self.max_steps = max_steps
    self._updater = self.llm.with_structured_output(_CellUpdate)

run 

run(task: str, **kwargs: Any) -> ArchitectureResult

Run the simulation. task is the INITIAL GRID as a multi-line string, with cells separated by | and rows by \n. Example: 'tree|tree|fire|empty\ntree|tree|tree|empty'

Source code in src/agentic_architectures/architectures/cellular_automata.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def run(self, task: str, **kwargs: Any) -> ArchitectureResult:
    """Run the simulation. `task` is the INITIAL GRID as a multi-line string,
    with cells separated by `|` and rows by `\\n`.
    Example: 'tree|tree|fire|empty\\ntree|tree|tree|empty'
    """
    # Parse initial grid
    rows = [row.split("|") for row in task.strip().split("\n")]
    # Validate dimensions
    if len(rows) != self.height:
        raise ValueError(f"Initial grid has {len(rows)} rows, expected {self.height}")
    if any(len(row) != self.width for row in rows):
        raise ValueError(f"Initial grid rows must have {self.width} columns")
    initial_grid: list[list[str]] = [[c.strip() for c in row] for row in rows]

    graph = self.build()
    config = {"recursion_limit": self.max_steps + 5}
    final_state = graph.invoke(
        {"grid": initial_grid, "step": 0, "history": [], "max_steps": self.max_steps},
        config=config,
    )

    history = final_state.get("history", []) + [final_state["grid"]]

    # Count state-label changes across history
    from collections import Counter

    per_step_counts = []
    for step_grid in history:
        counter: Counter[str] = Counter()
        for row in step_grid:
            for cell in row:
                counter[cell] += 1
        per_step_counts.append(dict(counter))

    return ArchitectureResult(
        output="\n\n".join(f"Step {i}:\n" + "\n".join("|".join(row) for row in g) for i, g in enumerate(history)),
        state={
            "final_grid": final_state["grid"],
            "history": history,
            "per_step_counts": per_step_counts,
        },
        trace=[{"type": "grid", "step": i, "grid": g} for i, g in enumerate(history)],
        metadata={
            "steps_run": len(history) - 1,
            "grid_size": f"{self.height}x{self.width}",
            "per_step_counts": per_step_counts,
            "allowed_states": self.allowed_states,
        },
    )

ReflexiveMetacognitive 

ReflexiveMetacognitive(self_model: str | None = None, capability_threshold: int = 3, **kwargs: Any)

Bases: Architecture

Self-aware agent that picks one of four routes per task.

Source code in src/agentic_architectures/architectures/reflexive_metacognitive.py 
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self,
    self_model: str | None = None,
    capability_threshold: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.self_model = self_model or DEFAULT_SELF_MODEL
    self.capability_threshold = capability_threshold
    self._decider = self.llm.with_structured_output(_MetaDecision)

Advanced patterns

Reflexion 

Reflexion(max_trials: int = 3, reflections_to_recall: int = 3, evaluator: Callable[[str, str], dict[str, Any]] | None = None, reflection_llm: BaseChatModel | None = None, episodic: EpisodicMemory | None = None, **kwargs: Any)

Bases: Architecture

Try → evaluate → verbal self-reflection → store → retry, with memory that persists across :meth:run calls on the same instance.

Parameters:

Name Type Description Default
max_trials int

hard cap on attempt → evaluate iterations per task.

 3
reflections_to_recall int

top-k past reflections to prepend on each new attempt's prompt.

 3
evaluator Callable[[str, str], dict[str, Any]] | None

(candidate, task) -> features dict. The dict MUST contain a passed: bool key — that's the deciding signal. Defaults to :func:default_haiku_checker.

 None
reflection_llm BaseChatModel | None

optionally use a stronger model for the reflection step (defaults to the same LLM as the agent).

 None
episodic EpisodicMemory | None

optionally inject a pre-existing :class:EpisodicMemory (for sharing lessons across multiple Reflexion instances).

 None
Source code in src/agentic_architectures/architectures/reflexion.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def __init__(
    self,
    max_trials: int = 3,
    reflections_to_recall: int = 3,
    evaluator: Callable[[str, str], dict[str, Any]] | None = None,
    reflection_llm: BaseChatModel | None = None,
    episodic: EpisodicMemory | None = None,
    **kwargs: Any,
) -> None:
    """
    Args:
        max_trials: hard cap on attempt → evaluate iterations per task.
        reflections_to_recall: top-k past reflections to prepend on each
            new attempt's prompt.
        evaluator: ``(candidate, task) -> features dict``. The dict MUST
            contain a ``passed: bool`` key — that's the deciding signal.
            Defaults to :func:`default_haiku_checker`.
        reflection_llm: optionally use a stronger model for the reflection
            step (defaults to the same LLM as the agent).
        episodic: optionally inject a pre-existing :class:`EpisodicMemory`
            (for sharing lessons across multiple Reflexion instances).
    """
    super().__init__(**kwargs)
    self.max_trials = max_trials
    self.reflections_to_recall = reflections_to_recall
    self.evaluator = evaluator if evaluator is not None else default_haiku_checker
    # Persistent across run() calls — the whole point of Reflexion.
    self.episodic = (
        episodic
        if episodic is not None
        else EpisodicMemory(
            collection_name="reflexion_lessons",
        )
    )
    self._reflector = (reflection_llm or self.llm).with_structured_output(_SelfReflection)

SelfDiscover 

SelfDiscover(modules: list[str] | None = None, **kwargs: Any)

Bases: Architecture

SELECT → ADAPT → IMPLEMENT → SOLVE — agent composes its reasoning recipe.

Parameters:

Name Type Description Default
modules list[str] | None

optional custom module library (replaces the default 16).

 None
Source code in src/agentic_architectures/architectures/self_discover.py 
163
164
165
166
167
168
169
170
171
172
173
174
def __init__(self, modules: list[str] | None = None, **kwargs: Any) -> None:
    """
    Args:
        modules: optional custom module library (replaces the default 16).
    """
    super().__init__(**kwargs)
    self.modules: list[str] = list(modules) if modules else list(MODULE_LIBRARY)
    # Pre-bind structured-output wrappers once per architecture instance.
    self._selector = self.llm.with_structured_output(_SelectedModules)
    self._adapter = self.llm.with_structured_output(_AdaptedModules)
    self._implementer = self.llm.with_structured_output(_ReasoningPlan)
    self._solver = self.llm.with_structured_output(_Solution)

ChainOfVerification 

ChainOfVerification(**kwargs: Any)

Bases: Architecture

BASELINE → PLAN questions → EXECUTE answers → REVISE.

Source code in src/agentic_architectures/architectures/chain_of_verification.py 
105
106
107
108
109
def __init__(self, **kwargs: Any) -> None:
    super().__init__(**kwargs)
    self._planner = self.llm.with_structured_output(_VerificationQuestions)
    self._executor = self.llm.with_structured_output(_VerificationAnswer)
    self._reviser = self.llm.with_structured_output(_RevisedResponse)

SelfConsistency 

SelfConsistency(n_samples: int = 5, sample_temperature: float = 0.8, **kwargs: Any)

Bases: Architecture

Sample N CoT paths, majority-vote the modal final answer.

Parameters:

Name Type Description Default
n_samples int

how many independent reasoning paths to draw.

 5
sample_temperature float

temperature for the sampling LLM. Higher = more diverse paths (and more likely to surface the correct one via majority).

 0.8
Source code in src/agentic_architectures/architectures/self_consistency.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(
    self,
    n_samples: int = 5,
    sample_temperature: float = 0.8,
    **kwargs: Any,
) -> None:
    """
    Args:
        n_samples: how many independent reasoning paths to draw.
        sample_temperature: temperature for the sampling LLM. Higher = more diverse
            paths (and more likely to surface the correct one via majority).
    """
    super().__init__(**kwargs)
    self.n_samples = n_samples
    self.sample_temperature = sample_temperature
    # Use the LLM with structured output to enforce the (reasoning, answer) shape.
    # We DON'T rebind temperature here — let the user inject an LLM at desired temp;
    # we override per-call below via .bind() so the sample-LLM is always warm.
    self._sampler_schema = _ReasoningSample

LATS 

LATS(max_iterations: int = 6, branching: int = 3, ucb_c: float = 1.4, max_depth: int = 5, **kwargs: Any)

Bases: Architecture

MCTS-style tree search over LLM-generated thoughts with deterministic-picker reward.

Source code in src/agentic_architectures/architectures/lats.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    max_iterations: int = 6,
    branching: int = 3,
    ucb_c: float = 1.4,
    max_depth: int = 5,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.max_iterations = max_iterations
    self.branching = branching
    self.ucb_c = ucb_c
    self.max_depth = max_depth
    self._expander = self.llm.with_structured_output(_ThoughtCandidates)
    self._evaluator = self.llm.with_structured_output(_LeafEvaluation)

AgenticRAG 

AgenticRAG(documents: list[str] | None = None, vector_memory: VectorMemory | None = None, max_iterations: int = 4, top_k: int = 3, **kwargs: Any)

Bases: Architecture

ReAct-style agent with a single retrieve-tool.

Parameters:

Name Type Description Default
documents list[str] | None

optional list of source documents to seed the corpus.

 None
vector_memory VectorMemory | None

a pre-built VectorMemory; takes precedence over documents.

 None
max_iterations int

hard cap on decide → retrieve cycles.

 4
top_k int

number of documents to return per retrieve call.

 3
Source code in src/agentic_architectures/architectures/agentic_rag.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
    self,
    documents: list[str] | None = None,
    vector_memory: VectorMemory | None = None,
    max_iterations: int = 4,
    top_k: int = 3,
    **kwargs: Any,
) -> None:
    """
    Args:
        documents: optional list of source documents to seed the corpus.
        vector_memory: a pre-built VectorMemory; takes precedence over `documents`.
        max_iterations: hard cap on decide → retrieve cycles.
        top_k: number of documents to return per retrieve call.
    """
    super().__init__(**kwargs)
    self.max_iterations = max_iterations
    self.top_k = top_k
    if vector_memory is not None:
        self.memory = vector_memory
    else:
        self.memory = VectorMemory(collection_name="agentic_rag_corpus")
        if documents:
            from langchain_core.documents import Document

            self.memory.add([Document(page_content=d) for d in documents])
    self._decider = self.llm.with_structured_output(_AgentDecision)

CorrectiveRAG 

CorrectiveRAG(documents: list[str] | None = None, vector_memory: VectorMemory | None = None, web_search_fn: Callable[[str], list[str]] | None = None, top_k: int = 3, relevance_threshold: float = 0.5, **kwargs: Any)

Bases: Architecture

Retrieve → grade per-doc → route → answer.

Parameters:

Name Type Description Default
documents list[str] | None

optional list to seed a fresh VectorMemory.

 None
vector_memory VectorMemory | None

pre-built memory; takes precedence over documents.

 None
web_search_fn Callable[[str], list[str]] | None

callable (query) -> list[str] returning web snippets. If None, web fallback returns an empty list.

 None
top_k int

how many docs to retrieve.

 3
relevance_threshold float

fraction of docs that must be 'relevant' to route directly to use_retrieved (else fall back / mix).

 0.5
Source code in src/agentic_architectures/architectures/corrective_rag.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    documents: list[str] | None = None,
    vector_memory: VectorMemory | None = None,
    web_search_fn: Callable[[str], list[str]] | None = None,
    top_k: int = 3,
    relevance_threshold: float = 0.5,
    **kwargs: Any,
) -> None:
    """
    Args:
        documents: optional list to seed a fresh VectorMemory.
        vector_memory: pre-built memory; takes precedence over `documents`.
        web_search_fn: callable `(query) -> list[str]` returning web snippets.
            If None, web fallback returns an empty list.
        top_k: how many docs to retrieve.
        relevance_threshold: fraction of docs that must be 'relevant' to
            route directly to use_retrieved (else fall back / mix).
    """
    super().__init__(**kwargs)
    self.top_k = top_k
    self.relevance_threshold = relevance_threshold
    self.web_search_fn = web_search_fn
    if vector_memory is not None:
        self.memory = vector_memory
    else:
        self.memory = VectorMemory(collection_name="corrective_rag_corpus")
        if documents:
            from langchain_core.documents import Document

            self.memory.add([Document(page_content=d) for d in documents])
    self._grader = self.llm.with_structured_output(_DocGrade)

SelfRAG 

SelfRAG(documents: list[str] | None = None, vector_memory: VectorMemory | None = None, top_k: int = 4, **kwargs: Any)

Bases: Architecture

Retrieve → reflection tokens per doc → Python keeps usable docs → answer.

Source code in src/agentic_architectures/architectures/self_rag.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __init__(
    self,
    documents: list[str] | None = None,
    vector_memory: VectorMemory | None = None,
    top_k: int = 4,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.top_k = top_k
    if vector_memory is not None:
        self.memory = vector_memory
    else:
        self.memory = VectorMemory(collection_name="self_rag_corpus")
        if documents:
            from langchain_core.documents import Document

            self.memory.add([Document(page_content=d) for d in documents])
    self._decider = self.llm.with_structured_output(_RetrieveDecision)
    self._reflector = self.llm.with_structured_output(_ReflectionTokens)

AdaptiveRAG 

AdaptiveRAG(documents: list[str] | None = None, vector_memory: VectorMemory | None = None, top_k: int = 3, multi_step_max_iterations: int = 3, **kwargs: Any)

Bases: Architecture

Pre-route by complexity → execute the matched RAG strategy.

Source code in src/agentic_architectures/architectures/adaptive_rag.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    documents: list[str] | None = None,
    vector_memory: VectorMemory | None = None,
    top_k: int = 3,
    multi_step_max_iterations: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.top_k = top_k
    self.multi_step_max_iterations = multi_step_max_iterations
    if vector_memory is not None:
        self.memory = vector_memory
    else:
        self.memory = VectorMemory(collection_name="adaptive_rag_corpus")
        if documents:
            from langchain_core.documents import Document

            self.memory.add([Document(page_content=d) for d in documents])
    self._classifier = self.llm.with_structured_output(_ComplexityClass)

GraphRAG 

GraphRAG(documents: list[str] | None = None, semantic_memory: SemanticMemory | None = None, max_communities: int = 6, **kwargs: Any)

Bases: Architecture

Build knowledge graph + community summaries; route queries local vs global.

Source code in src/agentic_architectures/architectures/graph_rag.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(
    self,
    documents: list[str] | None = None,
    semantic_memory: SemanticMemory | None = None,
    max_communities: int = 6,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.max_communities = max_communities
    self.semantic = semantic_memory if semantic_memory is not None else SemanticMemory()
    self._extractor = self.llm.with_structured_output(_IngestionResult)
    self._classifier = self.llm.with_structured_output(_QuestionScope)
    self.documents = documents or []
    self.communities: list[list[str]] = []
    self.community_summaries: list[str] = []
    if self.documents:
        self._ingest(self.documents)
        self._build_communities()

Debate 

Debate(n_agents: int = 3, n_rounds: int = 2, agent_personas: list[str] | None = None, sample_temperature: float = 0.7, **kwargs: Any)

Bases: Architecture

N agents × K rounds debate → majority vote on final answers.

Source code in src/agentic_architectures/architectures/debate.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __init__(
    self,
    n_agents: int = 3,
    n_rounds: int = 2,
    agent_personas: list[str] | None = None,
    sample_temperature: float = 0.7,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if n_agents < 2:
        raise ValueError("Debate needs at least 2 agents.")
    self.n_agents = n_agents
    self.n_rounds = n_rounds
    self.sample_temperature = sample_temperature
    self.personas = list(agent_personas) if agent_personas else list(self.AGENT_PERSONAS[:n_agents])
    if len(self.personas) < n_agents:
        # Pad with generic personas if user supplied too few
        self.personas += [
            f"You are Agent {chr(65 + i)}: thoughtful, careful, honest."
            for i in range(len(self.personas), n_agents)
        ]
    # Pre-bind structured-output sampler
    self._responder = self.llm.bind(temperature=self.sample_temperature).with_structured_output(_DebateResponse)

Voyager 

Voyager(**kwargs: Any)

Bases: Architecture

Persistent skill library — retrieve, reuse, or write new skills.

Source code in src/agentic_architectures/architectures/voyager.py 
132
133
134
135
136
137
138
def __init__(self, **kwargs: Any) -> None:
    super().__init__(**kwargs)
    self.skills: list[dict[str, str]] = []
    self._index = VectorMemory(collection_name="voyager_skills")
    self._decider = self.llm.with_structured_output(_SkillDecision)
    self._writer = self.llm.with_structured_output(_NewSkillSpec)
    self._applier = self.llm.with_structured_output(_ApplySkill)

STORM 

STORM(n_perspectives: int = 3, questions_per_perspective: int = 2, web_search_fn: Callable[[str], list[str]] | None = None, **kwargs: Any)

Bases: Architecture

Multi-perspective research → outline → article.

Source code in src/agentic_architectures/architectures/storm.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    n_perspectives: int = 3,
    questions_per_perspective: int = 2,
    web_search_fn: Callable[[str], list[str]] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.n_perspectives = n_perspectives
    self.questions_per_perspective = questions_per_perspective
    self.web_search_fn = web_search_fn
    self._perspectives = self.llm.with_structured_output(_Perspectives)
    self._questions = self.llm.with_structured_output(_Questions)
    self._outline = self.llm.with_structured_output(_Outline)
    self._writer = self.llm.with_structured_output(_ArticleSection)

MemGPT 

MemGPT(context_limit: int = 4, max_iterations: int = 5, archival: VectorMemory | None = None, **kwargs: Any)

Bases: Architecture

OS-style tiered memory: context (RAM) + archival (disk).

Source code in src/agentic_architectures/architectures/memgpt.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    context_limit: int = 4,
    max_iterations: int = 5,
    archival: VectorMemory | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.context_limit = context_limit
    self.max_iterations = max_iterations
    # Persistent across run() calls
    self.context_tier: list[str] = []
    self.archival: VectorMemory = archival or VectorMemory(collection_name="memgpt_archival")
    self.archival_count = 0
    self._decider = self.llm.with_structured_output(_MemAction)

ConstitutionalAI 

ConstitutionalAI(constitution: list[str] | None = None, max_iterations: int = 3, **kwargs: Any)

Bases: Architecture

Generate → critique against constitution → revise → loop until all pass.

Source code in src/agentic_architectures/architectures/constitutional_ai.py 
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    constitution: list[str] | None = None,
    max_iterations: int = 3,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.constitution: list[str] = list(constitution) if constitution else list(DEFAULT_CONSTITUTION)
    self.max_iterations = max_iterations
    self._critique = self.llm.with_structured_output(_CritiqueResult)

SWEAgent 

SWEAgent(working_dir: str | Path, max_iterations: int = 8, **kwargs: Any)

Bases: Architecture

Code-repo agent with sandboxed FS tools (list / read / write / run / answer).

Source code in src/agentic_architectures/architectures/swe_agent.py 
65
66
67
68
69
70
71
72
73
74
75
def __init__(
    self,
    working_dir: str | Path,
    max_iterations: int = 8,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.working_dir = Path(working_dir).resolve()
    self.working_dir.mkdir(parents=True, exist_ok=True)
    self.max_iterations = max_iterations
    self._decider = self.llm.with_structured_output(_SWEAction)

ComputerUse 

ComputerUse(initial_screen: dict[str, Any] | None = None, sensitive_patterns: list[str] | None = None, blocked_domains: list[str] | None = None, max_iterations: int = 6, **kwargs: Any)

Bases: Architecture

Mock-environment Computer-Use agent with hard safety gate.

Source code in src/agentic_architectures/architectures/computer_use.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    initial_screen: dict[str, Any] | None = None,
    sensitive_patterns: list[str] | None = None,
    blocked_domains: list[str] | None = None,
    max_iterations: int = 6,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.initial_screen = (
        dict(initial_screen)
        if initial_screen
        else {
            "url": "about:blank",
            "elements": [],
            "fields": {},
            "submitted": False,
        }
    )
    self.sensitive_patterns = [p.lower() for p in (sensitive_patterns or DEFAULT_SENSITIVE_PATTERNS)]
    self.blocked_domains = blocked_domains or []
    self.max_iterations = max_iterations
    self._decider = self.llm.with_structured_output(_ComputerAction)

BrowserAgent 

BrowserAgent(max_iterations: int = 6, headless: bool = True, page_text_chars: int = 2500, sensitive_patterns: list[str] | None = None, blocked_domains: list[str] | None = None, **kwargs: Any)

Bases: Architecture

Real headless-Chromium agent via Playwright, with Python safety gate.

Source code in src/agentic_architectures/architectures/browser_agent.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def __init__(
    self,
    max_iterations: int = 6,
    headless: bool = True,
    page_text_chars: int = 2500,
    sensitive_patterns: list[str] | None = None,
    blocked_domains: list[str] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.max_iterations = max_iterations
    self.headless = headless
    self.page_text_chars = page_text_chars
    self.sensitive_patterns = [p.lower() for p in (sensitive_patterns or DEFAULT_SENSITIVE_PATTERNS)]
    self.blocked_domains = blocked_domains or list(DEFAULT_BLOCKED_DOMAINS)
    self._decider = self.llm.with_structured_output(_BrowserAction)
    # Playwright is lazily opened on first navigate
    self._pw = None
    self._browser = None
    self._page = None

close 

close() -> None

Shut down the Playwright browser. Always call after run() is done.

Source code in src/agentic_architectures/architectures/browser_agent.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def close(self) -> None:
    """Shut down the Playwright browser. Always call after `run()` is done."""
    if self._page is not None:
        try:
            self._page.close()
        except Exception:
            pass
        self._page = None
    if self._browser is not None:
        try:
            self._browser.close()
        except Exception:
            pass
        self._browser = None
    if self._pw is not None:
        try:
            self._pw.stop()
        except Exception:
            pass
        self._pw = None

AgentWorkflowMemory 

AgentWorkflowMemory(**kwargs: Any)

Bases: Architecture

Retrieve workflow → use as prior → answer → extract new workflow.

Source code in src/agentic_architectures/architectures/agent_workflow_memory.py 
62
63
64
65
66
67
def __init__(self, **kwargs: Any) -> None:
    super().__init__(**kwargs)
    self.workflows: list[dict[str, Any]] = []
    self._index = VectorMemory(collection_name="awm_workflows")
    self._extractor = self.llm.with_structured_output(_Workflow)
    self._answerer = self.llm.with_structured_output(_Answer)