PlanActReflectAgent

PlanActReflectAgent is a self-improving agent that breaks tasks into steps, executes them, then explicitly evaluates whether the work is complete. If not, it iterates. This pattern solves a common problem in agent workflows: standard ReAct loops can get stuck or produce incomplete answers because they lack a clear evaluation step. By adding a dedicated critic that inspects all work and decides whether to continue or stop, this agent achieves higher task completion rates on complex assignments like research, analysis, and planning.

Import path: from tenxgraph.prebuilt.agent import PlanActReflectAgent

Install: pip install "10xgraph[openai]" (or [google-genai], [anthropic] for other providers)

When to use this agent

PlanActReflectAgent is best for tasks that require iterative refinement or verification. Use it when:

• The task has multiple steps. A plan phase ensures coherent structure before executing.
• Quality verification matters. The reflect phase catches incomplete or incorrect work before finishing.
• Tools are optional but helpful. It works equally well with or without external tools. With tools, the planner calls them when needed; without them, it performs pure reasoning.
• Iteration is acceptable. The agent may loop several times (bounded by max_iterations). If latency is critical, prefer a simpler agent like ReactAgent.
• You want explicit decision points. Unlike ReAct, which continues calling tools until it thinks it is done, PlanActReflect always evaluates completion against the full context.

Avoid this agent if you need single-shot responses or if the task is simple enough for a basic agent. For retrieval-heavy workloads, prefer RAGAgent. For team-based multi-agent orchestration, use SwarmAgent or SupervisorTeamAgent.

How it works

The agent runs three specialized LLM instances in sequence: a planner, an actor (tool executor), and a reflector.

The graph topology

flowchart TD
    START([START]) --> PLAN

    PLAN["PLAN\n(LLM + tools)"]
    ACT["ACT\n(ToolNode)"]
    REFLECT["REFLECT\n(LLM, no tools)"]
    INC["INCREMENT_ITERATIONS"]
    END_NODE([END])

    PLAN -- "has tool calls" --> ACT
    PLAN -- "no tool calls" --> REFLECT
    ACT --> REFLECT
    REFLECT -- "[DONE] or max_iterations reached" --> END_NODE
    REFLECT -- "not done" --> INC
    INC --> PLAN

Three separate Agent instances execute inside the graph, each with its own system prompt:

Routing decisions

At PLAN: If the planner emits tool calls, the graph routes to ACT; otherwise it routes directly to REFLECT. This branch lets the agent skip unnecessary tool execution when it can reason toward progress without external data.

At REFLECT: Two exit conditions:

1. The reflector’s output contains [DONE] (case-insensitive): signals successful completion.
2. The iteration counter reaches max_iterations: hard cap to prevent runaway loops.

If neither condition is met, the counter increments and the graph loops back to PLAN.

The reflect filter

To prevent long tool outputs from overflowing context, tool result messages (messages with role="tool") are hidden from the reflector. The planner still sees the full context on the next iteration, so information is not lost, only filtered from the critic’s view.

Default system prompts

Each phase has a built-in prompt that can be overridden (paraphrased here; see DEFAULT_PLAN_SYSTEM_PROMPT and DEFAULT_REFLECT_SYSTEM_PROMPT in tenxgraph.prebuilt.agent.plan_act_reflect for the exact text):

PLAN, “You are a strategic planner. Break the user’s task into clear, actionable steps and make progress toward it. When a step requires external information, call the appropriate tools. When you can make progress without tools, provide your analysis. Be concise.”

REFLECT, “You are a critical evaluator. Review all work done so far and decide whether the task is fully complete. If complete, provide a concise summary and end with [DONE]. If not complete, identify gaps and give clear guidance for the next planning step.”

Both are fully overridable via plan_system_prompt and reflect_system_prompt parameters.

No-tools mode

When called without tools, PLAN always routes directly to REFLECT, skipping the ACT node. The agent becomes a pure reasoning loop: plan → reflect → decide → iterate. This is useful for complex multi-step reasoning tasks that do not require external data.

Constructor parameters

Create a PlanActReflectAgent by passing a model and optionally tools and configuration:

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    tools=[google_web_search, fetch_url],
    max_iterations=4,
)
app = agent.compile()

Required parameters:

Optional parameters:

compile() parameters

Once you have a PlanActReflectAgent instance, call .compile() to create a runnable graph:

app = agent.compile(
    checkpointer=PgCheckpointer(...),
    interrupt_before=["REFLECT"],
)

Examples

Research task with tools

This example demonstrates the full plan-act-reflect loop on a research task. The agent breaks the task into steps, searches for information, and iterates until it has a complete answer.

import asyncio
from dotenv import load_dotenv
from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.prebuilt.tools import fetch_url, google_web_search
from tenxgraph.core.state import Message

load_dotenv()

def summarize_findings(text: str) -> str:
    """Compress a long text to key points."""
    return text[:2000] + "..." if len(text) > 2000 else text

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    tools=[fetch_url, google_web_search, summarize_findings],
    max_iterations=4,
)

app = agent.compile()

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message(
            "Research the current state of fusion energy and write a 3-paragraph summary."
        )]},
        config={"thread_id": "research-1"},
    )
    print(result["messages"][-1].text())

asyncio.run(main())

Custom system prompts

Override the default prompts to steer the agent’s behavior. This is useful for domain-specific tasks or when you want precise instructions for planning and evaluation.

from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.prebuilt.tools import fetch_url, google_web_search

agent = PlanActReflectAgent(
    model="gpt-4o",
    provider="openai",
    tools=[fetch_url, google_web_search],
    max_iterations=5,
    plan_system_prompt=[{
        "role": "system",
        "content": "You are a systematic researcher. Break each task into numbered steps.",
    }],
    reflect_system_prompt=[{
        "role": "system",
        "content": (
            "Review the work done. Is the research complete and accurate? "
            "If yes, write a summary and end with [DONE]. "
            "If not, list exactly what is still missing."
        ),
    }],
)

Pure reasoning (no tools)

For complex reasoning tasks that do not require external data, run the agent without tools. The plan phase generates reasoning, then the reflect phase evaluates and iterates if needed.

import asyncio
from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.core.state import Message

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    max_iterations=3,
)

app = agent.compile()

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("Devise three approaches to reduce LLM hallucination.")]},
        config={"thread_id": "reason-1"},
    )
    print(result["messages"][-1].text())

asyncio.run(main())

Persistent conversations with a checkpointer

Add a checkpointer to persist state across runs. Pass the same thread_id in the config to resume a conversation or retrieve the previous result.

import asyncio
from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.prebuilt.tools import fetch_url, google_web_search
from tenxgraph.storage.checkpointer import PgCheckpointer
from tenxgraph.core.state import Message

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    tools=[fetch_url, google_web_search],
    max_iterations=4,
)

# Requires: pip install "10xgraph[pg_checkpoint]"
checkpointer = PgCheckpointer(
    postgres_dsn="postgresql://user:pass@localhost/db",
    redis_url="redis://localhost:6379",
)
app = agent.compile(checkpointer=checkpointer)

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("Research recent breakthroughs in solid-state batteries.")]},
        config={"thread_id": "user-42-research"},
    )
    print(result["messages"][-1].text())

asyncio.run(main())

Using Google Gemini

PlanActReflectAgent works with any provider. Here is an example using Gemini with context trimming enabled to manage token usage on long conversations.

from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.prebuilt.tools import google_web_search

agent = PlanActReflectAgent(
    model="gemini-2.5-flash",
    provider="google",
    tools=[google_web_search],
    max_iterations=4,
    trim_context=True,
)

app = agent.compile()

Set your GOOGLE_API_KEY in the environment before running.

Streaming events

Stream execution events as they happen instead of waiting for the entire run to complete. Each event is printed as the agent plans, acts, and reflects.

import asyncio
from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.core.state import Message

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    max_iterations=3,
)
app = agent.compile()

async def main():
    async for event in app.astream(
        {"messages": [Message.text_message("Explain the trade-offs between RAG and fine-tuning.")]},
        config={"thread_id": "stream-1"},
    ):
        print(event)

asyncio.run(main())

Deploy with the API server and playground

To test your agent interactively or deploy it as a REST API, use the 10xgraph CLI.

graph.py, Define your agent:

from tenxgraph.prebuilt.agent import PlanActReflectAgent
from tenxgraph.prebuilt.tools import fetch_url, google_web_search, safe_calculator

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    provider="openai",
    tools=[fetch_url, google_web_search, safe_calculator],
    max_iterations=4,
)

app = agent.compile()

10xgraph.json, Point the CLI to your agent:

{
  "agent": "graph:app",
  "env": ".env",
  "auth": null,
  "checkpointer": null,
  "injectq": null,
  "store": null,
  "redis": null,
  "thread_name_generator": null
}

.env, Provide API keys:

OPENAI_API_KEY=sk-...
GOOGLE_API_KEY=AIza...

Launch the playground to test interactively:

10xgraph play

Launch the API server to serve the agent over HTTP:

10xgraph api

The server opens Swagger docs at http://localhost:8000/docs with full route reference.

Customization

Per-phase model selection

You can use different models for planning and reflection to balance cost and capability:

agent = PlanActReflectAgent(
    model="gpt-4o-mini",  # default
    plan_model="gpt-4o",  # more capable planner
    reflect_model="gpt-4o-mini",  # cheaper reflector
    tools=[...],
)

Extended reasoning for specific phases

Use plan_reasoning_config and reflect_reasoning_config to enable extended reasoning only where you need it:

agent = PlanActReflectAgent(
    model="gpt-4o",
    plan_reasoning_config={"effort": "high"},  # Enable thinking for planning
    reflect_reasoning_config=False,  # Skip for reflection to save tokens
    tools=[...],
)

Filter tools by tags

If you have a large set of tools, use tags to offer only a subset to the planner:

agent = PlanActReflectAgent(
    model="gpt-4o-mini",
    tools=[search_tool, fetch_tool, database_query_tool],
    tools_tags={"web", "research"},  # Only offer tools tagged with these
)

Human-in-the-loop approval

Pause before the REFLECT node so a human can review the planner’s proposed actions before evaluation:

app = agent.compile(interrupt_before=["REFLECT"])

# Run interactively
result = await app.ainvoke(
    {"messages": [Message.text_message("Research X")]},
    config={"thread_id": "t1"},
)
print(f"Plan state: {result['messages']}")

# Optionally add feedback, then resume
result = await app.ainvoke(
    {"messages": [Message.text_message("Also check Y")]},
    config={"thread_id": "t1"},
)

See also

• Full reference: /docs/reference/python/prebuilt-agents: comprehensive parameter tables.
• Other prebuilt agents: /docs/guides/prebuilt-agents: ReactAgent, RAGAgent, SwarmAgent, and more.
• Checkpointing: /docs/guides/set-up-checkpointing: persist state and resume conversations.
• Tools: /docs/guides/use-tool-decorator: define custom tools for your agent.
• Streaming: /docs/guides/stream-graph: handle partial results and token-by-token output.