Dynamic Planning
Dynamic planning creates the full plan on the fly during execution, as the agent learns more about the task. Each step's result informs what the next steps should be.
Static vs Dynamic Planning
Static planning: the full plan is created upfront before execution starts.
Best for: well-understood workflows with predictable steps.
Risk: if the plan is wrong, the agent may complete many wrong steps before detecting the problem.
Dynamic planning: the agent creates the next step based on the observation from the current step.
Best for: research tasks, exploratory workflows, and situations where the correct path is not known until execution starts.
Risk: the agent may take inefficient paths without a plan to guide it.
Hybrid: create a high-level plan upfront, then generate the specific steps within each phase dynamically. This combines the structure of static planning with the flexibility of dynamic planning.
Dynamic Planning Loop
# Dynamic planning: next step decided based on current result
class DynamicPlanner:
def __init__(self, tools: dict, max_steps: int = 15):
self.tools = tools
self.max_steps = max_steps
def decide_next_step(self, goal: str, history: list, last_result: dict) -> dict:
"""
In a real system, a language model decides the next step.
Here we use simple rules to simulate dynamic planning.
"""
step_count = len(history)
# No history yet — start with a search
if step_count == 0:
return {"action": "web_search",
"args": {"query": goal},
"description": "Initial research"}
# After search, extract key data
if step_count == 1 and last_result.get("status") == "success":
return {"action": "extract_data",
"args": {"text": last_result.get("result", "")},
"description": "Extract relevant information"}
# After extraction, summarise
if step_count == 2:
return {"action": "summarise",
"args": {"data": last_result.get("result", "")},
"description": "Summarise findings"}
# All steps done
return {"action": "FINISH",
"result": last_result.get("result", "Task complete")}
def run(self, goal: str) -> dict:
history = []
last_result = {}
for step_num in range(self.max_steps):
decision = self.decide_next_step(goal, history, last_result)
if decision["action"] == "FINISH":
return {"status": "complete", "steps": step_num,
"result": decision.get("result")}
tool_fn = self.tools.get(decision["action"])
if not tool_fn:
return {"status": "failed", "error": f"Tool not found: {decision['action']}"}
last_result = tool_fn(**decision["args"])
history.append({"step": step_num, **decision, "result": last_result})
print(f"Step {step_num+1}: {decision['description']}")
return {"status": "escalated", "reason": "max_steps_reached"}
tools = {
"web_search": lambda query: {"status": "success", "result": "Django FastAPI Flask"},
"extract_data": lambda text: {"status": "success", "result": ["Django", "FastAPI"]},
"summarise": lambda data: {"status": "success", "result": "Summary complete"},
}
planner = DynamicPlanner(tools)
result = planner.run("Top Python web frameworks")
print(f"Result: {result['status']} in {result['steps']} steps")