Skip to content

Plan act reflect

Classes:

Name Description
PlanActReflectAgent

Plan -> Act -> Reflect looping agent.

Attributes:

Name Type Description
StateT

Attributes

StateT module-attribute 

StateT = TypeVar('StateT', bound=AgentState)

Classes

PlanActReflectAgent

Plan -> Act -> Reflect looping agent.

Pattern

PLAN -> (condition) -> ACT | REFLECT | END ACT -> REFLECT REFLECT -> PLAN

Default condition (_should_act): - If last assistant message contains tool calls -> ACT - If last message is from a tool -> REFLECT - Else -> END

Provide a custom condition to override this heuristic and implement
  • Budget / depth limiting
  • Confidence-based early stop
  • Dynamic branch selection (e.g., different tool nodes)

Parameters (constructor): state: Optional initial state instance context_manager: Custom context manager publisher: Optional publisher for streaming / events id_generator: ID generation strategy container: InjectQ DI container

compile(...) arguments: plan_node: Callable (state -> state). Produces next thought / tool requests tool_node: ToolNode executing declared tools reflect_node: Callable (state -> state). Consumes tool results & may adjust plan condition: Optional Callable[[AgentState], str] returning next node name or END checkpointer/store/interrupt_before/interrupt_after/callback_manager: Standard graph compilation options

Returns:

Type Description

CompiledGraph ready for invoke / ainvoke.
Notes
  • Node names can be customized via (callable, "NAME") tuples.
  • condition must return one of: tool_node_name, reflect_node_name, END.

Methods:

Name Description
__init__
compile

Compile the Plan-Act-Reflect loop.
Source code in pyagenity/prebuilt/agent/plan_act_reflect.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 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
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
148
149
150
151
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
191
192
193
class PlanActReflectAgent[StateT: AgentState]:
    """Plan -> Act -> Reflect looping agent.

    Pattern:
        PLAN -> (condition) -> ACT | REFLECT | END
        ACT -> REFLECT
        REFLECT -> PLAN

    Default condition (_should_act):
        - If last assistant message contains tool calls -> ACT
        - If last message is from a tool -> REFLECT
        - Else -> END

    Provide a custom condition to override this heuristic and implement:
        * Budget / depth limiting
        * Confidence-based early stop
        * Dynamic branch selection (e.g., different tool nodes)

    Parameters (constructor):
        state: Optional initial state instance
        context_manager: Custom context manager
        publisher: Optional publisher for streaming / events
        id_generator: ID generation strategy
        container: InjectQ DI container

    compile(...) arguments:
        plan_node: Callable (state -> state). Produces next thought / tool requests
        tool_node: ToolNode executing declared tools
        reflect_node: Callable (state -> state). Consumes tool results & may adjust plan
        condition: Optional Callable[[AgentState], str] returning next node name or END
        checkpointer/store/interrupt_before/interrupt_after/callback_manager:
            Standard graph compilation options

    Returns:
        CompiledGraph ready for invoke / ainvoke.

    Notes:
        - Node names can be customized via (callable, "NAME") tuples.
        - condition must return one of: tool_node_name, reflect_node_name, END.
    """

    def __init__(
        self,
        state: StateT | None = None,
        context_manager: BaseContextManager[StateT] | None = None,
        publisher: BasePublisher | None = None,
        id_generator: BaseIDGenerator = DefaultIDGenerator(),
        container: InjectQ | None = None,
    ):
        self._graph = StateGraph[StateT](
            state=state,
            context_manager=context_manager,
            publisher=publisher,
            id_generator=id_generator,
            container=container,
        )

    def compile(
        self,
        plan_node: Callable | tuple[Callable, str],
        tool_node: ToolNode | tuple[ToolNode, str],
        reflect_node: Callable | tuple[Callable, str],
        *,
        condition: Callable[[AgentState], str] | None = None,
        checkpointer: BaseCheckpointer[StateT] | None = None,
        store: BaseStore | None = None,
        interrupt_before: list[str] | None = None,
        interrupt_after: list[str] | None = None,
        callback_manager: CallbackManager = CallbackManager(),
    ) -> CompiledGraph:
        """Compile the Plan-Act-Reflect loop.

        Args:
            plan_node: Callable or (callable, name)
            tool_node: ToolNode or (ToolNode, name)
            reflect_node: Callable or (callable, name)
            condition: Optional decision function. Defaults to internal heuristic.
            checkpointer/store/interrupt_* / callback_manager: Standard graph options.

        Returns:
            CompiledGraph
        """
        # PLAN
        if isinstance(plan_node, tuple):
            plan_func, plan_name = plan_node
            if not callable(plan_func):
                raise ValueError("plan_node[0] must be callable")
        else:
            plan_func = plan_node
            plan_name = "PLAN"
            if not callable(plan_func):
                raise ValueError("plan_node must be callable")

        # ACT
        if isinstance(tool_node, tuple):
            tool_func, tool_name = tool_node
            if not isinstance(tool_func, ToolNode):
                raise ValueError("tool_node[0] must be a ToolNode")
        else:
            tool_func = tool_node
            tool_name = "ACT"
            if not isinstance(tool_func, ToolNode):
                raise ValueError("tool_node must be a ToolNode")

        # REFLECT
        if isinstance(reflect_node, tuple):
            reflect_func, reflect_name = reflect_node
            if not callable(reflect_func):
                raise ValueError("reflect_node[0] must be callable")
        else:
            reflect_func = reflect_node
            reflect_name = "REFLECT"
            if not callable(reflect_func):
                raise ValueError("reflect_node must be callable")

        # Register nodes
        self._graph.add_node(plan_name, plan_func)
        self._graph.add_node(tool_name, tool_func)
        self._graph.add_node(reflect_name, reflect_func)

        # Decision
        decision_fn = condition or _should_act
        self._graph.add_conditional_edges(
            plan_name,
            decision_fn,
            {tool_name: tool_name, reflect_name: reflect_name, END: END},
        )

        # Loop edges
        self._graph.add_edge(tool_name, reflect_name)
        self._graph.add_edge(reflect_name, plan_name)

        # Entry
        self._graph.set_entry_point(plan_name)

        return self._graph.compile(
            checkpointer=checkpointer,
            store=store,
            interrupt_before=interrupt_before,
            interrupt_after=interrupt_after,
            callback_manager=callback_manager,
        )

Functions

__init__ 
__init__(state=None, context_manager=None, publisher=None, id_generator=DefaultIDGenerator(), container=None)
Source code in pyagenity/prebuilt/agent/plan_act_reflect.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    state: StateT | None = None,
    context_manager: BaseContextManager[StateT] | None = None,
    publisher: BasePublisher | None = None,
    id_generator: BaseIDGenerator = DefaultIDGenerator(),
    container: InjectQ | None = None,
):
    self._graph = StateGraph[StateT](
        state=state,
        context_manager=context_manager,
        publisher=publisher,
        id_generator=id_generator,
        container=container,
    )
compile 
compile(plan_node, tool_node, reflect_node, *, condition=None, checkpointer=None, store=None, interrupt_before=None, interrupt_after=None, callback_manager=CallbackManager())

Compile the Plan-Act-Reflect loop.

Parameters:

Name Type Description Default
plan_node
Callable | tuple[Callable, str]

Callable or (callable, name)

 required
tool_node
ToolNode | tuple[ToolNode, str]

ToolNode or (ToolNode, name)

 required
reflect_node
Callable | tuple[Callable, str]

Callable or (callable, name)

 required
condition
Callable[[AgentState], str] | None

Optional decision function. Defaults to internal heuristic.

 None
checkpointer/store/interrupt_* / callback_manager

Standard graph options.

 required

Returns:

Type Description
CompiledGraph

CompiledGraph
Source code in pyagenity/prebuilt/agent/plan_act_reflect.py 
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
148
149
150
151
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
191
192
193
def compile(
    self,
    plan_node: Callable | tuple[Callable, str],
    tool_node: ToolNode | tuple[ToolNode, str],
    reflect_node: Callable | tuple[Callable, str],
    *,
    condition: Callable[[AgentState], str] | None = None,
    checkpointer: BaseCheckpointer[StateT] | None = None,
    store: BaseStore | None = None,
    interrupt_before: list[str] | None = None,
    interrupt_after: list[str] | None = None,
    callback_manager: CallbackManager = CallbackManager(),
) -> CompiledGraph:
    """Compile the Plan-Act-Reflect loop.

    Args:
        plan_node: Callable or (callable, name)
        tool_node: ToolNode or (ToolNode, name)
        reflect_node: Callable or (callable, name)
        condition: Optional decision function. Defaults to internal heuristic.
        checkpointer/store/interrupt_* / callback_manager: Standard graph options.

    Returns:
        CompiledGraph
    """
    # PLAN
    if isinstance(plan_node, tuple):
        plan_func, plan_name = plan_node
        if not callable(plan_func):
            raise ValueError("plan_node[0] must be callable")
    else:
        plan_func = plan_node
        plan_name = "PLAN"
        if not callable(plan_func):
            raise ValueError("plan_node must be callable")

    # ACT
    if isinstance(tool_node, tuple):
        tool_func, tool_name = tool_node
        if not isinstance(tool_func, ToolNode):
            raise ValueError("tool_node[0] must be a ToolNode")
    else:
        tool_func = tool_node
        tool_name = "ACT"
        if not isinstance(tool_func, ToolNode):
            raise ValueError("tool_node must be a ToolNode")

    # REFLECT
    if isinstance(reflect_node, tuple):
        reflect_func, reflect_name = reflect_node
        if not callable(reflect_func):
            raise ValueError("reflect_node[0] must be callable")
    else:
        reflect_func = reflect_node
        reflect_name = "REFLECT"
        if not callable(reflect_func):
            raise ValueError("reflect_node must be callable")

    # Register nodes
    self._graph.add_node(plan_name, plan_func)
    self._graph.add_node(tool_name, tool_func)
    self._graph.add_node(reflect_name, reflect_func)

    # Decision
    decision_fn = condition or _should_act
    self._graph.add_conditional_edges(
        plan_name,
        decision_fn,
        {tool_name: tool_name, reflect_name: reflect_name, END: END},
    )

    # Loop edges
    self._graph.add_edge(tool_name, reflect_name)
    self._graph.add_edge(reflect_name, plan_name)

    # Entry
    self._graph.set_entry_point(plan_name)

    return self._graph.compile(
        checkpointer=checkpointer,
        store=store,
        interrupt_before=interrupt_before,
        interrupt_after=interrupt_after,
        callback_manager=callback_manager,
    )