""" backend/agents/planner.py The Planner Agent — the strategic brain of the system. Responsibilities: 1. Decompose complex tasks into ordered steps 2. Assign appropriate tools to each step 3. Define step dependencies (DAG structure) 4. Replan if Critic requests changes 5. Decide when task is complete Uses Gemini function calling to produce structured JSON plan. """ from __future__ import annotations import json import uuid from datetime import datetime, timezone from langchain_core.messages import SystemMessage, HumanMessage from ..core.llm import get_llm from ..state.graph_state import ( WorkflowState, TaskStatus, AgentRole, make_plan_step, make_agent_event, ) from ..core.logger import get_logger log = get_logger(__name__) PLANNER_SYSTEM = """You are a task planner. Decompose the task into 2-4 steps MAX. Tools: web_search, fetch_url, calculate, run_python, write_file, read_file, get_datetime, synthesize Always end with a synthesize step. Return ONLY JSON: {"thinking":"one line","plan":[{"step_id":"step_1","title":"short","description":"what to do","tool":"tool_name","depends_on":[]}],"estimated_complexity":"low|medium|high"} Rules: minimal steps, no redundancy, synthesize last, depends_on lists prerequisite step_ids.""" REPLAN_ADDITION = """ Replanning — fix these issues: {critique} Already done: {completed_steps} Only plan remaining steps.""" PLANNING_TOOL = { "type": "function", "function": { "name": "submit_plan", "description": "Submit the structured task plan", "parameters": { "type": "object", "required": ["thinking", "plan"], "properties": { "thinking": {"type": "string"}, "plan": { "type": "array", "items": { "type": "object", "required": ["step_id", "title", "description", "tool"], "properties": { "step_id": {"type": "string"}, "title": {"type": "string"}, "description": {"type": "string"}, "tool": {"type": "string"}, "depends_on": {"type": "array", "items": {"type": "string"}}, }, }, }, "estimated_complexity": {"type": "string", "enum": ["low", "medium", "high"]}, }, }, }, } def planner_node(state: WorkflowState) -> WorkflowState: """ LangGraph node — runs the Planner agent. Called at start and whenever Critic requests replanning. """ from ..core.config import get_settings settings = get_settings() llm = get_llm("planner", temperature=0.2) log.info("Planner running", task_id=state["task_id"], iteration=state["iteration"]) # Build prompt — include critique context if replanning system_content = PLANNER_SYSTEM if state.get("needs_replanning") and state.get("critique"): completed = [ s["title"] for s in state["plan"] if s["status"] == "done" ] system_content += REPLAN_ADDITION.format( critique=state["critique"], completed_steps="\n".join(f"- {t}" for t in completed) or "None", ) # Memory context memory_context = "" if state.get("memories"): mem_texts = [m["content"] for m in state["memories"][:3]] memory_context = "\n\nRelevant past experience:\n" + "\n".join(f"• {t}" for t in mem_texts) user_msg = f"Task: {state['task']}{memory_context}" if state.get("step_results"): results_summary = json.dumps( {k: str(v)[:200] for k, v in list(state["step_results"].items())[-3:]}, indent=2 ) user_msg += f"\n\nRecent results:\n{results_summary}" try: response = llm.invoke( [SystemMessage(content=system_content), HumanMessage(content=user_msg)], tools=[PLANNING_TOOL], tool_choice={"type": "function", "function": {"name": "submit_plan"}}, ) tool_call = response.tool_calls[0] if response.tool_calls else None if not tool_call: raise ValueError("No tool call in planner response") plan_data = tool_call["args"] thinking = plan_data.get("thinking", "") # Convert to plan steps (hard cap) new_plan = [] raw_steps = plan_data.get("plan", [])[:settings.max_plan_steps] for step in raw_steps: new_plan.append(make_plan_step( step_id=step.get("step_id", f"step_{len(new_plan)+1}"), title=step.get("title", ""), description=step.get("description", ""), tool=step.get("tool"), depends_on=step.get("depends_on", []), )) log.info("Plan created", steps=len(new_plan), thinking=thinking[:100]) # Preserve completed steps if replanning if state.get("needs_replanning"): completed = [s for s in state["plan"] if s["status"] == "done"] final_plan = completed + new_plan else: final_plan = new_plan tokens = response.usage_metadata.get("total_tokens", 0) if response.usage_metadata else 0 return { **state, "status": TaskStatus.EXECUTING, "plan": final_plan, "current_step_index": len([s for s in final_plan if s["status"] == "done"]), "needs_replanning": False, "iteration": state["iteration"] + 1, "total_tokens": state["total_tokens"] + tokens, "updated_at": datetime.now(timezone.utc).isoformat(), "events": state["events"] + [ make_agent_event( AgentRole.PLANNER, "plan_created", f"Created {len(new_plan)}-step plan: {thinking[:200]}", {"steps": [s["title"] for s in new_plan], "complexity": plan_data.get("estimated_complexity")}, ) ], } except Exception as e: log.error("Planner failed", error=str(e)) return { **state, "status": TaskStatus.FAILED, "error_message": f"Planner failed: {e}", "events": state["events"] + [ make_agent_event(AgentRole.PLANNER, "error", f"Planning failed: {e}") ], }