Planning in Agents

1Goal: Write a research report on transformer models2  ├── Gather sources3  │     ├── Search academic databases4  │     └── Search technical blogs5  ├── Synthesize findings6  │     ├── Summarize each source7  │     └── Identify common themes8  └── Write report9        ├── Draft outline10        ├── Write sections11        └── Edit and format

1"Let's think step by step:21. First, I need to understand what information is required.32. Then, I should search for the relevant facts.43. Next, I'll synthesize the findings.54. Finally, I'll produce the answer."

1PLAN:    Given goal G, generate plan [step_1, step_2, ..., step_n]2EXECUTE: Run step_1 using available tools, observe result_13EVALUATE: Did result_1 match expectations?4  IF YES: proceed to step_25  IF NO:  REPLAN given G and {step_1, result_1, failure_reason}

1from langchain_openai import ChatOpenAI2from langchain.schema import SystemMessage, HumanMessage3
4llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)5
6# ── Task decomposition via LLM ────────────────────────────────────────────────7def generate_plan(goal: str) -> list[str]:8    """Use LLM to decompose a goal into ordered sub-tasks."""9    messages = [10        SystemMessage(content="""You are a planning assistant. Given a goal, 11decompose it into a numbered list of concrete, executable sub-tasks. 12Be specific. Output only the numbered list, nothing else."""),13        HumanMessage(content=f"Goal: {goal}"),14    ]15    response = llm.invoke(messages)16    lines = response.content.strip().split("\n")17    steps = [line.split(". ", 1)[-1].strip() for line in lines if line.strip()]18    return steps19
20# ── Plan-Execute-Replan loop ──────────────────────────────────────────────────21def execute_step(step: str, tools: dict) -> tuple[str, bool]:22    """23    Mock executor: run a step using available tools.24    Returns (result, success).25    """26    # In a real system, this routes to appropriate tools27    if "search" in step.lower():28        return f"[search result for: {step}]", True29    elif "write" in step.lower() or "draft" in step.lower():30        return f"[drafted content for: {step}]", True31    else:32        return f"[completed: {step}]", True33
34
35def plan_execute_replan(goal: str, tools: dict, max_replans: int = 3):36    """Production-grade planning loop with replanning on failure."""37    plan = generate_plan(goal)38    print(f"Initial plan ({len(plan)} steps):")39    for i, step in enumerate(plan, 1):40        print(f"  {i}. {step}")41    42    results = []43    i = 044    replan_count = 045    46    while i < len(plan):47        step = plan[i]48        result, success = execute_step(step, tools)49        50        if success:51            results.append({"step": step, "result": result})52            print(f"\n✓ Step {i+1}: {step[:50]}...")53            i += 154        else:55            print(f"\n✗ Step {i+1} failed: {step[:50]}...")56            if replan_count < max_replans:57                # Replan from current state58                context = f"Completed: {results}\nFailed: {step}\nError: {result}"59                new_remaining = generate_plan(f"Continue goal: {goal}\nContext: {context}")60                plan = plan[:i] + new_remaining61                replan_count += 162                print(f"  → Replanned with {len(new_remaining)} new steps")63            else:64                print(f"  → Max replans reached. Aborting.")65                break66    67    return results68
69# Example usage70goal = "Research and summarize the top 3 use cases for RAG systems in enterprise settings"71tools = {"search": lambda q: f"results for {q}", "write": lambda t: f"text about {t}"}72results = plan_execute_replan(goal, tools)73print(f"\nCompleted {len(results)} steps")