jatin gyass
update for the new web data source
4db2d34
Raw
History Blame Contribute Delete
6.5 kB
"""
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}")
],
}