Course Content
Live Coding Interview Prep
7 sections · 50 lessons
Build a planning + execution agent workflow.
What you need to know
Plan-and-execute separates deciding from doing:
ReAct loop
- One model call per step
- Adapts after every observation
- Cost grows with steps
- Hard to review before it acts
Plan-and-execute
- One call to plan, one to answer
- Plan fixed before any result exists
- Cheaper, and steps can run in parallel
- The plan can be shown to a human first
The plan is a directed acyclic graph (DAG): steps are nodes, depends_on lists are edges. A step can run once all its dependencies are done. Running steps in such an order is a topological sort. If at some point no remaining step is runnable, the graph has a cycle (1 waits for 2, 2 waits for 1) or depends on a step that does not exist.
Results flow between steps with placeholders: "{{1}}" inside an argument means "the result of step 1".
1import json, re2from collections.abc import Callable34PLAN_PROMPT = (5 "Break the goal into at most 5 steps. Reply with JSON only:\n"6 '{"steps": [{"id": 1, "tool": "search", "args": {"query": "..."}, "depends_on": []}]}\n'7 'Write "{{2}}" inside args to use the result of step 2.\n\nGoal: '8)9REF = re.compile(r"\{\{(\d+)\}\}")1011def substitute(value, results: dict[int, object]):12 """Replace {{n}} placeholders, recursively, with step results."""13 if isinstance(value, str):14 return REF.sub(lambda m: str(results[int(m.group(1))]), value)15 if isinstance(value, dict):16 return {k: substitute(v, results) for k, v in value.items()}17 if isinstance(value, list):18 return [substitute(v, results) for v in value]19 return value2021def execute_plan(steps: list[dict], tools: dict[str, Callable]) -> dict[int, object]:22 """Run steps in dependency order; failures are recorded, not raised."""23 ids = {s["id"] for s in steps}24 for s in steps:25 if missing := set(s.get("depends_on", [])) - ids:26 raise ValueError(f"step {s['id']} depends on unknown steps {sorted(missing)}")27 results: dict[int, object] = {}28 remaining = list(steps)29 while remaining:30 runnable = [s for s in remaining if set(s.get("depends_on", [])) <= results.keys()]31 if not runnable:32 raise ValueError("plan has a dependency cycle")33 for step in runnable: # these are independent: could run in parallel34 fn = tools.get(step["tool"])35 try:36 if fn is None:37 raise KeyError(f"unknown tool {step['tool']}")38 results[step["id"]] = fn(**substitute(step.get("args", {}), results))39 except Exception as exc:40 results[step["id"]] = f"error: {type(exc).__name__}: {exc}"41 remaining.remove(step)42 return results4344def plan_and_execute(goal: str, llm_fn: Callable[[str], str], tools: dict) -> str:45 steps = json.loads(llm_fn(PLAN_PROMPT + goal))["steps"][:5]46 results = execute_plan(steps, tools)47 summary = "\n".join(f"Step {k}: {v}" for k, v in sorted(results.items()))48 return llm_fn(f"Goal: {goal}\n\nResults:\n{summary}\n\nWrite the final answer.")The tricky parts:
- Validate ids up front. A step that depends on step 9 when there is no step 9 would otherwise look exactly like a cycle; checking first gives a clear error.
results.keys()works as a set, soset(depends_on) <= results.keys()asks "are all dependencies done?".- Failed steps still produce a result (the error text), so dependent steps can run and the final answer can say what failed. A stricter version skips dependants of a failed step.
- Each
runnablebatch has no internal dependencies, which is why it could go to a thread pool.
Complexity: with V steps and E dependency edges, this simple loop scans the remaining steps once per round, and a straight chain needs V rounds, so it is O(V·(V + E)) in the worst case. Kahn's algorithm (tracking how many unfinished dependencies each step has) brings it to O(V + E). With 5 steps either is instant; saying which is which is what the interviewer is checking. Model calls: 2, plus 1 per replan.
A real-life example
Goal: "Total price of an iPhone 16 and AirPods in INR". The planner returns three steps, with step 3 depending on steps 1 and 2:
1PRICES = {"iphone 16": 79900, "airpods 4": 12900}2tools = {"price": lambda item: PRICES[item.lower()],3 "calculator": lambda expression: sum(int(x) for x in expression.split("+"))}4plan = [{"id": 1, "tool": "price", "args": {"item": "iPhone 16"}, "depends_on": []},5 {"id": 2, "tool": "price", "args": {"item": "AirPods 4"}, "depends_on": []},6 {"id": 3, "tool": "calculator", "args": {"expression": "{{1}}+{{2}}"}, "depends_on": [1, 2]}]7print(execute_plan(plan, tools)) # {1: 79900, 2: 12900, 3: 92800}89cyclic = [{"id": 1, "tool": "price", "args": {}, "depends_on": [2]},10 {"id": 2, "tool": "price", "args": {}, "depends_on": [1]}]11try:12 execute_plan(cyclic, tools)13except ValueError as e:14 print(e) # plan has a dependency cycle| round | done before | runnable | results after |
|---|---|---|---|
| 1 | {} | steps 1, 2 | {1: 79900, 2: 12900} |
| 2 | {1, 2} | step 3, args become "79900+12900" | {…, 3: 92800} |
In the cyclic plan, round 1 finds no step whose dependencies are done, and the executor raises instead of looping forever.
Travel-booking agents use this shape: search flights and hotels in parallel, then compute the total, then ask the user to approve before booking anything.
Follow-up questions to expect
- "The plan was wrong because step 1 returned something unexpected — now what?" — Replan: send the goal, the old plan and the results so far back to the planner, and cap it at one or two replans.
- "How do you stop the model writing a dangerous plan?" — Validate every step against the tool allowlist and argument rules before executing anything, and require human approval for destructive steps. The plan being inspectable is the main safety benefit of this pattern.
- "When is ReAct better?" — When each step depends on reading the previous result in a way you cannot predict, such as exploring an unknown website.